feat: entity relationship refactor — notification trackers, command system, chat actions

Rework entity schema: rename Tracker→NotificationTracker, add CommandConfig/
CommandTracker/CommandTrackerListener entities for decoupled command handling.
Commands now resolve through CommandTracker→CommandConfig instead of
TelegramBot.commands_config. Smart ref-counted bot polling based on active
listeners. Add chat_action to telegram targets. Full frontend CRUD pages
for command configs and command trackers. Idempotent SQLite migrations.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

packages/server/src/notify_bridge_server/services/scheduler.py

@@ -26,9 +26,9 @@ async def start_scheduler() -> None:
 
     await _load_tracker_jobs()
 
-    # Start Telegram bot polling for bots in polling mode
-    from .telegram_poller import start_bot_polling
-    await start_bot_polling()
+    # Start Telegram bot polling for bots with active command listeners
+    from .telegram_poller import start_command_listener_polling
+    await start_command_listener_polling()
 
 
 async def _load_tracker_jobs() -> None:
@@ -36,13 +36,13 @@ async def _load_tracker_jobs() -> None:
     from sqlmodel import select
     from sqlmodel.ext.asyncio.session import AsyncSession
     from ..database.engine import get_engine
-    from ..database.models import Tracker
+    from ..database.models import NotificationTracker
 
     engine = get_engine()
     scheduler = get_scheduler()
 
     async with AsyncSession(engine) as session:
-        result = await session.exec(select(Tracker).where(Tracker.enabled == True))
+        result = await session.exec(select(NotificationTracker).where(NotificationTracker.enabled == True))
         trackers = result.all()
 
     for tracker in trackers:

packages/server/src/notify_bridge_server/services/telegram_poller.py

@@ -3,6 +3,9 @@
 Uses APScheduler to run getUpdates periodically for each bot
 with update_mode == "polling". Processes updates identically
 to the webhook handler (auto-save chat, dispatch commands).
