moved

cache/cache.py

@@ -0,0 +1,342 @@
+import asyncio
+import json
+from typing import List
+
+from sqlalchemy import select, join, and_
+from orm.author import Author, AuthorFollower
+from orm.topic import Topic, TopicFollower
+from orm.shout import Shout, ShoutAuthor, ShoutTopic
+from services.db import local_session
+from utils.encoders import CustomJSONEncoder
+from cache.rediscache import redis
+from utils.logger import root_logger as logger
+
+DEFAULT_FOLLOWS = {
+    "topics": [],
+    "authors": [],
+    "communities": [{"id": 1, "name": "Дискурс", "slug": "discours", "pic": ""}],
+}
+
+
+# Кэширование данных темы
+async def cache_topic(topic: dict):
+    payload = json.dumps(topic, cls=CustomJSONEncoder)
+    # Одновременное кэширование по id и slug для быстрого доступа
+    await asyncio.gather(
+        redis.execute("SET", f"topic:id:{topic['id']}", payload),
+        redis.execute("SET", f"topic:slug:{topic['slug']}", payload),
+    )
+
+
+# Кэширование данных автора
+async def cache_author(author: dict):
+    payload = json.dumps(author, cls=CustomJSONEncoder)
+    # Кэширование данных автора по user и id
+    await asyncio.gather(
+        redis.execute("SET", f"author:user:{author['user'].strip()}", str(author["id"])),
+        redis.execute("SET", f"author:id:{author['id']}", payload),
+    )
+
+
+async def get_cached_topic(topic_id: int):
+    """
+    Получает информацию о теме из кэша или базы данных.
+
+    Args:
+        topic_id (int): Идентификатор темы.
+
+    Returns:
+        dict: Данные темы в формате словаря или None, если тема не найдена.
+    """
+    # Ключ для кэширования темы в Redis
+    topic_key = f"topic:id:{topic_id}"
+    cached_topic = await redis.get(topic_key)
+    if cached_topic:
+        return json.loads(cached_topic)
+
+    # Если данных о теме нет в кэше, загружаем из базы данных
+    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.set(topic_key, json.dumps(topic_dict, cls=CustomJSONEncoder))
+            return topic_dict
+
+    return None
+
+
+async def get_cached_shout_authors(shout_id: int):
+    """
+    Retrieves a list of authors for a given shout from the cache or database if not present.
+
+    Args:
+        shout_id (int): The ID of the shout for which to retrieve authors.
+
+    Returns:
+        List[dict]: A list of dictionaries containing author data.
+    """
+    # Attempt to retrieve cached author IDs for the shout
+    rkey = f"shout:authors:{shout_id}"
+    cached_author_ids = await redis.get(rkey)
+    if cached_author_ids:
+        author_ids = json.loads(cached_author_ids)
+    else:
+        # If not in cache, fetch from the database and cache the result
+        with local_session() as session:
+            query = (
+                select(ShoutAuthor.author)
+                .where(ShoutAuthor.shout == shout_id)
+                .join(Author, ShoutAuthor.author == Author.id)
+                .filter(Author.deleted_at.is_(None))
+            )
+            author_ids = [author_id for (author_id,) in session.execute(query).all()]
+            await redis.execute("set", rkey, json.dumps(author_ids))
+
+    # Retrieve full author details from cached IDs
+    if author_ids:
+        authors = await get_cached_authors_by_ids(author_ids)
+        logger.debug(f"Shout#{shout_id} authors fetched and cached: {len(authors)} authors found.")
+        return authors
+
+    return []
+
+
+# Кэширование данных о подписках
+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.get(key)
+    follows = json.loads(follows_str) if follows_str else []
+    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))
+    update_follower_stat(follower_id, entity_type, len(follows))
+
+
+# Обновление статистики подписчика
+async def update_follower_stat(follower_id, entity_type, count):
+    follower_key = f"author:id:{follower_id}"
+    follower_str = await redis.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)
+
+
+# Получение автора из кэша
+async def get_cached_author(author_id: int):
+    author_key = f"author:id:{author_id}"
+    result = await redis.get(author_key)
+    if result:
+        return json.loads(result)
+    # Загрузка из базы данных, если не найдено в кэше
+    with local_session() as session:
+        author = session.execute(select(Author).where(Author.id == author_id)).scalar_one_or_none()
+        if author:
+            await cache_author(author.dict())
+            return author.dict()
+    return None
+
+
+# Получение темы по slug из кэша
+async def get_cached_topic_by_slug(slug: str):
+    topic_key = f"topic:slug:{slug}"
+    result = await redis.get(topic_key)
+    if result:
+        return json.loads(result)
+    # Загрузка из базы данных, если не найдено в кэше
+    with local_session() as session:
+        topic = session.execute(select(Topic).where(Topic.slug == slug)).scalar_one_or_none()
+        if topic:
+            await cache_topic(topic.dict())
+            return topic.dict()
+    return None
+
+
+# Получение списка авторов по ID из кэша
+async def get_cached_authors_by_ids(author_ids: List[int]) -> List[dict]:
+    # Одновременное получение данных всех авторов
+    keys = [f"author:id:{author_id}" for author_id in author_ids]
+    results = await asyncio.gather(*(redis.get(key) for key in keys))
+    authors = [json.loads(result) if result else None for result in results]
+    # Загрузка отсутствующих авторов из базы данных и кэширование
+    missing_indices = [index for index, author in enumerate(authors) if author is None]
+    if missing_indices:
+        missing_ids = [author_ids[index] for index in missing_indices]
+        with local_session() as session:
+            query = select(Author).where(Author.id.in_(missing_ids))
+            missing_authors = session.execute(query).scalars().all()
+            await asyncio.gather(*(cache_author(author.dict()) for author in missing_authors))
+            authors = [author.dict() for author in missing_authors]
+    return authors
+
+
+async def get_cached_topic_followers(topic_id: int):
+    # Попытка извлечь кэшированные данные
+    cached = await redis.get(f"topic:followers:{topic_id}")
+    if cached:
+        followers = json.loads(cached)
+        logger.debug(f"Cached followers for topic#{topic_id}: {len(followers)}")
+        return followers
+
+    # Загрузка из базы данных и кэширование результатов
+    with local_session() as session:
+        followers_ids = [
+            f[0]
+            for f in session.query(Author.id)
+            .join(TopicFollower, TopicFollower.follower == Author.id)
+            .filter(TopicFollower.topic == topic_id)
+            .all()
+        ]
+        await redis.execute("SET", f"topic:followers:{topic_id}", json.dumps(followers_ids))
+        followers = await get_cached_authors_by_ids(followers_ids)
+        return followers
+
+
+async def get_cached_author_followers(author_id: int):
+    # Проверяем кэш на наличие данных
+    cached = await redis.get(f"author:followers:{author_id}")
+    if 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
+
+    # Запрос в базу данных если кэш пуст
+    with local_session() as session:
+        followers_ids = [
+            f[0]
+            for f in session.query(Author.id)
+            .join(AuthorFollower, AuthorFollower.follower == Author.id)
+            .filter(AuthorFollower.author == author_id, Author.id != author_id)
+            .all()
+        ]
+        await redis.execute("SET", f"author:followers:{author_id}", json.dumps(followers_ids))
+        followers = await get_cached_authors_by_ids(followers_ids)
+        return followers
+
+
+async def get_cached_follower_authors(author_id: int):
+    # Попытка получить авторов из кэша
+    cached = await redis.get(f"author:follows-authors:{author_id}")
+    if cached:
+        authors_ids = json.loads(cached)
+    else:
+        # Запрос авторов из базы данных
+        with local_session() as session:
+            authors_ids = [
+                a[0]
+                for a in session.execute(
+                    select(Author.id)
+                    .select_from(join(Author, AuthorFollower, Author.id == AuthorFollower.author))
+                    .where(AuthorFollower.follower == author_id)
+                ).all()
+            ]
+            await redis.execute("SET", f"author:follows-authors:{author_id}", json.dumps(authors_ids))
+
+    authors = await get_cached_authors_by_ids(authors_ids)
+    return authors
+
+
+async def get_cached_follower_topics(author_id: int):
+    # Попытка получить темы из кэша
+    cached = await redis.get(f"author:follows-topics:{author_id}")
+    if cached:
+        topics_ids = json.loads(cached)
+    else:
+        # Загрузка тем из базы данных и их кэширование
+        with local_session() as session:
+            topics_ids = [
+                t[0]
+                for t in session.query(Topic.id)
+                .join(TopicFollower, TopicFollower.topic == Topic.id)
+                .where(TopicFollower.follower == author_id)
+                .all()
+            ]
+            await redis.execute("SET", f"author:follows-topics:{author_id}", json.dumps(topics_ids))
+
+    topics = []
+    for topic_id in topics_ids:
+        topic_str = await redis.get(f"topic:id:{topic_id}")
+        if topic_str:
+            topic = json.loads(topic_str)
+            if topic and topic not in topics:
+                topics.append(topic)
+
+    logger.debug(f"Cached topics for author#{author_id}: {len(topics)}")
+    return topics
+
+
+async def get_cached_author_by_user_id(user_id: str):
+    """
+    Получает информацию об авторе по его user_id, сначала проверяя кэш, а затем базу данных.
+
+    Args:
+        user_id (str): Идентификатор пользователя, по которому нужно получить автора.
+
+    Returns:
+        dict: Словарь с данными автора или None, если автор не найден.
+    """
+    # Пытаемся найти ID автора по user_id в кэше Redis
+    author_id = await redis.get(f"author:user:{user_id.strip()}")
+    if author_id:
+        # Если ID найден, получаем полные данные автора по его ID
+        author_data = await redis.get(f"author:id:{author_id}")
+        if author_data:
+            return json.loads(author_data)
+
+    # Если данные в кэше не найдены, выполняем запрос к базе данных
+    with local_session() as session:
+        author = session.execute(select(Author).where(Author.user == user_id)).scalar_one_or_none()
+
+        if author:
+            # Кэшируем полученные данные автора
+            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}", json.dumps(author_dict)),
+            )
+            return author_dict
+
+    # Возвращаем None, если автор не найден
+    return None
+
+
+async def get_cached_topic_authors(topic_id: int):
+    """
+    Получает список авторов для заданной темы, используя кэш или базу данных.
+
+    Args:
+        topic_id (int): Идентификатор темы, для которой нужно получить авторов.
+
+    Returns:
+        List[dict]: Список словарей, содержащих данные авторов.
+    """
+    # Пытаемся получить список ID авторов из кэша
+    rkey = f"topic:authors:{topic_id}"
+    cached_authors_ids = await redis.get(rkey)
+    if cached_authors_ids:
+        authors_ids = json.loads(cached_authors_ids)
+    else:
+        # Если кэш пуст, получаем данные из базы данных
+        with local_session() as session:
+            query = (
+                select(ShoutAuthor.author)
+                .select_from(join(ShoutTopic, Shout, ShoutTopic.shout == Shout.id))
+                .join(ShoutAuthor, ShoutAuthor.shout == Shout.id)
+                .where(and_(ShoutTopic.topic == topic_id, Shout.published_at.is_not(None), Shout.deleted_at.is_(None)))
+            )
+            authors_ids = [author_id for (author_id,) in session.execute(query).all()]
+            # Кэшируем полученные ID авторов
+            await redis.execute("set", rkey, json.dumps(authors_ids))
+
+    # Получаем полные данные авторов по кэшированным ID
+    if authors_ids:
+        authors = await get_cached_authors_by_ids(authors_ids)
+        logger.debug(f"Topic#{topic_id} authors fetched and cached: {len(authors)} authors found.")
+        return authors
+
+    return []

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

