diff --git a/backend/core/src/core/services/user.py b/backend/core/src/core/services/user.py index 9b05cf23..fc44ad3a 100644 --- a/backend/core/src/core/services/user.py +++ b/backend/core/src/core/services/user.py @@ -1,4 +1,5 @@ -from typing import Iterable, Any +import logging +from typing import Any, Iterable from sqlalchemy.orm import joinedload, load_only, QueryableAttribute @@ -6,6 +7,8 @@ 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) diff --git a/backend/indexer_gifts/actions/user.py b/backend/indexer_gifts/actions/user.py index 651a8de9..db98344c 100644 --- a/backend/indexer_gifts/actions/user.py +++ b/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}." + ) diff --git a/backend/indexer_gifts/indexers/bot_api.py b/backend/indexer_gifts/indexers/bot_api.py index 75bd7a07..3732e0ea 100644 --- a/backend/indexer_gifts/indexers/bot_api.py +++ b/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) diff --git a/backend/indexer_gifts/tasks.py b/backend/indexer_gifts/tasks.py index 2a8e975a..5c4cd6a0 100644 --- a/backend/indexer_gifts/tasks.py +++ b/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 ===