"""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