@@ -0,0 +1,150 @@
+import json
+
+from sqlalchemy import and_, join, select
+
+from orm.author import Author, AuthorFollower
+from orm.shout import Shout, ShoutAuthor, ShoutTopic
+from orm.topic import Topic, TopicFollower
+from resolvers.stat import get_with_stat
+from cache.cache import cache_author, cache_topic
+from services.db import local_session
+from utils.encoders import CustomJSONEncoder
+from utils.logger import root_logger as logger
+from cache.rediscache import redis
+
+
+async def precache_authors_followers(author_id, session):
+    # Precache author followers
+    authors_followers = set()
+    followers_query = select(AuthorFollower.follower).where(AuthorFollower.author == author_id)
+    result = session.execute(followers_query)
+
+    for row in result:
+        follower_id = row[0]
+        if follower_id:
+            authors_followers.add(follower_id)
+
+    followers_payload = json.dumps(
+        [f for f in authors_followers],
+        cls=CustomJSONEncoder,
+    )
+    await redis.execute("SET", f"author:followers:{author_id}", followers_payload)
+
+
+async def precache_authors_follows(author_id, session):
+    # Precache topics followed by author
+    follows_topics = set()
+    follows_topics_query = select(TopicFollower.topic).where(TopicFollower.follower == author_id)
+    result = session.execute(follows_topics_query)
+
+    for row in result:
+        followed_topic_id = row[0]
+        if followed_topic_id:
+            follows_topics.add(followed_topic_id)
+
+    topics_payload = json.dumps([t for t in follows_topics], cls=CustomJSONEncoder)
+    await redis.execute("SET", f"author:follows-topics:{author_id}", topics_payload)
+
+    # Precache authors followed by author
+    follows_authors = set()
+    follows_authors_query = select(AuthorFollower.author).where(AuthorFollower.follower == author_id)
+    result = session.execute(follows_authors_query)
+
+    for row in result:
+        followed_author_id = row[0]
+        if followed_author_id:
+            follows_authors.add(followed_author_id)
+
+    authors_payload = json.dumps([a for a in follows_authors], cls=CustomJSONEncoder)
+    await redis.execute("SET", f"author:follows-authors:{author_id}", authors_payload)
+
+
+async def precache_topics_authors(topic_id: int, session):
+    # Precache topic authors
+    topic_authors = set()
+    topic_authors_query = (
+        select(ShoutAuthor.author)
+        .select_from(join(ShoutTopic, Shout, ShoutTopic.shout == Shout.id))
+        .join(ShoutAuthor, ShoutAuthor.shout == Shout.id)
+        .filter(
+            and_(
+                ShoutTopic.topic == topic_id,
+                Shout.published_at.is_not(None),
+                Shout.deleted_at.is_(None),
+            )
+        )
+    )
+    result = session.execute(topic_authors_query)
+
+    for row in result:
+        author_id = row[0]
+        if author_id:
+            topic_authors.add(author_id)
+
+    authors_payload = json.dumps([a for a in topic_authors], cls=CustomJSONEncoder)
+    await redis.execute("SET", f"topic:authors:{topic_id}", authors_payload)
+
+
+async def precache_topics_followers(topic_id: int, session):
+    # Precache topic followers
+    topic_followers = set()
+    followers_query = select(TopicFollower.follower).where(TopicFollower.topic == topic_id)
+    result = session.execute(followers_query)
+
+    for row in result:
+        follower_id = row[0]
+        if follower_id:
+            topic_followers.add(follower_id)
+
+    followers_payload = json.dumps([f for f in topic_followers], cls=CustomJSONEncoder)
+    await redis.execute("SET", f"topic:followers:{topic_id}", followers_payload)
+
+
+async def precache_data():
+    try:
+        # cache reset
+        await redis.execute("FLUSHDB")
+        logger.info("redis flushed")
+
+        # topics
+        topics_by_id = {}
+        topics = get_with_stat(select(Topic))
+        for topic in topics:
+            topic_profile = topic.dict() if not isinstance(topic, dict) else topic
+            await cache_topic(topic_profile)
+        logger.info(f"{len(topics)} topics precached")
+
+        # followings for topics
+        with local_session() as session:
+            for topic_id in topics_by_id.keys():
+                await precache_topics_followers(topic_id, session)
+                await precache_topics_authors(topic_id, session)
+        logger.info("topics followings precached")
+
+        # authors
+        authors_by_id = {}
+        authors = get_with_stat(select(Author).where(Author.user.is_not(None)))
+        logger.debug(f"{len(authors)} authors found in database")
+        c = 0
+        for author in authors:
+            if isinstance(author, Author):
+                profile = author.dict()
+                author_id = profile.get("id")
+                user_id = profile.get("user", "").strip()
+                if author_id and user_id:
+                    authors_by_id[author_id] = profile
+                    await cache_author(profile)
+                    c += 1
+            else:
+                logger.error(f"fail caching {author}")
+
+        logger.info(f"{c} authors precached")
+
+        # followings for authors
+        with local_session() as session:
+            for author_id in authors_by_id.keys():
+                await precache_authors_followers(author_id, session)
+                await precache_authors_follows(author_id, session)
+        logger.info("authors followings precached")
+    except Exception as exc:
+        logger.error(exc)

