feat: optimize blockchain indexer with single-query NFT and last_activity skip

backend/community_manager/actions/chat.py

@@ -13,7 +13,11 @@
     RPCError,
 )
 from telethon.utils import get_peer_id
-from aiogram.utils.markdown import text as fmt_text, bold as fmt_bold
+from aiogram.utils.markdown import (
+    text as fmt_text,
+    bold as fmt_bold,
+    markdown_decoration,
+)
 
 from community_manager.dtos.chat import TargetChatMembersDTO
 from community_manager.events import ChatAdminChangeEventBuilder
@@ -344,7 +348,11 @@ async def _refresh(self, chat: TelegramChat) -> TelegramChat:
             logger.warning(
                 f"Chat {chat.id!r} has insufficient permissions set. Disabling it..."
             )
-            self.telegram_chat_service.set_insufficient_privileges(chat_id=chat.id)
+            if not chat.insufficient_privileges:
+                self.telegram_chat_service.set_insufficient_privileges(chat_id=chat.id)
+                await self._notify_insufficient_privileges(
+                    chat_id=chat.id, chat_title=chat.title
+                )
             raise
 
         except (
@@ -629,12 +637,58 @@ async def on_bot_chat_member_update(
                 self.telegram_chat_service.set_insufficient_privileges(
                     chat_id=chat.id, value=True
                 )
+                await self._notify_insufficient_privileges(
+                    chat_id=chat.id, chat_title=chat.title
+                )
         elif chat.insufficient_privileges:
             logger.info("Sufficient permissions for the bot in chat %d", chat.id)
             self.telegram_chat_service.set_insufficient_privileges(
                 chat_id=chat.id, value=False
             )
 
+    async def _notify_insufficient_privileges(
+        self, chat_id: int, chat_title: str
+    ) -> None:
+        try:
+            managers = (
+                self.db_session.query(TelegramChatUser)
+                .filter(
+                    TelegramChatUser.chat_id == chat_id,
+                    TelegramChatUser.is_manager_admin.is_(True),
+                )
+                .all()
+            )
+            telegram_ids = [m.user.telegram_id for m in managers if m.user]
+
+            if not telegram_ids:
+                logger.warning(f"No manager admins found for chat {chat_id} to notify.")
+                return
+
+            text = fmt_text(
+                "⚠️ ",
+                fmt_bold("Insufficient Privileges"),
+                "\n\n",
+                "The bot no longer has sufficient administrative privileges to manage the chat ",
+                fmt_bold(chat_title),
+                " \\(ID: ",
+                markdown_decoration.quote(str(chat_id)),
+                "\\)\\.\n\n",
+                "Please ensure the bot is added as an administrator with the required permissions\\.",
+                sep="",
+            )
+            async with TelegramBotApiService() as bot_service:
+                for tg_id in telegram_ids:
+                    try:
+                        await bot_service.send_message(chat_id=tg_id, text=text)
+                    except Exception as e:
+                        logger.warning(
+                            f"Failed to notify manager {tg_id} for chat {chat_id}: {e}"
+                        )
+        except Exception as e:
+            logger.error(
+                f"Error in _notify_insufficient_privileges for chat {chat_id}: {e}"
+            )
+
     async def on_join_request(
         self,
         telegram_user_id: int,

backend/core/src/core/dtos/chat/__init__.py

@@ -34,10 +34,10 @@ class BaseTelegramChatDTO(TelegramChatPreviewDTO):
     username: str | None
     is_enabled: bool
     join_url: str | None = None
+    insufficient_privileges: bool
 
 
 class TelegramChatDTO(BaseTelegramChatDTO):
-    insufficient_privileges: bool
     is_full_control: bool
 
     @classmethod
@@ -128,6 +128,7 @@ def from_object(
                 join_url=join_url,
                 members_count=members_count,
                 is_enabled=obj.is_enabled,
+                insufficient_privileges=obj.insufficient_privileges,
             )
         else:
             return cls(
@@ -143,6 +144,7 @@ def from_object(
                 is_enabled=obj.is_enabled,
                 is_eligible=False,
                 join_url=None,
+                insufficient_privileges=obj.insufficient_privileges,
             )
 
 

backend/core/src/core/migrations/versions/1750345690-ec82e8aa0d67-telegram_chat_rule_group.py

@@ -138,6 +138,9 @@ def set_proper_rule_group_id_in_table(
         for chat_id, group_id in chat_id_group_id_mapping.items()
     ]
 
+    if not params:
+        return
+
     # Use executemany for better performance with many records
     connection.execute(
         sa.text(

backend/core/src/core/migrations/versions/1780501788-105b4511d5ca-add_last_activity_to_user_wallet.py

@@ -0,0 +1,32 @@
+"""add_last_activity_to_user_wallet
+
+Revision ID: 105b4511d5ca
+Revises: 6c6b45ffe090
+Create Date: 2026-06-03 15:49:48.553736
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision: str = "105b4511d5ca"
+down_revision: Union[str, None] = "6c6b45ffe090"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "user_wallet", sa.Column("last_activity", mysql.BIGINT(), nullable=True)
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("user_wallet", "last_activity")
+    # ### end Alembic commands ###

backend/core/src/core/models/wallet.py

@@ -27,6 +27,11 @@ class UserWallet(Base):
     address = mapped_column(BlockchainAddressRawField, primary_key=True)
     user_id = mapped_column(ForeignKey("user.id"), nullable=False)
     balance = mapped_column(BIGINT, nullable=True, doc="Balance of the wallet in TONs")
+    last_activity = mapped_column(
+        BIGINT,
+        nullable=True,
+        doc="Last activity timestamp of the wallet on the blockchain",
+    )
     # DEPRECATED attribute
     hide_wallet = mapped_column(Boolean, default=False, nullable=False)
 

backend/core/src/core/services/wallet.py

@@ -101,7 +101,9 @@ def turn_visibility_off(self, user_id: int) -> None:
         ).update({"hide_wallet": True})
         self.db_session.flush()
 
-    def set_balance(self, address_raw: str, balance: int) -> None:
+    def set_balance(
+        self, address_raw: str, balance: int, last_activity: int | None = None
+    ) -> None:
         """
         Updates the balance for a specific wallet address using the database session.
 
@@ -111,10 +113,15 @@ def set_balance(self, address_raw: str, balance: int) -> None:
 
         :param address_raw: The wallet address whose balance needs to be updated.
         :param balance: The new balance to be set for the given wallet address in nano
+        :param last_activity: Optional last activity timestamp of the wallet
         """
+        updates = {"balance": balance}
+        if last_activity is not None:
+            updates["last_activity"] = last_activity
+
         self.db_session.query(UserWallet).filter(
             UserWallet.address == address_raw,
-        ).update({"balance": balance})
+        ).update(updates)
 
     def count(self) -> int:
         return self.db_session.query(UserWallet).count()

backend/indexer_blockchain/tasks.py

@@ -22,14 +22,14 @@
 
 
 async def get_all_nfts_per_user(
-    blockchain_service: TonApiService, address: str, nft_collections: list[str]
+    blockchain_service: TonApiService, address: str
 ) -> NftItems:
     nft_items = []
-    for collection_address in nft_collections:
-        async for batch in blockchain_service.get_all_nft_items_for_user(
-            wallet_address=address, collection_address=collection_address
-        ):
-            nft_items.extend(batch.nft_items)
+    # Query all NFT items in paginated batches
+    async for batch in blockchain_service.get_all_nft_items_for_user(
+        wallet_address=address, collection_address=None
+    ):
+        nft_items.extend(batch.nft_items)
     return NftItems(nft_items=nft_items)
 
 
@@ -46,6 +46,20 @@ def fetch_wallet_details(address: str) -> None:
     blockchain_service = TonApiService()
 
     account_info = asyncio.run(blockchain_service.get_account_info(address))
+    current_last_activity = account_info.last_activity
+    raw_address = account_info.address.to_raw()
+
+    with DBService().db_session() as db_session:
+        wallet_service = WalletService(db_session)
+        wallet = wallet_service.get_user_wallet(raw_address)
+        stored_last_activity = wallet.last_activity if wallet else None
+
+    if stored_last_activity is not None and current_last_activity is not None:
+        if stored_last_activity == current_last_activity:
+            logger.info(
+                f"Skipping wallet {address!r} sync: last_activity has not changed ({current_last_activity})."
+            )
+            return
 
     jettons_balances: JettonsBalances = asyncio.run(
         blockchain_service.get_all_jetton_balances(address)
@@ -64,16 +78,25 @@ def fetch_wallet_details(address: str) -> None:
         get_all_nfts_per_user(
             blockchain_service=blockchain_service,
             address=address,
-            nft_collections=whitelist_collection_addresses,
         )
     )
 
+    # Pre-filter fetched NFT items against whitelisted collections in memory
+    whitelist_set = set(whitelist_collection_addresses)
+    filtered_nfts = [
+        item
+        for item in nft_items.nft_items
+        if item.collection and item.collection.address.to_raw() in whitelist_set
+    ]
+    nft_items = NftItems(nft_items=filtered_nfts)
+
     with DBService().db_session() as db_session:
         wallet_service = WalletService(db_session)
         wallet_service.set_balance(
-            account_info.address.to_raw(),
+            raw_address,
             # It already contains the balance in nano
             int(str(account_info.balance)),
+            last_activity=current_last_activity,
         )
 
         jetton_service = JettonService(db_session)