debug(reader.py): query_with_stat(info) always

.gitea/workflows/main.yml

@@ -29,7 +29,16 @@ jobs:
         if: github.ref == 'refs/heads/dev'
         uses: dokku/github-action@master
         with:
-          branch: 'dev'
+          branch: 'main'
           force: true
           git_remote_url: 'ssh://dokku@v2.discours.io:22/core'
           ssh_private_key: ${{ secrets.SSH_PRIVATE_KEY }}
+
+      - name: Push to dokku for staging branch
+        if: github.ref == 'refs/heads/staging'
+        uses: dokku/github-action@master
+        with:
+          branch: 'dev'
+          git_remote_url: 'ssh://dokku@staging.discours.io:22/core'
+          ssh_private_key: ${{ secrets.SSH_PRIVATE_KEY }}
+          git_push_flags: '--force'

.gitignore

@@ -161,4 +161,5 @@ views.json
 *.key
 *.crt
 *cache.json
-.cursor
+.cursor
+.devcontainer/

CHANGELOG.md

@@ -1,73 +1,3 @@
-#### [0.4.15] - 2025-03-22
-- Upgraded caching system described `docs/caching.md`
-- Module `cache/memorycache.py` removed
-- Enhanced caching system with backward compatibility:
-  - Unified cache key generation with support for existing naming patterns
-  - Improved Redis operation function with better error handling
-  - Updated precache module to use consistent Redis interface
-  - Integrated revalidator with the invalidation system for better performance
-  - Added comprehensive documentation for the caching system
-  - Enhanced cached_query to support template-based cache keys
-  - Standardized error handling across all cache operations
-- Optimized cache invalidation system:
-  - Added targeted invalidation for individual entities (authors, topics)
-  - Improved revalidation manager with individual object processing
-  - Implemented batched processing for high-volume invalidations
-  - Reduced Redis operations by using precise key invalidation instead of prefix-based wipes
-  - Added special handling for slug changes in topics
-- Unified caching system for all models:
-  - Implemented abstract functions `cache_data`, `get_cached_data` and `invalidate_cache_by_prefix`
-  - Added `cached_query` function for unified approach to query caching
-  - Updated resolvers `author.py` and `topic.py` to use the new caching API
-  - Improved logging for cache operations to simplify debugging
-  - Optimized Redis memory usage through key format unification
-- Improved caching and sorting in Topic and Author modules:
-  - Added support for dictionary sorting parameters in `by` for both modules
-  - Optimized cache key generation for stable behavior with various parameters
-  - Enhanced sorting logic with direction support and arbitrary fields
-  - Added `by` parameter support in the API for getting topics by community
-- Performance optimizations for author-related queries:
-  - Added SQLAlchemy-managed indexes to `Author`, `AuthorFollower`, `AuthorRating` and `AuthorBookmark` models
-  - Implemented persistent Redis caching for author queries without TTL (invalidated only on changes)
-  - Optimized author retrieval with separate endpoints:
-    - `get_authors_all` - returns all non-deleted authors without statistics
-    - `get_authors_paginated` - returns authors with statistics and pagination support
-    - `load_authors_by` - optimized to use caching and efficient sorting
-  - Improved SQL queries with optimized JOIN conditions and efficient filtering
-  - Added pre-aggregation of statistics (shouts count, followers count) in single efficient queries
-  - Implemented robust cache invalidation on author updates
-  - Created necessary indexes for author lookups by user ID, slug, and timestamps
-
-#### [0.4.14] - 2025-03-21
-- Significant performance improvements for topic queries:
-  - Added database indexes to optimize JOIN operations
-  - Implemented persistent Redis caching for topic queries (no TTL, invalidated only on changes)
-  - Optimized topic retrieval with separate endpoints for different use cases:
-    - `get_topics_all` - returns all topics without statistics for lightweight listing
-    - `get_topics_paginated` - returns topics with statistics and pagination support
-    - `get_topics_by_community` - adds pagination and optimized filtering by community
-  - Added SQLAlchemy-managed indexes directly in ORM models for automatic schema maintenance
-  - Created `sync_indexes()` function for automatic index synchronization during app startup
-  - Reduced database load by pre-aggregating statistics in optimized SQL queries
-  - Added robust cache invalidation on topic create/update/delete operations
-  - Improved query optimization with proper JOIN conditions and specific partial indexes
-
-#### [0.4.13] - 2025-03-20
-- Fixed Topic objects serialization error in cache/memorycache.py
-- Improved CustomJSONEncoder to support SQLAlchemy models with dict() method
-- Enhanced error handling in cache_on_arguments decorator
-- Modified `load_reactions_by` to include deleted reactions when `include_deleted=true` for proper comment tree building
-- Fixed featured/unfeatured logic in reaction processing:
-  - Dislike reactions now properly take precedence over likes
-  - Featured status now requires more than 4 likes from users with featured articles
-  - Removed unnecessary filters for deleted reactions since rating reactions are physically deleted
-  - Author's featured status now based on having non-deleted articles with featured_at
-
-#### [0.4.12] - 2025-03-19
-- `delete_reaction` detects comments and uses `deleted_at` update
-- `check_to_unfeature` etc. update
-- dogpile dep in `services/memorycache.py` optimized
-
 #### [0.4.11] - 2025-02-12
 - `create_draft` resolver requires draft_id fixed
 - `create_draft` resolver defaults body and title fields to empty string
@@ -298,4 +228,22 @@
 
 
 #### [0.2.7]