cache/rediscache.py

@@ -0,0 +1,63 @@
+import logging
+
+import redis.asyncio as aredis
+
+from settings import REDIS_URL
+
+# Set redis logging level to suppress DEBUG messages
+logger = logging.getLogger("redis")
+logger.setLevel(logging.WARNING)
+
+
+class RedisCache:
+    def __init__(self, uri=REDIS_URL):
+        self._uri: str = uri
+        self.pubsub_channels = []
+        self._client = None
+
+    async def connect(self):
+        self._client = aredis.Redis.from_url(self._uri, decode_responses=True)
+
+    async def disconnect(self):
+        if self._client:
+            await self._client.close()
+
+    async def execute(self, command, *args, **kwargs):
+        if self._client:
+            try:
+                logger.debug(f"{command}")  # {args[0]}") # {args} {kwargs}")
+                for arg in args:
+                    if isinstance(arg, dict):
+                        if arg.get("_sa_instance_state"):
+                            del arg["_sa_instance_state"]
+                r = await self._client.execute_command(command, *args, **kwargs)
+                # logger.debug(type(r))
+                # logger.debug(r)
+                return r
+            except Exception as e:
+                logger.error(e)
+
+    async def subscribe(self, *channels):
+        if self._client:
+            async with self._client.pubsub() as pubsub:
+                for channel in channels:
+                    await pubsub.subscribe(channel)
+                    self.pubsub_channels.append(channel)
+
+    async def unsubscribe(self, *channels):
+        if not self._client:
+            return
+        async with self._client.pubsub() as pubsub:
+            for channel in channels:
+                await pubsub.unsubscribe(channel)
+                self.pubsub_channels.remove(channel)
+
+    async def publish(self, channel, data):
+        if not self._client:
+            return
+        await self._client.publish(channel, data)
+
+
+redis = RedisCache()
+
+__all__ = ["redis"]

