feat(bot): add MAX notifications adapter and operator handoff

Refs #83
2026-10-04 16:10:30 +03:00
parent 28967c7c37
commit 7780b484f3
8 changed files with 966 additions and 0 deletions
@@ -0,0 +1,224 @@
+"""Notifications-only MAX adapter; no patient text enters the dialog engine or an LLM.
+
+Webhook transactions persist receipts and replies together, then return immediately.
+The worker delivers from identity DB with leases and bounded retries. MAX has no send
+idempotency key: a crash after API acceptance and before commit can duplicate a message.
+"""
+
+import asyncio
+import contextlib
+import json
+import logging
+from functools import lru_cache
+
+import yaml
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+
+from app.channels.max_client import MaxApiError, MaxClient
+from app.config import get_settings
+from app.core import clock
+from app.db.max_notifications import MaxBinding, MaxReceipt
+from app.db.session import get_identity_session
+from app.dialog.contracts import Reply
+from app.integrations import max_store as store
+from app.integrations.base import Outbox, integrations
+
+log = logging.getLogger(__name__)
+
+
+class Texts(BaseModel):
+    connected: str = Field(max_length=4000)
+    not_connected: str = Field(max_length=4000)
+    invalid: str = Field(max_length=4000)
+    stopped: str = Field(max_length=4000)
+    help: str = Field(max_length=4000)
+    notification: str = Field(max_length=4000)
+    completed: str = Field(max_length=4000)
+    open_site: str = Field(max_length=128)
+
+
+@lru_cache
+def texts() -> Texts:
+    path = get_settings().config_dir / "texts" / "max.yaml"
+    return Texts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
+
+
+def _event(event: dict) -> tuple[str, int, str, str] | None:
+    """Normalize only documented private-user events; unknown/malformed events are ignored."""
+    kind = event.get("update_type")
+    stamp = event.get("timestamp")
+    if type(stamp) is not int or stamp < 0:
+        return None
+    message_id = ""
+    if kind == "message_created":
+        message = event.get("message")
+        if not isinstance(message, dict):
+            return None
+        recipient, body = message.get("recipient"), message.get("body")
+        if not isinstance(recipient, dict) or recipient.get("chat_type") != "dialog" or not isinstance(body, dict):
+            return None
+        user, text, message_id = message.get("sender"), body.get("text"), body.get("mid")
+        if not isinstance(message_id, str) or not message_id or not isinstance(text, str):
+            return None
+    elif kind in ("bot_started", "bot_stopped", "dialog_removed"):
+        user, text = event.get("user"), event.get("payload") or ""
+    else:
+        return None
+    if not isinstance(user, dict) or user.get("is_bot") is not False:
+        return None
+    uid = user.get("user_id")
+    if type(uid) is not int or not 0 < uid < 2**63 or not isinstance(text, str) or len(text) > 4000:
+        return None
+    # Names, profile details and raw text are neither persisted nor logged.
+    receipt = store.digest(f"{kind}:{uid}:{message_id or stamp}")
+    return kind, uid, text.strip(), receipt
+
+
+async def handle_update(event: dict) -> None:
+    if not get_settings().max_enabled or (parsed := _event(event)) is None:
+        return
+    kind, user_id, text, receipt = parsed
+    now = clock.now()
+    try:
+        async with get_identity_session() as session:
+            if await session.get(MaxReceipt, receipt):
+                return
+            session.add(MaxReceipt(digest=receipt, received_at=now))
+            await session.flush()
+            binding = await session.scalar(select(MaxBinding).where(MaxBinding.user_id == user_id))
+            reply_token = None
+            command, _, payload = text.partition(" ")
+            if kind in ("bot_stopped", "dialog_removed") or command.lower() == "/stop":
+                if binding:
+                    await store.disconnect(session, binding.token)
+                # When MAX itself stops/deletes the bot, do not attempt a reply.
+                if kind != "message_created":
+                    return
+                reply = "stopped"
+            elif kind == "bot_started" or command.lower() == "/start":
+                code = text if kind == "bot_started" else payload.strip()
+                if code:
+                    token = await store.code_token(session, code, now) if len(code) <= 128 else None
+                    linked = (
+                        token and await store.active_patient(token) and await store.bind(session, code, user_id, now)
+                    )
+                    reply = "connected" if linked else "invalid"
+                    reply_token = token if linked else None
+                else:
+                    reply = "connected" if binding and binding.enabled else "not_connected"
+                    reply_token = binding.token if binding and binding.enabled else None
+            else:
+                reply = "help"
+            await store.enqueue(session, key=receipt, user_id=user_id, token=reply_token, kind=reply, now=now)
+    except IntegrityError:
+        # A concurrent delivery of the same webhook can race on its receipt primary key.
+        async with get_identity_session() as session:
+            if await session.get(MaxReceipt, receipt) is None:
+                raise
+
+
+class MaxOutbox:
+    def __init__(self, fallback: Outbox):
+        self.fallback = fallback
+
+    async def push(self, token: str, replies: list[Reply]) -> None:
+        if replies and get_settings().max_enabled and await store.active_patient(token):
+            async with get_identity_session() as session:
+                binding = await session.get(MaxBinding, token)
+                if binding and binding.enabled:
+                    # Hash the event for retry deduplication; never store/transmit its clinical text.
+                    event = json.dumps([r.model_dump() for r in replies], sort_keys=True, ensure_ascii=False)
+                    key = store.digest(f"{token}:{binding.connected_at.isoformat()}:{clock.today()}:{event}")
+                    await store.enqueue(
+                        session, key=key, user_id=binding.user_id, token=token, kind="notification", now=clock.now()
+                    )
+        await self.fallback.push(token, replies)
+
+
+async def deliver_one(client: MaxClient) -> bool:
+    async with get_identity_session() as session:
+        row = await store.claim(session, clock.now())
+    if row is None:
+        return False
+    if row.token is not None:
+        async with get_identity_session() as session:
+            binding = await session.get(MaxBinding, row.token)
+            valid = (
+                binding
+                and binding.enabled
+                and binding.user_id == row.user_id
+                and binding.connected_at <= row.created_at
+            )
+        if not valid or not await store.active_patient(row.token):
+            async with get_identity_session() as session:
+                await store.finish(session, row, state="cancelled", now=clock.now())
+            return True
+    settings, messages = get_settings(), texts()
+    try:
+        await client.send(
+            row.user_id, getattr(messages, row.kind), site_url=settings.max_site_url, button=messages.open_site
+        )
+    except MaxApiError as exc:
+        retry = exc.retryable and row.attempts < store.MAX_ATTEMPTS
+        delay = max(exc.retry_after, min(300, 5 * 2 ** (row.attempts - 1)))
+        async with get_identity_session() as session:
+            await store.finish(
+                session,
+                row,
+                state="queued" if retry else "failed",
+                now=clock.now(),
+                error=exc.code,
+                retry_seconds=delay if retry else 0,
+            )
+        log.warning("MAX delivery deferred=%s code=%s", retry, exc.code)
+    else:
+        async with get_identity_session() as session:
+            await store.finish(session, row, state="sent", now=clock.now())
+    return True
+
+
+async def _run(client: MaxClient) -> None:
+    tick = 0
+    while True:
+        try:
+            if tick % 30 == 0:
+                async with get_identity_session() as session:
+                    await store.collect_completed(session, clock.now())
+            await deliver_one(client)
+        except Exception as exc:
+            # No exception text: SQL/HTTP errors may contain identifiers or parameters.
+            log.error("MAX worker failed type=%s", type(exc).__name__)
+        tick += 1
+        # One send per second per process, below MAX's two-per-second per-dialog limit.
+        await asyncio.sleep(1)
+
+
+_client: MaxClient | None = None
+_task: asyncio.Task | None = None
+_wrapper: MaxOutbox | None = None
+
+
+async def start() -> None:
+    global _client, _task, _wrapper
+    if not get_settings().max_enabled or _task is not None:
+        return
+    texts()  # Validate templates before replacing a working outbox.
+    _client = MaxClient(get_settings())
+    _wrapper = MaxOutbox(integrations.outbox)
+    integrations.outbox = _wrapper
+    _task = asyncio.create_task(_run(_client), name="max-notifications")
+
+
+async def stop() -> None:
+    global _client, _task, _wrapper
+    if _task:
+        _task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await _task
+    if _wrapper and integrations.outbox is _wrapper:
+        integrations.outbox = _wrapper.fallback
+    if _client:
+        await _client.close()
+    _client = _task = _wrapper = None