+
+Ref-counted: only starts/stops polling for bots that have active
+CommandTrackerListeners with enabled CommandTrackers.
 """
 
 from __future__ import annotations
@@ -17,7 +20,7 @@ from sqlmodel.ext.asyncio.session import AsyncSession
 from notify_bridge_core.notifications.telegram.media import TELEGRAM_API_BASE_URL
 
 from ..database.engine import get_engine
-from ..database.models import TelegramBot
+from ..database.models import CommandTracker, CommandTrackerListener, TelegramBot
 from ..services.telegram import save_chat_from_webhook
 from .scheduler import get_scheduler
 
@@ -27,18 +30,82 @@ _LOGGER = logging.getLogger(__name__)
 _last_update_id: dict[int, int] = {}
 
 
-async def start_bot_polling() -> None:
-    """Schedule polling jobs for all bots with update_mode == 'polling'."""
+async def _get_bot_ids_with_active_listeners() -> set[int]:
+    """Return bot IDs that have at least one active command tracker listener.
+
+    A bot is "active" if there is a CommandTrackerListener with
+    listener_type="telegram_bot" pointing to it, AND the associated
+    CommandTracker is enabled.
+    """
     engine = get_engine()
     async with AsyncSession(engine) as session:
         result = await session.exec(
-            select(TelegramBot).where(TelegramBot.update_mode == "polling")
+            select(CommandTrackerListener).where(
+                CommandTrackerListener.listener_type == "telegram_bot"
+            )
+        )
+        listeners = result.all()
+
+        active_bot_ids: set[int] = set()
+        for listener in listeners:
+            tracker = await session.get(CommandTracker, listener.command_tracker_id)
+            if tracker and tracker.enabled:
+                active_bot_ids.add(listener.listener_id)
+
+        return active_bot_ids
+
+
+async def start_command_listener_polling() -> None:
+    """Schedule polling jobs only for bots with active command tracker listeners."""
+    active_bot_ids = await _get_bot_ids_with_active_listeners()
+    if not active_bot_ids:
+        _LOGGER.info("No bots with active command listeners to poll")
+        return
+
+    engine = get_engine()
+    async with AsyncSession(engine) as session:
+        result = await session.exec(
+            select(TelegramBot).where(
+                TelegramBot.update_mode == "polling",
+                TelegramBot.id.in_(active_bot_ids),
+            )
         )
         bots = result.all()
 
     for bot in bots:
         schedule_bot_polling(bot.id)
 
+    _LOGGER.info("Started command listener polling for %d bot(s)", len(bots))
+
+
+async def start_bot_polling() -> None:
+    """Schedule polling jobs for all bots with update_mode == 'polling'.
+
+    Deprecated: prefer start_command_listener_polling() which only starts
+    bots with active command tracker listeners.
+    """
+    await start_command_listener_polling()
+
+
+async def start_bot_if_needed(bot_id: int) -> None:
+    """Start polling for a bot if it has active listeners and is not already running."""
+    engine = get_engine()
+    async with AsyncSession(engine) as session:
+        bot = await session.get(TelegramBot, bot_id)
+        if not bot or bot.update_mode != "polling":
+            return
+
+    active_bot_ids = await _get_bot_ids_with_active_listeners()
+    if bot_id in active_bot_ids:
+        schedule_bot_polling(bot_id)
+
+
+async def stop_bot_if_unused(bot_id: int) -> None:
+    """Stop polling for a bot if it has no enabled command tracker listeners."""
+    active_bot_ids = await _get_bot_ids_with_active_listeners()
+    if bot_id not in active_bot_ids:
+        unschedule_bot_polling(bot_id)
+
 
 def schedule_bot_polling(bot_id: int) -> None:
     """Add a polling job for a bot (idempotent)."""
@@ -70,76 +137,82 @@ def unschedule_bot_polling(bot_id: int) -> None:
 async def _poll_bot(bot_id: int) -> None:
     """Fetch updates from Telegram and process them."""
     engine = get_engine()
+
+    # Eagerly load bot data and close session before aiohttp work
+    # (cannot nest aiohttp inside active SQLAlchemy async session)
     async with AsyncSession(engine) as session:
         bot = await session.get(TelegramBot, bot_id)
         if not bot or bot.update_mode != "polling":
             unschedule_bot_polling(bot_id)
             return
+        # Extract what we need before closing session
+        bot_token = bot.token
+        bot_obj = bot
 
-        offset = _last_update_id.get(bot_id, 0)
-        params: dict[str, Any] = {
-            "timeout": 0,
-            "limit": 50,
-            "allowed_updates": '["message"]',
-        }
-        if offset:
-            params["offset"] = offset + 1
+    offset = _last_update_id.get(bot_id, 0)
+    params: dict[str, Any] = {
+        "timeout": 0,
+        "limit": 50,
+        "allowed_updates": '["message"]',
+    }
+    if offset:
+        params["offset"] = offset + 1
 
+    try:
+        async with aiohttp.ClientSession() as http:
+            async with http.get(
+                f"{TELEGRAM_API_BASE_URL}{bot_token}/getUpdates",
+                params=params,
+                timeout=aiohttp.ClientTimeout(total=10),
+            ) as resp:
+                data = await resp.json()
+                if not data.get("ok"):
+                    return
+                updates = data.get("result", [])
+    except Exception as e:
+        _LOGGER.debug("Polling error for bot %d: %s", bot_id, e)
+        return
+
+    if not updates:
+        return
+
+    # Update offset to latest
+    _last_update_id[bot_id] = updates[-1]["update_id"]
+
+    # Process each update
+    from ..commands.handler import handle_command, send_media_group
+
+    for update in updates:
+        message = update.get("message")
+        if not message:
+            continue
+
+        chat_info = message.get("chat", {})
+        chat_id = str(chat_info.get("id", ""))
+        text = message.get("text", "")
+
+        if not chat_id:
+            continue
+
+        # Auto-persist chat (fresh session per save)
         try:
-            async with aiohttp.ClientSession() as http:
-                async with http.get(
-                    f"{TELEGRAM_API_BASE_URL}{bot.token}/getUpdates",
-                    params=params,
-                    timeout=aiohttp.ClientTimeout(total=10),
-                ) as resp:
-                    data = await resp.json()
-                    if not data.get("ok"):
-                        return
-                    updates = data.get("result", [])
-        except Exception as e:
-            _LOGGER.debug("Polling error for bot %d: %s", bot_id, e)
-            return
+            async with AsyncSession(engine) as save_session:
+                await save_chat_from_webhook(save_session, bot_obj.id, chat_info)
+                await save_session.commit()
+        except Exception:
+            _LOGGER.debug("Failed to auto-save chat %s", chat_id, exc_info=True)
 
-        if not updates:
-            return
-
-        # Update offset to latest
-        _last_update_id[bot_id] = updates[-1]["update_id"]
-
-        # Process each update
-        from ..commands.handler import handle_command, send_media_group
-
-        for update in updates:
-            message = update.get("message")
-            if not message:
-                continue
-
-            chat_info = message.get("chat", {})
-            chat_id = str(chat_info.get("id", ""))
-            text = message.get("text", "")
-
-            if not chat_id:
-                continue
-
-            # Auto-persist chat
+        # Dispatch commands
+        if text and text.startswith("/"):
             try:
-                async with AsyncSession(engine) as save_session:
-                    await save_chat_from_webhook(save_session, bot.id, chat_info)
-                    await save_session.commit()
+                cmd_response = await handle_command(bot_obj, chat_id, text)
+                if cmd_response is not None:
+                    if isinstance(cmd_response, list):
+                        await send_media_group(bot_token, chat_id, cmd_response)
+                    else:
+                        await _send_reply(bot_token, chat_id, cmd_response)
             except Exception:
-                _LOGGER.debug("Failed to auto-save chat %s", chat_id, exc_info=True)
-
-            # Dispatch commands
-            if text and text.startswith("/"):
-                try:
-                    cmd_response = await handle_command(bot, chat_id, text)
-                    if cmd_response is not None:
-                        if isinstance(cmd_response, list):
-                            await send_media_group(bot.token, chat_id, cmd_response)
-                        else:
-                            await _send_reply(bot.token, chat_id, cmd_response)
-                except Exception:
-                    _LOGGER.error("Error handling command from bot %d", bot_id, exc_info=True)
+                _LOGGER.error("Error handling command from bot %d", bot_id, exc_info=True)
 
 
 async def _send_reply(bot_token: str, chat_id: str, text: str) -> None:

packages/server/src/notify_bridge_server/services/watcher.py

@@ -19,11 +19,11 @@ from ..database.engine import get_engine
 from ..database.models import (
     EventLog,
     NotificationTarget,
+    NotificationTracker,
+    NotificationTrackerState,
+    NotificationTrackerTarget,
     ServiceProvider,
     TemplateConfig,
-    Tracker,
-    TrackerState,
-    TrackerTarget,
     TrackingConfig,
 )
 
@@ -89,7 +89,7 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
 
     # Load all DB data eagerly before entering aiohttp context
     async with AsyncSession(engine) as session:
-        tracker = await session.get(Tracker, tracker_id)
+        tracker = await session.get(NotificationTracker, tracker_id)
         if not tracker or not tracker.enabled:
             return {"status": "skipped", "reason": "disabled or not found"}
 
@@ -99,7 +99,7 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
 
         # Load tracker state
         result = await session.exec(
-            select(TrackerState).where(TrackerState.tracker_id == tracker_id)
+            select(NotificationTrackerState).where(NotificationTrackerState.tracker_id == tracker_id)
         )
         states = result.all()
         state_dict: dict[str, Any] = {}
@@ -113,7 +113,7 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
 
         # Load tracker-target links (replaces old target_ids JSON array)
         tt_result = await session.exec(
-            select(TrackerTarget).where(TrackerTarget.tracker_id == tracker_id)
+            select(NotificationTrackerTarget).where(NotificationTrackerTarget.tracker_id == tracker_id)
         )
         tracker_targets = tt_result.all()
 
@@ -188,7 +188,7 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
                 existing.shared = cstate.get("shared", False)
                 session.add(existing)
             else:
-                new_ts = TrackerState(
+                new_ts = NotificationTrackerState(
                     tracker_id=tracker_id,
                     collection_id=cid,
                     collection_name=cstate.get("name", ""),