cache/revalidator.py

@@ -0,0 +1,48 @@
+import asyncio
+from utils.logger import root_logger as logger
+from cache.cache import get_cached_author, cache_author, cache_topic, get_cached_topic
+
+
+class CacheRevalidationManager:
+    """Управление периодической ревалидацией кэша."""
+
+    def __init__(self):
+        self.items_to_revalidate = {"authors": set(), "topics": set()}
+        self.revalidation_interval = 60  # Интервал ревалидации в секундах
+        self.loop = None
+
+    def start(self):
+        self.loop = asyncio.get_event_loop()
+        self.loop.run_until_complete(self.revalidate_cache())
+        self.loop.run_forever()
+        logger.info("[services.revalidator] started infinite loop")
+
+    async def revalidate_cache(self):
+        """Периодическая ревалидация кэша."""
+        while True:
+            await asyncio.sleep(self.revalidation_interval)
+            await self.process_revalidation()
+
+    async def process_revalidation(self):
+        """Ревалидация кэша для отмеченных сущностей."""
+        for entity_type, ids in self.items_to_revalidate.items():
+            for entity_id in ids:
+                if entity_type == "authors":
+                    # Ревалидация кэша автора
+                    author = await get_cached_author(entity_id)
+                    if author:
+                        await cache_author(author)
+                elif entity_type == "topics":
+                    # Ревалидация кэша темы
+                    topic = await get_cached_topic(entity_id)
+                    if topic:
+                        await cache_topic(topic)
+            ids.clear()
+
+    def mark_for_revalidation(self, entity_id, entity_type):
+        """Отметить сущность для ревалидации."""
+        self.items_to_revalidate[entity_type].add(entity_id)
+
+
+# Инициализация менеджера ревалидации
+revalidation_manager = CacheRevalidationManager()

