feat(backend): alerts come enriched and paged with an unseen count and a live alert frame

backend/src/api/realtime.py

@@ -2,7 +2,7 @@ import asyncio
 import json
 import logging
 from datetime import UTC, datetime
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import asyncpg
 
@@ -10,6 +10,10 @@ from utils.events import BG_EVENTS_CHANNEL
 from utils.listen import Listener
 from utils.read import chats as chats_read
 from utils.read import presence as presence_read
+from utils.read import watches as watches_read
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
 
 logger = logging.getLogger(__name__)
 
@@ -27,8 +31,15 @@ class EventHub:
     def __init__(self) -> None:
         self._subscribers: set[Subscriber] = set()
         self._pool: asyncpg.Pool | None = None
-        self._listener = Listener("realtime", {BG_EVENTS_CHANNEL: self._on_notify})
+        self._listener = Listener(
+            "realtime",
+            {
+                BG_EVENTS_CHANNEL: self._on_notify,
+                watches_read.ALERTS_CHANGED_CHANNEL: self._on_alert,
+            },
+        )
         self._tasks: set[asyncio.Task] = set()
+        self.alert_listeners: list[Callable[[], None]] = []
 
     def subscribe(self, account_id: int, chat_id: int | None) -> Subscriber:
         sub = Subscriber(account_id, chat_id)
@@ -52,6 +63,39 @@ class EventHub:
         self._tasks.add(task)
         task.add_done_callback(self._tasks.discard)
 
+    def _on_alert(self, payload: str) -> None:
+        for listener in self.alert_listeners:
+            listener()
+        task = asyncio.create_task(self._dispatch_alert(payload))
+        self._tasks.add(task)
+        task.add_done_callback(self._tasks.discard)
+
+    async def _dispatch_alert(self, payload: str) -> None:
+        try:
+            event = json.loads(payload)
+        except json.JSONDecodeError:
+            return
+        account_id = event.get("account_id")
+        targets = [
+            sub
+            for sub in self._subscribers
+            if sub.account_id == account_id and sub.chat_id is None
+        ]
+        if not targets or self._pool is None:
+            return
+        alert_id = event.get("alert_id")
+        item = await watches_read.get_alert(self._pool, alert_id) if alert_id else None
+        frame = {
+            "type": "alert",
+            "alert": item.model_dump(mode="json") if item else None,
+            "unseen": await watches_read.unseen_count(self._pool, account_id),
+        }
+        for sub in targets:
+            try:
+                sub.queue.put_nowait(frame)
+            except asyncio.QueueFull:
+                logger.warning("Dropping alert for slow subscriber")
+
     async def _dispatch(self, payload: str) -> None:
         try:
             event = json.loads(payload)