-- `loadFollowedReactions` now with `
+- `loadFollowedReactions` now with `login_required`
+- notifier service api draft
+- added `shout` visibility kind in schema
+- community isolated from author in orm
+
+
+#### [0.2.6]
+- redis connection pool
+- auth context fixes
+- communities orm, resolvers, schema
+
+
+#### [0.2.5]
+- restructured
+- all users have their profiles as authors in core
+- `gittask`, `inbox` and `auth` logics removed
+- `settings` moved to base and now smaller
+- new outside auth schema
+- removed `gittask`, `auth`, `inbox`, `migration`

Dockerfile

@@ -13,6 +13,4 @@ RUN pip install -r requirements.txt
 
 COPY . .
 
-EXPOSE 8000
-
-CMD ["python", "-m", "granian", "main:app", "--interface", "asgi", "--host", "0.0.0.0", "--port", "8000"]
+CMD ["python", "server.py"]

cache/cache.py

@@ -1,37 +1,7 @@
-"""
-Caching system for the Discours platform
-----------------------------------------
-
-This module provides a comprehensive caching solution with these key components:
-
-1. KEY NAMING CONVENTIONS:
-   - Entity-based keys: "entity:property:value" (e.g., "author:id:123")
-   - Collection keys: "entity:collection:params" (e.g., "authors:stats:limit=10:offset=0")
-   - Special case keys: Maintained for backwards compatibility (e.g., "topic_shouts_123")
-
-2. CORE FUNCTIONS:
-   - cached_query(): High-level function for retrieving cached data or executing queries
-
-3. ENTITY-SPECIFIC FUNCTIONS:
-   - cache_author(), cache_topic(): Cache entity data
-   - get_cached_author(), get_cached_topic(): Retrieve entity data from cache
-   - invalidate_cache_by_prefix(): Invalidate all keys with a specific prefix
-
-4. CACHE INVALIDATION STRATEGY:
-   - Direct invalidation via invalidate_* functions for immediate changes
-   - Delayed invalidation via revalidation_manager for background processing
-   - Event-based triggers for automatic cache updates (see triggers.py)
-
-To maintain consistency with the existing codebase, this module preserves
-the original key naming patterns while providing a more structured approach
-for new cache operations.
-"""
-
 import asyncio
 import json
-from typing import Any, Dict, List, Optional, Union
+from typing import List
 
-import orjson
 from sqlalchemy import and_, join, select
 
 from orm.author import Author, AuthorFollower
@@ -49,10 +19,8 @@ DEFAULT_FOLLOWS = {
     "communities": [{"id": 1, "name": "Дискурс", "slug": "discours", "pic": ""}],
 }
 
-CACHE_TTL = 300  # 5 minutes
+CACHE_TTL = 300  # 5 минут
 
-# Key templates for common entity types
-# These are used throughout the codebase and should be maintained for compatibility
 CACHE_KEYS = {
     "TOPIC_ID": "topic:id:{}",
     "TOPIC_SLUG": "topic:slug:{}",
@@ -69,8 +37,8 @@ CACHE_KEYS = {
 async def cache_topic(topic: dict):
     payload = json.dumps(topic, cls=CustomJSONEncoder)
     await asyncio.gather(
-        redis.execute("SET", f"topic:id:{topic['id']}", payload),
-        redis.execute("SET", f"topic:slug:{topic['slug']}", payload),
+        redis_operation("SET", f"topic:id:{topic['id']}", payload),
+        redis_operation("SET", f"topic:slug:{topic['slug']}", payload),
     )
 
 
@@ -78,30 +46,30 @@ async def cache_topic(topic: dict):
 async def cache_author(author: dict):
     payload = json.dumps(author, cls=CustomJSONEncoder)
     await asyncio.gather(
-        redis.execute("SET", f"author:user:{author['user'].strip()}", str(author["id"])),
-        redis.execute("SET", f"author:id:{author['id']}", payload),
+        redis_operation("SET", f"author:user:{author['user'].strip()}", str(author["id"])),
+        redis_operation("SET", f"author:id:{author['id']}", payload),
     )
 
 
 # Cache follows data
 async def cache_follows(follower_id: int, entity_type: str, entity_id: int, is_insert=True):
     key = f"author:follows-{entity_type}s:{follower_id}"
-    follows_str = await redis.execute("GET", key)
-    follows = orjson.loads(follows_str) if follows_str else DEFAULT_FOLLOWS[entity_type]
+    follows_str = await redis_operation("GET", key)
+    follows = json.loads(follows_str) if follows_str else DEFAULT_FOLLOWS[entity_type]
     if is_insert:
         if entity_id not in follows:
             follows.append(entity_id)
     else:
         follows = [eid for eid in follows if eid != entity_id]
-    await redis.execute("SET", key, json.dumps(follows, cls=CustomJSONEncoder))
+    await redis_operation("SET", key, json.dumps(follows, cls=CustomJSONEncoder))
     await update_follower_stat(follower_id, entity_type, len(follows))
 
 
 # Update follower statistics
 async def update_follower_stat(follower_id, entity_type, count):
     follower_key = f"author:id:{follower_id}"
-    follower_str = await redis.execute("GET", follower_key)
-    follower = orjson.loads(follower_str) if follower_str else None
+    follower_str = await redis_operation("GET", follower_key)
+    follower = json.loads(follower_str) if follower_str else None
     if follower:
         follower["stat"] = {f"{entity_type}s": count}
         await cache_author(follower)
@@ -110,9 +78,9 @@ async def update_follower_stat(follower_id, entity_type, count):
 # Get author from cache
 async def get_cached_author(author_id: int, get_with_stat):
     author_key = f"author:id:{author_id}"
-    result = await redis.execute("GET", author_key)
+    result = await redis_operation("GET", author_key)
     if result:
-        return orjson.loads(result)
+        return json.loads(result)
     # Load from database if not found in cache
     q = select(Author).where(Author.id == author_id)
     authors = get_with_stat(q)
@@ -135,16 +103,16 @@ async def get_cached_topic(topic_id: int):
         dict: Topic data or None if not found.
     """
     topic_key = f"topic:id:{topic_id}"
-    cached_topic = await redis.execute("GET", topic_key)
+    cached_topic = await redis_operation("GET", topic_key)
     if cached_topic:
-        return orjson.loads(cached_topic)
+        return json.loads(cached_topic)
 
     # If not in cache, fetch from the database
     with local_session() as session:
         topic = session.execute(select(Topic).where(Topic.id == topic_id)).scalar_one_or_none()
         if topic:
             topic_dict = topic.dict()
-            await redis.execute("SET", topic_key, json.dumps(topic_dict, cls=CustomJSONEncoder))
+            await redis_operation("SET", topic_key, json.dumps(topic_dict, cls=CustomJSONEncoder))
             return topic_dict
 
     return None
@@ -153,9 +121,9 @@ async def get_cached_topic(topic_id: int):
 # Get topic by slug from cache
 async def get_cached_topic_by_slug(slug: str, get_with_stat):
     topic_key = f"topic:slug:{slug}"
-    result = await redis.execute("GET", topic_key)
+    result = await redis_operation("GET", topic_key)
     if result:
-        return orjson.loads(result)
+        return json.loads(result)
     # Load from database if not found in cache
     topic_query = select(Topic).where(Topic.slug == slug)
     topics = get_with_stat(topic_query)
@@ -170,8 +138,8 @@ async def get_cached_topic_by_slug(slug: str, get_with_stat):
 async def get_cached_authors_by_ids(author_ids: List[int]) -> List[dict]:
     # Fetch all author data concurrently
     keys = [f"author:id:{author_id}" for author_id in author_ids]
-    results = await asyncio.gather(*(redis.execute("GET", key) for key in keys))
-    authors = [orjson.loads(result) if result else None for result in results]
+    results = await asyncio.gather(*(redis_operation("GET", key) for key in keys))
+    authors = [json.loads(result) if result else None for result in results]
     # Load missing authors from database and cache
     missing_indices = [index for index, author in enumerate(authors) if author is None]
     if missing_indices:
@@ -197,10 +165,10 @@ async def get_cached_topic_followers(topic_id: int):
     """
     try:
         cache_key = CACHE_KEYS["TOPIC_FOLLOWERS"].format(topic_id)
-        cached = await redis.execute("GET", cache_key)
+        cached = await redis_operation("GET", cache_key)
 
         if cached:
-            followers_ids = orjson.loads(cached)
+            followers_ids = json.loads(cached)
             logger.debug(f"Found {len(followers_ids)} cached followers for topic #{topic_id}")
             return await get_cached_authors_by_ids(followers_ids)
 
@@ -213,7 +181,7 @@ async def get_cached_topic_followers(topic_id: int):
                 .all()
             ]
 
-            await redis.execute("SETEX", cache_key, CACHE_TTL, orjson.dumps(followers_ids))
+            await redis_operation("SETEX", cache_key, value=json.dumps(followers_ids), ttl=CACHE_TTL)
             followers = await get_cached_authors_by_ids(followers_ids)
             logger.debug(f"Cached {len(followers)} followers for topic #{topic_id}")
             return followers
@@ -226,9 +194,9 @@ async def get_cached_topic_followers(topic_id: int):
 # Get cached author followers
 async def get_cached_author_followers(author_id: int):
     # Check cache for data
-    cached = await redis.execute("GET", f"author:followers:{author_id}")
+    cached = await redis_operation("GET", f"author:followers:{author_id}")
     if cached:
-        followers_ids = orjson.loads(cached)
+        followers_ids = json.loads(cached)
         followers = await get_cached_authors_by_ids(followers_ids)
         logger.debug(f"Cached followers for author #{author_id}: {len(followers)}")
         return followers
@@ -242,7 +210,7 @@ async def get_cached_author_followers(author_id: int):
             .filter(AuthorFollower.author == author_id, Author.id != author_id)
             .all()
         ]
-        await redis.execute("SET", f"author:followers:{author_id}", orjson.dumps(followers_ids))
+        await redis_operation("SET", f"author:followers:{author_id}", json.dumps(followers_ids))
         followers = await get_cached_authors_by_ids(followers_ids)
         return followers
 
@@ -250,9 +218,9 @@ async def get_cached_author_followers(author_id: int):
 # Get cached follower authors
 async def get_cached_follower_authors(author_id: int):
     # Attempt to retrieve authors from cache
-    cached = await redis.execute("GET", f"author:follows-authors:{author_id}")
+    cached = await redis_operation("GET", f"author:follows-authors:{author_id}")
     if cached:
-        authors_ids = orjson.loads(cached)
+        authors_ids = json.loads(cached)
     else:
         # Query authors from database
         with local_session() as session:
@@ -264,7 +232,7 @@ async def get_cached_follower_authors(author_id: int):
                     .where(AuthorFollower.follower == author_id)
                 ).all()
             ]
-            await redis.execute("SET", f"author:follows-authors:{author_id}", orjson.dumps(authors_ids))
+            await redis_operation("SET", f"author:follows-authors:{author_id}", json.dumps(authors_ids))
 
     authors = await get_cached_authors_by_ids(authors_ids)
     return authors
@@ -273,9 +241,9 @@ async def get_cached_follower_authors(author_id: int):
 # Get cached follower topics
 async def get_cached_follower_topics(author_id: int):
     # Attempt to retrieve topics from cache
-    cached = await redis.execute("GET", f"author:follows-topics:{author_id}")
+    cached = await redis_operation("GET", f"author:follows-topics:{author_id}")
     if cached:
-        topics_ids = orjson.loads(cached)
+        topics_ids = json.loads(cached)
     else:
         # Load topics from database and cache them
         with local_session() as session:
@@ -286,13 +254,13 @@ async def get_cached_follower_topics(author_id: int):
                 .where(TopicFollower.follower == author_id)
                 .all()
             ]
-            await redis.execute("SET", f"author:follows-topics:{author_id}", orjson.dumps(topics_ids))
+            await redis_operation("SET", f"author:follows-topics:{author_id}", json.dumps(topics_ids))
 
     topics = []
     for topic_id in topics_ids:
-        topic_str = await redis.execute("GET", f"topic:id:{topic_id}")
+        topic_str = await redis_operation("GET", f"topic:id:{topic_id}")
         if topic_str:
-            topic = orjson.loads(topic_str)
+            topic = json.loads(topic_str)
             if topic and topic not in topics:
                 topics.append(topic)
 
@@ -312,12 +280,12 @@ async def get_cached_author_by_user_id(user_id: str, get_with_stat):
         dict: Dictionary with author data or None if not found.
     """
     # Attempt to find author ID by user_id in Redis cache
-    author_id = await redis.execute("GET", f"author:user:{user_id.strip()}")
+    author_id = await redis_operation("GET", f"author:user:{user_id.strip()}")
     if author_id:
         # If ID is found, get full author data by ID
-        author_data = await redis.execute("GET", f"author:id:{author_id}")
+        author_data = await redis_operation("GET", f"author:id:{author_id}")
         if author_data:
-            return orjson.loads(author_data)
+            return json.loads(author_data)
 
     # If data is not found in cache, query the database
     author_query = select(Author).where(Author.user == user_id)
@@ -327,8 +295,8 @@ async def get_cached_author_by_user_id(user_id: str, get_with_stat):
         author = authors[0]
         author_dict = author.dict()
         await asyncio.gather(
-            redis.execute("SET", f"author:user:{user_id.strip()}", str(author.id)),
-            redis.execute("SET", f"author:id:{author.id}", orjson.dumps(author_dict)),
+            redis_operation("SET", f"author:user:{user_id.strip()}", str(author.id)),
+            redis_operation("SET", f"author:id:{author.id}", json.dumps(author_dict)),
         )
         return author_dict
 
@@ -349,9 +317,9 @@ async def get_cached_topic_authors(topic_id: int):
     """
     # Attempt to get a list of author IDs from cache
     rkey = f"topic:authors:{topic_id}"
-    cached_authors_ids = await redis.execute("GET", rkey)
+    cached_authors_ids = await redis_operation("GET", rkey)
     if cached_authors_ids:
-        authors_ids = orjson.loads(cached_authors_ids)
+        authors_ids = json.loads(cached_authors_ids)
     else:
         # If cache is empty, get data from the database
         with local_session() as session:
@@ -363,7 +331,7 @@ async def get_cached_topic_authors(topic_id: int):
             )
             authors_ids = [author_id for (author_id,) in session.execute(query).all()]
             # Cache the retrieved author IDs
-            await redis.execute("SET", rkey, orjson.dumps(authors_ids))
+            await redis_operation("SET", rkey, json.dumps(authors_ids))
 
     # Retrieve full author details from cached IDs
     if authors_ids:
@@ -384,11 +352,11 @@ async def invalidate_shouts_cache(cache_keys: List[str]):
             cache_key = f"shouts:{key}"
 
             # Удаляем основной кэш
-            await redis.execute("DEL", cache_key)
+            await redis_operation("DEL", cache_key)
             logger.debug(f"Invalidated cache key: {cache_key}")
 
             # Добавляем ключ в список инвалидированных с TTL
-            await redis.execute("SETEX", f"{cache_key}:invalidated", CACHE_TTL, "1")
+            await redis_operation("SETEX", f"{cache_key}:invalidated", value="1", ttl=CACHE_TTL)
 
             # Если это кэш темы, инвалидируем также связанные ключи
             if key.startswith("topic_"):
@@ -400,7 +368,7 @@ async def invalidate_shouts_cache(cache_keys: List[str]):
                     f"topic:stats:{topic_id}",
                 ]
                 for related_key in related_keys:
-                    await redis.execute("DEL", related_key)
+                    await redis_operation("DEL", related_key)
                     logger.debug(f"Invalidated related key: {related_key}")
 
         except Exception as e:
@@ -411,15 +379,15 @@ async def cache_topic_shouts(topic_id: int, shouts: List[dict]):
     """Кэширует список публикаций для темы"""
     key = f"topic_shouts_{topic_id}"
     payload = json.dumps(shouts, cls=CustomJSONEncoder)
-    await redis.execute("SETEX", key, CACHE_TTL, payload)
+    await redis_operation("SETEX", key, value=payload, ttl=CACHE_TTL)
 
 
 async def get_cached_topic_shouts(topic_id: int) -> List[dict]:
     """Получает кэшированный список публикаций для темы"""
     key = f"topic_shouts_{topic_id}"
-    cached = await redis.execute("GET", key)
+    cached = await redis_operation("GET", key)
     if cached:
-        return orjson.loads(cached)
+        return json.loads(cached)
     return None
 
 
@@ -463,7 +431,27 @@ async def invalidate_shout_related_cache(shout: Shout, author_id: int):
     await invalidate_shouts_cache(list(cache_keys))
 
 
-# Function removed - direct Redis calls used throughout the module instead
+async def redis_operation(operation: str, key: str, value=None, ttl=None):
+    """
+    Унифицированная функция для работы с Redis
+
+    Args:
+        operation: 'GET', 'SET', 'DEL', 'SETEX'
+        key: ключ
+        value: значение (для SET/SETEX)
+        ttl: время жизни в секундах (для SETEX)
+    """
+    try:
+        if operation == "GET":
+            return await redis.execute("GET", key)
+        elif operation == "SET":
+            await redis.execute("SET", key, value)
+        elif operation == "SETEX":
+            await redis.execute("SETEX", key, ttl or CACHE_TTL, value)
+        elif operation == "DEL":
+            await redis.execute("DEL", key)
+    except Exception as e:
+        logger.error(f"Redis {operation} error for key {key}: {e}")
 
 
 async def get_cached_entity(entity_type: str, entity_id: int, get_method, cache_method):
@@ -477,9 +465,9 @@ async def get_cached_entity(entity_type: str, entity_id: int, get_method, cache_
         cache_method: метод кэширования
     """
     key = f"{entity_type}:id:{entity_id}"
-    cached = await redis.execute("GET", key)
+    cached = await redis_operation("GET", key)
     if cached:
-        return orjson.loads(cached)
+        return json.loads(cached)
 
     entity = await get_method(entity_id)
     if entity:
@@ -508,120 +496,3 @@ async def cache_by_id(entity, entity_id: int, cache_method):
     d = x.dict()
     await cache_method(d)
     return d
-
-
-# Универсальная функция для сохранения данных в кеш
-async def cache_data(key: str, data: Any, ttl: Optional[int] = None) -> None:
-    """
-    Сохраняет данные в кеш по указанному ключу.
-
-    Args:
-        key: Ключ кеша
-        data: Данные для сохранения
-        ttl: Время жизни кеша в секундах (None - бессрочно)
-    """
-    try:
-        payload = json.dumps(data, cls=CustomJSONEncoder)
-        if ttl:
-            await redis.execute("SETEX", key, ttl, payload)
-        else:
-            await redis.execute("SET", key, payload)
-        logger.debug(f"Данные сохранены в кеш по ключу {key}")
-    except Exception as e:
-        logger.error(f"Ошибка при сохранении данных в кеш: {e}")
-
-
-# Универсальная функция для получения данных из кеша
-async def get_cached_data(key: str) -> Optional[Any]:
-    """
-    Получает данные из кеша по указанному ключу.
-
-    Args:
-        key: Ключ кеша
-
-    Returns:
-        Any: Данные из кеша или None, если данных нет
-    """
-    try:
-        cached_data = await redis.execute("GET", key)
-        if cached_data:
-            logger.debug(f"Данные получены из кеша по ключу {key}")
-            return orjson.loads(cached_data)
-        return None
-    except Exception as e:
-        logger.error(f"Ошибка при получении данных из кеша: {e}")
-        return None
-
-
-# Универсальная функция для инвалидации кеша по префиксу
-async def invalidate_cache_by_prefix(prefix: str) -> None:
-    """
-    Инвалидирует все ключи кеша с указанным префиксом.
-
-    Args:
-        prefix: Префикс ключей кеша для инвалидации
-    """
-    try:
-        keys = await redis.execute("KEYS", f"{prefix}:*")
-        if keys:
-            await redis.execute("DEL", *keys)
-            logger.debug(f"Удалено {len(keys)} ключей кеша с префиксом {prefix}")
-    except Exception as e:
-        logger.error(f"Ошибка при инвалидации кеша: {e}")
-
-
-# Универсальная функция для получения и кеширования данных
-async def cached_query(
-    cache_key: str,
-    query_func: callable,
-    ttl: Optional[int] = None,
-    force_refresh: bool = False,
-    use_key_format: bool = True,
-    **query_params,
-) -> Any:
-    """
-    Gets data from cache or executes query and saves result to cache.
-    Supports existing key formats for compatibility.
-
-    Args:
-        cache_key: Cache key or key template from CACHE_KEYS
-        query_func: Function to execute the query
-        ttl: Cache TTL in seconds (None - indefinite)
-        force_refresh: Force cache refresh
-        use_key_format: Whether to check if cache_key matches a key template in CACHE_KEYS
-        **query_params: Parameters to pass to the query function
-
-    Returns:
-        Any: Data from cache or query result
-    """
-    # Check if cache_key matches a pattern in CACHE_KEYS
-    actual_key = cache_key
-    if use_key_format and "{}" in cache_key:
-        # Look for a template match in CACHE_KEYS
-        for key_name, key_format in CACHE_KEYS.items():
-            if cache_key == key_format:
-                # We have a match, now look for the id or value to format with
-                for param_name, param_value in query_params.items():
-                    if param_name in ["id", "slug", "user", "topic_id", "author_id"]:
-                        actual_key = cache_key.format(param_value)
-                        break
-
-    # If not forcing refresh, try to get data from cache
-    if not force_refresh:
-        cached_result = await get_cached_data(actual_key)
-        if cached_result is not None:
-            return cached_result
-
-    # If data not in cache or refresh required, execute query
-    try:
-        result = await query_func(**query_params)
-        if result is not None:
-            # Save result to cache
-            await cache_data(actual_key, result, ttl)
-        return result
-    except Exception as e:
-        logger.error(f"Error executing query for caching: {e}")
-        # In case of error, return data from cache if not forcing refresh
-        if not force_refresh:
-            return await get_cached_data(actual_key)
-        raise

cache/memorycache.py

@@ -0,0 +1,11 @@
+from dogpile.cache import make_region
+
+from settings import REDIS_URL
+
+# Создание региона кэша с TTL
+cache_region = make_region()
+cache_region.configure(
+    "dogpile.cache.redis",
+    arguments={"url": f"{REDIS_URL}/1"},
+    expiration_time=3600,  # Cache expiration time in seconds
+)

cache/precache.py

@@ -86,15 +86,11 @@ async def precache_data():
 
         # Преобразуем словарь в список аргументов для HSET
         if value:
-            # Если значение - словарь, преобразуем его в плоский список для HSET
-            if isinstance(value, dict):
-                flattened = []
-                for field, val in value.items():
-                    flattened.extend([field, val])
-                await redis.execute("HSET", key, *flattened)
-            else:
-                # Предполагаем, что значение уже содержит список
-                await redis.execute("HSET", key, *value)
+            flattened = []
+            for field, val in value.items():
+                flattened.extend([field, val])
+
+            await redis.execute("HSET", key, *flattened)
             logger.info(f"redis hash '{key}' was restored")
 
         with local_session() as session:

cache/revalidator.py

@@ -1,26 +1,17 @@
 import asyncio
 
-from cache.cache import (
-    cache_author,
-    cache_topic,
-    get_cached_author,
-    get_cached_topic,
-    invalidate_cache_by_prefix,
-)
+from cache.cache import cache_author, cache_topic, get_cached_author, get_cached_topic
 from resolvers.stat import get_with_stat
 from utils.logger import root_logger as logger
 
-CACHE_REVALIDATION_INTERVAL = 300  # 5 minutes
-
 
 class CacheRevalidationManager:
-    def __init__(self, interval=CACHE_REVALIDATION_INTERVAL):
+    def __init__(self, interval=60):
         """Инициализация менеджера с заданным интервалом проверки (в секундах)."""
         self.interval = interval
         self.items_to_revalidate = {"authors": set(), "topics": set(), "shouts": set(), "reactions": set()}
         self.lock = asyncio.Lock()
         self.running = True
-        self.MAX_BATCH_SIZE = 10  # Максимальное количество элементов для поштучной обработки
 
     async def start(self):
         """Запуск фонового воркера для ревалидации кэша."""
@@ -41,107 +32,22 @@ class CacheRevalidationManager:
         """Обновление кэша для всех сущностей, требующих ревалидации."""
         async with self.lock:
             # Ревалидация кэша авторов
-            if self.items_to_revalidate["authors"]:
-                logger.debug(f"Revalidating {len(self.items_to_revalidate['authors'])} authors")
-                for author_id in self.items_to_revalidate["authors"]:
-                    if author_id == "all":
-                        await invalidate_cache_by_prefix("authors")
-                        break
-                    author = await get_cached_author(author_id, get_with_stat)
-                    if author:
-                        await cache_author(author)
-                self.items_to_revalidate["authors"].clear()
+            for author_id in self.items_to_revalidate["authors"]:
+                author = await get_cached_author(author_id, get_with_stat)
+                if author:
+                    await cache_author(author)
+            self.items_to_revalidate["authors"].clear()
 
             # Ревалидация кэша тем
-            if self.items_to_revalidate["topics"]:
-                logger.debug(f"Revalidating {len(self.items_to_revalidate['topics'])} topics")
-                for topic_id in self.items_to_revalidate["topics"]:
-                    if topic_id == "all":
-                        await invalidate_cache_by_prefix("topics")
-                        break
-                    topic = await get_cached_topic(topic_id)
-                    if topic:
-                        await cache_topic(topic)
-                self.items_to_revalidate["topics"].clear()
-
-            # Ревалидация шаутов (публикаций)
-            if self.items_to_revalidate["shouts"]:
-                shouts_count = len(self.items_to_revalidate["shouts"])
-                logger.debug(f"Revalidating {shouts_count} shouts")
-
-                # Проверяем наличие специального флага 'all'
-                if "all" in self.items_to_revalidate["shouts"]:
-                    await invalidate_cache_by_prefix("shouts")
-                # Если элементов много, но не 'all', используем специфический подход
-                elif shouts_count > self.MAX_BATCH_SIZE:
-                    # Инвалидируем только collections keys, которые затрагивают много сущностей
-                    collection_keys = await asyncio.create_task(self._redis.execute("KEYS", "shouts:*"))
-                    if collection_keys:
-                        await self._redis.execute("DEL", *collection_keys)
-                        logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей шаутов")
-
-                    # Обновляем кеш каждого конкретного шаута
-                    for shout_id in self.items_to_revalidate["shouts"]:
-                        if shout_id != "all":
-                            # Точечная инвалидация для каждого shout_id
-                            specific_keys = [f"shout:id:{shout_id}"]
-                            for key in specific_keys:
-                                await self._redis.execute("DEL", key)
-                                logger.debug(f"Удален ключ кеша {key}")
-                else:
-                    # Если элементов немного, обрабатываем каждый
-                    for shout_id in self.items_to_revalidate["shouts"]:
-                        if shout_id != "all":
-                            # Точечная инвалидация для каждого shout_id
-                            specific_keys = [f"shout:id:{shout_id}"]
-                            for key in specific_keys:
-                                await self._redis.execute("DEL", key)
-                                logger.debug(f"Удален ключ кеша {key}")
-
-                self.items_to_revalidate["shouts"].clear()
-
-            # Аналогично для реакций - точечная инвалидация
-            if self.items_to_revalidate["reactions"]:
-                reactions_count = len(self.items_to_revalidate["reactions"])
-                logger.debug(f"Revalidating {reactions_count} reactions")
-
-                if "all" in self.items_to_revalidate["reactions"]:
-                    await invalidate_cache_by_prefix("reactions")
-                elif reactions_count > self.MAX_BATCH_SIZE:
-                    # Инвалидируем только collections keys для реакций
-                    collection_keys = await asyncio.create_task(self._redis.execute("KEYS", "reactions:*"))
-                    if collection_keys:
-                        await self._redis.execute("DEL", *collection_keys)
-                        logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей реакций")
-
-                    # Точечная инвалидация для каждой реакции
-                    for reaction_id in self.items_to_revalidate["reactions"]:
-                        if reaction_id != "all":
-                            specific_keys = [f"reaction:id:{reaction_id}"]
-                            for key in specific_keys:
-                                await self._redis.execute("DEL", key)
-                                logger.debug(f"Удален ключ кеша {key}")
-                else:
-                    # Точечная инвалидация для каждой реакции
-                    for reaction_id in self.items_to_revalidate["reactions"]:
-                        if reaction_id != "all":
-                            specific_keys = [f"reaction:id:{reaction_id}"]
-                            for key in specific_keys:
-                                await self._redis.execute("DEL", key)
-                                logger.debug(f"Удален ключ кеша {key}")
-
-                self.items_to_revalidate["reactions"].clear()
+            for topic_id in self.items_to_revalidate["topics"]:
+                topic = await get_cached_topic(topic_id)
+                if topic:
+                    await cache_topic(topic)
+            self.items_to_revalidate["topics"].clear()
 
     def mark_for_revalidation(self, entity_id, entity_type):
         """Отметить сущность для ревалидации."""
-        if entity_id and entity_type:
-            self.items_to_revalidate[entity_type].add(entity_id)
-
-    def invalidate_all(self, entity_type):
-        """Пометить для инвалидации все элементы указанного типа."""
-        logger.debug(f"Marking all {entity_type} for invalidation")
-        # Особый флаг для полной инвалидации
-        self.items_to_revalidate[entity_type].add("all")
+        self.items_to_revalidate[entity_type].add(entity_id)
 
     async def stop(self):
         """Остановка фонового воркера."""
@@ -154,4 +60,4 @@ class CacheRevalidationManager:
                 pass
 
 
-revalidation_manager = CacheRevalidationManager()
+revalidation_manager = CacheRevalidationManager(interval=300)  # Ревалидация каждые 5 минут

docs/caching.md

@@ -1,279 +0,0 @@
-# Система кеширования Discours
-
-## Общее описание
-
-Система кеширования Discours - это комплексное решение для повышения производительности платформы. Она использует Redis для хранения часто запрашиваемых данных и уменьшения нагрузки на основную базу данных.
-
-Кеширование реализовано как многоуровневая система, состоящая из нескольких модулей:
-
-- `cache.py` - основной модуль с функциями кеширования
-- `revalidator.py` - асинхронный менеджер ревалидации кеша
-- `triggers.py` - триггеры событий SQLAlchemy для автоматической ревалидации
-- `precache.py` - предварительное кеширование данных при старте приложения
-
-## Ключевые компоненты
-
-### 1. Форматы ключей кеша
-
-Система поддерживает несколько форматов ключей для обеспечения совместимости и удобства использования:
-
-- **Ключи сущностей**: `entity:property:value` (например, `author:id:123`)
-- **Ключи коллекций**: `entity:collection:params` (например, `authors:stats:limit=10:offset=0`)
-- **Специальные ключи**: для обратной совместимости (например, `topic_shouts_123`)
-
-Все стандартные форматы ключей хранятся в словаре `CACHE_KEYS`:
-
-```python
-CACHE_KEYS = {
-    "TOPIC_ID": "topic:id:{}",
-    "TOPIC_SLUG": "topic:slug:{}",
-    "AUTHOR_ID": "author:id:{}",
-    # и другие...
-}
-```
-
-### 2. Основные функции кеширования
-
-#### Структура ключей
-
-Вместо генерации ключей через вспомогательные функции, система следует строгим конвенциям формирования ключей:
-
-1. **Ключи для отдельных сущностей** строятся по шаблону:
-   ```
-   entity:property:value
-   ```
-   Например:
-   - `topic:id:123` - тема с ID 123
-   - `author:slug:john-doe` - автор со слагом "john-doe"
-   - `shout:id:456` - публикация с ID 456
-
-2. **Ключи для коллекций** строятся по шаблону:
-   ```
-   entity:collection[:filter1=value1:filter2=value2:...]
-   ```
-   Например:
-   - `topics:all:basic` - базовый список всех тем
-   - `authors:stats:limit=10:offset=0:sort=name` - отсортированный список авторов с пагинацией
-   - `shouts:feed:limit=20:community=1` - лента публикаций с фильтром по сообществу
-
-3. **Специальные форматы ключей** для обратной совместимости:
-   ```
-   entity_action_id
-   ```
-   Например:
-   - `topic_shouts_123` - публикации для темы с ID 123
-
-Во всех модулях системы разработчики должны явно формировать ключи в соответствии с этими конвенциями, что обеспечивает единообразие и предсказуемость кеширования.
-
-#### Работа с данными в кеше
-
-```python
-async def cache_data(key, data, ttl=None)
-async def get_cached_data(key)
-```
-
-Эти функции предоставляют универсальный интерфейс для сохранения и получения данных из кеша. Они напрямую используют Redis через вызовы `redis.execute()`.
-
-#### Высокоуровневое кеширование запросов
-
-```python
-async def cached_query(cache_key, query_func, ttl=None, force_refresh=False, **query_params)
-```
-
-Функция `cached_query` объединяет получение данных из кеша и выполнение запроса в случае отсутствия данных в кеше. Это основная функция, которую следует использовать в резолверах для кеширования результатов запросов.
-
-### 3. Кеширование сущностей
-
-Для основных типов сущностей реализованы специальные функции:
-
-```python
-async def cache_topic(topic: dict)
-async def cache_author(author: dict)
-async def get_cached_topic(topic_id: int)
-async def get_cached_author(author_id: int, get_with_stat)
-```
-
-Эти функции упрощают работу с часто используемыми типами данных и обеспечивают единообразный подход к их кешированию.
-
-### 4. Работа со связями
-
-Для работы со связями между сущностями предназначены функции:
-
-```python
-async def cache_follows(follower_id, entity_type, entity_id, is_insert=True)
-async def get_cached_topic_followers(topic_id)
-async def get_cached_author_followers(author_id)
-async def get_cached_follower_topics(author_id)
-```
-
-Они позволяют эффективно кешировать и получать информацию о подписках, связях между авторами, темами и публикациями.
-
-## Система инвалидации кеша
-
-### 1. Прямая инвалидация
-
-Система поддерживает два типа инвалидации кеша:
-
-#### 1.1. Инвалидация по префиксу
-
-```python
-async def invalidate_cache_by_prefix(prefix)
-```
-
-Позволяет инвалидировать все ключи кеша, начинающиеся с указанного префикса. Используется в резолверах для инвалидации группы кешей при массовых изменениях.
-
-#### 1.2. Точечная инвалидация
-
-```python
-async def invalidate_authors_cache(author_id=None)
-async def invalidate_topics_cache(topic_id=None)
-```
-
-Эти функции позволяют инвалидировать кеш только для конкретной сущности, что снижает нагрузку на Redis и предотвращает ненужную потерю кешированных данных. Если ID сущности не указан, используется инвалидация по префиксу.
-
-Примеры использования точечной инвалидации:
-
-```python
-# Инвалидация кеша только для автора с ID 123
-await invalidate_authors_cache(123)
-
-# Инвалидация кеша только для темы с ID 456
-await invalidate_topics_cache(456)
-```
-
-### 2. Отложенная инвалидация
-
-Модуль `revalidator.py` реализует систему отложенной инвалидации кеша через класс `CacheRevalidationManager`:
-
-```python
-class CacheRevalidationManager:
-    # ...
-    async def process_revalidation(self):
-        # ...
-    def mark_for_revalidation(self, entity_id, entity_type):
-        # ...
-```
-
-Менеджер ревалидации работает как асинхронный фоновый процесс, который периодически (по умолчанию каждые 5 минут) проверяет наличие сущностей для ревалидации.
-
-Особенности реализации:
-- Для авторов и тем используется поштучная ревалидация каждой записи
-- Для шаутов и реакций используется батчевая обработка, с порогом в 10 элементов
-- При достижении порога система переключается на инвалидацию коллекций вместо поштучной обработки
-- Специальный флаг `all` позволяет запустить полную инвалидацию всех записей типа
-
-### 3. Автоматическая инвалидация через триггеры
-
-Модуль `triggers.py` регистрирует обработчики событий SQLAlchemy, которые автоматически отмечают сущности для ревалидации при изменении данных в базе:
-
-```python
-def events_register():
-    event.listen(Author, "after_update", mark_for_revalidation)
-    event.listen(Topic, "after_update", mark_for_revalidation)
-    # и другие...
-```
-
-Триггеры имеют следующие особенности:
-- Реагируют на события вставки, обновления и удаления
-- Отмечают затронутые сущности для отложенной ревалидации
-- Учитывают связи между сущностями (например, при изменении темы обновляются связанные шауты)
-
-## Предварительное кеширование
-
-Модуль `precache.py` реализует предварительное кеширование часто используемых данных при старте приложения:
-
-```python
-async def precache_data():
-    # ...
-```
-
-Эта функция выполняется при запуске приложения и заполняет кеш данными, которые будут часто запрашиваться пользователями.
-
-## Примеры использования
-
-### Простое кеширование результата запроса
-
-```python
-async def get_topics_with_stats(limit=10, offset=0, by="title"):
-    # Формирование ключа кеша по конвенции
-    cache_key = f"topics:stats:limit={limit}:offset={offset}:sort={by}"
-    
-    cached_data = await get_cached_data(cache_key)
-    if cached_data:
-        return cached_data
-        
-    # Выполнение запроса к базе данных
-    result = ... # логика получения данных
-    
-    await cache_data(cache_key, result, ttl=300)
-    return result
-```
-
-### Использование обобщенной функции cached_query
-
-```python
-async def get_topics_with_stats(limit=10, offset=0, by="title"):
-    async def fetch_data(limit, offset, by):
-        # Логика получения данных
-        return result
-        
-    # Формирование ключа кеша по конвенции
-    cache_key = f"topics:stats:limit={limit}:offset={offset}:sort={by}"
-    
-    return await cached_query(
-        cache_key, 
-        fetch_data, 
-        ttl=300, 
-        limit=limit, 
-        offset=offset, 
-        by=by
-    )
-```
-
-### Точечная инвалидация кеша при изменении данных
-
-```python
-async def update_topic(topic_id, new_data):
-    # Обновление данных в базе
-    # ...
-    
-    # Точечная инвалидация кеша только для измененной темы
-    await invalidate_topics_cache(topic_id)
-    
-    return updated_topic
-```
-
-## Отладка и мониторинг
-
-Система кеширования использует логгер для отслеживания операций:
-
-```python
-logger.debug(f"Данные получены из кеша по ключу {key}")
-logger.debug(f"Удалено {len(keys)} ключей кеша с префиксом {prefix}")
-logger.error(f"Ошибка при инвалидации кеша: {e}")
-```
-
-Это позволяет отслеживать работу кеша и выявлять возможные проблемы на ранних стадиях.
-
-## Рекомендации по использованию
-
-1. **Следуйте конвенциям формирования ключей** - это критически важно для консистентности и предсказуемости кеша.
-2. **Не создавайте собственные форматы ключей** - используйте существующие шаблоны для обеспечения единообразия.
-3. **Не забывайте об инвалидации** - всегда инвалидируйте кеш при изменении данных.
-4. **Используйте точечную инвалидацию** - вместо инвалидации по префиксу для снижения нагрузки на Redis.
-5. **Устанавливайте разумные TTL** - используйте разные значения TTL в зависимости от частоты изменения данных.
-6. **Не кешируйте большие объемы данных** - кешируйте только то, что действительно необходимо для повышения производительности.
-
-## Технические детали реализации
-
-- **Сериализация данных**: используется `orjson` для эффективной сериализации и десериализации данных.
-- **Форматирование даты и времени**: для корректной работы с датами используется `CustomJSONEncoder`.
-- **Асинхронность**: все операции кеширования выполняются асинхронно для минимального влияния на производительность API.
-- **Прямое взаимодействие с Redis**: все операции выполняются через прямые вызовы `redis.execute()` с обработкой ошибок.
-- **Батчевая обработка**: для массовых операций используется пороговое значение, после которого применяются оптимизированные стратегии.
-
-## Известные ограничения
-
-1. **Согласованность данных** - система не гарантирует абсолютную согласованность данных в кеше и базе данных.
-2. **Память** - необходимо следить за объемом данных в кеше, чтобы избежать проблем с памятью Redis.
-3. **Производительность Redis** - при большом количестве операций с кешем может стать узким местом.

docs/features.md

@@ -6,20 +6,14 @@
 
 ## Мультидоменная авторизация
 
-- Поддержка авторизации для разных доменов
+- Поддержка авторизации для разных доменов:
+  - *.dscrs.site (включая testing.dscrs.site)
+  - localhost[:port]
+  - testingdiscoursio-git-*-discoursio.vercel.app
+  - *.discours.io
 - Автоматическое определение сервера авторизации
 - Корректная обработка CORS для всех поддерживаемых доменов
 
-## Система кеширования
-
-- Redis используется в качестве основного механизма кеширования
-- Поддержка как синхронных, так и асинхронных функций в декораторе cache_on_arguments
-- Автоматическая сериализация/десериализация данных в JSON с использованием CustomJSONEncoder
-- Резервная сериализация через pickle для сложных объектов
-- Генерация уникальных ключей кеша на основе сигнатуры функции и переданных аргументов
-- Настраиваемое время жизни кеша (TTL)
-- Возможность ручной инвалидации кеша для конкретных функций и аргументов
-
 ## Webhooks
 
 - Автоматическая регистрация вебхука для события user.login
@@ -31,6 +25,10 @@
 
 ## CORS Configuration
 
+- Поддерживаются домены:
+  - *.dscrs.site (включая testing.dscrs.site, core.dscrs.site)
+  - *.discours.io (включая testing.discours.io)
+  - localhost (включая порты)
 - Поддерживаемые методы: GET, POST, OPTIONS
 - Настроена поддержка credentials
 - Разрешенные заголовки: Authorization, Content-Type, X-Requested-With, DNT, Cache-Control

main.py

@@ -17,7 +17,8 @@ from cache.revalidator import revalidation_manager
 from services.exception import ExceptionHandlerMiddleware
 from services.redis import redis
 from services.schema import create_all_tables, resolvers
-from services.search import search_service
+#from services.search import search_service
+from services.search import search_service, initialize_search_index
 from services.viewed import ViewedStorage
 from services.webhook import WebhookEndpoint, create_webhook_endpoint
 from settings import DEV_SERVER_PID_FILE_NAME, MODE
@@ -34,24 +35,67 @@ async def start():
                 f.write(str(os.getpid()))
     print(f"[main] process started in {MODE} mode")
 
+async def check_search_service():
+    """Check if search service is available and log result"""
+    info = await search_service.info()
+    if info.get("status") in ["error", "unavailable"]:
+        print(f"[WARNING] Search service unavailable: {info.get('message', 'unknown reason')}")
+    else:
+        print(f"[INFO] Search service is available: {info}")
 
+
+# indexing DB data
+# async def indexing():
+#     from services.db import fetch_all_shouts
+#     all_shouts = await fetch_all_shouts()
+#     await initialize_search_index(all_shouts)
 async def lifespan(_app):
     try:
+        print("[lifespan] Starting application initialization")
         create_all_tables()
         await asyncio.gather(
             redis.connect(),
             precache_data(),
             ViewedStorage.init(),
             create_webhook_endpoint(),
-            search_service.info(),
+            check_search_service(),
             start(),
             revalidation_manager.start(),
         )
+        print("[lifespan] Basic initialization complete")
+
+        # Add a delay before starting the intensive search indexing
+        print("[lifespan] Waiting for system stabilization before search indexing...")
+        await asyncio.sleep(10)  # 10-second delay to let the system stabilize
+
+        # Start search indexing as a background task with lower priority
+        asyncio.create_task(initialize_search_index_background())
+
         yield
     finally:
+        print("[lifespan] Shutting down application services")
         tasks = [redis.disconnect(), ViewedStorage.stop(), revalidation_manager.stop()]
         await asyncio.gather(*tasks, return_exceptions=True)
+        print("[lifespan] Shutdown complete")
 
+# Initialize search index in the background
+async def initialize_search_index_background():
+    """Run search indexing as a background task with low priority"""
+    try:
+        print("[search] Starting background search indexing process")
+        from services.db import fetch_all_shouts
+        
+        # Get total count first (optional)
+        all_shouts = await fetch_all_shouts()
+        total_count = len(all_shouts) if all_shouts else 0
+        print(f"[search] Fetched {total_count} shouts for background indexing")
+        
+        # Start the indexing process with the fetched shouts
+        print("[search] Beginning background search index initialization...")
+        await initialize_search_index(all_shouts)
+        print("[search] Background search index initialization complete")
+    except Exception as e:
+        print(f"[search] Error in background search indexing: {str(e)}")
 
 # Создаем экземпляр GraphQL
 graphql_app = GraphQL(schema, debug=True)

orm/author.py

@@ -1,6 +1,6 @@
 import time
 
-from sqlalchemy import JSON, Boolean, Column, ForeignKey, Index, Integer, String
+from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
 
 from services.db import Base
 
@@ -8,15 +8,6 @@ from services.db import Base
 
 
 class AuthorRating(Base):
-    """
-    Рейтинг автора от другого автора.
-
-    Attributes:
-        rater (int): ID оценивающего автора
-        author (int): ID оцениваемого автора
-        plus (bool): Положительная/отрицательная оценка
-    """
-
     __tablename__ = "author_rating"
 
     id = None  # type: ignore
@@ -24,26 +15,8 @@ class AuthorRating(Base):
     author = Column(ForeignKey("author.id"), primary_key=True)
     plus = Column(Boolean)
 
-    # Определяем индексы
-    __table_args__ = (
-        # Индекс для быстрого поиска всех оценок конкретного автора
-        Index("idx_author_rating_author", "author"),
-        # Индекс для быстрого поиска всех оценок, оставленных конкретным автором
-        Index("idx_author_rating_rater", "rater"),
-    )
-
 
 class AuthorFollower(Base):
-    """
-    Подписка одного автора на другого.
-
-    Attributes:
-        follower (int): ID подписчика
-        author (int): ID автора, на которого подписываются
-        created_at (int): Время создания подписки
-        auto (bool): Признак автоматической подписки
-    """
-
     __tablename__ = "author_follower"
 
     id = None  # type: ignore
@@ -52,57 +25,16 @@ class AuthorFollower(Base):
     created_at = Column(Integer, nullable=False, default=lambda: int(time.time()))
     auto = Column(Boolean, nullable=False, default=False)
 
-    # Определяем индексы
-    __table_args__ = (
-        # Индекс для быстрого поиска всех подписчиков автора
-        Index("idx_author_follower_author", "author"),
-        # Индекс для быстрого поиска всех авторов, на которых подписан конкретный автор
-        Index("idx_author_follower_follower", "follower"),
-    )
-
 
 class AuthorBookmark(Base):
-    """
-    Закладка автора на публикацию.
-
-    Attributes:
-        author (int): ID автора
-        shout (int): ID публикации
-    """
-
     __tablename__ = "author_bookmark"
 
     id = None  # type: ignore
     author = Column(ForeignKey("author.id"), primary_key=True)
     shout = Column(ForeignKey("shout.id"), primary_key=True)
 
-    # Определяем индексы
-    __table_args__ = (
-        # Индекс для быстрого поиска всех закладок автора
-        Index("idx_author_bookmark_author", "author"),
-        # Индекс для быстрого поиска всех авторов, добавивших публикацию в закладки
-        Index("idx_author_bookmark_shout", "shout"),
-    )
-
 
 class Author(Base):
-    """
-    Модель автора в системе.
-
-    Attributes:
-        user (str): Идентификатор пользователя в системе авторизации
-        name (str): Отображаемое имя
-        slug (str): Уникальный строковый идентификатор
-        bio (str): Краткая биография/статус
-        about (str): Полное описание
-        pic (str): URL изображения профиля
-        links (dict): Ссылки на социальные сети и сайты
-        created_at (int): Время создания профиля
-        last_seen (int): Время последнего посещения
-        updated_at (int): Время последнего обновления
-        deleted_at (int): Время удаления (если профиль удален)
-    """
-
     __tablename__ = "author"
 
     user = Column(String)  # unbounded link with authorizer's User type
@@ -121,17 +53,3 @@ class Author(Base):
     # search_vector = Column(
     #    TSVectorType("name", "slug", "bio", "about", regconfig="pg_catalog.russian")
     # )
-
-    # Определяем индексы
-    __table_args__ = (
-        # Индекс для быстрого поиска по slug
-        Index("idx_author_slug", "slug"),
-        # Индекс для быстрого поиска по идентификатору пользователя
-        Index("idx_author_user", "user"),
-        # Индекс для фильтрации неудаленных авторов
-        Index("idx_author_deleted_at", "deleted_at", postgresql_where=deleted_at.is_(None)),
-        # Индекс для сортировки по времени создания (для новых авторов)
-        Index("idx_author_created_at", "created_at"),
-        # Индекс для сортировки по времени последнего посещения
-        Index("idx_author_last_seen", "last_seen"),
-    )

orm/shout.py

@@ -1,6 +1,6 @@
 import time
 
-from sqlalchemy import JSON, Boolean, Column, ForeignKey, Index, Integer, String
+from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
 from sqlalchemy.orm import relationship
 
 from orm.author import Author
@@ -10,15 +10,6 @@ from services.db import Base
 
 
 class ShoutTopic(Base):
-    """
-    Связь между публикацией и темой.
-
-    Attributes:
-        shout (int): ID публикации
-        topic (int): ID темы
-        main (bool): Признак основной темы
-    """
-
     __tablename__ = "shout_topic"
 
     id = None  # type: ignore
@@ -26,12 +17,6 @@ class ShoutTopic(Base):
     topic = Column(ForeignKey("topic.id"), primary_key=True, index=True)
     main = Column(Boolean, nullable=True)
 
-    # Определяем дополнительные индексы
-    __table_args__ = (
-        # Оптимизированный составной индекс для запросов, которые ищут публикации по теме
-        Index("idx_shout_topic_topic_shout", "topic", "shout"),
-    )
-
 
 class ShoutReactionsFollower(Base):
     __tablename__ = "shout_reactions_followers"
@@ -45,15 +30,6 @@ class ShoutReactionsFollower(Base):
 
 
 class ShoutAuthor(Base):
-    """
-    Связь между публикацией и автором.
-
-    Attributes:
-        shout (int): ID публикации
-        author (int): ID автора
-        caption (str): Подпись автора
-    """
-
     __tablename__ = "shout_author"
 
     id = None  # type: ignore
@@ -61,18 +37,8 @@ class ShoutAuthor(Base):
     author = Column(ForeignKey("author.id"), primary_key=True, index=True)
     caption = Column(String, nullable=True, default="")
 
-    # Определяем дополнительные индексы
-    __table_args__ = (
-        # Оптимизированный индекс для запросов, которые ищут публикации по автору
-        Index("idx_shout_author_author_shout", "author", "shout"),
-    )
-
 
 class Shout(Base):
-    """
-    Публикация в системе.
-    """
-
     __tablename__ = "shout"
 
     created_at: int = Column(Integer, nullable=False, default=lambda: int(time.time()))
@@ -108,20 +74,3 @@ class Shout(Base):
     seo: str | None = Column(String, nullable=True)  # JSON
 
     draft: int | None = Column(ForeignKey("draft.id"), nullable=True)
-
-    # Определяем индексы
-    __table_args__ = (
-        # Индекс для быстрого поиска неудаленных публикаций
-        Index("idx_shout_deleted_at", "deleted_at", postgresql_where=deleted_at.is_(None)),
-        # Индекс для быстрой фильтрации по community
-        Index("idx_shout_community", "community"),
-        # Индекс для быстрого поиска по slug
-        Index("idx_shout_slug", "slug"),
-        # Составной индекс для фильтрации опубликованных неудаленных публикаций
-        Index(
-            "idx_shout_published_deleted",
-            "published_at",
-            "deleted_at",
-            postgresql_where=published_at.is_not(None) & deleted_at.is_(None),
-        ),
-    )

orm/topic.py

@@ -1,21 +1,11 @@
 import time
 
-from sqlalchemy import JSON, Boolean, Column, ForeignKey, Index, Integer, String
+from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
 
 from services.db import Base
 
 
 class TopicFollower(Base):
-    """
-    Связь между топиком и его подписчиком.
-
-    Attributes:
-        follower (int): ID подписчика
-        topic (int): ID топика
-        created_at (int): Время создания связи
-        auto (bool): Автоматическая подписка
-    """
-
     __tablename__ = "topic_followers"
 
     id = None  # type: ignore
@@ -24,29 +14,8 @@ class TopicFollower(Base):
     created_at = Column(Integer, nullable=False, default=int(time.time()))
     auto = Column(Boolean, nullable=False, default=False)
 
-    # Определяем индексы
-    __table_args__ = (
-        # Индекс для быстрого поиска всех подписчиков топика
-        Index("idx_topic_followers_topic", "topic"),
-        # Индекс для быстрого поиска всех топиков, на которые подписан автор
-        Index("idx_topic_followers_follower", "follower"),
-    )
-
 
 class Topic(Base):
-    """
-    Модель топика (темы) публикаций.
-
-    Attributes:
-        slug (str): Уникальный строковый идентификатор темы
-        title (str): Название темы
-        body (str): Описание темы
-        pic (str): URL изображения темы
-        community (int): ID сообщества
-        oid (str): Старый ID
-        parent_ids (list): IDs родительских тем
-    """
-
     __tablename__ = "topic"
 
     slug = Column(String, unique=True)
@@ -55,12 +24,5 @@ class Topic(Base):
     pic = Column(String, nullable=True, comment="Picture")
     community = Column(ForeignKey("community.id"), default=1)
     oid = Column(String, nullable=True, comment="Old ID")
-    parent_ids = Column(JSON, nullable=True, comment="Parent Topic IDs")
 
-    # Определяем индексы
-    __table_args__ = (
-        # Индекс для быстрого поиска по slug
-        Index("idx_topic_slug", "slug"),
-        # Индекс для быстрого поиска по сообществу
-        Index("idx_topic_community", "community"),
-    )
+    parent_ids = Column(JSON, nullable=True, comment="Parent Topic IDs")

requirements.dev.txt

@@ -1,6 +0,0 @@
-fakeredis
-pytest
-pytest-asyncio
-pytest-cov
-mypy
-ruff

requirements.txt

@@ -2,10 +2,13 @@
 bcrypt
 authlib
 passlib
-opensearch-py
+
 google-analytics-data
+dogpile-cache
+opensearch-py
 colorlog
 psycopg2-binary
+dogpile-cache
 httpx
 redis[hiredis]
 sentry-sdk[starlette,sqlalchemy]
@@ -13,5 +16,14 @@ starlette
 gql
 ariadne
 granian
-orjson
-pydantic
+
+# NLP and search
+httpx
+
+pydantic
+fakeredis
+pytest
+pytest-asyncio
+pytest-cov
+mypy
+ruff

resolvers/author.py

@@ -1,196 +1,25 @@
 import asyncio
 import time
-from typing import Optional
 
-from sqlalchemy import select, text
+from sqlalchemy import desc, select, text
 
 from cache.cache import (
     cache_author,
-    cached_query,
     get_cached_author,
     get_cached_author_by_user_id,
     get_cached_author_followers,
     get_cached_follower_authors,
     get_cached_follower_topics,
-    invalidate_cache_by_prefix,
 )
 from orm.author import Author
+from orm.shout import ShoutAuthor, ShoutTopic
+from orm.topic import Topic
 from resolvers.stat import get_with_stat
 from services.auth import login_required
 from services.db import local_session
-from services.redis import redis
 from services.schema import mutation, query
 from utils.logger import root_logger as logger
 
-DEFAULT_COMMUNITIES = [1]
-
-
-# Вспомогательная функция для получения всех авторов без статистики
-async def get_all_authors():
-    """
-    Получает всех авторов без статистики.
-    Используется для случаев, когда нужен полный список авторов без дополнительной информации.
-
-    Returns:
-        list: Список всех авторов без статистики
-    """
-    cache_key = "authors:all:basic"
-
-    # Функция для получения всех авторов из БД
-    async def fetch_all_authors():
-        logger.debug("Получаем список всех авторов из БД и кешируем результат")
-
-        with local_session() as session:
-            # Запрос на получение базовой информации об авторах
-            authors_query = select(Author).where(Author.deleted_at.is_(None))
-            authors = session.execute(authors_query).scalars().all()
-
-            # Преобразуем авторов в словари
-            return [author.dict() for author in authors]
-
-    # Используем универсальную функцию для кеширования запросов
-    return await cached_query(cache_key, fetch_all_authors)
-
-
-# Вспомогательная функция для получения авторов со статистикой с пагинацией
-async def get_authors_with_stats(limit=50, offset=0, by: Optional[str] = None):
-    """
-    Получает авторов со статистикой с пагинацией.
-
-    Args:
-        limit: Максимальное количество возвращаемых авторов
-        offset: Смещение для пагинации
-        by: Опциональный параметр сортировки (new/active)
-
-    Returns:
-        list: Список авторов с их статистикой
-    """
-    # Формируем ключ кеша с помощью универсальной функции
-    cache_key = f"authors:stats:limit={limit}:offset={offset}"
-
-    # Функция для получения авторов из БД
-    async def fetch_authors_with_stats():
-        logger.debug(f"Выполняем запрос на получение авторов со статистикой: limit={limit}, offset={offset}, by={by}")
-
-        with local_session() as session:
-            # Базовый запрос для получения авторов
-            base_query = select(Author).where(Author.deleted_at.is_(None))
-
-            # Применяем сортировку
-            if by:
-                if isinstance(by, dict):
-                    # Обработка словаря параметров сортировки
-                    from sqlalchemy import asc, desc
-
-                    for field, direction in by.items():
-                        column = getattr(Author, field, None)
-                        if column:
-                            if direction.lower() == "desc":
-                                base_query = base_query.order_by(desc(column))
-                            else:
-                                base_query = base_query.order_by(column)
-                elif by == "new":
-                    base_query = base_query.order_by(desc(Author.created_at))
-                elif by == "active":
-                    base_query = base_query.order_by(desc(Author.last_seen))
-                else:
-                    # По умолчанию сортируем по времени создания
-                    base_query = base_query.order_by(desc(Author.created_at))
-            else:
-                base_query = base_query.order_by(desc(Author.created_at))
-
-            # Применяем лимит и смещение
-            base_query = base_query.limit(limit).offset(offset)
-
-            # Получаем авторов
-            authors = session.execute(base_query).scalars().all()
-            author_ids = [author.id for author in authors]
-
-            if not author_ids:
-                return []
-
-            # Оптимизированный запрос для получения статистики по публикациям для авторов
-            shouts_stats_query = f"""
-            SELECT sa.author, COUNT(DISTINCT s.id) as shouts_count
-            FROM shout_author sa
-            JOIN shout s ON sa.shout = s.id AND s.deleted_at IS NULL AND s.published_at IS NOT NULL
-            WHERE sa.author IN ({",".join(map(str, author_ids))})
-            GROUP BY sa.author
-            """
-            shouts_stats = {row[0]: row[1] for row in session.execute(text(shouts_stats_query))}
-
-            # Запрос на получение статистики по подписчикам для авторов
-            followers_stats_query = f"""
-            SELECT author, COUNT(DISTINCT follower) as followers_count
-            FROM author_follower
-            WHERE author IN ({",".join(map(str, author_ids))})
-            GROUP BY author
-            """
-            followers_stats = {row[0]: row[1] for row in session.execute(text(followers_stats_query))}
-
-            # Формируем результат с добавлением статистики
-            result = []
-            for author in authors:
-                author_dict = author.dict()
-                author_dict["stat"] = {
-                    "shouts": shouts_stats.get(author.id, 0),
-                    "followers": followers_stats.get(author.id, 0),
-                }
-                result.append(author_dict)
-
-                # Кешируем каждого автора отдельно для использования в других функциях
-                await cache_author(author_dict)
-
-            return result
-
-    # Используем универсальную функцию для кеширования запросов
-    return await cached_query(cache_key, fetch_authors_with_stats)
-
-
-# Функция для инвалидации кеша авторов
-async def invalidate_authors_cache(author_id=None):
-    """
-    Инвалидирует кеши авторов при изменении данных.
-
-    Args:
-        author_id: Опциональный ID автора для точечной инвалидации.
-               Если не указан, инвалидируются все кеши авторов.
-    """
-    if author_id:
-        # Точечная инвалидация конкретного автора
-        logger.debug(f"Инвалидация кеша для автора #{author_id}")
-        specific_keys = [
-            f"author:id:{author_id}",
-            f"author:followers:{author_id}",
-            f"author:follows-authors:{author_id}",
-            f"author:follows-topics:{author_id}",
-            f"author:follows-shouts:{author_id}",
-        ]
-
-        # Получаем user_id автора, если есть
-        with local_session() as session:
-            author = session.query(Author).filter(Author.id == author_id).first()
-            if author and author.user:
-                specific_keys.append(f"author:user:{author.user.strip()}")
-
-        # Удаляем конкретные ключи
-        for key in specific_keys:
-            try:
-                await redis.execute("DEL", key)
-                logger.debug(f"Удален ключ кеша {key}")
-            except Exception as e:
-                logger.error(f"Ошибка при удалении ключа {key}: {e}")
-
-        # Также ищем и удаляем ключи коллекций, содержащих данные об этом авторе
-        collection_keys = await redis.execute("KEYS", "authors:stats:*")
-        if collection_keys:
-            await redis.execute("DEL", *collection_keys)
-            logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей авторов")
-    else:
-        # Общая инвалидация всех кешей авторов
-        logger.debug("Полная инвалидация кеша авторов")
-        await invalidate_cache_by_prefix("authors")
-
 
 @mutation.field("update_author")
 @login_required
@@ -222,30 +51,10 @@ async def update_author(_, info, profile):
 
 
 @query.field("get_authors_all")
-async def get_authors_all(_, _info):
-    """
-    Получает список всех авторов без статистики.
-
-    Returns:
-        list: Список всех авторов
-    """
-    return await get_all_authors()
-
-
-@query.field("get_authors_paginated")
-async def get_authors_paginated(_, _info, limit=50, offset=0, by=None):
-    """
-    Получает список авторов с пагинацией и статистикой.
-
-    Args:
-        limit: Максимальное количество возвращаемых авторов
-        offset: Смещение для пагинации
-        by: Параметр сортировки (new/active)
-
-    Returns:
-        list: Список авторов с их статистикой
-    """
-    return await get_authors_with_stats(limit, offset, by)
+def get_authors_all(_, _info):
+    with local_session() as session:
+        authors = session.query(Author).all()
+        return authors
 
 
 @query.field("get_author")
@@ -296,105 +105,145 @@ async def get_author_id(_, _info, user: str):
                 asyncio.create_task(cache_author(author_dict))
                 return author_with_stat
     except Exception as exc:
-        logger.error(f"Error getting author: {exc}")
-    return None
+        import traceback
+
+        traceback.print_exc()
+        logger.error(exc)
 
 
 @query.field("load_authors_by")
 async def load_authors_by(_, _info, by, limit, offset):
-    """
-    Загружает авторов по заданному критерию с пагинацией.
+    logger.debug(f"loading authors by {by}")
+    authors_query = select(Author)
 
-    Args:
-        by: Критерий сортировки авторов (new/active)
-        limit: Максимальное количество возвращаемых авторов
-        offset: Смещение для пагинации
+    if by.get("slug"):
+        authors_query = authors_query.filter(Author.slug.ilike(f"%{by['slug']}%"))
+    elif by.get("name"):
+        authors_query = authors_query.filter(Author.name.ilike(f"%{by['name']}%"))
+    elif by.get("topic"):
+        authors_query = (
+            authors_query.join(ShoutAuthor)  # Первое соединение ShoutAuthor
+            .join(ShoutTopic, ShoutAuthor.shout == ShoutTopic.shout)
+            .join(Topic, ShoutTopic.topic == Topic.id)
+            .filter(Topic.slug == str(by["topic"]))
+        )
 
-    Returns:
-        list: Список авторов с учетом критерия
-    """
-    # Используем оптимизированную функцию для получения авторов
-    return await get_authors_with_stats(limit, offset, by)
+    if by.get("last_seen"):  # в unix time
+        before = int(time.time()) - by["last_seen"]
+        authors_query = authors_query.filter(Author.last_seen > before)
+    elif by.get("created_at"):  # в unix time
+        before = int(time.time()) - by["created_at"]
+        authors_query = authors_query.filter(Author.created_at > before)
+
+    authors_query = authors_query.limit(limit).offset(offset)
+
+    with local_session() as session:
+        authors_nostat = session.execute(authors_query).all()
+        authors = []
+        for a in authors_nostat:
+            if isinstance(a, Author):
+                author_dict = await get_cached_author(a.id, get_with_stat)
+                if author_dict and isinstance(author_dict.get("shouts"), int):
+                    authors.append(author_dict)
+
+    # order
+    order = by.get("order")
+    if order in ["shouts", "followers"]:
+        authors_query = authors_query.order_by(desc(text(f"{order}_stat")))
+
+    # group by
+    authors = get_with_stat(authors_query)
+    return authors or []
 
 
 def get_author_id_from(slug="", user=None, author_id=None):
-    try:
-        author_id = None
-        if author_id:
-            return author_id
-        with local_session() as session:
-            author = None
-            if slug:
-                author = session.query(Author).filter(Author.slug == slug).first()
-                if author:
-                    author_id = author.id
-                    return author_id
-            if user:
-                author = session.query(Author).filter(Author.user == user).first()
-                if author:
-                    author_id = author.id
-    except Exception as exc:
-        logger.error(exc)
+    if not slug and not user and not author_id:
+        raise ValueError("One of slug, user, or author_id must be provided")
+
+    author_query = select(Author.id)
+    if user:
+        author_query = author_query.filter(Author.user == user)
+    elif slug:
+        author_query = author_query.filter(Author.slug == slug)
+    elif author_id:
+        author_query = author_query.filter(Author.id == author_id)
+
+    with local_session() as session:
+        author_id_result = session.execute(author_query).first()
+        author_id = author_id_result[0] if author_id_result else None
+
+    if not author_id:
+        raise ValueError("Author not found")
+
     return author_id
 
 
 @query.field("get_author_follows")
 async def get_author_follows(_, _info, slug="", user=None, author_id=0):
-    logger.debug(f"getting follows for @{slug}")
-    author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
-    if not author_id:
-        return {}
+    try:
+        author_id = get_author_id_from(slug, user, author_id)
 
-    followed_authors = await get_cached_follower_authors(author_id)
-    followed_topics = await get_cached_follower_topics(author_id)
+        if bool(author_id):
+            logger.debug(f"getting {author_id} follows authors")
+            authors = await get_cached_follower_authors(author_id)
+            topics = await get_cached_follower_topics(author_id)
+        return {
+            "topics": topics,
+            "authors": authors,
+            "communities": [{"id": 1, "name": "Дискурс", "slug": "discours", "pic": ""}],
+        }
+    except Exception:
+        import traceback
 
-    # TODO: Get followed communities too
-    return {
-        "authors": followed_authors,
-        "topics": followed_topics,
-        "communities": DEFAULT_COMMUNITIES,
-        "shouts": [],
-    }
+        traceback.print_exc()
+        return {"error": "Author not found"}
 
 
 @query.field("get_author_follows_topics")
 async def get_author_follows_topics(_, _info, slug="", user=None, author_id=None):
-    logger.debug(f"getting followed topics for @{slug}")
-    author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
-    if not author_id:
-        return []
-    followed_topics = await get_cached_follower_topics(author_id)
-    return followed_topics
+    try:
+        follower_id = get_author_id_from(slug, user, author_id)
+        topics = await get_cached_follower_topics(follower_id)
+        return topics
+    except Exception:
+        import traceback
+
+        traceback.print_exc()
 
 
 @query.field("get_author_follows_authors")
 async def get_author_follows_authors(_, _info, slug="", user=None, author_id=None):
-    logger.debug(f"getting followed authors for @{slug}")
-    author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
-    if not author_id:
-        return []
-    followed_authors = await get_cached_follower_authors(author_id)
-    return followed_authors
+    try:
+        follower_id = get_author_id_from(slug, user, author_id)
+        return await get_cached_follower_authors(follower_id)
+    except Exception:
+        import traceback
+
+        traceback.print_exc()
 
 
 def create_author(user_id: str, slug: str, name: str = ""):
-    author = Author()
-    author.user = user_id  # Связь с user_id из системы авторизации
-    author.slug = slug  # Идентификатор из системы авторизации
-    author.created_at = author.updated_at = int(time.time())
-    author.name = name or slug  # если не указано
-
     with local_session() as session:
-        session.add(author)
-        session.commit()
-        return author
+        try:
+            author = None
+            if user_id:
+                author = session.query(Author).filter(Author.user == user_id).first()
+            elif slug:
+                author = session.query(Author).filter(Author.slug == slug).first()
+            if not author:
+                new_author = Author(user=user_id, slug=slug, name=name)
+                session.add(new_author)
+                session.commit()
+                logger.info(f"author created by webhook {new_author.dict()}")
+        except Exception as exc:
+            logger.debug(exc)
 
 
 @query.field("get_author_followers")
 async def get_author_followers(_, _info, slug: str = "", user: str = "", author_id: int = 0):
-    logger.debug(f"getting followers for author @{slug} or ID:{author_id}")
+    logger.debug(f"getting followers for @{slug}")
     author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
-    if not author_id:
-        return []
-    followers = await get_cached_author_followers(author_id)
+    followers = []
+    if author_id:
+        followers = await get_cached_author_followers(author_id)
     return followers

resolvers/draft.py

@@ -1,5 +1,4 @@
 import time
-from operator import or_
 
 from sqlalchemy.sql import and_
 
@@ -56,11 +55,7 @@ async def load_drafts(_, info):
         return {"error": "User ID and author ID are required"}
 
     with local_session() as session:
-        drafts = (
-            session.query(Draft)
-            .filter(or_(Draft.authors.any(Author.id == author_id), Draft.created_by == author_id))
-            .all()
-        )
+        drafts = session.query(Draft).filter(Draft.authors.any(Author.id == author_id)).all()
     return {"drafts": drafts}
 
 
@@ -101,7 +96,7 @@ async def create_draft(_, info, draft_input):
     # Проверяем обязательные поля
     if "body" not in draft_input or not draft_input["body"]:
         draft_input["body"] = ""  # Пустая строка вместо NULL
-
+        
     if "title" not in draft_input or not draft_input["title"]:
         draft_input["title"] = ""  # Пустая строка вместо NULL
 
@@ -125,29 +120,23 @@ async def create_draft(_, info, draft_input):
 
 @mutation.field("update_draft")
 @login_required
-async def update_draft(_, info, draft_id: int, draft_input):
-    """Обновляет черновик публикации.
-
-    Args:
-        draft_id: ID черновика для обновления
-        draft_input: Данные для обновления черновика
-
-    Returns:
-        dict: Обновленный черновик или сообщение об ошибке
-    """
+async def update_draft(_, info, draft_input):
     user_id = info.context.get("user_id")
     author_dict = info.context.get("author", {})
     author_id = author_dict.get("id")
-
+    draft_id = draft_input.get("id")
+    if not draft_id:
+        return {"error": "Draft ID is required"}
     if not user_id or not author_id:
         return {"error": "Author ID are required"}
 
     with local_session() as session:
         draft = session.query(Draft).filter(Draft.id == draft_id).first()
+        del draft_input["id"]
+        Draft.update(draft, {**draft_input})
         if not draft:
             return {"error": "Draft not found"}
 
-        Draft.update(draft, draft_input)
         draft.updated_at = int(time.time())
         session.commit()
         return {"draft": draft}

resolvers/editor.py

@@ -1,6 +1,6 @@
+import json
 import time
 
-import orjson
 from sqlalchemy import and_, desc, select
 from sqlalchemy.orm import joinedload
 from sqlalchemy.sql.functions import coalesce
@@ -106,7 +106,7 @@ async def get_my_shout(_, info, shout_id: int):
         if hasattr(shout, "media") and shout.media:
             if isinstance(shout.media, str):
                 try:
-                    shout.media = orjson.loads(shout.media)
+                    shout.media = json.loads(shout.media)
                 except Exception as e:
                     logger.error(f"Error parsing shout media: {e}")
                     shout.media = []

resolvers/notifier.py

@@ -1,7 +1,7 @@
+import json
 import time
 from typing import List, Tuple
 
-import orjson
 from sqlalchemy import and_, select
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import aliased
@@ -115,7 +115,7 @@ def get_notifications_grouped(author_id: int, after: int = 0, limit: int = 10, o
         if (groups_amount + offset) >= limit:
             break
 
-        payload = orjson.loads(str(notification.payload))
+        payload = json.loads(str(notification.payload))
 
         if str(notification.entity) == NotificationEntity.SHOUT.value:
             shout = payload
@@ -177,7 +177,7 @@ def get_notifications_grouped(author_id: int, after: int = 0, limit: int = 10, o
 
             elif str(notification.entity) == "follower":
                 thread_id = "followers"
-                follower = orjson.loads(payload)
+                follower = json.loads(payload)
                 group = groups_by_thread.get(thread_id)
                 if group:
                     if str(notification.action) == "follow":
@@ -293,11 +293,11 @@ async def notifications_seen_thread(_, info, thread: str, after: int):
             )
             exclude = set()
             for nr in removed_reaction_notifications:
-                reaction = orjson.loads(str(nr.payload))
+                reaction = json.loads(str(nr.payload))
                 reaction_id = reaction.get("id")
                 exclude.add(reaction_id)
             for n in new_reaction_notifications:
-                reaction = orjson.loads(str(n.payload))
+                reaction = json.loads(str(n.payload))
                 reaction_id = reaction.get("id")
                 if (
                     reaction_id not in exclude

resolvers/reaction.py

@@ -97,23 +97,20 @@ def get_reactions_with_stat(q, limit, offset):
 
 def is_featured_author(session, author_id) -> bool:
     """
-    Check if an author has at least one non-deleted featured article.
+    Check if an author has at least one featured article.
 
     :param session: Database session.
     :param author_id: Author ID.
     :return: True if the author has a featured article, else False.
     """
     return session.query(
-        session.query(Shout)
-        .where(Shout.authors.any(id=author_id))
-        .filter(Shout.featured_at.is_not(None), Shout.deleted_at.is_(None))
-        .exists()
+        session.query(Shout).where(Shout.authors.any(id=author_id)).filter(Shout.featured_at.is_not(None)).exists()
     ).scalar()
 
 
 def check_to_feature(session, approver_id, reaction) -> bool:
     """
-    Make a shout featured if it receives more than 4 votes from authors.
+    Make a shout featured if it receives more than 4 votes.
 
     :param session: Database session.
     :param approver_id: Approver author ID.
@@ -121,78 +118,46 @@ def check_to_feature(session, approver_id, reaction) -> bool:
     :return: True if shout should be featured, else False.
     """
     if not reaction.reply_to and is_positive(reaction.kind):
-        # Проверяем, не содержит ли пост более 20% дизлайков
-        # Если да, то не должен быть featured независимо от количества лайков
-        if check_to_unfeature(session, reaction):
-            return False
-
-        # Собираем всех авторов, поставивших лайк
-        author_approvers = set()
+        approvers = {approver_id}
+        # Count the number of approvers
         reacted_readers = (
             session.query(Reaction.created_by)
-            .filter(
-                Reaction.shout == reaction.shout,
-                is_positive(Reaction.kind),
-                # Рейтинги (LIKE, DISLIKE) физически удаляются, поэтому фильтр deleted_at не нужен
-            )
+            .filter(Reaction.shout == reaction.shout, is_positive(Reaction.kind), Reaction.deleted_at.is_(None))
             .distinct()
-            .all()
         )
 
-        # Добавляем текущего одобряющего
-        approver = session.query(Author).filter(Author.id == approver_id).first()
-        if approver and is_featured_author(session, approver_id):
-            author_approvers.add(approver_id)
-
-        # Проверяем, есть ли у реагировавших авторов featured публикации
-        for (reader_id,) in reacted_readers:
+        for reader_id in reacted_readers:
             if is_featured_author(session, reader_id):
-                author_approvers.add(reader_id)
-
-        # Публикация становится featured при наличии более 4 лайков от авторов
-        logger.debug(f"Публикация {reaction.shout} имеет {len(author_approvers)} лайков от авторов")
-        return len(author_approvers) > 4
+                approvers.add(reader_id)
+        return len(approvers) > 4
     return False
 
 
-def check_to_unfeature(session, reaction) -> bool:
+def check_to_unfeature(session, rejecter_id, reaction) -> bool:
     """
     Unfeature a shout if 20% of reactions are negative.
 
     :param session: Database session.
+    :param rejecter_id: Rejecter author ID.
     :param reaction: Reaction object.
     :return: True if shout should be unfeatured, else False.
     """
-    if not reaction.reply_to:
-        # Проверяем соотношение дизлайков, даже если текущая реакция не дизлайк
+    if not reaction.reply_to and is_negative(reaction.kind):
         total_reactions = (
             session.query(Reaction)
             .filter(
-                Reaction.shout == reaction.shout,
-                Reaction.reply_to.is_(None),
-                Reaction.kind.in_(RATING_REACTIONS),
-                # Рейтинги физически удаляются при удалении, поэтому фильтр deleted_at не нужен
+                Reaction.shout == reaction.shout, Reaction.kind.in_(RATING_REACTIONS), Reaction.deleted_at.is_(None)
             )
             .count()
         )
 
         negative_reactions = (
             session.query(Reaction)
-            .filter(
-                Reaction.shout == reaction.shout,
-                is_negative(Reaction.kind),
-                Reaction.reply_to.is_(None),
-                # Рейтинги физически удаляются при удалении, поэтому фильтр deleted_at не нужен
-            )
+            .filter(Reaction.shout == reaction.shout, is_negative(Reaction.kind), Reaction.deleted_at.is_(None))
             .count()
         )
 
-        # Проверяем, составляют ли отрицательные реакции 20% или более от всех реакций
-        negative_ratio = negative_reactions / total_reactions if total_reactions > 0 else 0
-        logger.debug(
-            f"Публикация {reaction.shout}: {negative_reactions}/{total_reactions} отрицательных реакций ({negative_ratio:.2%})"
-        )
-        return total_reactions > 0 and negative_ratio >= 0.2
+        return total_reactions > 0 and (negative_reactions / total_reactions) >= 0.2
     return False
 
 
@@ -231,8 +196,8 @@ async def _create_reaction(session, shout_id: int, is_author: bool, author_id: i
     Create a new reaction and perform related actions such as updating counters and notification.
 
     :param session: Database session.
-    :param shout_id: Shout ID.
-    :param is_author: Flag indicating if the user is the author of the shout.
+    :param info: GraphQL context info.
+    :param shout: Shout object.
     :param author_id: Author ID.
     :param reaction: Dictionary with reaction data.
     :return: Dictionary with created reaction data.
@@ -252,14 +217,10 @@ async def _create_reaction(session, shout_id: int, is_author: bool, author_id: i
 
     # Handle rating
     if r.kind in RATING_REACTIONS:
-        # Проверяем сначала условие для unfeature (дизлайки имеют приоритет)
-        if check_to_unfeature(session, r):
+        if check_to_unfeature(session, author_id, r):
             set_unfeatured(session, shout_id)
-            logger.info(f"Публикация {shout_id} потеряла статус featured из-за высокого процента дизлайков")
-        # Только если не было unfeature, проверяем условие для feature
         elif check_to_feature(session, author_id, r):
             await set_featured(session, shout_id)
-            logger.info(f"Публикация {shout_id} получила статус featured благодаря лайкам от авторов")
 
     # Notify creation
     await notify_reaction(rdict, "create")
@@ -393,7 +354,7 @@ async def update_reaction(_, info, reaction):
 
             result = session.execute(reaction_query).unique().first()
             if result:
-                r, author, _shout, commented_stat, rating_stat = result
+                r, author, shout, commented_stat, rating_stat = result
                 if not r or not author:
                     return {"error": "Invalid reaction ID or unauthorized"}
 
@@ -445,24 +406,15 @@ async def delete_reaction(_, info, reaction_id: int):
             if r.created_by != author_id and "editor" not in roles:
                 return {"error": "Access denied"}
 
-            if r.kind == ReactionKind.COMMENT.value:
-                r.deleted_at = int(time.time())
-                update_author_stat(author.id)
-                session.add(r)
-                session.commit()
-            elif r.kind == ReactionKind.PROPOSE.value:
-                r.deleted_at = int(time.time())
-                session.add(r)
-                session.commit()
-                # TODO: add more reaction types here
-            else:
-                logger.debug(f"{user_id} user removing his #{reaction_id} reaction")
-                session.delete(r)
-                session.commit()
-                if check_to_unfeature(session, r):
-                    set_unfeatured(session, r.shout)
-
+            logger.debug(f"{user_id} user removing his #{reaction_id} reaction")
             reaction_dict = r.dict()
+            session.delete(r)
+            session.commit()
+
+            # Update author stat
+            if r.kind == ReactionKind.COMMENT.value:
+                update_author_stat(author.id)
+
             await notify_reaction(reaction_dict, "delete")
 
             return {"error": None, "reaction": reaction_dict}
@@ -533,9 +485,7 @@ async def load_reactions_by(_, _info, by, limit=50, offset=0):
     # Add statistics and apply filters
     q = add_reaction_stat_columns(q)
     q = apply_reaction_filters(by, q)
-
-    # Include reactions with deleted_at for building comment trees
-    # q = q.where(Reaction.deleted_at.is_(None))
+    q = q.where(Reaction.deleted_at.is_(None))
 
     # Group and sort
     q = q.group_by(Reaction.id, Author.id, Shout.id)

resolvers/reader.py

@@ -1,4 +1,5 @@
-import orjson
+import json
+
 from graphql import GraphQLResolveInfo
 from sqlalchemy import and_, nulls_last, text
 from sqlalchemy.orm import aliased
@@ -221,7 +222,7 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
                         if has_field(info, "stat"):
                             stat = {}
                             if isinstance(row.stat, str):
-                                stat = orjson.loads(row.stat)
+                                stat = json.loads(row.stat)
                             elif isinstance(row.stat, dict):
                                 stat = row.stat
                             viewed = ViewedStorage.get_shout(shout_id=shout_id) or 0
@@ -230,7 +231,7 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
                         # Обработка main_topic и topics
                         topics = None
                         if has_field(info, "topics") and hasattr(row, "topics"):
-                            topics = orjson.loads(row.topics) if isinstance(row.topics, str) else row.topics
+                            topics = json.loads(row.topics) if isinstance(row.topics, str) else row.topics
                             # logger.debug(f"Shout#{shout_id} topics: {topics}")
                             shout_dict["topics"] = topics
 
@@ -239,7 +240,7 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
                             if hasattr(row, "main_topic"):
                                 # logger.debug(f"Raw main_topic for shout#{shout_id}: {row.main_topic}")
                                 main_topic = (
-                                    orjson.loads(row.main_topic) if isinstance(row.main_topic, str) else row.main_topic
+                                    json.loads(row.main_topic) if isinstance(row.main_topic, str) else row.main_topic
                                 )
                                 # logger.debug(f"Parsed main_topic for shout#{shout_id}: {main_topic}")
 
@@ -252,14 +253,14 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
                                     "is_main": True,
                                 }
                             elif not main_topic:
-                                logger.warning(f"No main_topic and no topics found for shout#{shout_id}")
+                                logger.debug(f"No main_topic and no topics found for shout#{shout_id}")
                                 main_topic = {"id": 0, "title": "no topic", "slug": "notopic", "is_main": True}
                             shout_dict["main_topic"] = main_topic
-                            # logger.debug(f"Final main_topic for shout#{shout_id}: {main_topic}")
+                            logger.debug(f"Final main_topic for shout#{shout_id}: {main_topic}")
 
                         if has_field(info, "authors") and hasattr(row, "authors"):
                             shout_dict["authors"] = (
-                                orjson.loads(row.authors) if isinstance(row.authors, str) else row.authors
+                                json.loads(row.authors) if isinstance(row.authors, str) else row.authors
                             )
 
                         if has_field(info, "media") and shout.media:
@@ -267,8 +268,8 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
                             media_data = shout.media
                             if isinstance(media_data, str):
                                 try:
-                                    media_data = orjson.loads(media_data)
-                                except orjson.JSONDecodeError:
+                                    media_data = json.loads(media_data)
+                                except json.JSONDecodeError:
                                     media_data = []
                             shout_dict["media"] = [media_data] if isinstance(media_data, dict) else media_data
 
@@ -412,18 +413,26 @@ async def load_shouts_search(_, info, text, options):
                 scores[shout_id] = sr.get("score")
                 hits_ids.append(shout_id)
 
-        q = (
+        """ q = (
             query_with_stat(info)
             if has_field(info, "stat")
             else select(Shout).filter(and_(Shout.published_at.is_not(None), Shout.deleted_at.is_(None)))
-        )
+        ) """
+        q = query_with_stat(info)
+
         q = q.filter(Shout.id.in_(hits_ids))
         q = apply_filters(q, options)
-        q = apply_sorting(q, options)
+
+        # added this to join topics
+        topic_join = aliased(ShoutTopic)
+        topic = aliased(Topic)
+        q = q.outerjoin(topic_join, topic_join.shout == Shout.id)
+        q = q.outerjoin(topic, topic.id == topic_join.topic)
+
         shouts = get_shouts_with_links(info, q, limit, offset)
         for shout in shouts:
-            shout.score = scores[f"{shout.id}"]
-        shouts.sort(key=lambda x: x.score, reverse=True)
+            shout["score"] = scores[f"{shout['id']}"]
+        shouts.sort(key=lambda x: x["score"], reverse=True)
         return shouts
     return []
 

resolvers/topic.py

@@ -1,240 +1,44 @@
-from sqlalchemy import desc, select, text
+from sqlalchemy import select
 
 from cache.cache import (
-    cache_topic,
-    cached_query,
     get_cached_topic_authors,
     get_cached_topic_by_slug,
     get_cached_topic_followers,
-    invalidate_cache_by_prefix,
 )
+from cache.memorycache import cache_region
 from orm.author import Author
 from orm.topic import Topic
 from resolvers.stat import get_with_stat
 from services.auth import login_required
 from services.db import local_session
-from services.redis import redis
 from services.schema import mutation, query
 from utils.logger import root_logger as logger
 
 
-# Вспомогательная функция для получения всех тем без статистики
-async def get_all_topics():
-    """
-    Получает все темы без статистики.
-    Используется для случаев, когда нужен полный список тем без дополнительной информации.
-
-    Returns:
-        list: Список всех тем без статистики
-    """
-    cache_key = "topics:all:basic"
-
-    # Функция для получения всех тем из БД
-    async def fetch_all_topics():
-        logger.debug("Получаем список всех тем из БД и кешируем результат")
-
-        with local_session() as session:
-            # Запрос на получение базовой информации о темах
-            topics_query = select(Topic)
-            topics = session.execute(topics_query).scalars().all()
-
-            # Преобразуем темы в словари
-            return [topic.dict() for topic in topics]
-
-    # Используем универсальную функцию для кеширования запросов
-    return await cached_query(cache_key, fetch_all_topics)
-
-
-# Вспомогательная функция для получения тем со статистикой с пагинацией
-async def get_topics_with_stats(limit=100, offset=0, community_id=None, by=None):
-    """
-    Получает темы со статистикой с пагинацией.
-
-    Args:
-        limit: Максимальное количество возвращаемых тем
-        offset: Смещение для пагинации
-        community_id: Опциональный ID сообщества для фильтрации
-        by: Опциональный параметр сортировки
-
-    Returns:
-        list: Список тем с их статистикой
-    """
-    # Формируем ключ кеша с помощью универсальной функции
-    cache_key = f"topics:stats:limit={limit}:offset={offset}:community_id={community_id}"
-
-    # Функция для получения тем из БД
-    async def fetch_topics_with_stats():
-        logger.debug(f"Выполняем запрос на получение тем со статистикой: limit={limit}, offset={offset}")
-
-        with local_session() as session:
-            # Базовый запрос для получения тем
-            base_query = select(Topic)
-
-            # Добавляем фильтр по сообществу, если указан
-            if community_id:
-                base_query = base_query.where(Topic.community == community_id)
-
-            # Применяем сортировку на основе параметра by
-            if by:
-                if isinstance(by, dict):
-                    # Обработка словаря параметров сортировки
-                    for field, direction in by.items():
-                        column = getattr(Topic, field, None)
-                        if column:
-                            if direction.lower() == "desc":
-                                base_query = base_query.order_by(desc(column))
-                            else:
-                                base_query = base_query.order_by(column)
-                elif by == "popular":
-                    # Сортировка по популярности (количеству публикаций)
-                    # Примечание: это требует дополнительного запроса или подзапроса
-                    base_query = base_query.order_by(
-                        desc(Topic.id)
-                    )  # Временно, нужно заменить на proper implementation
-                else:
-                    # По умолчанию сортируем по ID в обратном порядке
-                    base_query = base_query.order_by(desc(Topic.id))
-            else:
-                # По умолчанию сортируем по ID в обратном порядке
-                base_query = base_query.order_by(desc(Topic.id))
-
-            # Применяем лимит и смещение
-            base_query = base_query.limit(limit).offset(offset)
-
-            # Получаем темы
-            topics = session.execute(base_query).scalars().all()
-            topic_ids = [topic.id for topic in topics]
-
-            if not topic_ids:
-                return []
-
-            # Запрос на получение статистики по публикациям для выбранных тем
-            shouts_stats_query = f"""
-            SELECT st.topic, COUNT(DISTINCT s.id) as shouts_count
-            FROM shout_topic st
-            JOIN shout s ON st.shout = s.id AND s.deleted_at IS NULL
-            WHERE st.topic IN ({",".join(map(str, topic_ids))})
-            GROUP BY st.topic
-            """
-            shouts_stats = {row[0]: row[1] for row in session.execute(text(shouts_stats_query))}
-
-            # Запрос на получение статистики по подписчикам для выбранных тем
-            followers_stats_query = f"""
-            SELECT topic, COUNT(DISTINCT follower) as followers_count
-            FROM topic_followers
-            WHERE topic IN ({",".join(map(str, topic_ids))})
-            GROUP BY topic
-            """
-            followers_stats = {row[0]: row[1] for row in session.execute(text(followers_stats_query))}
-
-            # Формируем результат с добавлением статистики
-            result = []
-            for topic in topics:
-                topic_dict = topic.dict()
-                topic_dict["stat"] = {
-                    "shouts": shouts_stats.get(topic.id, 0),
-                    "followers": followers_stats.get(topic.id, 0),
-                }
-                result.append(topic_dict)
-
-                # Кешируем каждую тему отдельно для использования в других функциях
-                await cache_topic(topic_dict)
-
-            return result
-
-    # Используем универсальную функцию для кеширования запросов
-    return await cached_query(cache_key, fetch_topics_with_stats)
-
-
-# Функция для инвалидации кеша тем
-async def invalidate_topics_cache(topic_id=None):
-    """
-    Инвалидирует кеши тем при изменении данных.
-
-    Args:
-        topic_id: Опциональный ID темы для точечной инвалидации.
-               Если не указан, инвалидируются все кеши тем.
-    """
-    if topic_id:
-        # Точечная инвалидация конкретной темы
-        logger.debug(f"Инвалидация кеша для темы #{topic_id}")
-        specific_keys = [
-            f"topic:id:{topic_id}",
-            f"topic:authors:{topic_id}",
-            f"topic:followers:{topic_id}",
-            f"topic_shouts_{topic_id}",
-        ]
-
-        # Получаем slug темы, если есть
-        with local_session() as session:
-            topic = session.query(Topic).filter(Topic.id == topic_id).first()
-            if topic and topic.slug:
-                specific_keys.append(f"topic:slug:{topic.slug}")
-
-        # Удаляем конкретные ключи
-        for key in specific_keys:
-            try:
-                await redis.execute("DEL", key)
-                logger.debug(f"Удален ключ кеша {key}")
-            except Exception as e:
-                logger.error(f"Ошибка при удалении ключа {key}: {e}")
-
-        # Также ищем и удаляем ключи коллекций, содержащих данные об этой теме
-        collection_keys = await redis.execute("KEYS", "topics:stats:*")
-        if collection_keys:
-            await redis.execute("DEL", *collection_keys)
-            logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей тем")
-    else:
-        # Общая инвалидация всех кешей тем
-        logger.debug("Полная инвалидация кеша тем")
-        await invalidate_cache_by_prefix("topics")
-
-
 # Запрос на получение всех тем
 @query.field("get_topics_all")
-async def get_topics_all(_, _info):
-    """
-    Получает список всех тем без статистики.
+def get_topics_all(_, _info):
+    cache_key = "get_topics_all"  # Ключ для кеша
 
-    Returns:
-        list: Список всех тем
-    """
-    return await get_all_topics()
+    @cache_region.cache_on_arguments(cache_key)
+    def _get_topics_all():
+        topics_query = select(Topic)
+        return get_with_stat(topics_query)  # Получение тем с учетом статистики
 
-
-# Запрос на получение тем с пагинацией и статистикой
-@query.field("get_topics_paginated")
-async def get_topics_paginated(_, _info, limit=100, offset=0, by=None):
-    """
-    Получает список тем с пагинацией и статистикой.
-
-    Args:
-        limit: Максимальное количество возвращаемых тем
-        offset: Смещение для пагинации
-        by: Опциональные параметры сортировки
-
-    Returns:
-        list: Список тем с их статистикой
-    """
-    return await get_topics_with_stats(limit, offset, None, by)
+    return _get_topics_all()
 
 
 # Запрос на получение тем по сообществу
 @query.field("get_topics_by_community")
-async def get_topics_by_community(_, _info, community_id: int, limit=100, offset=0, by=None):
-    """
-    Получает список тем, принадлежащих указанному сообществу с пагинацией и статистикой.
+def get_topics_by_community(_, _info, community_id: int):
+    cache_key = f"get_topics_by_community_{community_id}"  # Ключ для кеша
 
-    Args:
-        community_id: ID сообщества
-        limit: Максимальное количество возвращаемых тем
-        offset: Смещение для пагинации
-        by: Опциональные параметры сортировки
+    @cache_region.cache_on_arguments(cache_key)
+    def _get_topics_by_community():
+        topics_by_community_query = select(Topic).where(Topic.community == community_id)
+        return get_with_stat(topics_by_community_query)
 
-    Returns:
-        list: Список тем с их статистикой
-    """
-    return await get_topics_with_stats(limit, offset, community_id, by)
+    return _get_topics_by_community()
 
 
 # Запрос на получение тем по автору
@@ -270,9 +74,6 @@ async def create_topic(_, _info, topic_input):
         session.add(new_topic)
         session.commit()
 
-        # Инвалидируем кеш всех тем
-        await invalidate_topics_cache()
-
         return {"topic": new_topic}
 
 
@@ -286,19 +87,10 @@ async def update_topic(_, _info, topic_input):
         if not topic:
             return {"error": "topic not found"}
         else:
-            old_slug = topic.slug
             Topic.update(topic, topic_input)
             session.add(topic)
             session.commit()
 
-            # Инвалидируем кеш только для этой конкретной темы
-            await invalidate_topics_cache(topic.id)
-
-            # Если slug изменился, удаляем старый ключ
-            if old_slug != topic.slug:
-                await redis.execute("DEL", f"topic:slug:{old_slug}")
-                logger.debug(f"Удален ключ кеша для старого slug: {old_slug}")
-
             return {"topic": topic}
 
 
@@ -319,11 +111,6 @@ async def delete_topic(_, info, slug: str):
             session.delete(t)
             session.commit()
 
-            # Инвалидируем кеш всех тем и конкретной темы
-            await invalidate_topics_cache()
-            await redis.execute("DEL", f"topic:slug:{slug}")
-            await redis.execute("DEL", f"topic:id:{t.id}")
-
             return {}
     return {"error": "access denied"}
 

schema/type.graphql

@@ -207,6 +207,7 @@ type CommonResult {
 }
 
 type SearchResult {
+    id: Int!
     slug: String!
     title: String!
     cover: String

server.py

@@ -0,0 +1,34 @@
+import sys
+from pathlib import Path
+
+from granian.constants import Interfaces
+from granian.log import LogLevels
+from granian.server import Server
+
+from settings import PORT
+from utils.logger import root_logger as logger
+
+if __name__ == "__main__":
+    logger.info("started")
+    try:
+
+        granian_instance = Server(
+            "main:app",
+            address="0.0.0.0",
+            port=PORT,
+            interface=Interfaces.ASGI,
+            workers=1,
+            websockets=False,
+            log_level=LogLevels.debug,
+            backlog=2048,
+        )
+
+        if "dev" in sys.argv:
+            logger.info("dev mode, building ssl context")
+            granian_instance.build_ssl_context(cert=Path("localhost.pem"), key=Path("localhost-key.pem"), password=None)
+        granian_instance.serve()
+    except Exception as error:
+        logger.error(error, exc_info=True)
+        raise
+    finally:
+        logger.info("stopped")

services/db.py

@@ -1,23 +1,21 @@
+import json
 import math
 import time
 import traceback
 import warnings
 from typing import Any, Callable, Dict, TypeVar
 
-import orjson
 import sqlalchemy
 from sqlalchemy import (
     JSON,
     Column,
     Engine,
-    Index,
     Integer,
     create_engine,
     event,
     exc,
     func,
     inspect,
-    text,
 )
 from sqlalchemy.orm import Session, configure_mappers, declarative_base
 from sqlalchemy.sql.schema import Table
@@ -58,82 +56,6 @@ def create_table_if_not_exists(engine, table):
         logger.info(f"Table '{table.__tablename__}' ok.")
 
 
-def sync_indexes():
-    """
-    Синхронизирует индексы в БД с индексами, определенными в моделях SQLAlchemy.
-    Создает недостающие индексы, если они определены в моделях, но отсутствуют в БД.
-
-    Использует pg_catalog для PostgreSQL для получения списка существующих индексов.
-    """
-    if not DB_URL.startswith("postgres"):
-        logger.warning("Функция sync_indexes поддерживается только для PostgreSQL.")
-        return
-
-    logger.info("Начинаем синхронизацию индексов в базе данных...")
-
-    # Получаем все существующие индексы в БД
-    with local_session() as session:
-        existing_indexes_query = text("""
-            SELECT 
-                t.relname AS table_name,
-                i.relname AS index_name
-            FROM 
-                pg_catalog.pg_class i
-            JOIN 
-                pg_catalog.pg_index ix ON ix.indexrelid = i.oid
-            JOIN 
-                pg_catalog.pg_class t ON t.oid = ix.indrelid
-            JOIN 
-                pg_catalog.pg_namespace n ON n.oid = i.relnamespace
-            WHERE 
-                i.relkind = 'i'
-                AND n.nspname = 'public'
-                AND t.relkind = 'r'
-            ORDER BY 
-                t.relname, i.relname;
-        """)
-
-        existing_indexes = {row[1].lower() for row in session.execute(existing_indexes_query)}
-        logger.debug(f"Найдено {len(existing_indexes)} существующих индексов в БД")
-
-        # Проверяем каждую модель и её индексы
-        for _model_name, model_class in REGISTRY.items():
-            if hasattr(model_class, "__table__") and hasattr(model_class, "__table_args__"):
-                table_args = model_class.__table_args__
-
-                # Если table_args - это кортеж, ищем в нём объекты Index
-                if isinstance(table_args, tuple):
-                    for arg in table_args:
-                        if isinstance(arg, Index):
-                            index_name = arg.name.lower()
-
-                            # Проверяем, существует ли индекс в БД
-                            if index_name not in existing_indexes:
-                                logger.info(
-                                    f"Создаем отсутствующий индекс {index_name} для таблицы {model_class.__tablename__}"
-                                )
-
-                                # Создаем индекс если он отсутствует
-                                try:
-                                    arg.create(engine)
-                                    logger.info(f"Индекс {index_name} успешно создан")
-                                except Exception as e:
-                                    logger.error(f"Ошибка при создании индекса {index_name}: {e}")
-                            else:
-                                logger.debug(f"Индекс {index_name} уже существует")
-
-        # Анализируем таблицы для оптимизации запросов
-        for model_name, model_class in REGISTRY.items():
-            if hasattr(model_class, "__tablename__"):
-                try:
-                    session.execute(text(f"ANALYZE {model_class.__tablename__}"))
-                    logger.debug(f"Таблица {model_class.__tablename__} проанализирована")
-                except Exception as e:
-                    logger.error(f"Ошибка при анализе таблицы {model_class.__tablename__}: {e}")
-
-    logger.info("Синхронизация индексов завершена.")
-
-
 # noinspection PyUnusedLocal
 def local_session(src=""):
     return Session(bind=engine, expire_on_commit=False)
@@ -162,8 +84,8 @@ class Base(declarative_base()):
                 # Check if the value is JSON and decode it if necessary
                 if isinstance(value, (str, bytes)) and isinstance(self.__table__.columns[column_name].type, JSON):
                     try:
-                        data[column_name] = orjson.loads(value)
-                    except (TypeError, orjson.JSONDecodeError) as e:
+                        data[column_name] = json.loads(value)
+                    except (TypeError, json.JSONDecodeError) as e:
                         logger.error(f"Error decoding JSON for column '{column_name}': {e}")
                         data[column_name] = value
                 else:
@@ -259,3 +181,27 @@ def get_json_builder():
 
 # Используем их в коде
 json_builder, json_array_builder, json_cast = get_json_builder()
+
+async def fetch_all_shouts(session=None):
+    """Fetch all published shouts for search indexing"""
+    from orm.shout import Shout
+    
+    close_session = False
+    if session is None:
+        session = local_session()
+        close_session = True
+    
+    try:
+        # Fetch only published and non-deleted shouts
+        query = session.query(Shout).filter(
+            Shout.published_at.is_not(None),
+            Shout.deleted_at.is_(None)
+        )
+        shouts = query.all()
+        return shouts
+    except Exception as e:
+        logger.error(f"Error fetching shouts for search indexing: {e}")
+        return []
+    finally:
+        if close_session:
+            session.close()

services/notify.py

@@ -1,4 +1,4 @@
-import orjson
+import json
 
 from orm.notification import Notification
 from services.db import local_session
@@ -18,7 +18,7 @@ async def notify_reaction(reaction, action: str = "create"):
     data = {"payload": reaction, "action": action}
     try:
         save_notification(action, channel_name, data.get("payload"))
-        await redis.publish(channel_name, orjson.dumps(data))
+        await redis.publish(channel_name, json.dumps(data))
     except Exception as e:
         logger.error(f"Failed to publish to channel {channel_name}: {e}")
 
@@ -28,7 +28,7 @@ async def notify_shout(shout, action: str = "update"):
     data = {"payload": shout, "action": action}
     try:
         save_notification(action, channel_name, data.get("payload"))
-        await redis.publish(channel_name, orjson.dumps(data))
+        await redis.publish(channel_name, json.dumps(data))
     except Exception as e:
         logger.error(f"Failed to publish to channel {channel_name}: {e}")
 
@@ -43,7 +43,7 @@ async def notify_follower(follower: dict, author_id: int, action: str = "follow"
         save_notification(action, channel_name, data.get("payload"))
 
         # Convert data to JSON string
-        json_data = orjson.dumps(data)
+        json_data = json.dumps(data)
 
         # Ensure the data is not empty before publishing
         if json_data:

services/search.py

@@ -2,231 +2,571 @@ import asyncio
 import json
 import logging
 import os
+import httpx
+import time
+import random
 
-import orjson
-from opensearchpy import OpenSearch
-
-from services.redis import redis
-from utils.encoders import CustomJSONEncoder
-
-# Set redis logging level to suppress DEBUG messages
+# Set up proper logging
 logger = logging.getLogger("search")
-logger.setLevel(logging.WARNING)
+logger.setLevel(logging.INFO)  # Change to INFO to see more details
 
-ELASTIC_HOST = os.environ.get("ELASTIC_HOST", "").replace("https://", "")
-ELASTIC_USER = os.environ.get("ELASTIC_USER", "")
-ELASTIC_PASSWORD = os.environ.get("ELASTIC_PASSWORD", "")
-ELASTIC_PORT = os.environ.get("ELASTIC_PORT", 9200)
-ELASTIC_URL = os.environ.get(
-    "ELASTIC_URL",
-    f"https://{ELASTIC_USER}:{ELASTIC_PASSWORD}@{ELASTIC_HOST}:{ELASTIC_PORT}",
-)
-REDIS_TTL = 86400  # 1 день в секундах
-
-index_settings = {
-    "settings": {
-        "index": {"number_of_shards": 1, "auto_expand_replicas": "0-all"},
-        "analysis": {
-            "analyzer": {
-                "ru": {
-                    "tokenizer": "standard",
-                    "filter": ["lowercase", "ru_stop", "ru_stemmer"],
-                }
-            },
-            "filter": {
-                "ru_stemmer": {"type": "stemmer", "language": "russian"},
-                "ru_stop": {"type": "stop", "stopwords": "_russian_"},
-            },
-        },
-    },
-    "mappings": {
-        "properties": {
-            "body": {"type": "text", "analyzer": "ru"},
-            "title": {"type": "text", "analyzer": "ru"},
-            "subtitle": {"type": "text", "analyzer": "ru"},
-            "lead": {"type": "text", "analyzer": "ru"},
-            "media": {"type": "text", "analyzer": "ru"},
-        }
-    },
-}
-
-expected_mapping = index_settings["mappings"]
-
-# Создание цикла событий
-search_loop = asyncio.get_event_loop()
-
-# В начале файла добавим флаг
-SEARCH_ENABLED = bool(os.environ.get("ELASTIC_HOST", ""))
-
-
-def get_indices_stats():
-    indices_stats = search_service.client.cat.indices(format="json")
-    for index_info in indices_stats:
-        index_name = index_info["index"]
-        if not index_name.startswith("."):
-            index_health = index_info["health"]
-            index_status = index_info["status"]
-            pri_shards = index_info["pri"]
-            rep_shards = index_info["rep"]
-            docs_count = index_info["docs.count"]
-            docs_deleted = index_info["docs.deleted"]
-            store_size = index_info["store.size"]
-            pri_store_size = index_info["pri.store.size"]
-
-            logger.info(f"Index: {index_name}")
-            logger.info(f"Health: {index_health}")
-            logger.info(f"Status: {index_status}")
-            logger.info(f"Primary Shards: {pri_shards}")
-            logger.info(f"Replica Shards: {rep_shards}")
-            logger.info(f"Documents Count: {docs_count}")
-            logger.info(f"Deleted Documents: {docs_deleted}")
-            logger.info(f"Store Size: {store_size}")
-            logger.info(f"Primary Store Size: {pri_store_size}")
+# Configuration for search service
+SEARCH_ENABLED = bool(os.environ.get("SEARCH_ENABLED", "true").lower() in ["true", "1", "yes"])
+TXTAI_SERVICE_URL = os.environ.get("TXTAI_SERVICE_URL", "none")
+MAX_BATCH_SIZE = int(os.environ.get("SEARCH_MAX_BATCH_SIZE", "25"))
 
 
 class SearchService:
-    def __init__(self, index_name="search_index"):
-        logger.info("Инициализируем поиск...")
-        self.index_name = index_name
-        self.client = None
-        self.lock = asyncio.Lock()
-
-        # Инициализация клиента OpenSearch только если поиск включен
-        if SEARCH_ENABLED:
-            try:
-                self.client = OpenSearch(
-                    hosts=[{"host": ELASTIC_HOST, "port": ELASTIC_PORT}],
-                    http_compress=True,
-                    http_auth=(ELASTIC_USER, ELASTIC_PASSWORD),
-                    use_ssl=True,
-                    verify_certs=False,
-                    ssl_assert_hostname=False,
-                    ssl_show_warn=False,
-                )
-                logger.info("Клиент OpenSearch.org подключен")
-                search_loop.create_task(self.check_index())
-            except Exception as exc:
-                logger.warning(f"Поиск отключен из-за ошибки подключения: {exc}")
-                self.client = None
-        else:
-            logger.info("Поиск отключен (ELASTIC_HOST не установлен)")
+    def __init__(self):
+        logger.info(f"Initializing search service with URL: {TXTAI_SERVICE_URL}")
+        self.available = SEARCH_ENABLED
+        # Use different timeout settings for indexing and search requests
+        self.client = httpx.AsyncClient(timeout=30.0, base_url=TXTAI_SERVICE_URL)
+        self.index_client = httpx.AsyncClient(timeout=120.0, base_url=TXTAI_SERVICE_URL)
 
+        if not self.available:
+            logger.info("Search disabled (SEARCH_ENABLED = False)")
+            
     async def info(self):
-        if not SEARCH_ENABLED:
+        """Return information about search service"""
+        if not self.available:
             return {"status": "disabled"}
 
         try:
-            return get_indices_stats()
+            response = await self.client.get("/info")
+            response.raise_for_status()
+            result = response.json()
+            logger.info(f"Search service info: {result}")
+            return result
         except Exception as e:
             logger.error(f"Failed to get search info: {e}")
             return {"status": "error", "message": str(e)}
 
-    def delete_index(self):
-        if self.client:
-            logger.warning(f"[!!!] Удаляем индекс {self.index_name}")
-            self.client.indices.delete(index=self.index_name, ignore_unavailable=True)
-
-    def create_index(self):
-        if self.client:
-            logger.info(f"Создается индекс: {self.index_name}")
-            self.client.indices.create(index=self.index_name, body=index_settings)
-            logger.info(f"Индекс {self.index_name} создан")
-
-    async def check_index(self):
-        if self.client:
-            logger.info(f"Проверяем индекс {self.index_name}...")
-            if not self.client.indices.exists(index=self.index_name):
-                self.create_index()
-                self.client.indices.put_mapping(index=self.index_name, body=expected_mapping)
-            else:
-                logger.info(f"Найден существующий индекс {self.index_name}")
-                # Проверка и обновление структуры индекса, если необходимо
-                result = self.client.indices.get_mapping(index=self.index_name)
-                if isinstance(result, str):
-                    result = orjson.loads(result)
-                if isinstance(result, dict):
-                    mapping = result.get(self.index_name, {}).get("mappings")
-                    logger.info(f"Найдена структура индексации: {mapping['properties'].keys()}")
-                    expected_keys = expected_mapping["properties"].keys()
-                    if mapping and mapping["properties"].keys() != expected_keys:
-                        logger.info(f"Ожидаемая структура индексации: {expected_mapping}")
-                        logger.warning("[!!!] Требуется переиндексация всех данных")
-                        self.delete_index()
-                        self.client = None
-        else:
-            logger.error("клиент не инициализован, невозможно проверить индекс")
+    def is_ready(self):
+        """Check if service is available"""
+        return self.available
+    
+    async def verify_docs(self, doc_ids):
+        """Verify which documents exist in the search index"""
+        if not self.available:
+            return {"status": "disabled"}
+            
+        try:
+            logger.info(f"Verifying {len(doc_ids)} documents in search index")
+            response = await self.client.post(
+                "/verify-docs",
+                json={"doc_ids": doc_ids},
+                timeout=60.0  # Longer timeout for potentially large ID lists
+            )
+            response.raise_for_status()
+            result = response.json()
+            
+            # Log summary of verification results
+            missing_count = len(result.get("missing", []))
+            logger.info(f"Document verification complete: {missing_count} missing out of {len(doc_ids)} total")
+            
+            return result
+        except Exception as e:
+            logger.error(f"Document verification error: {e}")
+            return {"status": "error", "message": str(e)}
 
     def index(self, shout):
-        if not SEARCH_ENABLED:
+        """Index a single document"""
+        if not self.available:
             return
 
-        if self.client:
-            logger.info(f"Индексируем пост {shout.id}")
-            index_body = {
-                "body": shout.body,
-                "title": shout.title,
-                "subtitle": shout.subtitle,
-                "lead": shout.lead,
-                "media": shout.media,
-            }
-            asyncio.create_task(self.perform_index(shout, index_body))
+        logger.info(f"Indexing post {shout.id}")
+        # Start in background to not block
+        asyncio.create_task(self.perform_index(shout))
 
-    async def perform_index(self, shout, index_body):
-        if self.client:
+    async def perform_index(self, shout):
+        """Actually perform the indexing operation"""
+        if not self.available:
+            return
+        
+        try:
+            # Combine all text fields
+            text = " ".join(filter(None, [
+                shout.title or "", 
+                shout.subtitle or "", 
+                shout.lead or "", 
+                shout.body or "",
+                shout.media or ""
+            ]))
+            
+            if not text.strip():
+                logger.warning(f"No text content to index for shout {shout.id}")
+                return
+                
+            logger.info(f"Indexing document: ID={shout.id}, Text length={len(text)}")
+            
+            # Send to txtai service
+            response = await self.client.post(
+                "/index",
+                json={"id": str(shout.id), "text": text}
+            )
+            response.raise_for_status()
+            result = response.json()
+            logger.info(f"Post {shout.id} successfully indexed: {result}")
+        except Exception as e:
+            logger.error(f"Indexing error for shout {shout.id}: {e}")
+
+    async def bulk_index(self, shouts):
+        """Index multiple documents at once with adaptive batch sizing"""
+        if not self.available or not shouts:
+            logger.warning(f"Bulk indexing skipped: available={self.available}, shouts_count={len(shouts) if shouts else 0}")
+            return
+        
+        start_time = time.time()
+        logger.info(f"Starting bulk indexing of {len(shouts)} documents")
+
+        MAX_TEXT_LENGTH = 4000  # Maximum text length to send in a single request
+        max_batch_size = MAX_BATCH_SIZE
+        total_indexed = 0
+        total_skipped = 0
+        total_truncated = 0
+        total_retries = 0
+        
+        # Group documents by size to process smaller documents in larger batches
+        small_docs = []
+        medium_docs = []
+        large_docs = []
+        
+        # First pass: prepare all documents and categorize by size
+        for shout in shouts:
             try:
-                await asyncio.wait_for(
-                    self.client.index(index=self.index_name, id=str(shout.id), body=index_body), timeout=40.0
-                )
-            except asyncio.TimeoutError:
-                logger.error(f"Indexing timeout for shout {shout.id}")
+                text_fields = []
+                for field_name in ['title', 'subtitle', 'lead', 'body']:
+                    field_value = getattr(shout, field_name, None)
+                    if field_value and isinstance(field_value, str) and field_value.strip():
+                        text_fields.append(field_value.strip())
+                
+                # Media field processing remains the same
+                media = getattr(shout, 'media', None)
+                if media:
+                    # Your existing media processing logic
+                    if isinstance(media, str):
+                        try:
+                            media_json = json.loads(media)
+                            if isinstance(media_json, dict):
+                                if 'title' in media_json:
+                                    text_fields.append(media_json['title'])
+                                if 'body' in media_json:
+                                    text_fields.append(media_json['body'])
+                        except json.JSONDecodeError:
+                            text_fields.append(media)
+                    elif isinstance(media, dict):
+                        if 'title' in media:
+                            text_fields.append(media['title'])
+                        if 'body' in media:
+                            text_fields.append(media['body'])
+                
+                text = " ".join(text_fields)
+                
+                if not text.strip():
+                    logger.debug(f"Skipping shout {shout.id}: no text content")
+                    total_skipped += 1
+                    continue
+
+                # Truncate text if it exceeds the maximum length
+                original_length = len(text)
+                if original_length > MAX_TEXT_LENGTH:
+                    text = text[:MAX_TEXT_LENGTH]
+                    logger.info(f"Truncated document {shout.id} from {original_length} to {MAX_TEXT_LENGTH} chars")
+                    total_truncated += 1
+                
+                document = {
+                    "id": str(shout.id),
+                    "text": text
+                }
+                
+                # Categorize by size
+                text_len = len(text)
+                if text_len > 5000:
+                    large_docs.append(document)
+                elif text_len > 2000:
+                    medium_docs.append(document)
+                else:
+                    small_docs.append(document)
+                
+                total_indexed += 1
+                
             except Exception as e:
-                logger.error(f"Indexing error for shout {shout.id}: {e}")
+                logger.error(f"Error processing shout {getattr(shout, 'id', 'unknown')} for indexing: {e}")
+                total_skipped += 1
+        
+        # Process each category with appropriate batch sizes
+        logger.info(f"Documents categorized: {len(small_docs)} small, {len(medium_docs)} medium, {len(large_docs)} large")
+        
+        # Process small documents (larger batches)
+        if small_docs:
+            batch_size = min(max_batch_size, 15)
+            await self._process_document_batches(small_docs, batch_size, "small")
+            
+        # Process medium documents (medium batches)
+        if medium_docs:
+            batch_size = min(max_batch_size, 10)
+            await self._process_document_batches(medium_docs, batch_size, "medium")
+            
+        # Process large documents (small batches)
+        if large_docs:
+            batch_size = min(max_batch_size, 3)
+            await self._process_document_batches(large_docs, batch_size, "large")
+        
+        elapsed = time.time() - start_time
+        logger.info(f"Bulk indexing completed in {elapsed:.2f}s: {total_indexed} indexed, {total_skipped} skipped, {total_truncated} truncated, {total_retries} retries")
+
+    async def _process_document_batches(self, documents, batch_size, size_category):
+            """Process document batches with retry logic"""
+            # Check for possible database corruption before starting
+            db_error_count = 0
+            
+            for i in range(0, len(documents), batch_size):
+                batch = documents[i:i+batch_size]
+                batch_id = f"{size_category}-{i//batch_size + 1}"
+                logger.info(f"Processing {size_category} batch {batch_id} of {len(batch)} documents")
+                
+                retry_count = 0
+                max_retries = 3
+                success = False
+                
+                # Process with retries
+                while not success and retry_count < max_retries:
+                    try:
+                        if batch:
+                            sample = batch[0]
+                            logger.info(f"Sample document in batch {batch_id}: id={sample['id']}, text_length={len(sample['text'])}")
+                        
+                        logger.info(f"Sending batch {batch_id} of {len(batch)} documents to search service (attempt {retry_count+1})")
+                        response = await self.index_client.post(
+                            "/bulk-index",
+                            json=batch,
+                            timeout=120.0  # Explicit longer timeout for large batches
+                        )
+                        
+                        # Handle 422 validation errors - these won't be fixed by retrying
+                        if response.status_code == 422:
+                            error_detail = response.json()
+                            truncated_error = self._truncate_error_detail(error_detail)
+                            logger.error(f"Validation error from search service for batch {batch_id}: {truncated_error}")
+                            break
+                        
+                        # Handle 500 server errors - these might be fixed by retrying with smaller batches
+                        elif response.status_code == 500:
+                            db_error_count += 1
+                            
+                            # If we've seen multiple 500s, log a critical error
+                            if db_error_count >= 3:
+                                logger.critical(f"Multiple server errors detected (500). The search service may need manual intervention. Stopping batch {batch_id} processing.")
+                                break
+                                
+                            # Try again with exponential backoff
+                            if retry_count < max_retries - 1:
+                                retry_count += 1
+                                wait_time = (2 ** retry_count) + (random.random() * 0.5)  # Exponential backoff with jitter
+                                logger.warning(f"Server error for batch {batch_id}, retrying in {wait_time:.1f}s (attempt {retry_count+1}/{max_retries})")
+                                await asyncio.sleep(wait_time)
+                                continue
+                            
+                            # Final retry, split the batch
+                            elif len(batch) > 1:
+                                logger.warning(f"Splitting batch {batch_id} after repeated failures")
+                                mid = len(batch) // 2
+                                await self._process_single_batch(batch[:mid], f"{batch_id}-A")
+                                await self._process_single_batch(batch[mid:], f"{batch_id}-B")
+                                break
+                            else:
+                                # Can't split a single document
+                                logger.error(f"Failed to index document {batch[0]['id']} after {max_retries} attempts")
+                                break
+                        
+                        # Normal success case
+                        response.raise_for_status()
+                        result = response.json()
+                        logger.info(f"Batch {batch_id} indexed successfully: {result}")
+                        success = True
+                        db_error_count = 0  # Reset error counter on success
+                        
+                    except Exception as e:
+                        # Check if it looks like a database corruption error
+                        error_str = str(e).lower()
+                        if "duplicate key" in error_str or "unique constraint" in error_str or "nonetype" in error_str:
+                            db_error_count += 1
+                            if db_error_count >= 2:
+                                logger.critical(f"Potential database corruption detected: {error_str}. The search service may need manual intervention. Stopping batch {batch_id} processing.")
+                                break
+                        
+                        if retry_count < max_retries - 1:
+                            retry_count += 1
+                            wait_time = (2 ** retry_count) + (random.random() * 0.5)
+                            logger.warning(f"Error for batch {batch_id}, retrying in {wait_time:.1f}s: {str(e)[:200]}")
+                            await asyncio.sleep(wait_time)
+                        else:
+                            # Last resort - try to split the batch
+                            if len(batch) > 1:
+                                logger.warning(f"Splitting batch {batch_id} after exception: {str(e)[:200]}")
+                                mid = len(batch) // 2
+                                await self._process_single_batch(batch[:mid], f"{batch_id}-A")
+                                await self._process_single_batch(batch[mid:], f"{batch_id}-B")
+                            else:
+                                logger.error(f"Failed to index document {batch[0]['id']} after {max_retries} attempts: {e}")
+                            break
+
+    async def _process_single_batch(self, documents, batch_id):
+        """Process a single batch with maximum reliability"""
+        max_retries = 3
+        retry_count = 0
+        
+        while retry_count < max_retries:
+            try:
+                if not documents:
+                    return
+                    
+                logger.info(f"Processing sub-batch {batch_id} with {len(documents)} documents")
+                response = await self.index_client.post(
+                    "/bulk-index",
+                    json=documents,
+                    timeout=90.0
+                )
+                response.raise_for_status()
+                result = response.json()
+                logger.info(f"Sub-batch {batch_id} indexed successfully: {result}")
+                return  # Success, exit the retry loop
+                
+            except Exception as e:
+                error_str = str(e).lower()
+                retry_count += 1
+                
+                # Check if it's a transient error that txtai might recover from internally
+                if "dictionary changed size" in error_str or "transaction error" in error_str:
+                    wait_time = (2 ** retry_count) + (random.random() * 0.5)
+                    logger.warning(f"Transient txtai error in sub-batch {batch_id}, waiting {wait_time:.1f}s for recovery: {str(e)[:200]}")
+                    await asyncio.sleep(wait_time)  # Wait for txtai to recover
+                    continue  # Try again
+                
+                # For other errors or final retry failure
+                logger.error(f"Error indexing sub-batch {batch_id} (attempt {retry_count}/{max_retries}): {str(e)[:200]}")
+                
+                # Only try one-by-one on the final retry
+                if retry_count >= max_retries and len(documents) > 1:
+                    logger.info(f"Processing documents in sub-batch {batch_id} individually")
+                    for i, doc in enumerate(documents):
+                        try:
+                            resp = await self.index_client.post("/index", json=doc, timeout=30.0)
+                            resp.raise_for_status()
+                            logger.info(f"Indexed document {doc['id']} individually")
+                        except Exception as e2:
+                            logger.error(f"Failed to index document {doc['id']} individually: {str(e2)[:100]}")
+                    return  # Exit after individual processing attempt
+
+    def _truncate_error_detail(self, error_detail):
+        """Truncate error details for logging"""
+        truncated_detail = error_detail.copy() if isinstance(error_detail, dict) else error_detail
+        
+        if isinstance(truncated_detail, dict) and 'detail' in truncated_detail and isinstance(truncated_detail['detail'], list):
+            for i, item in enumerate(truncated_detail['detail']):
+                if isinstance(item, dict) and 'input' in item:
+                    if isinstance(item['input'], dict) and any(k in item['input'] for k in ['documents', 'text']):
+                        # Check for documents list
+                        if 'documents' in item['input'] and isinstance(item['input']['documents'], list):
+                            for j, doc in enumerate(item['input']['documents']):
+                                if 'text' in doc and isinstance(doc['text'], str) and len(doc['text']) > 100:
+                                    item['input']['documents'][j]['text'] = f"{doc['text'][:100]}... [truncated, total {len(doc['text'])} chars]"
+                        
+                        # Check for direct text field
+                        if 'text' in item['input'] and isinstance(item['input']['text'], str) and len(item['input']['text']) > 100:
+                            item['input']['text'] = f"{item['input']['text'][:100]}... [truncated, total {len(item['input']['text'])} chars]"
+        
+        return truncated_detail
 
     async def search(self, text, limit, offset):
-        if not SEARCH_ENABLED:
+        """Search documents"""
+        if not self.available:
+            logger.warning("Search not available")
             return []
-
-        logger.info(f"Ищем: {text} {offset}+{limit}")
-        search_body = {
-            "query": {"multi_match": {"query": text, "fields": ["title", "lead", "subtitle", "body", "media"]}}
-        }
-
-        if self.client:
-            search_response = self.client.search(
-                index=self.index_name,
-                body=search_body,
-                size=limit,
-                from_=offset,
-                _source=False,
-                _source_excludes=["title", "body", "subtitle", "media", "lead", "_index"],
+        
+        if not isinstance(text, str) or not text.strip():
+            logger.warning(f"Invalid search text: {text}")
+            return []
+            
+        logger.info(f"Searching for: '{text}' (limit={limit}, offset={offset})")
+            
+        try:
+            logger.info(f"Sending search request: text='{text}', limit={limit}, offset={offset}")
+            response = await self.client.post(
+                "/search",
+                json={"text": text, "limit": limit, "offset": offset}
             )
-            hits = search_response["hits"]["hits"]
-            results = [{"id": hit["_id"], "score": hit["_score"]} for hit in hits]
-
-            # если результаты не пустые
-            if results:
-                # Кэширование в Redis с TTL
-                redis_key = f"search:{text}:{offset}+{limit}"
-                await redis.execute(
-                    "SETEX",
-                    redis_key,
-                    REDIS_TTL,
-                    json.dumps(results, cls=CustomJSONEncoder),
-                )
-            return results
-        return []
+            response.raise_for_status()
+            
+            logger.info(f"Raw search response: {response.text}")
+            result = response.json()
+            logger.info(f"Parsed search response: {result}")
+            
+            formatted_results = result.get("results", [])
+            logger.info(f"Search for '{text}' returned {len(formatted_results)} results")
+            
+            if formatted_results:
+                logger.info(f"Sample result: {formatted_results[0]}")
+            else:
+                logger.warning(f"No results found for '{text}'")
+            
+            return formatted_results
+        except Exception as e:
+            logger.error(f"Search error for '{text}': {e}", exc_info=True)
+            return []
+    
+    async def check_index_status(self):
+        """Get detailed statistics about the search index health"""
+        if not self.available:
+            return {"status": "disabled"}
+            
+        try:
+            response = await self.client.get("/index-status")
+            response.raise_for_status()
+            result = response.json()
+            logger.info(f"Index status check: {result['status']}, {result['documents_count']} documents")
+            
+            # Log warnings for any inconsistencies
+            if result.get("consistency", {}).get("status") != "ok":
+                null_count = result.get("consistency", {}).get("null_embeddings_count", 0)
+                if null_count > 0:
+                    logger.warning(f"Found {null_count} documents with NULL embeddings")
+                    
+            return result
+        except Exception as e:
+            logger.error(f"Failed to check index status: {e}")
+            return {"status": "error", "message": str(e)}
 
 
+# Create the search service singleton
 search_service = SearchService()
 
 
+# API-compatible function to perform a search
 async def search_text(text: str, limit: int = 50, offset: int = 0):
     payload = []
-    if search_service.client:
-        # Использование метода search_post из OpenSearchService
+    if search_service.available:
         payload = await search_service.search(text, limit, offset)
     return payload
 
 
-# Проверить что URL корректный
-OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "rc1a-3n5pi3bhuj9gieel.mdb.yandexcloud.net")
+async def initialize_search_index(shouts_data):
+    """Initialize search index with existing data during application startup"""
+    if not SEARCH_ENABLED:
+        logger.info("Search indexing skipped (SEARCH_ENABLED=False)")
+        return
+        
+    if not shouts_data:
+        logger.warning("No shouts data provided for search indexing")
+        return
+        
+    logger.info(f"Checking search index status for {len(shouts_data)} documents")
+    
+    # Get the current index info
+    info = await search_service.info()
+    if info.get("status") in ["error", "unavailable", "disabled"]:
+        logger.error(f"Cannot initialize search index: {info}")
+        return
+    
+    # Check if index has approximately right number of documents
+    index_stats = info.get("index_stats", {})
+    indexed_doc_count = index_stats.get("document_count", 0)
+
+    # Add a more detailed status check
+    index_status = await search_service.check_index_status()
+    if index_status.get("status") == "healthy":
+        logger.info("Index status check passed")
+    elif index_status.get("status") == "inconsistent":
+        logger.warning("Index status check found inconsistencies")
+        
+        # Get documents with null embeddings
+        problem_ids = index_status.get("consistency", {}).get("null_embeddings_sample", [])
+                
+        if problem_ids:
+            logger.info(f"Repairing {len(problem_ids)} documents with NULL embeddings")
+            problem_docs = [shout for shout in shouts_data if str(shout.id) in problem_ids]
+            if problem_docs:
+                await search_service.bulk_index(problem_docs)
+    
+    # Log database document summary
+    db_ids = [str(shout.id) for shout in shouts_data]
+    logger.info(f"Database contains {len(shouts_data)} documents. Sample IDs: {', '.join(db_ids[:5])}...")
+    
+    # Calculate summary by ID range to understand the coverage
+    try:
+        # Parse numeric IDs where possible to analyze coverage
+        numeric_ids = [int(sid) for sid in db_ids if sid.isdigit()]
+        if numeric_ids:
+            min_id = min(numeric_ids)
+            max_id = max(numeric_ids)
+            id_range = max_id - min_id + 1
+            coverage_pct = (len(numeric_ids) / id_range) * 100 if id_range > 0 else 0
+            logger.info(f"ID range analysis: min_id={min_id}, max_id={max_id}, range={id_range}, " 
+                        f"coverage={coverage_pct:.1f}% ({len(numeric_ids)}/{id_range})")
+    except Exception as e:
+        logger.warning(f"Could not analyze ID ranges: {e}")
+    
+    # If counts are significantly different, do verification
+    if abs(indexed_doc_count - len(shouts_data)) > 10:
+        logger.info(f"Document count mismatch: {indexed_doc_count} in index vs {len(shouts_data)} in database. Verifying...")
+        
+        # Get all document IDs from your database
+        doc_ids = [str(shout.id) for shout in shouts_data]
+        
+        # Verify which ones are missing from the index
+        verification = await search_service.verify_docs(doc_ids)
+        
+        if verification.get("status") == "error":
+            logger.error(f"Document verification failed: {verification.get('message')}")
+            return
+            
+        # Index only missing documents
+        missing_ids = verification.get("missing", [])
+        if missing_ids:
+            logger.info(f"Found {len(missing_ids)} documents missing from index. Indexing them...")
+            logger.info(f"Sample missing IDs: {', '.join(missing_ids[:10])}...")
+            missing_docs = [shout for shout in shouts_data if str(shout.id) in missing_ids]
+            await search_service.bulk_index(missing_docs)
+        else:
+            logger.info("All documents are already indexed.")
+    else:
+        logger.info(f"Search index appears to be in sync ({indexed_doc_count} documents indexed).")
+        
+        # Optional sample verification (can be slow with large document sets)
+        # Uncomment if you want to periodically check a random sample even when counts match
+        """
+        sample_size = 10
+        if len(db_ids) > sample_size:
+            sample_ids = random.sample(db_ids, sample_size)
+            logger.info(f"Performing random sample verification on {sample_size} documents...")
+            verification = await search_service.verify_docs(sample_ids)
+            if verification.get("missing"):
+                missing_count = len(verification.get("missing", []))
+                logger.warning(f"Random verification found {missing_count}/{sample_size} missing docs " 
+                              f"despite count match. Consider full verification.")
+            else:
+                logger.info("Random document sample verification passed.")
+        """
+    
+    # Verify with test query
+    try:
+        test_query = "test"
+        logger.info(f"Verifying search index with query: '{test_query}'")
+        test_results = await search_text(test_query, 5)
+        
+        if test_results:
+            logger.info(f"Search verification successful: found {len(test_results)} results")
+            # Log categories covered by search results
+            categories = set()
+            for result in test_results:
+                result_id = result.get("id")
+                matching_shouts = [s for s in shouts_data if str(s.id) == result_id]
+                if matching_shouts and hasattr(matching_shouts[0], 'category'):
+                    categories.add(getattr(matching_shouts[0], 'category', 'unknown'))
+            if categories:
+                logger.info(f"Search results cover categories: {', '.join(categories)}")
+        else:
+            logger.warning("Search verification returned no results. Index may be empty or not working.")
+    except Exception as e:
+        logger.error(f"Error verifying search index: {e}")

services/viewed.py

@@ -1,11 +1,10 @@
 import asyncio
+import json
 import os
 import time
 from datetime import datetime, timedelta, timezone
 from typing import Dict
 
-import orjson
-
 # ga
 from google.analytics.data_v1beta import BetaAnalyticsDataClient
 from google.analytics.data_v1beta.types import (
@@ -85,7 +84,7 @@ class ViewedStorage:
                 logger.warn(f" * {viewfile_path} is too old: {self.start_date}")
 
             with open(viewfile_path, "r") as file:
-                precounted_views = orjson.loads(file.read())
+                precounted_views = json.load(file)
                 self.precounted_by_slug.update(precounted_views)
                 logger.info(f" * {len(precounted_views)} shouts with views was loaded.")
 

settings.py

@@ -1,24 +1,18 @@
 import sys
 from os import environ
 
-MODE = "development" if "dev" in sys.argv else "production"
-DEV_SERVER_PID_FILE_NAME = "dev-server.pid"
-
-PORT = environ.get("PORT") or 8000
-
-# storages
+PORT = 8000
 DB_URL = (
     environ.get("DATABASE_URL", "").replace("postgres://", "postgresql://")
     or environ.get("DB_URL", "").replace("postgres://", "postgresql://")
     or "sqlite:///discoursio.db"
 )
 REDIS_URL = environ.get("REDIS_URL") or "redis://127.0.0.1"
-
-# debug
+AUTH_URL = environ.get("AUTH_URL") or ""
 GLITCHTIP_DSN = environ.get("GLITCHTIP_DSN")
+DEV_SERVER_PID_FILE_NAME = "dev-server.pid"
+MODE = "development" if "dev" in sys.argv else "production"
 
-# authorizer.dev
-AUTH_URL = environ.get("AUTH_URL") or "https://auth.discours.io/graphql"
 ADMIN_SECRET = environ.get("AUTH_SECRET") or "nothing"
 WEBHOOK_SECRET = environ.get("WEBHOOK_SECRET") or "nothing-else"
 

utils/encoders.py

@@ -1,28 +1,9 @@
+import json
 from decimal import Decimal
-from json import JSONEncoder
 
 
-class CustomJSONEncoder(JSONEncoder):
-    """
-    Расширенный JSON энкодер с поддержкой сериализации объектов SQLAlchemy.
-
-    Примеры:
-    >>> import json
-    >>> from decimal import Decimal
-    >>> from orm.topic import Topic
-    >>> json.dumps(Decimal("10.50"), cls=CustomJSONEncoder)
-    '"10.50"'
-    >>> topic = Topic(id=1, slug="test")
-    >>> json.dumps(topic, cls=CustomJSONEncoder)
-    '{"id": 1, "slug": "test", ...}'
-    """
-
+class CustomJSONEncoder(json.JSONEncoder):
     def default(self, obj):
         if isinstance(obj, Decimal):
             return str(obj)
-
-        # Проверяем, есть ли у объекта метод dict() (как у моделей SQLAlchemy)
-        if hasattr(obj, "dict") and callable(obj.dict):
-            return obj.dict()
-
         return super().default(obj)

utils/logger.py

@@ -13,7 +13,7 @@ def filter(record: logging.LogRecord):
     record.emoji = (
         "🔍"
         if record.levelno == logging.DEBUG
-        else "ℹ︎"
+        else "🖊️"
         if record.levelno == logging.INFO
         else "🚧"
         if record.levelno == logging.WARNING