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