chore: Improve gifts indexing

backend/core/src/core/services/user.py

@@ -1,11 +1,14 @@
-from typing import Iterable, Any
+import logging
+from typing import Any, Iterable
 
 from sqlalchemy.orm import joinedload, load_only, QueryableAttribute
 
 from core.dtos.user import TelegramUserDTO
 from core.models.user import User
 from core.services.base import BaseService
 
+logger = logging.getLogger(__name__)
+
 
 class UserService(BaseService):
     def get_all(
@@ -27,9 +30,20 @@ def get_all_telegram_ids(self) -> list[int]:
             u.telegram_id
             for u in self.db_session.query(User.telegram_id)
             .filter(User.telegram_id.is_not(None))
+            .filter(User.is_blocked.is_(False))
             .all()
         ]
 
+    def mark_as_blocked(self, telegram_ids: list[int]) -> None:
+        if not telegram_ids:
+            return
+
+        logger.info(f"Marking {len(telegram_ids)} users as blocked.")
+        self.db_session.query(User).filter(User.telegram_id.in_(telegram_ids)).update(
+            {"is_blocked": True}, synchronize_session=False
+        )
+        self.db_session.flush()
+
     def get_by_telegram_id(self, telegram_id: int) -> User:
         return (
             self.db_session.query(User)

backend/indexer_gifts/actions/user.py

@@ -2,10 +2,11 @@
 from sqlalchemy.orm import Session
 
 from core.actions.base import BaseAction
-from core.services.gift.item import GiftUniqueService
+from core.constants import UPDATED_GIFT_USER_IDS
 from core.services.gift.collection import GiftCollectionService
+from core.services.gift.item import GiftUniqueService
 from core.services.superredis import RedisService
-from core.constants import UPDATED_GIFT_USER_IDS
+from core.services.user import UserService
 from indexer_gifts.indexers.bot_api import BotApiGiftIndexer
 
 logger = logging.getLogger(__name__)
@@ -17,16 +18,51 @@ def __init__(self, db_session: Session):
         self.gift_service = GiftUniqueService(db_session)
         self.gift_collection_service = GiftCollectionService(db_session)
         self.redis_service = RedisService()
+        self.user_service = UserService(db_session)
+
+    async def sync_single_user_gifts(self, telegram_user_id: int) -> None:
+        """Public method for single-user sync, manages the indexer lifecycle."""
+        async with BotApiGiftIndexer() as indexer:
+            try:
+                await self._sync_user_gifts(telegram_user_id, indexer=indexer)
+            except Exception as e:
+                if "user not found" in str(e).lower():
+                    self.user_service.mark_as_blocked([telegram_user_id])
+                raise
+
+    async def sync_users_batch(
+        self,
+        telegram_user_ids: list[int],
+        existing_collection_ids: set[int] | None = None,
+    ) -> None:
+        """Public method for batch sync, manages a single indexer across multiple users."""
+        blocked_user_ids = []
+        async with BotApiGiftIndexer() as indexer:
+            for user_id in telegram_user_ids:
+                try:
+                    await self._sync_user_gifts(
+                        user_id,
+                        indexer=indexer,
+                        existing_collection_ids=existing_collection_ids,
+                    )
+                except Exception as e:
+                    logger.error(f"Error refreshing gifts for user {user_id}: {e}")
+                    if "user not found" in str(e).lower():
+                        blocked_user_ids.append(user_id)
+
+        if blocked_user_ids:
+            self.user_service.mark_as_blocked(blocked_user_ids)
 
-    async def sync_user_gifts(
-        self, telegram_user_id: int, existing_collection_ids: set[int] | None = None
+    async def _sync_user_gifts(
+        self,
+        telegram_user_id: int,
+        indexer: BotApiGiftIndexer,
+        existing_collection_ids: set[int] | None = None,
     ) -> None:
         """
-        Syncs unique gifts for a user using the Telegram Bot API.
-        Identifies gifts the user lost and updates their database ownership.
+        Internal business logic to sync unique gifts for a user using the provided indexer.
         """
         logger.info(f"Starting gifts synchronizing for user {telegram_user_id}")
-        indexer = BotApiGiftIndexer()
         current_owned_slugs = set()
         lost_ownership_user_ids = set()
         created_count = 0
@@ -37,102 +73,99 @@ async def sync_user_gifts(
         if existing_collection_ids is None:
             existing_collection_ids = self.gift_collection_service.get_all_ids()
 
-        try:
-            # 1. Fetch current gifts from Telegram API
-            async for owned_gift in indexer.iter_user_gifts(telegram_user_id):
-                ug = owned_gift.gift
-                current_owned_slugs.add(ug.name)
-
-                # Check if it exists
-                existing = self.gift_service.find(ug.name)
-
-                if existing:
-                    # If it was owned by someone else, record they lost it
-                    if (
-                        existing.telegram_owner_id
-                        and existing.telegram_owner_id != telegram_user_id
-                    ):
-                        lost_ownership_user_ids.add(existing.telegram_owner_id)
-
-                    self.gift_service.update_ownership(
-                        slug=ug.name,
-                        telegram_owner_id=telegram_user_id,
-                        blockchain_address=existing.blockchain_address,
-                        owner_address=existing.owner_address,
-                    )
-                    updated_count += 1
-                else:
-                    # Ensure collection exists using in-memory cache
-                    collection_id = int(ug.gift_id)
-                    if collection_id not in existing_collection_ids:
-                        logger.info(
-                            f"Collection {collection_id} missing, creating stub collection."
-                        )
-                        self.gift_collection_service.create(
-                            id=collection_id,
-                            title=ug.base_name or f"Gift {collection_id}",
-                            preview_url=None,
-                            supply=0,
-                            upgraded_count=0,
-                        )
-                        existing_collection_ids.add(collection_id)
-
-                    # Extract string identifiers for visual attributes if available
-                    model_name = (
-                        getattr(ug.model, "name", None)
-                        if getattr(ug, "model", None)
-                        else None
-                    )
-                    backdrop_name = (
-                        getattr(ug.backdrop, "name", None)
-                        if getattr(ug, "backdrop", None)
-                        else None
-                    )
-                    pattern_name = (
-                        getattr(ug.symbol, "name", None)
-                        if getattr(ug, "symbol", None)
-                        else None
-                    )
+        # 1. Fetch current gifts from Telegram API
+        async for owned_gift in indexer.iter_user_gifts(telegram_user_id):
+            ug = owned_gift.gift
+            current_owned_slugs.add(ug.name)
+
+            # Check if it exists
+            existing = self.gift_service.find(ug.name)
+
+            if existing:
+                # If it was owned by someone else, record they lost it
+                if (
+                    existing.telegram_owner_id
+                    and existing.telegram_owner_id != telegram_user_id
+                ):
+                    lost_ownership_user_ids.add(existing.telegram_owner_id)
 
-                    # Create new gift
-                    self.gift_service.create(
-                        slug=ug.name,
-                        collection_id=collection_id,
-                        number=ug.number,
-                        model=model_name,
-                        backdrop=backdrop_name,
-                        pattern=pattern_name,
-                        telegram_owner_id=telegram_user_id,
-                        blockchain_address=None,
-                        owner_address=None,
+                self.gift_service.update_ownership(
+                    slug=ug.name,
+                    telegram_owner_id=telegram_user_id,
+                    blockchain_address=existing.blockchain_address,
+                    owner_address=existing.owner_address,
+                )
+                updated_count += 1
+            else:
+                # Ensure collection exists using in-memory cache
+                collection_id = int(ug.gift_id)
+                if collection_id not in existing_collection_ids:
+                    logger.info(
+                        f"Collection {collection_id} missing, creating stub collection."
                     )
-                    created_count += 1
-
-            # 2. Fetch all gifts currently assigned to the user in the database
-            db_gifts = self.gift_service.get_all(telegram_user_id=telegram_user_id)
-            for db_gift in db_gifts:
-                if db_gift.slug not in current_owned_slugs:
-                    # User no longer owns this gift
-                    self.gift_service.update_ownership(
-                        slug=db_gift.slug,
-                        telegram_owner_id=None,
-                        blockchain_address=db_gift.blockchain_address,
-                        owner_address=db_gift.owner_address,
+                    self.gift_collection_service.create(
+                        id=collection_id,
+                        title=ug.base_name or f"Gift {collection_id}",
+                        preview_url=None,
+                        supply=0,
+                        upgraded_count=0,
                     )
-                    lost_ownership_user_ids.add(telegram_user_id)
-                    lost_count += 1
+                    existing_collection_ids.add(collection_id)
 
-            # 3. Add users who lost ownership to the Redis set for community manager checking
-            if lost_ownership_user_ids:
-                logger.info(
-                    f"Adding {len(lost_ownership_user_ids)} users to Redis for community manager check due to lost gifts."
+                # Extract string identifiers for visual attributes if available
+                model_name = (
+                    getattr(ug.model, "name", None)
+                    if getattr(ug, "model", None)
+                    else None
+                )
+                backdrop_name = (
+                    getattr(ug.backdrop, "name", None)
+                    if getattr(ug, "backdrop", None)
+                    else None
+                )
+                pattern_name = (
+                    getattr(ug.symbol, "name", None)
+                    if getattr(ug, "symbol", None)
+                    else None
                 )
-                for uid in lost_ownership_user_ids:
-                    self.redis_service.add_to_set(UPDATED_GIFT_USER_IDS, str(uid))
 
+                # Create new gift
+                self.gift_service.create(
+                    slug=ug.name,
+                    collection_id=collection_id,
+                    number=ug.number,
+                    model=model_name,
+                    backdrop=backdrop_name,
+                    pattern=pattern_name,
+                    telegram_owner_id=telegram_user_id,
+                    blockchain_address=None,
+                    owner_address=None,
+                )
+                created_count += 1
+
+        # 2. Fetch all gifts currently assigned to the user in the database
+        db_gifts = self.gift_service.get_all(telegram_user_id=telegram_user_id)
+        for db_gift in db_gifts:
+            if db_gift.slug not in current_owned_slugs:
+                # User no longer owns this gift
+                self.gift_service.update_ownership(
+                    slug=db_gift.slug,
+                    telegram_owner_id=None,
+                    blockchain_address=db_gift.blockchain_address,
+                    owner_address=db_gift.owner_address,
+                )
+                lost_ownership_user_ids.add(telegram_user_id)
+                lost_count += 1
+
+        # 3. Add users who lost ownership to the Redis set for community manager checking
+        if lost_ownership_user_ids:
             logger.info(
-                f"Finished gifts synchronizing for user {telegram_user_id}. "
-                f"Created: {created_count}, Updated: {updated_count}, Lost: {lost_count}."
+                f"Adding {len(lost_ownership_user_ids)} users to Redis for community manager check due to lost gifts."
             )
-        finally:
-            await indexer.close()
+            for uid in lost_ownership_user_ids:
+                self.redis_service.add_to_set(UPDATED_GIFT_USER_IDS, str(uid))
+
+        logger.info(
+            f"Finished gifts synchronizing for user {telegram_user_id}. "
+            f"Created: {created_count}, Updated: {updated_count}, Lost: {lost_count}."
+        )

backend/indexer_gifts/indexers/bot_api.py

@@ -1,4 +1,5 @@
 import logging
+import asyncio
 from typing import AsyncGenerator
 
 from aiogram import Bot
@@ -59,7 +60,15 @@ async def iter_user_gifts(
 
             except Exception as e:
                 logger.error(f"Error fetching gifts for user {telegram_user_id}: {e}")
-                break
+                raise
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.close()
 
     async def close(self):
         await self.bot.session.close()
+        # Allow underlying aiohttp SSL connections to close gracefully
+        await asyncio.sleep(0.250)

backend/indexer_gifts/tasks.py

@@ -3,10 +3,10 @@
 
 from core.constants import (
     CELERY_GIFT_COLLECTIONS_SYNC_QUEUE,
-    CELERY_GIFT_USER_PRIORITY_QUEUE,
     CELERY_GIFT_USER_BATCH_QUEUE,
-    DEFAULT_CELERY_TASK_RETRY_DELAY,
+    CELERY_GIFT_USER_PRIORITY_QUEUE,
     DEFAULT_CELERY_TASK_MAX_RETRIES,
+    DEFAULT_CELERY_TASK_RETRY_DELAY,
 )
 from core.services.db import DBService
 from core.services.user import UserService
@@ -47,7 +47,7 @@ def sync_gift_collections_from_api() -> None:
 def index_user_gifts(telegram_user_id: int) -> bool:
     with DBService().db_session() as db_session:
         action = IndexerUserGiftAction(db_session)
-        asyncio.run(action.sync_user_gifts(telegram_user_id))
+        asyncio.run(action.sync_single_user_gifts(telegram_user_id))
     return True
 
 
@@ -65,15 +65,7 @@ def refresh_users_gifts_batch(telegram_user_ids: list[int]) -> None:
         action = IndexerUserGiftAction(db_session)
         existing_collection_ids = action.gift_collection_service.get_all_ids()
 
-        for user_id in telegram_user_ids:
-            try:
-                asyncio.run(
-                    action.sync_user_gifts(
-                        user_id, existing_collection_ids=existing_collection_ids
-                    )
-                )
-            except Exception as e:
-                logger.error(f"Error refreshing gifts for user {user_id}: {e}")
+        asyncio.run(action.sync_users_batch(telegram_user_ids, existing_collection_ids))
 
 
 # === Flow C: Master Dispatcher Task ===