topics+authors-reimplemented-cache

cache/revalidator.py

@@ -1,17 +1,26 @@
 import asyncio
 
-from cache.cache import cache_author, cache_topic, get_cached_author, get_cached_topic
+from cache.cache import (
+    cache_author,
+    cache_topic,
+    get_cached_author,
+    get_cached_topic,
+    invalidate_cache_by_prefix,
+)
 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=60):
+    def __init__(self, interval=CACHE_REVALIDATION_INTERVAL):
         """Инициализация менеджера с заданным интервалом проверки (в секундах)."""
         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):
         """Запуск фонового воркера для ревалидации кэша."""
@@ -32,22 +41,107 @@ class CacheRevalidationManager:
         """Обновление кэша для всех сущностей, требующих ревалидации."""
         async with self.lock:
             # Ревалидация кэша авторов
-            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["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 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()
+            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()
 
     def mark_for_revalidation(self, entity_id, entity_type):
         """Отметить сущность для ревалидации."""
-        self.items_to_revalidate[entity_type].add(entity_id)
+        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")
 
     async def stop(self):
         """Остановка фонового воркера."""
@@ -60,4 +154,4 @@ class CacheRevalidationManager:
                 pass
 
 
-revalidation_manager = CacheRevalidationManager(interval=300)  # Ревалидация каждые 5 минут
+revalidation_manager = CacheRevalidationManager()