PTSRaiAgent/app/services/questionnaire.py

"""Transactional questionnaire storage and a durable doctor notification queue.

Notifier acceptance is not delivery or a doctor's acknowledgement. At-least-once
retries can duplicate delivery if a process dies after sending but before commit.
"""

import asyncio
from dataclasses import asdict
from datetime import timedelta
from uuid import uuid4
from zoneinfo import ZoneInfo

from sqlalchemy import select

from app.core import clock
from app.core.intake_questions import content, cssrs_answers, group_answers
from app.core.intake_runtime import load_runtime, month_before
from app.core.questionnaire import evaluate_intake, load_policy, score_alcohol, score_cssrs_recent
from app.core.scales import load_scale, score
from app.core.types import AlertLevel
from app.db import repo
from app.db.clinical import Alert, DialogState
from app.db.intake import QuestionnaireNotice, QuestionnaireRecord, QuestionnaireTask
from app.db.session import get_clinical_session
from app.integrations.base import MemoryNotifier, integrations


async def records(session, token):
    return list(
        (
            await session.scalars(
                select(QuestionnaireRecord)
                .where(QuestionnaireRecord.token == token)
                .order_by(QuestionnaireRecord.started_at, QuestionnaireRecord.id)
            )
        ).all()
    )


async def create_record(ctx, kind, *, task=None):
    profile = await ctx.profile()
    runtime = load_runtime()
    prior = next(
        (r for r in reversed(await records(ctx.session, ctx.token)) if r.kind == "intake" and r.status == "complete"),
        None,
    )
    sex = profile.sex if profile else None
    if sex is None and prior:
        sex = prior.details.get("sex")
    day = ctx.now.astimezone(ZoneInfo(runtime.timezone)).date()
    row = QuestionnaireRecord(
        id=uuid4().hex,
        token=ctx.token,
        kind=kind,
        version=content()["version"],
        policy_version=load_policy().version,
        started_at=ctx.now,
        day=day,
        status="active",
        answers={},
        details={
            "runtime_version": runtime.version,
            "modules": list(profile.modules or []) if profile else [],
            "sex": sex,
            "medication_active": bool(
                (profile and profile.therapy_start and profile.therapy_start <= ctx.now.date())
                or (prior and prior.answers.get("psychotropics") == 1)
            ),
            "suicide_monitoring": bool(profile and profile.enhanced_monitoring),
            "weekly_pcl": kind == "pcl5" and bool((ctx.data.get("crisis") or {}).get("closed_at")),
            "answer_metadata": {},
            "task_id": task.id if task else None,
            # Days since stage 1: rotates the PHQ-9 items of the depression diary module (3 a day).
            "day_index": (day - prior.day).days if prior else 0,
        },
    )
    ctx.session.add(row)
    await ctx.session.flush()
    if task:
        task.record_id = row.id
        task.status = "active"
    return row


def save_answer(row, question, value, now, *, derived=False):
    """Anchor recall windows to administration, never to the day of a resumed answer."""
    day = row.day
    starts = {
        "past_month": month_before(day),
        "past_3_months": month_before(day, 3),
        "past_week": day - timedelta(days=7),
        "past_day": day - timedelta(days=1),
        "последние 2 недели": day - timedelta(days=14),
        "последний месяц": month_before(day),
        "последняя неделя": day - timedelta(days=7),
        "прошлая неделя": day - timedelta(days=7),
        "последние 12 месяцев": month_before(day, 12),
    }
    row.answers = {**row.answers, question.id: value}
    if question.id == "sex" and value in ("m", "f"):
        row.details = {**row.details, "sex": value}
    row.details = {
        **row.details,
        "answer_metadata": {
            **row.details.get("answer_metadata", {}),
            question.id: {
                "answered_at": now.isoformat(),
                "period": question.period,
                "period_start": starts[question.period].isoformat() if question.period in starts else None,
                "period_end": day.isoformat(),
                "source": "derived" if derived else "patient",
                "derived": derived,
            },
        },
    }


def decision(row):
    a = row.answers
    sex = row.details.get("sex") or (a.get("sex") if a.get("sex") in ("m", "f") else None)
    return evaluate_intake(
        phq9=group_answers(a, "phq9"),
        tsq=group_answers(a, "tsq"),
        pcl5=group_answers(a, "pcl5"),
        alcohol=group_answers(a, "rus_audit"),
        cssrs=cssrs_answers(a),
        sex=sex,
        trauma_exposure=bool(a["trauma"]) if "trauma" in a else None,
        traumatic_brain_injury=a.get("tbi") == 1,
        psychotropics=a.get("psychotropics") == 1,
    )


