From 64655ad9341f615415235eab35a5190f5a4fc12c Mon Sep 17 00:00:00 2001 From: Pavel Eliseev Date: Sun, 4 Oct 2026 15:59:27 +0300 Subject: [PATCH 1/2] feat(backend): add durable MAX linking and notification queue Refs #82 --- app/api/max_notifications.py | 110 ++++++++++++++++++ app/api/routes.py | 3 +- app/db/bootstrap.py | 1 + app/db/max_notifications.py | 50 ++++++++ app/integrations/max_store.py | 201 +++++++++++++++++++++++++++++++++ tests/test_api_max.py | 92 +++++++++++++++ tests/test_db.py | 10 +- tests/test_integrations_max.py | 123 ++++++++++++++++++++ 8 files changed, 588 insertions(+), 2 deletions(-) create mode 100644 app/api/max_notifications.py create mode 100644 app/db/max_notifications.py create mode 100644 app/integrations/max_store.py create mode 100644 tests/test_api_max.py create mode 100644 tests/test_integrations_max.py diff --git a/app/api/max_notifications.py b/app/api/max_notifications.py new file mode 100644 index 0000000..c232976 --- /dev/null +++ b/app/api/max_notifications.py @@ -0,0 +1,110 @@ +"""Server API for a future MAX connect UI; webhook processing is supplied by the channel adapter.""" + +import hmac +import json +import re +from datetime import datetime +from typing import Annotated, Literal + +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +from pydantic import BaseModel, ConfigDict +from sqlalchemy import select + +from app.config import get_settings +from app.core import clock +from app.db import repo +from app.db.max_notifications import MaxBinding, MaxDelivery +from app.db.session import get_clinical_session, get_identity_session +from app.integrations import max_store + +router = APIRouter(tags=["max-notifications"]) +bearer = HTTPBearer(auto_error=False) + + +async def patient_token(auth: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]) -> str: + if auth is None or auth.scheme.lower() != "bearer" or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", auth.credentials): + raise HTTPException(401, "patient authentication required", headers={"WWW-Authenticate": "Bearer"}) + async with get_clinical_session() as session: + if await repo.get_profile(session, auth.credentials) is None: + raise HTTPException(401, "patient authentication required") + return auth.credentials + + +PatientToken = Annotated[str, Depends(patient_token)] + + +class MaxStatus(BaseModel): + available: bool + connected: bool + last_delivery: str | None = None + + +class LinkConsent(BaseModel): + model_config = ConfigDict(extra="forbid") + consent: Literal[True] + + +class MaxLink(BaseModel): + url: str + expires_at: datetime + + +@router.get("/me/notifications/max", response_model=MaxStatus) +async def status(token: PatientToken) -> MaxStatus: + async with get_identity_session() as session: + binding = await session.get(MaxBinding, token) + last = await session.scalar( + select(MaxDelivery.state) + .where(MaxDelivery.token == token) + .order_by(MaxDelivery.created_at.desc(), MaxDelivery.id.desc()) + .limit(1) + ) + return MaxStatus( + available=get_settings().max_enabled, connected=bool(binding and binding.enabled), last_delivery=last + ) + + +@router.post("/me/notifications/max/link", response_model=MaxLink) +async def link(body: LinkConsent, token: PatientToken) -> MaxLink: + settings = get_settings() + if not settings.max_enabled: + raise HTTPException(503, "MAX is not configured") + if not await max_store.active_patient(token): + raise HTTPException(403, "active consent and adult confirmation required") + async with get_identity_session() as session: + raw, expires = await max_store.issue_code(session, token, clock.now(), settings.max_link_ttl_seconds) + return MaxLink(url=f"https://max.ru/{settings.max_bot_username}?start={raw}", expires_at=expires) + + +@router.delete("/me/notifications/max", response_model=MaxStatus) +async def unlink(token: PatientToken) -> MaxStatus: + async with get_identity_session() as session: + await max_store.disconnect(session, token) + return MaxStatus(available=get_settings().max_enabled, connected=False) + + +@router.post("/webhooks/max") +async def webhook(request: Request) -> dict[str, bool]: + settings = get_settings() + if not settings.max_enabled: + raise HTTPException(503, "MAX is not configured") + expected = settings.max_webhook_secret.get_secret_value().encode() + supplied = request.headers.get("X-Max-Bot-Api-Secret", "").encode() + if not expected or not hmac.compare_digest(supplied, expected): + raise HTTPException(403, "invalid webhook secret") + raw = bytearray() + async for chunk in request.stream(): + raw.extend(chunk) + if len(raw) > 65536: + raise HTTPException(413, "event is too large") + try: + event = json.loads(raw) + except (ValueError, UnicodeError) as exc: + raise HTTPException(400, "invalid JSON") from exc + if not isinstance(event, dict): + raise HTTPException(422, "event must be an object") + from app.channels.max_bot import handle_update + + await handle_update(event) + return {"success": True} diff --git a/app/api/routes.py b/app/api/routes.py index 8d616b8..88266a7 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -15,7 +15,7 @@ from sqlalchemy import select from app.api import demo as console from app.api import doctor_views as views -from app.api import fixtures +from app.api import fixtures, max_notifications from app.api.patient_web import router as patient_web_router from app.api.questionnaire import router as questionnaire_router from app.api.schemas import ( @@ -46,6 +46,7 @@ from app.scheduler import jobs as scheduler from app.services.monitoring import refresh_status router = APIRouter(prefix="/api/v1") +router.include_router(max_notifications.router) router.include_router(questionnaire_router) router.include_router(patient_web_router) diff --git a/app/db/bootstrap.py b/app/db/bootstrap.py index f4f163d..1b6724b 100644 --- a/app/db/bootstrap.py +++ b/app/db/bootstrap.py @@ -5,6 +5,7 @@ The only module (besides registration code) that touches both schemas — and on from app.db import audit as _audit # noqa: F401 (registers audit_log in clinical metadata) from app.db import intake as _intake # noqa: F401 +from app.db import max_notifications as _max_notifications # noqa: F401 from app.db import scheduler as _scheduler # noqa: F401 from app.db.clinical import ClinicalBase from app.db.identity import IdentityBase diff --git a/app/db/max_notifications.py b/app/db/max_notifications.py new file mode 100644 index 0000000..c8c4391 --- /dev/null +++ b/app/db/max_notifications.py @@ -0,0 +1,50 @@ +"""MAX identifiers and delivery metadata live only in the identity DB; no medical message bodies.""" + +from datetime import datetime + +from sqlalchemy import BigInteger, Boolean, Integer, String +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.identity import IdentityBase +from app.db.types import TZDateTime + + +class MaxBinding(IdentityBase): + __tablename__ = "max_bindings" + + # No FK: anonymous web profiles intentionally have no patients row in identity. + token: Mapped[str] = mapped_column(String(64), primary_key=True) + user_id: Mapped[int] = mapped_column(BigInteger, unique=True) + enabled: Mapped[bool] = mapped_column(Boolean, default=True) + connected_at: Mapped[datetime] = mapped_column(TZDateTime) + + +class MaxLinkCode(IdentityBase): + __tablename__ = "max_link_codes" + + digest: Mapped[str] = mapped_column(String(64), primary_key=True) + token: Mapped[str] = mapped_column(String(64), unique=True) + expires_at: Mapped[datetime] = mapped_column(TZDateTime, index=True) + used_at: Mapped[datetime | None] = mapped_column(TZDateTime) + + +class MaxReceipt(IdentityBase): + __tablename__ = "max_webhook_receipts" + + digest: Mapped[str] = mapped_column(String(64), primary_key=True) + received_at: Mapped[datetime] = mapped_column(TZDateTime, index=True) + + +class MaxDelivery(IdentityBase): + __tablename__ = "max_deliveries" + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + token: Mapped[str | None] = mapped_column(String(64), index=True) + user_id: Mapped[int] = mapped_column(BigInteger) + kind: Mapped[str] = mapped_column(String(24)) + state: Mapped[str] = mapped_column(String(16), default="queued", index=True) + attempts: Mapped[int] = mapped_column(Integer, default=0) + created_at: Mapped[datetime] = mapped_column(TZDateTime) + next_attempt_at: Mapped[datetime] = mapped_column(TZDateTime, index=True) + lease: Mapped[str | None] = mapped_column(String(32)) + error_code: Mapped[str | None] = mapped_column(String(32)) diff --git a/app/integrations/max_store.py b/app/integrations/max_store.py new file mode 100644 index 0000000..76b8495 --- /dev/null +++ b/app/integrations/max_store.py @@ -0,0 +1,201 @@ +"""Registration boundary and durable MAX queue. Callers own the identity transaction. + +Only this boundary joins a pseudonymous profile to a MAX recipient. Clinical modules never import it. +Queue rows contain a template name, never replies, scores, prescriptions or patient-auth tokens in URLs. +""" + +import hashlib +import secrets +from datetime import datetime, timedelta + +from sqlalchemy import delete, select, update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db import repo +from app.db.clinical import ScaleResult +from app.db.intake import QuestionnaireRecord +from app.db.max_notifications import MaxBinding, MaxDelivery, MaxLinkCode +from app.db.session import get_clinical_session + +MAX_ATTEMPTS = 6 +LEASE_SECONDS = 60 + + +def digest(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +async def active_patient(token: str) -> bool: + """Same consent/adult gate used by the scheduler, checked again at delivery time.""" + async with get_clinical_session() as session: + state = await repo.get_dialog_state(session, token) + return bool( + await repo.get_profile(session, token) + and state + and state.data.get("consent") is True + and state.data.get("adult") is True + ) + + +async def issue_code(session: AsyncSession, token: str, now: datetime, ttl: int) -> tuple[str, datetime]: + raw = secrets.token_urlsafe(32) + expires = now + timedelta(seconds=ttl) + await session.execute(delete(MaxLinkCode).where(MaxLinkCode.token == token)) + session.add(MaxLinkCode(digest=digest(raw), token=token, expires_at=expires)) + await session.flush() + return raw, expires + + +async def code_token(session: AsyncSession, raw: str, now: datetime) -> str | None: + row = await session.get(MaxLinkCode, digest(raw)) + return row.token if row and row.used_at is None and row.expires_at > now else None + + +async def bind(session: AsyncSession, raw: str, user_id: int, now: datetime) -> str | None: + """Atomically consume a code. Existing active identities cannot be silently reassigned.""" + token = await code_token(session, raw, now) + if token is None: + return None + owner = await session.scalar(select(MaxBinding).where(MaxBinding.user_id == user_id)) + binding = await session.get(MaxBinding, token) + if owner and owner.token != token and owner.enabled: + return None + if binding and binding.user_id != user_id and binding.enabled: + return None + used = await session.execute( + update(MaxLinkCode) + .where(MaxLinkCode.digest == digest(raw), MaxLinkCode.used_at.is_(None), MaxLinkCode.expires_at > now) + .values(used_at=now) + .returning(MaxLinkCode.token) + ) + if used.scalar_one_or_none() is None: + return None + if owner and owner.token != token: + await session.delete(owner) + await session.flush() + if binding is None: + binding = MaxBinding(token=token, user_id=user_id, connected_at=now) + session.add(binding) + # Cancel stale deliveries even when the same account reconnects later. + await cancel_pending(session, token) + binding.user_id, binding.enabled, binding.connected_at = user_id, True, now + await session.flush() + return token + + +async def cancel_pending(session: AsyncSession, token: str) -> None: + await session.execute( + update(MaxDelivery) + .where(MaxDelivery.token == token, MaxDelivery.state.in_(("queued", "sending"))) + .values(state="cancelled", lease=None) + ) + + +async def disconnect(session: AsyncSession, token: str) -> None: + binding = await session.get(MaxBinding, token) + if binding: + binding.enabled = False + await session.execute(delete(MaxLinkCode).where(MaxLinkCode.token == token)) + await cancel_pending(session, token) + + +async def enqueue( + session: AsyncSession, *, key: str, user_id: int, kind: str, now: datetime, token: str | None = None +) -> None: + if await session.get(MaxDelivery, key) is not None: + return + try: + async with session.begin_nested(): + session.add( + MaxDelivery(id=key, user_id=user_id, token=token, kind=kind, created_at=now, next_attempt_at=now) + ) + await session.flush() + except IntegrityError: + # Another worker may enqueue the same event between SELECT and INSERT. + if await session.get(MaxDelivery, key) is None: + raise + + +async def collect_completed(session: AsyncSession, now: datetime) -> None: + """Recover completion events from committed clinical records, including after a restart. + + No clinical hooks depend on identity. Read only IDs/dates, then enqueue a generic template. + A binding's activation time bounds history; repeated scans are idempotent. + """ + bindings = (await session.scalars(select(MaxBinding).where(MaxBinding.enabled.is_(True)))).all() + for binding in bindings: + if not await active_patient(binding.token): + continue + async with get_clinical_session() as clinical: + forms = ( + await clinical.scalars( + select(QuestionnaireRecord.id).where( + QuestionnaireRecord.token == binding.token, + QuestionnaireRecord.status == "complete", + QuestionnaireRecord.completed_at >= binding.connected_at, + ) + ) + ).all() + scales = ( + await clinical.scalars( + select(ScaleResult.id).where( + ScaleResult.token == binding.token, + ScaleResult.complete.is_(True), + ScaleResult.ts >= binding.connected_at, + ) + ) + ).all() + for source, ids in (("form", forms), ("scale", scales)): + for record_id in ids: + key = digest(f"completed:{binding.token}:{binding.connected_at.isoformat()}:{source}:{record_id}") + await enqueue(session, key=key, token=binding.token, user_id=binding.user_id, kind="completed", now=now) + + +async def claim(session: AsyncSession, now: datetime) -> MaxDelivery | None: + """One lease per row; recovery after a worker crash is bounded by the lease deadline.""" + ready = (MaxDelivery.state.in_(("queued", "sending"))) & (MaxDelivery.next_attempt_at <= now) + row = await session.scalar( + select(MaxDelivery) + .where(ready) + .order_by(MaxDelivery.created_at, MaxDelivery.id) + .with_for_update(skip_locked=True) + .limit(1) + ) + if row is None: + return None + if row.attempts >= MAX_ATTEMPTS: + row.state, row.error_code, row.lease = "failed", "attempt_limit", None + return None + lease = secrets.token_hex(16) + result = await session.execute( + update(MaxDelivery) + .where(MaxDelivery.id == row.id, ready) + .values( + state="sending", + attempts=MaxDelivery.attempts + 1, + lease=lease, + next_attempt_at=now + timedelta(seconds=LEASE_SECONDS), + ) + .returning(MaxDelivery.id) + ) + if result.scalar_one_or_none() is None: + return None + await session.refresh(row) + return row + + +async def finish( + session: AsyncSession, + row: MaxDelivery, + *, + state: str, + now: datetime, + error: str | None = None, + retry_seconds: float = 0, +) -> None: + await session.execute( + update(MaxDelivery) + .where(MaxDelivery.id == row.id, MaxDelivery.lease == row.lease, MaxDelivery.state == "sending") + .values(state=state, lease=None, error_code=error, next_attempt_at=now + timedelta(seconds=retry_seconds)) + ) diff --git a/tests/test_api_max.py b/tests/test_api_max.py new file mode 100644 index 0000000..549ff33 --- /dev/null +++ b/tests/test_api_max.py @@ -0,0 +1,92 @@ +"""MAX endpoints authenticate before reading bindings or accepting events.""" + +from urllib.parse import parse_qs, urlsplit + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.config import get_settings +from app.core import clock +from app.db import repo, session +from app.db.bootstrap import init_db +from app.integrations import max_store +from app.main import app + +BASE = "/api/v1/me/notifications/max" +HOOK = "/api/v1/webhooks/max" +TOKEN = "synthetic-patient" +AUTH = {"Authorization": f"Bearer {TOKEN}"} +SECRET = "test-secret-" * 4 + + +@pytest.fixture +async def client(monkeypatch): + for key, value in { + "MAX_ENABLED": "true", + "MAX_BOT_TOKEN": "fake-token", + "MAX_BOT_USERNAME": "test_bot", + "MAX_WEBHOOK_SECRET": SECRET, + "MAX_WEBHOOK_URL": "https://api.example.org" + HOOK, + "MAX_SITE_URL": "https://example.org", + }.items(): + monkeypatch.setenv(key, value) + get_settings.cache_clear() + await session.configure(identity_url="sqlite+aiosqlite:///:memory:", clinical_url="sqlite+aiosqlite:///:memory:") + await init_db() + async with session.get_clinical_session() as s: + await repo.upsert_profile(s, token=TOKEN, doctor_id="") + await repo.set_dialog_state( + s, token=TOKEN, state="idle", data={"consent": True, "adult": True}, now=clock.now() + ) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + await session.configure() + get_settings.cache_clear() + + +@pytest.mark.parametrize("auth", [{}, {"Authorization": "Bearer unknown"}, {"Authorization": "Basic test"}]) +async def test_auth_required(client, auth): + assert (await client.get(BASE, headers=auth)).status_code == 401 + assert (await client.post(BASE + "/link", headers=auth, json={"consent": True})).status_code == 401 + assert (await client.delete(BASE, headers=auth)).status_code == 401 + + +async def test_link_requires_consent_and_hides_patient_credential(client): + for body in ({}, {"consent": False}, {"consent": True, "user_id": 100}): + assert (await client.post(BASE + "/link", headers=AUTH, json=body)).status_code == 422 + response = await client.post(BASE + "/link", headers=AUTH, json={"consent": True}) + assert response.status_code == 200 + assert TOKEN not in response.text + url = urlsplit(response.json()["url"]) + assert url.netloc == "max.ru" and url.path == "/test_bot" + code = parse_qs(url.query)["start"][0] + async with session.get_identity_session() as s: + assert await max_store.bind(s, code, 100, clock.now()) == TOKEN + assert (await client.get(BASE, headers=AUTH)).json()["connected"] is True + assert (await client.delete(BASE, headers=AUTH)).json()["connected"] is False + + +async def test_revoke_consent_prevents_linking(client): + async with session.get_clinical_session() as s: + await repo.set_dialog_state( + s, token=TOKEN, state="idle", data={"consent": False, "adult": True}, now=clock.now() + ) + assert (await client.post(BASE + "/link", headers=AUTH, json={"consent": True})).status_code == 403 + + +async def test_disabled_service_still_allows_disconnect(client, monkeypatch): + monkeypatch.setenv("MAX_ENABLED", "false") + get_settings.cache_clear() + assert (await client.get(BASE, headers=AUTH)).json()["available"] is False + assert (await client.post(BASE + "/link", headers=AUTH, json={"consent": True})).status_code == 503 + assert (await client.post(HOOK, json={})).status_code == 503 + assert (await client.delete(BASE, headers=AUTH)).status_code == 200 + + +async def test_webhook_secret_checked_before_body_or_adapter(client): + assert (await client.post(HOOK, content="not-json")).status_code == 403 + assert (await client.post(HOOK, headers={"X-Max-Bot-Api-Secret": "wrong"}, json={})).status_code == 403 + headers = {"X-Max-Bot-Api-Secret": SECRET} + assert (await client.post(HOOK, headers=headers, content="not-json")).status_code == 400 + assert (await client.post(HOOK, headers=headers, json=[])).status_code == 422 + assert (await client.post(HOOK, headers=headers, content=" " * 65537)).status_code == 413 diff --git a/tests/test_db.py b/tests/test_db.py index 15d5516..ae7d01e 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -53,7 +53,15 @@ async def _tables(engine_fn: Callable[[], AsyncEngine]) -> set[str]: async def test_create_all_puts_tables_in_the_right_db(dbs: None) -> None: identity_tables = await _tables(session.identity_engine) clinical_tables = await _tables(session.clinical_engine) - assert identity_tables == {"patients", "doctors", "connect_codes"} + assert identity_tables == { + "patients", + "doctors", + "connect_codes", + "max_bindings", + "max_link_codes", + "max_webhook_receipts", + "max_deliveries", + } assert { "patient_profiles", "item_responses", diff --git a/tests/test_integrations_max.py b/tests/test_integrations_max.py new file mode 100644 index 0000000..e63965b --- /dev/null +++ b/tests/test_integrations_max.py @@ -0,0 +1,123 @@ +"""Persistent, one-time linking and queue recovery; synthetic identities only.""" + +from datetime import timedelta + +import pytest +from sqlalchemy import select + +from app.core import clock +from app.db import repo, session +from app.db.bootstrap import init_db +from app.db.clinical import ScaleResult +from app.db.max_notifications import MaxBinding, MaxDelivery, MaxLinkCode +from app.integrations import max_store as store + + +@pytest.fixture(autouse=True) +async def database(): + await session.configure(identity_url="sqlite+aiosqlite:///:memory:", clinical_url="sqlite+aiosqlite:///:memory:") + await init_db() + yield + await session.configure() + + +async def test_codes_are_hashed_expiring_one_time_and_replaced(): + now = clock.now() + async with session.get_identity_session() as s: + old, _ = await store.issue_code(s, "patient", now, 60) + code, expires = await store.issue_code(s, "patient", now, 60) + assert await store.code_token(s, old, now) is None + row = (await s.scalars(select(MaxLinkCode))).one() + assert row.digest != code and row.digest == store.digest(code) + assert await store.code_token(s, code, expires) is None + assert await store.bind(s, code, 100, now) == "patient" + assert await store.bind(s, code, 100, now) is None + async with session.get_identity_session() as s: + assert (await s.get(MaxBinding, "patient")).user_id == 100 + + +async def test_active_binding_cannot_be_reassigned_in_either_direction(): + now = clock.now() + async with session.get_identity_session() as s: + code, _ = await store.issue_code(s, "one", now, 60) + await store.bind(s, code, 100, now) + code, _ = await store.issue_code(s, "two", now, 60) + assert await store.bind(s, code, 100, now) is None + assert await store.code_token(s, code, now) == "two" + code, _ = await store.issue_code(s, "one", now, 60) + assert await store.bind(s, code, 200, now) is None + await store.disconnect(s, "one") + assert await store.code_token(s, code, now) is None + code, _ = await store.issue_code(s, "two", now, 60) + assert await store.bind(s, code, 100, now) == "two" + + +async def test_disconnect_cancels_pending_delivery_and_prevents_stale_ack(): + now = clock.now() + async with session.get_identity_session() as s: + await store.enqueue(s, key="job", user_id=100, token="one", kind="notification", now=now) + row = await store.claim(s, now) + async with session.get_identity_session() as s: + await store.disconnect(s, "one") + await store.finish(s, row, state="sent", now=now) + async with session.get_identity_session() as s: + assert (await s.get(MaxDelivery, "job")).state == "cancelled" + + +async def test_queue_survives_sessions_deduplicates_and_recovers_expired_lease(): + now = clock.now() + async with session.get_identity_session() as s: + for _ in range(2): + await store.enqueue(s, key="job", user_id=100, kind="help", now=now) + async with session.get_identity_session() as s: + first = await store.claim(s, now) + assert first.attempts == 1 + async with session.get_identity_session() as s: + assert await store.claim(s, now + timedelta(seconds=59)) is None + async with session.get_identity_session() as s: + second = await store.claim(s, now + timedelta(seconds=61)) + assert second.id == first.id and second.lease != first.lease and second.attempts == 2 + async with session.get_identity_session() as s: + await store.finish(s, first, state="sent", now=now) + async with session.get_identity_session() as s: + assert (await s.get(MaxDelivery, "job")).state == "sending" + await store.finish(s, second, state="sent", now=now) + async with session.get_identity_session() as s: + assert await store.claim(s, now + timedelta(days=1)) is None + + +async def test_retry_deadline_and_attempt_limit(): + now = clock.now() + async with session.get_identity_session() as s: + await store.enqueue(s, key="job", user_id=100, kind="help", now=now) + for attempt in range(store.MAX_ATTEMPTS): + when = now + timedelta(seconds=attempt * 120) + async with session.get_identity_session() as s: + row = await store.claim(s, when) + async with session.get_identity_session() as s: + await store.finish(s, row, state="queued", now=when, retry_seconds=120, error="http_429") + async with session.get_identity_session() as s: + assert await store.claim(s, when + timedelta(seconds=119)) is None + async with session.get_identity_session() as s: + assert await store.claim(s, now + timedelta(days=1)) is None + async with session.get_identity_session() as s: + row = await s.get(MaxDelivery, "job") + assert row.state == "failed" and row.attempts == store.MAX_ATTEMPTS + + +async def test_completion_scan_ignores_old_and_incomplete_tests_and_recovers_without_duplicates(): + now = clock.now() + async with session.get_clinical_session() as s: + await repo.upsert_profile(s, token="one", doctor_id="") + await repo.set_dialog_state(s, token="one", state="idle", data={"consent": True, "adult": True}, now=now) + for ts, complete in ((now - timedelta(days=1), True), (now, True), (now, False)): + s.add(ScaleResult(token="one", ts=ts, scale="phq9", total=0, severity="none", complete=complete)) + async with session.get_identity_session() as s: + s.add(MaxBinding(token="one", user_id=100, enabled=True, connected_at=now)) + for _ in range(2): + async with session.get_identity_session() as s: + await store.collect_completed(s, now) + async with session.get_identity_session() as s: + row = (await s.scalars(select(MaxDelivery))).one() + assert row.kind == "completed" and row.token == "one" + assert "phq9" not in row.id From 461c3d02ea9cddd9c158693a1d84da30ffb19e52 Mon Sep 17 00:00:00 2001 From: Pavel Eliseev Date: Sun, 4 Oct 2026 16:03:01 +0300 Subject: [PATCH 2/2] fix(backend): redact retryable MAX storage failures Refs #82 --- app/api/max_notifications.py | 7 ++++++- tests/test_api_max.py | 11 +++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/app/api/max_notifications.py b/app/api/max_notifications.py index c232976..c8e759e 100644 --- a/app/api/max_notifications.py +++ b/app/api/max_notifications.py @@ -10,6 +10,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from pydantic import BaseModel, ConfigDict from sqlalchemy import select +from sqlalchemy.exc import SQLAlchemyError from app.config import get_settings from app.core import clock @@ -106,5 +107,9 @@ async def webhook(request: Request) -> dict[str, bool]: raise HTTPException(422, "event must be an object") from app.channels.max_bot import handle_update - await handle_update(event) + try: + await handle_update(event) + except SQLAlchemyError: + # Let MAX retry; do not log DB exception parameters containing identity data. + raise HTTPException(503, "notification storage unavailable") from None return {"success": True} diff --git a/tests/test_api_max.py b/tests/test_api_max.py index 549ff33..123f13a 100644 --- a/tests/test_api_max.py +++ b/tests/test_api_max.py @@ -1,9 +1,13 @@ """MAX endpoints authenticate before reading bindings or accepting events.""" +import sys +from types import SimpleNamespace +from unittest.mock import AsyncMock from urllib.parse import parse_qs, urlsplit import pytest from httpx import ASGITransport, AsyncClient +from sqlalchemy.exc import OperationalError from app.config import get_settings from app.core import clock @@ -90,3 +94,10 @@ async def test_webhook_secret_checked_before_body_or_adapter(client): assert (await client.post(HOOK, headers=headers, content="not-json")).status_code == 400 assert (await client.post(HOOK, headers=headers, json=[])).status_code == 422 assert (await client.post(HOOK, headers=headers, content=" " * 65537)).status_code == 413 + + +async def test_webhook_storage_failure_is_retryable_without_identity_details(client, monkeypatch): + handler = AsyncMock(side_effect=OperationalError("private SQL", {"token": TOKEN}, Exception("private"))) + monkeypatch.setitem(sys.modules, "app.channels.max_bot", SimpleNamespace(handle_update=handler)) + response = await client.post(HOOK, headers={"X-Max-Bot-Api-Secret": SECRET}, json={}) + assert response.status_code == 503 and TOKEN not in response.text and "private" not in response.text