From 908628d492eeebcd5674f33c200a7c7e806dea16 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Fri, 25 Sep 2026 22:15:29 +0200 Subject: [PATCH 1/5] fix(indexer): handle flood control with retry logic in BotApiGiftIndexer - Wraps get_user_gifts with _safe_request to catch TelegramRetryAfter - Sleeps for retry_after seconds and retries up to 3 times before raising - Removes duplicate generic exception logging in pagination loop --- backend/indexer_gifts/indexers/bot_api.py | 77 ++++++++++++++--------- 1 file changed, 47 insertions(+), 30 deletions(-) diff --git a/backend/indexer_gifts/indexers/bot_api.py b/backend/indexer_gifts/indexers/bot_api.py index 3732e0ea..451534a1 100644 --- a/backend/indexer_gifts/indexers/bot_api.py +++ b/backend/indexer_gifts/indexers/bot_api.py @@ -1,15 +1,18 @@ -import logging import asyncio -from typing import AsyncGenerator +import logging +from typing import Any, AsyncGenerator from aiogram import Bot from aiogram.client.default import DefaultBotProperties +from aiogram.exceptions import TelegramRetryAfter from aiogram.types import OwnedGifts, OwnedGiftUnique from core.settings import core_settings logger = logging.getLogger(__name__) +MAX_FLOOD_RETRIES = 3 + class BotApiGiftIndexer: def __init__(self): @@ -19,48 +22,62 @@ def __init__(self): default=DefaultBotProperties(parse_mode="HTML"), ) + async def _safe_request(self, func, *args, **kwargs) -> Any: + """ + Wraps a request with retry logic for Telegram flood control (429). + """ + for attempt in range(1, MAX_FLOOD_RETRIES + 1): + try: + return await func(*args, **kwargs) + except TelegramRetryAfter as e: + if attempt == MAX_FLOOD_RETRIES: + logger.error( + f"Flood control: exceeded {MAX_FLOOD_RETRIES} retries, giving up." + ) + raise + logger.warning( + f"Flood control: sleeping {e.retry_after}s " + f"(attempt {attempt}/{MAX_FLOOD_RETRIES})" + ) + await asyncio.sleep(e.retry_after) + raise RuntimeError("Unexpected state in _safe_request") + async def iter_user_gifts( self, telegram_user_id: int ) -> AsyncGenerator[OwnedGiftUnique, None]: """ Yields OwnedGiftUnique objects owned by the user via pagination. - Handles rate limits or pagination as needed. + Handles flood control by sleeping and retrying up to MAX_FLOOD_RETRIES times. """ offset = "" page_num = 1 logger.info(f"Starting to fetch gifts for user {telegram_user_id}...") while True: - try: - logger.debug( - f"Fetching page {page_num} for user {telegram_user_id} (offset: {offset!r})" - ) - # get_user_gifts is available in aiogram 3.24.0 - user_gifts: OwnedGifts = await self.bot.get_user_gifts( - user_id=telegram_user_id, offset=offset, limit=100 - ) + user_gifts: OwnedGifts = await self._safe_request( + self.bot.get_user_gifts, + user_id=telegram_user_id, + offset=offset, + limit=100, + ) - logger.info( - f"Fetched {len(user_gifts.gifts)} gifts for user {telegram_user_id} on page {page_num}." - ) + logger.info( + f"Fetched {len(user_gifts.gifts)} gifts for user " + f"{telegram_user_id} on page {page_num}." + ) - # Yield only unique gifts from the current page - for gift in user_gifts.gifts: - if isinstance(gift, OwnedGiftUnique): - yield gift + for gift in user_gifts.gifts: + if isinstance(gift, OwnedGiftUnique): + yield gift - # Check if there are more pages - if not user_gifts.next_offset: - logger.info( - f"Finished fetching all gifts for user {telegram_user_id}. Total count: {user_gifts.total_count}" - ) - break - - offset = user_gifts.next_offset - page_num += 1 + if not user_gifts.next_offset: + logger.info( + f"Finished fetching all gifts for user {telegram_user_id}. " + f"Total count: {user_gifts.total_count}" + ) + break - except Exception as e: - logger.error(f"Error fetching gifts for user {telegram_user_id}: {e}") - raise + offset = user_gifts.next_offset + page_num += 1 async def __aenter__(self): return self From ca83d3096027054dac726c6ac6391a7a00fb0681 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Fri, 25 Sep 2026 22:15:53 +0200 Subject: [PATCH 2/5] feat(gifts): track user writeability and skip un-synced collections - Skips gifts belonging to collections not yet synced rather than creating empty stubs - Uses allows_write_to_pm flag instead of is_blocked to represent user bot writeability - Marks users as not writable on Telegram 'user not found' error during gifts indexing - Marks users as writable when interacting with bot via /start - Preserves allows_write_to_pm during profile update from auth - Adds unit tests for get_all_writable_telegram_ids and status toggles --- backend/community_manager/handlers/bot.py | 8 ++++ backend/core/src/core/services/user.py | 19 +++++--- backend/indexer_gifts/actions/user.py | 25 ++++------ backend/indexer_gifts/tasks.py | 2 +- backend/tests/unit/core/services/test_user.py | 47 ++++++++++++++++--- 5 files changed, 71 insertions(+), 30 deletions(-) diff --git a/backend/community_manager/handlers/bot.py b/backend/community_manager/handlers/bot.py index 9d2613a5..c447e256 100644 --- a/backend/community_manager/handlers/bot.py +++ b/backend/community_manager/handlers/bot.py @@ -1,5 +1,8 @@ from telethon import events, types +from core.services.db import DBService +from core.services.user import UserService + START_MESSAGE = """**Access bot** What I can do: @@ -14,6 +17,11 @@ async def handle_start_message(event: events.NewMessage()) -> None: + # Mark user as writable since they interacted with the bot + with DBService().db_session() as db_session: + user_service = UserService(db_session) + user_service.mark_as_writable(event.sender_id) + await event.respond( START_MESSAGE, buttons=types.ReplyInlineMarkup( diff --git a/backend/core/src/core/services/user.py b/backend/core/src/core/services/user.py index fc44ad3a..7c9b9b57 100644 --- a/backend/core/src/core/services/user.py +++ b/backend/core/src/core/services/user.py @@ -25,22 +25,30 @@ def get_all( return query.all() - def get_all_telegram_ids(self) -> list[int]: + def get_all_writable_telegram_ids(self) -> list[int]: return [ 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)) + .filter(User.allows_write_to_pm.is_(True)) .all() ] - def mark_as_blocked(self, telegram_ids: list[int]) -> None: + def mark_as_not_writable(self, telegram_ids: list[int]) -> None: if not telegram_ids: return - logger.info(f"Marking {len(telegram_ids)} users as blocked.") + logger.info( + f"Marking {len(telegram_ids)} users as not writable (allows_write_to_pm=False)." + ) self.db_session.query(User).filter(User.telegram_id.in_(telegram_ids)).update( - {"is_blocked": True}, synchronize_session=False + {"allows_write_to_pm": False}, synchronize_session=False + ) + self.db_session.flush() + + def mark_as_writable(self, telegram_id: int) -> None: + self.db_session.query(User).filter(User.telegram_id == telegram_id).update( + {"allows_write_to_pm": True} ) self.db_session.flush() @@ -88,7 +96,6 @@ def update(self, user: User, telegram_user: TelegramUserDTO) -> User: user.last_name = telegram_user.last_name user.username = telegram_user.username user.is_premium = bool(telegram_user.is_premium) - user.allows_write_to_pm = telegram_user.allow_write_to_pm # TODO add photo_url self.db_session.add(user) self.db_session.flush() diff --git a/backend/indexer_gifts/actions/user.py b/backend/indexer_gifts/actions/user.py index 43110f8d..9a8f7a18 100644 --- a/backend/indexer_gifts/actions/user.py +++ b/backend/indexer_gifts/actions/user.py @@ -27,7 +27,8 @@ async def sync_single_user_gifts(self, telegram_user_id: int) -> None: 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]) + self.user_service.mark_as_not_writable([telegram_user_id]) + return raise async def sync_users_batch( @@ -36,7 +37,7 @@ async def sync_users_batch( existing_collection_ids: set[int] | None = None, ) -> None: """Public method for batch sync, manages a single indexer across multiple users.""" - blocked_user_ids = [] + not_writable_user_ids = [] async with BotApiGiftIndexer() as indexer: for user_id in telegram_user_ids: try: @@ -48,10 +49,10 @@ async def sync_users_batch( 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) + not_writable_user_ids.append(user_id) - if blocked_user_ids: - self.user_service.mark_as_blocked(blocked_user_ids) + if not_writable_user_ids: + self.user_service.mark_as_not_writable(not_writable_user_ids) async def _sync_user_gifts( self, @@ -95,20 +96,12 @@ async def _sync_user_gifts( ) 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." + logger.debug( + f"Skipping gift {ug.name}: collection {collection_id} not yet synced." ) - 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) + continue # Extract string identifiers for visual attributes if available model_name = ( diff --git a/backend/indexer_gifts/tasks.py b/backend/indexer_gifts/tasks.py index 5c4cd6a0..0e738818 100644 --- a/backend/indexer_gifts/tasks.py +++ b/backend/indexer_gifts/tasks.py @@ -78,7 +78,7 @@ def refresh_all_user_gifts() -> None: """Master dispatcher: chunks all registered users and enqueues batch tasks.""" with DBService().db_session() as db_session: user_service = UserService(db_session) - all_telegram_ids = user_service.get_all_telegram_ids() + all_telegram_ids = user_service.get_all_writable_telegram_ids() if not all_telegram_ids: logger.info("No registered Telegram users found for gift refresh.") diff --git a/backend/tests/unit/core/services/test_user.py b/backend/tests/unit/core/services/test_user.py index 04856a90..511d96fe 100644 --- a/backend/tests/unit/core/services/test_user.py +++ b/backend/tests/unit/core/services/test_user.py @@ -2,19 +2,52 @@ from tests.factories.user import UserFactory -def test_get_all_telegram_ids(db_session): +def test_get_all_writable_telegram_ids(db_session): user_service = UserService(db_session) - # Create some users with and without telegram_ids - UserFactory.with_session(db_session).create(telegram_id=10001) - UserFactory.with_session(db_session).create(telegram_id=10002) - UserFactory.with_session(db_session).create(telegram_id=None) + # Create some users with and without telegram_ids and allows_write_to_pm + UserFactory.with_session(db_session).create( + telegram_id=10001, allows_write_to_pm=True + ) + UserFactory.with_session(db_session).create( + telegram_id=10002, allows_write_to_pm=True + ) + UserFactory.with_session(db_session).create( + telegram_id=10003, allows_write_to_pm=False + ) + UserFactory.with_session(db_session).create( + telegram_id=None, allows_write_to_pm=True + ) db_session.flush() - telegram_ids = user_service.get_all_telegram_ids() + telegram_ids = user_service.get_all_writable_telegram_ids() assert 10001 in telegram_ids assert 10002 in telegram_ids + assert 10003 not in telegram_ids assert None not in telegram_ids - assert set([10001, 10002]).issubset(set(telegram_ids)) + +def test_mark_as_not_writable_and_writable(db_session): + user_service = UserService(db_session) + + user = UserFactory.with_session(db_session).create( + telegram_id=20001, allows_write_to_pm=True + ) + db_session.flush() + + assert user.allows_write_to_pm is True + + # Mark as not writable + user_service.mark_as_not_writable([20001]) + db_session.refresh(user) + assert user.allows_write_to_pm is False + + # Mark back as writable + user_service.mark_as_writable(20001) + db_session.refresh(user) + assert user.allows_write_to_pm is True + + # Calling mark_as_writable on a non-existent user should be a safe no-op + user_service.mark_as_writable(999999999) + user_service.mark_as_not_writable([999999999]) From f79cd91382062131a7234cd7b2fdb3b4c557e03a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Fri, 25 Sep 2026 22:16:14 +0200 Subject: [PATCH 3/5] chore(scheduler): adjust intervals for gift collections and user gifts sync - Runs sync-gift-collections-from-api hourly at :00 - Runs refresh-all-user-gifts every 4 hours at :02, giving collection sync time to complete --- backend/scheduler/celery_app.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/scheduler/celery_app.py b/backend/scheduler/celery_app.py index 99c4c601..95a7ba6c 100644 --- a/backend/scheduler/celery_app.py +++ b/backend/scheduler/celery_app.py @@ -56,12 +56,12 @@ def create_app() -> Celery: # }, "sync-gift-collections-from-api": { "task": "sync-gift-collections-from-api", - "schedule": crontab(hour="1", minute="0"), # Daily at 1:00 AM + "schedule": crontab(hour="*/1", minute="0"), # Hourly at :00 "options": {"queue": CELERY_GIFT_COLLECTIONS_SYNC_QUEUE}, }, "refresh-all-user-gifts": { "task": "refresh-all-user-gifts", - "schedule": crontab(hour="*/1", minute="15"), # Hourly at :15 + "schedule": crontab(hour="*/4", minute="2"), # Every 4 hours at :02 "options": {"queue": CELERY_GIFT_USER_BATCH_QUEUE}, }, "refresh-prices": { From 1a2be7002ca721f0b6054da9108a534b8b5c69e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Fri, 25 Sep 2026 22:20:11 +0200 Subject: [PATCH 4/5] chore(scheduler): temporary update sync-gift-collections-from-api schedule to run every 10 minutes --- backend/scheduler/celery_app.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/backend/scheduler/celery_app.py b/backend/scheduler/celery_app.py index 95a7ba6c..88aa3870 100644 --- a/backend/scheduler/celery_app.py +++ b/backend/scheduler/celery_app.py @@ -49,14 +49,10 @@ def create_app() -> Celery: "schedule": crontab(minute="*/10"), # Every 10 minutes "options": {"queue": CELERY_STICKER_FETCH_QUEUE_NAME}, }, - # "fetch-gift-ownerships": { - # "task": "fetch-gift-ownership-details", - # "schedule": crontab(hour="*/1", minute="0"), # Every hour - # "options": {"queue": CELERY_GIFT_COLLECTIONS_SYNC_QUEUE}, - # }, "sync-gift-collections-from-api": { "task": "sync-gift-collections-from-api", - "schedule": crontab(hour="*/1", minute="0"), # Hourly at :00 + # "schedule": crontab(hour="1", minute="0"), # Daily at 01:00 + "schedule": crontab(minute="*/10"), # Every 15 minutes "options": {"queue": CELERY_GIFT_COLLECTIONS_SYNC_QUEUE}, }, "refresh-all-user-gifts": { From 26c2a3f2831e3b578678175d6c0786636a85cb1c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Fri, 25 Sep 2026 22:28:18 +0200 Subject: [PATCH 5/5] fix(indexer): simplify _safe_request retry loop and remove RuntimeError --- backend/indexer_gifts/indexers/bot_api.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/backend/indexer_gifts/indexers/bot_api.py b/backend/indexer_gifts/indexers/bot_api.py index 451534a1..f97ccd75 100644 --- a/backend/indexer_gifts/indexers/bot_api.py +++ b/backend/indexer_gifts/indexers/bot_api.py @@ -26,11 +26,12 @@ async def _safe_request(self, func, *args, **kwargs) -> Any: """ Wraps a request with retry logic for Telegram flood control (429). """ - for attempt in range(1, MAX_FLOOD_RETRIES + 1): + attempt = 1 + while True: try: return await func(*args, **kwargs) except TelegramRetryAfter as e: - if attempt == MAX_FLOOD_RETRIES: + if attempt >= MAX_FLOOD_RETRIES: logger.error( f"Flood control: exceeded {MAX_FLOOD_RETRIES} retries, giving up." ) @@ -40,7 +41,7 @@ async def _safe_request(self, func, *args, **kwargs) -> Any: f"(attempt {attempt}/{MAX_FLOOD_RETRIES})" ) await asyncio.sleep(e.retry_after) - raise RuntimeError("Unexpected state in _safe_request") + attempt += 1 async def iter_user_gifts( self, telegram_user_id: int