def summary(row):
    a = row.answers
    totals = {}
    for sid in ("phq9", "tsq", "pcl5"):
        result = score(sid, group_answers(a, sid))
        totals[sid] = {
            "complete": result.complete,
            "total": result.total if result.complete else None,
            "answered": result.answered,
        }
    sex = row.details.get("sex") or (a.get("sex") if a.get("sex") in ("m", "f") else None)
    for sid in ("rus_audit_s", "rus_audit"):
        values = {k: a[k] for k in load_scale(sid).item_ids if k in a}
        totals[sid] = asdict(score_alcohol(values, short=sid.endswith("_s"), sex=sex))
    totals["cssrs_recent"] = asdict(score_cssrs_recent(cssrs_answers(a)))
    return totals


async def finish(ctx, row, status, *, extra=None):
    row.status = status
    row.completed_at = ctx.now if status == "complete" else None
    row.details = {**row.details, **(extra or {})}
    if row.details.get("task_id"):
        task = await ctx.session.get(QuestionnaireTask, row.details["task_id"])
        if task:
            task.status = "done"


async def alert(session, token, key, code, now, *, level=AlertLevel.MODERATE, params=None):
    existing = await session.scalar(
        select(QuestionnaireNotice).where(QuestionnaireNotice.token == token, QuestionnaireNotice.event_key == key)
    )
    if existing:
        return existing
    row = await repo.add_alert(session, token=token, level=level, code=code, reason=code, ts=now, params=params)
    notice = QuestionnaireNotice(token=token, event_key=key, alert_id=row.id, created_at=now)
    session.add(notice)
    await session.flush()
    return notice


async def cancel_pending(session, token):
    for row in await records(session, token):
        if row.status in ("active", "paused"):
            row.status = "cancelled"
    tasks = await session.scalars(
        select(QuestionnaireTask).where(
            QuestionnaireTask.token == token, QuestionnaireTask.status.in_(("pending", "active"))
        )
    )
    for task in tasks:
        task.status = "cancelled"
        task.delivery = []
    notices = await session.scalars(
        select(QuestionnaireNotice).where(QuestionnaireNotice.token == token, QuestionnaireNotice.status == "pending")
    )
    for notice in notices:
        notice.status = "cancelled"


async def dispatch_notices(token=None):
    now = clock.now()
    async with get_clinical_session() as session:
        stmt = (
            select(QuestionnaireNotice.id, QuestionnaireNotice.token)
            .where(QuestionnaireNotice.status == "pending")
            .order_by(QuestionnaireNotice.id)
            .limit(100)
        )
        if token:
            stmt = stmt.where(QuestionnaireNotice.token == token)
        candidates = (await session.execute(stmt)).all()
    for notice_id, patient_token in candidates:
        async with get_clinical_session() as session:
            # Same lock order as withdrawal: patient first, then notice.
            await lock_patient(session, patient_token)
            notice = await session.scalar(
                select(QuestionnaireNotice).where(QuestionnaireNotice.id == notice_id).with_for_update()
            )
            if notice.status != "pending":
                continue
            if notice.attempted_at and now < notice.attempted_at + timedelta(
                minutes=min(60, 2 ** min(notice.attempts, 6))
            ):
                continue
            state = await repo.get_dialog_state(session, notice.token)
            if state is None or state.data.get("consent") is not True:
                notice.status = "cancelled"
                continue
            profile = await repo.get_profile(session, notice.token)
            if not profile or not profile.doctor_id:
                continue
            item = await session.get(Alert, notice.alert_id)
            notice.attempted_at, notice.attempts = now, notice.attempts + 1
            try:
                await asyncio.wait_for(
                    integrations.notifier.send(profile.doctor_id, item.level, item.code, token=notice.token), timeout=10
                )
            except Exception as exc:
                notice.error_code = type(exc).__name__[:80]
            else:
                notice.error_code = None
                notice.accepted_at = now
                notice.status = (
                    "test_sink" if isinstance(integrations.notifier, MemoryNotifier) else "accepted_by_adapter"
                )


async def queue_task(session, token, kind, key, now):
    existing = await session.scalar(
        select(QuestionnaireTask).where(QuestionnaireTask.token == token, QuestionnaireTask.task_key == key)
    )
    if existing:
        return existing
    # Only one outstanding administration per form, however many ticks were missed.
    existing = await session.scalar(
        select(QuestionnaireTask).where(
            QuestionnaireTask.token == token,
            QuestionnaireTask.kind == kind,
            QuestionnaireTask.status.in_(("pending", "active")),
        )
    )
    if existing:
        return existing
    row = QuestionnaireTask(token=token, kind=kind, task_key=key, due_at=now)
    session.add(row)
    await session.flush()
    return row


async def outstanding_task(session, token, kind):
    return await session.scalar(
        select(QuestionnaireTask).where(
            QuestionnaireTask.token == token,
            QuestionnaireTask.kind == kind,
            QuestionnaireTask.status.in_(("pending", "active")),
        )
    )


async def lock_patient(session, token):
    # PostgreSQL serializes callbacks from different workers. SQLite tests additionally
    # use the engine's in-process lock. Consent and answers share this transaction.
    await session.scalar(select(DialogState).where(DialogState.token == token).with_for_update())