225 lines
9.1 KiB
Python
225 lines
9.1 KiB
Python
"""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
|