cache/triggers.py

@@ -0,0 +1,85 @@
+from sqlalchemy import event
+from orm.author import Author, AuthorFollower
+from orm.reaction import Reaction
+from orm.shout import Shout, ShoutAuthor
+from orm.topic import Topic, TopicFollower
+from cache.revalidator import revalidation_manager
+from utils.logger import root_logger as logger
+
+
+def after_update_handler(mapper, connection, target):
+    """Обработчик обновления сущности."""
+    entity_type = "authors" if isinstance(target, Author) else "topics" if isinstance(target, Topic) else "shouts"
+    revalidation_manager.mark_for_revalidation(target.id, entity_type)
+
+
+def after_follower_insert_update_handler(mapper, connection, target):
+    """Обработчик добавления или обновления подписки."""
+    if isinstance(target, AuthorFollower):
+        # Пометить автора и подписчика для ревалидации
+        revalidation_manager.mark_for_revalidation(target.author_id, "authors")
+        revalidation_manager.mark_for_revalidation(target.follower_id, "authors")
+    elif isinstance(target, TopicFollower):
+        # Пометить тему и подписчика для ревалидации
+        revalidation_manager.mark_for_revalidation(target.topic_id, "topics")
+        revalidation_manager.mark_for_revalidation(target.follower_id, "authors")
+
+
+def after_follower_delete_handler(mapper, connection, target):
+    """Обработчик удаления подписки."""
+    if isinstance(target, AuthorFollower):
+        # Пометить автора и подписчика для ревалидации
+        revalidation_manager.mark_for_revalidation(target.author_id, "authors")
+        revalidation_manager.mark_for_revalidation(target.follower_id, "authors")
+    elif isinstance(target, TopicFollower):
+        # Пометить тему и подписчика для ревалидации
+        revalidation_manager.mark_for_revalidation(target.topic_id, "topics")
+        revalidation_manager.mark_for_revalidation(target.follower_id, "authors")
+
+
+def after_reaction_update_handler(mapper, connection, reaction):
+    """Обработчик изменений реакций."""
+    # Пометить shout для ревалидации
+    revalidation_manager.mark_for_revalidation(reaction.shout_id, "shouts")
+    # Пометить автора реакции для ревалидации
+    revalidation_manager.mark_for_revalidation(reaction.created_by, "authors")
+
+
+def after_shout_author_insert_update_handler(mapper, connection, target):
+    """Обработчик добавления или обновления авторства публикации."""
+    # Пометить shout и автора для ревалидации
+    revalidation_manager.mark_for_revalidation(target.shout_id, "shouts")
+    revalidation_manager.mark_for_revalidation(target.author_id, "authors")
+
+
+def after_shout_author_delete_handler(mapper, connection, target):
+    """Обработчик удаления авторства публикации."""
+    # Пометить shout и автора для ревалидации
+    revalidation_manager.mark_for_revalidation(target.shout_id, "shouts")
+    revalidation_manager.mark_for_revalidation(target.author_id, "authors")
+
+
+def events_register():
+    """Регистрация обработчиков событий для всех сущностей."""
+    event.listen(ShoutAuthor, "after_insert", after_shout_author_insert_update_handler)
+    event.listen(ShoutAuthor, "after_update", after_shout_author_insert_update_handler)
+    event.listen(ShoutAuthor, "after_delete", after_shout_author_delete_handler)
+
+    event.listen(AuthorFollower, "after_insert", after_follower_insert_update_handler)
+    event.listen(AuthorFollower, "after_update", after_follower_insert_update_handler)
+    event.listen(AuthorFollower, "after_delete", after_follower_delete_handler)
+    event.listen(TopicFollower, "after_insert", after_follower_insert_update_handler)
+    event.listen(TopicFollower, "after_update", after_follower_insert_update_handler)
+    event.listen(TopicFollower, "after_delete", after_follower_delete_handler)
+    event.listen(Reaction, "after_update", after_reaction_update_handler)
+
+    event.listen(Author, "after_update", after_update_handler)
+    event.listen(Topic, "after_update", after_update_handler)
+    event.listen(Shout, "after_update", after_update_handler)
+    event.listen(
+        Reaction,
+        "after_update",
+        lambda mapper, connection, target: revalidation_manager.mark_for_revalidation(target.shout, "shouts"),
+    )
+
+    logger.info("Event handlers registered successfully.")

cache/unread.py

@@ -0,0 +1,24 @@
+import json
+
+from cache.rediscache import redis
+
+
+async def get_unread_counter(chat_id: str, author_id: int) -> int:
+    r = await redis.execute("LLEN", f"chats/{chat_id}/unread/{author_id}")
+    if isinstance(r, str):
+        return int(r)
+    elif isinstance(r, int):
+        return r
+    else:
+        return 0
+
+
+async def get_total_unread_counter(author_id: int) -> int:
+    chats_set = await redis.execute("SMEMBERS", f"chats_by_author/{author_id}")
+    s = 0
+    if isinstance(chats_set, str):
+        chats_set = json.loads(chats_set)
+    if isinstance(chats_set, list):
+        for chat_id in chats_set:
+            s += await get_unread_counter(chat_id, author_id)
+    return s