diff --git a/backend/src/cyclone/store/__init__.py b/backend/src/cyclone/store/__init__.py index 141dfdf..32a1a39 100644 --- a/backend/src/cyclone/store/__init__.py +++ b/backend/src/cyclone/store/__init__.py @@ -80,6 +80,14 @@ from .acks import ( list_acks, list_ta1_acks, ) +from .claim_acks import ( + add_claim_ack as _add_claim_ack, + batch_envelope_index as _batch_envelope_index, + find_ack_orphans as _find_ack_orphans, + list_acks_for_claim as _list_acks_for_claim, + list_claims_for_ack as _list_claims_for_ack, + remove_claim_ack as _remove_claim_ack, +) from .backups import add_backup_pending from .exceptions import AlreadyMatchedError, InvalidStateError, NotMatchedError from .inbox import list_unmatched, manual_match, manual_unmatch @@ -102,6 +110,7 @@ from .providers import ( from .records import BatchKind, BatchRecord, BatchRecord837, BatchRecord835 from .ui import ( to_ui_ack, + to_ui_claim_ack, to_ui_claim_detail, to_ui_claim_from_orm, to_ui_provider, @@ -287,6 +296,38 @@ class CycloneStore: def get_277ca_ack(self, ack_id): return get_277ca_ack(ack_id) + # -- SP28: claim↔ack auto-link join table -------------------------- + + def add_claim_ack(self, **kwargs): + """Persist one claim_acks link row. + + Mirrors the publish-from-store contract used by the ACK paths: + when ``event_bus`` is passed, the matching ``claim_ack_written`` + event fires after commit so live-tail subscribers on both the + claim and the ack side see the row immediately. + """ + return _add_claim_ack(**kwargs) + + def list_acks_for_claim(self, claim_id): + """Return every ClaimAck row for one claim (newest first).""" + return _list_acks_for_claim(claim_id) + + def list_claims_for_ack(self, kind, ack_id): + """Return every ClaimAck row for one ack.""" + return _list_claims_for_ack(kind, ack_id) + + def find_ack_orphans(self, kind): + """Return acks with no resolvable link (Inbox ack-orphans lane).""" + return _find_ack_orphans(kind) + + def remove_claim_ack(self, link_id, *, event_bus=None): + """Unlink one row. Publishes ``claim_ack_dropped`` on the bus.""" + return _remove_claim_ack(link_id, event_bus=event_bus) + + def batch_envelope_index(self): + """Return a {envelope.control_number: batch.id} map for D10 Pass 1.""" + return _batch_envelope_index() + # -- SP17: encrypted DB backups ------------------------------------- def add_backup_pending(self, *, filename: str, backup_dir: str): diff --git a/backend/src/cyclone/store/claim_acks.py b/backend/src/cyclone/store/claim_acks.py new file mode 100644 index 0000000..cedd35e --- /dev/null +++ b/backend/src/cyclone/store/claim_acks.py @@ -0,0 +1,360 @@ +"""SP28: claim_acks persistence + batch envelope index (D10). + +Five methods on top of the new ``ClaimAck`` ORM table: + +* ``add_claim_ack`` — insert one link row (manual or auto) and + publish ``claim_ack_written`` on the bus. +* ``list_acks_for_claim`` — every link row for one claim (per-claim + only; TA1 batch-level rows are filtered out by the API layer). +* ``list_claims_for_ack`` — every link row for one ack. +* ``find_ack_orphans`` — acks with no resolvable claim (Inbox + ack-orphans lane). +* ``remove_claim_ack`` — unlink. Publishes ``claim_ack_dropped``. +* ``batch_envelope_index`` — D10 in-memory map of + ``Batch.envelope.control_number → batch.id`` (cheap to rebuild; + re-built once per ingest). + +Each mutating method opens its own ``db.SessionLocal()()`` session +so callers don't have to manage session lifecycles. Publishes are +best-effort (wrapped in :func:`_safe_publish`) so a failing bus +subscriber cannot roll back the persisted row. +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone +from typing import TYPE_CHECKING + +from cyclone import db +from cyclone.db import ( + Ack, + Batch, + Claim, + ClaimAck, + Ta1Ack, + Two77caAck, +) + +from . import utcnow +from .ui import to_ui_claim_ack +from .write import _sync_publish + +if TYPE_CHECKING: + from cyclone.pubsub import EventBus + + +log = logging.getLogger(__name__) + + +def _safe_publish(event_bus: "EventBus | None", kind: str, payload: dict) -> None: + """Best-effort publish wrapped in try/except. + + Mirrors ``cyclone.store.acks._safe_publish`` — never raises, + never rolls back the persisted row. + """ + if event_bus is None: + return + try: + _sync_publish(event_bus, kind, payload) + except Exception: # noqa: BLE001 + log.exception("claim_acks store: %s publish failed", kind) + + +# --------------------------------------------------------------------------- +# D10 batch envelope index +# --------------------------------------------------------------------------- + + +def batch_envelope_index() -> dict[str, str]: + """Build a ``{envelope.control_number: batch.id}`` map. + + D10 primary join key (spec §D10). Built once per ingest from + every Batch row whose ``raw_result_json`` carries an envelope — + cost is O(N batches), currently ~16, so trivial. Kept as a + plain dict so callers can ``index.get(scn)`` to resolve. + """ + out: dict[str, str] = {} + with db.SessionLocal()() as s: + rows = ( + s.query(Batch.id, Batch.raw_result_json) + .filter(Batch.kind == "837") + .all() + ) + for bid, raw in rows: + env = (raw or {}).get("envelope") or {} + ctrl = env.get("control_number") + if isinstance(ctrl, str) and ctrl: + out[ctrl] = bid + return out + + +# --------------------------------------------------------------------------- +# Mutators +# --------------------------------------------------------------------------- + + +def add_claim_ack( + *, + claim_id: str | None, + batch_id: str | None, + ack_id: int, + ack_kind: str, + ak2_index: int | None = None, + set_control_number: str | None = None, + set_accept_reject_code: str | None = None, + linked_by: str = "auto", + event_bus: "EventBus | None" = None, + now: datetime | None = None, +) -> ClaimAck: + """Persist one link row and return it. + + The DB-level unique index + ``ux_claim_acks_dedup(claim_id, ack_kind, ack_id, ak2_index) + WHERE claim_id IS NOT NULL AND ak2_index IS NOT NULL`` enforces + idempotency for the per-AK2 case. The 277CA (ak2_index NULL) + and TA1 (claim_id NULL) paths are deduplicated by application + code — see :func:`cyclone.claim_acks._link_exists`. + + When ``event_bus`` is provided, publishes one ``claim_ack_written`` + event (with the full :func:`cyclone.store.ui.to_ui_claim_ack` + payload) so the live-tail subscribers on both + ``/api/claims/{id}/acks/stream`` and + ``/api/acks/{kind}/{id}/claims/stream`` see new rows the moment + they land. + """ + if ack_kind not in ("999", "277ca", "ta1"): + raise ValueError(f"add_claim_ack: unknown ack_kind={ack_kind!r}") + if linked_by not in ("auto", "manual"): + raise ValueError(f"add_claim_ack: linked_by={linked_by!r}") + if claim_id is None and batch_id is None: + raise ValueError( + "add_claim_ack: at least one of claim_id/batch_id must be set" + ) + ts = now or utcnow() + with db.SessionLocal()() as s: + row = ClaimAck( + claim_id=claim_id, + batch_id=batch_id, + ack_id=ack_id, + ack_kind=ack_kind, + ak2_index=ak2_index, + set_control_number=set_control_number, + set_accept_reject_code=set_accept_reject_code, + linked_at=ts, + linked_by=linked_by, + ) + s.add(row) + s.commit() + s.refresh(row) + row_id = row.id + + if event_bus is not None: + with db.SessionLocal()() as s: + payload = to_ui_claim_ack(s.get(ClaimAck, row_id)) + _safe_publish(event_bus, "claim_ack_written", payload) + + return row + + +def remove_claim_ack( + link_id: int, + *, + event_bus: "EventBus | None" = None, +) -> bool: + """Unlink. Returns ``True`` when a row was deleted. + + Publishes ``claim_ack_dropped`` with ``{"id", "claim_id"}`` so + the live-tail subscribers can remove the link from their local + store. Does NOT touch ``Claim.state`` — the unlink is purely + about the link row, per spec §D6. + """ + with db.SessionLocal()() as s: + row = s.get(ClaimAck, link_id) + if row is None: + return False + payload = { + "id": row.id, + "claim_id": row.claim_id, + "ack_id": row.ack_id, + "ack_kind": row.ack_kind, + } + s.delete(row) + s.commit() + + if event_bus is not None: + _safe_publish(event_bus, "claim_ack_dropped", payload) + return True + + +# --------------------------------------------------------------------------- +# Readers +# --------------------------------------------------------------------------- + + +def list_acks_for_claim(claim_id: str) -> list[ClaimAck]: + """Return every ClaimAck row for one claim (newest first).""" + with db.SessionLocal()() as s: + return ( + s.query(ClaimAck) + .filter(ClaimAck.claim_id == claim_id) + .order_by(ClaimAck.id.desc()) + .all() + ) + + +def list_claims_for_ack(kind: str, ack_id: int) -> list[ClaimAck]: + """Return every ClaimAck row for one ack (any kind). + + For 999 / 277CA: returns 0..N rows (one per AK2 / ClaimStatus). + For TA1: returns 0..1 row (envelope-level, populated batch_id). + """ + if kind not in ("999", "277ca", "ta1"): + raise ValueError(f"list_claims_for_ack: unknown kind={kind!r}") + with db.SessionLocal()() as s: + return ( + s.query(ClaimAck) + .filter( + ClaimAck.ack_kind == kind, + ClaimAck.ack_id == ack_id, + ) + .order_by(ClaimAck.id.asc()) + .all() + ) + + +# --------------------------------------------------------------------------- +# Orphan detection (Inbox "Ack orphans" lane — spec §D7) +# --------------------------------------------------------------------------- + + +def find_ack_orphans(kind: str) -> list[dict]: + """Return acks with no resolvable Claim row of their own kind. + + Used by the Inbox ack-orphans lane for the operator's manual + reconciliation flow. The downstream ``/api/inbox/ack-orphans`` + endpoint calls this and returns the rendered shape. + + "Orphan" means: for 999 / 277CA, ``ack_kind=kind AND ack_id IN + (acks_with_no_claim_acks_link)``. For TA1, "orphan" means the + TA1 row exists but no Batch with matching sender/receiver was + resolved (so no link row was created). + + Output dict shape (one row per orphan ack, rendered by + :func:`to_ui_claim_ack`-style serialization): + + * ``kind`` — "999" / "277ca" / "ta1" + * ``ack_id`` — the ack row's id + * ``control_number`` — for 999/277CA, envelope.control_number; + for TA1, ta1.control_number + * ``set_control_numbers`` — empty list when no claims match + * ``raw_summary`` — flat copy of the ack's UI shape for the + lane-header counts + """ + if kind not in ("999", "277ca", "ta1"): + raise ValueError(f"find_ack_orphans: unknown kind={kind!r}") + + out: list[dict] = [] + if kind == "999": + ack_table = Ack + ctrl_attr = None + elif kind == "277ca": + ack_table = Two77caAck + ctrl_attr = "control_number" + else: + ack_table = Ta1Ack + ctrl_attr = "control_number" + + with db.SessionLocal()() as s: + # Every ack row of the given kind, with a LEFT JOIN against + # any claim_acks link; orphan when NO link was created. + if kind == "999": + # For 999, the "ack has no link" means no ClaimAck row + # was emitted at all (the auto-linker emits one per AK2 + # even when the AK2 is rejected, so 999 with at least + # one AK2 that resolved to a claim is never an orphan). + # We treat a 999 as orphan when it has zero ClaimAck + # rows tied to its id. + all_acks = s.query(Ack).order_by(Ack.id.desc()).all() + for ack_row in all_acks: + count = ( + s.query(ClaimAck) + .filter(ClaimAck.ack_kind == "999", + ClaimAck.ack_id == ack_row.id) + .count() + ) + if count == 0: + out.append({ + "kind": "999", + "ack_id": ack_row.id, + "control_number": _ack_control_number(ack_row, "999"), + "parsed_at": ( + ack_row.parsed_at.isoformat().replace( + "+00:00", "Z" + ) if ack_row.parsed_at else None + ), + }) + elif kind == "277ca": + all_acks = s.query(Two77caAck).order_by(Two77caAck.id.desc()).all() + for ack_row in all_acks: + count = ( + s.query(ClaimAck) + .filter(ClaimAck.ack_kind == "277ca", + ClaimAck.ack_id == ack_row.id) + .count() + ) + if count == 0: + out.append({ + "kind": "277ca", + "ack_id": ack_row.id, + "control_number": ack_row.control_number or "", + "parsed_at": ( + ack_row.parsed_at.isoformat().replace( + "+00:00", "Z" + ) if ack_row.parsed_at else None + ), + }) + else: + all_acks = s.query(Ta1Ack).order_by(Ta1Ack.id.desc()).all() + for ack_row in all_acks: + count = ( + s.query(ClaimAck) + .filter(ClaimAck.ack_kind == "ta1", + ClaimAck.ack_id == ack_row.id) + .count() + ) + if count == 0: + out.append({ + "kind": "ta1", + "ack_id": ack_row.id, + "control_number": ack_row.control_number or "", + "parsed_at": ( + ack_row.parsed_at.isoformat().replace( + "+00:00", "Z" + ) if ack_row.parsed_at else None + ), + }) + return out + + +def _ack_control_number(ack_row: Ack, kind: str) -> str: + """Best-effort control-number lookup for a 999 ack row. + + The 999 ORM row doesn't carry the envelope's control_number in + a dedicated column; we re-derive it from ``raw_json`` (the same + source :func:`cyclone.store.ui.to_ui_ack` uses for the patient + control number). + """ + raw = ack_row.raw_json or {} + env = raw.get("envelope") or {} + return env.get("control_number") or "" + + +__all__ = [ + "add_claim_ack", + "batch_envelope_index", + "find_ack_orphans", + "list_acks_for_claim", + "list_claims_for_ack", + "remove_claim_ack", +] diff --git a/backend/src/cyclone/store/ui.py b/backend/src/cyclone/store/ui.py index 70c85da..32e4f12 100644 --- a/backend/src/cyclone/store/ui.py +++ b/backend/src/cyclone/store/ui.py @@ -8,6 +8,15 @@ Also hosts ``utcnow()`` (the tz-aware UTC now) because it's a small free function that several modules want and there's no better home. SP25: ``to_ui_ack`` / ``to_ui_ta1_ack`` / ``to_ui_two77ca_ack`` were +moved here from ``api_routers/acks.py`` / ``api_routers/ta1_acks.py`` +/ ``api.py`` so the live-tail event payload matches the list endpoint +shape byte-for-byte. The seam between persistence and streaming +depends on the two halves staying in sync. + +SP28 adds ``to_ui_claim_ack`` for the same reason — the +``claim_ack_written`` event payload must match the +``GET /api/acks/{kind}/{id}/claims`` list shape byte-for-byte. + moved here from ``api_routers/acks.py`` / ``api_routers/ta1_acks.py`` / ``api.py`` so the live-tail event payload can match the list endpoint shape byte-for-byte. The seam between persistence and streaming @@ -20,7 +29,7 @@ from datetime import datetime, timezone from decimal import Decimal from cyclone import db -from cyclone.db import Claim, Remittance +from cyclone.db import Claim, ClaimAck, Remittance from cyclone.parsers.models import ClaimOutput from cyclone.parsers.models_835 import ClaimPayment from cyclone.parsers.payer import PayerConfig835 @@ -626,4 +635,49 @@ def _date_in_bounds( return False if date_to is not None and date_part > date_to: return False - return True \ No newline at end of file + return True + + +def to_ui_claim_ack(row: ClaimAck) -> dict: + """SP28: map a ClaimAck ORM row to the API shape. + + Mirrors the rest of the to_ui_* serializers. The wire shape is + identical between the matching list endpoint (``GET /api/acks/{kind}/ + {id}/claims``) and the ``claim_ack_written`` pubsub event so the + live-tail subscribers can rehydrate the snapshot from the bus + without diverging from the API response. + + ``claim_state`` is queried via a session round-trip + (``SELECT state FROM claims WHERE id = :claim_id``) so the drawer + panel can render the colored ClaimStateBadge inline. For TA1 rows + with ``claim_id IS NULL`` (batch-level envelope link), + ``claim_state`` is ``"n/a"``. + """ + claim_state: str = "n/a" + if row.claim_id: + with db.SessionLocal()() as lookup_s: + crow = lookup_s.get(Claim, row.claim_id) + if crow is not None: + claim_state = ( + crow.state.value + if hasattr(crow.state, "value") + else str(crow.state) + ) + linked_iso = ( + row.linked_at.isoformat().replace("+00:00", "Z") + if row.linked_at is not None + else "" + ) + return { + "id": row.id, + "claim_id": row.claim_id, + "batch_id": row.batch_id, + "ack_id": row.ack_id, + "ack_kind": row.ack_kind, + "ak2_index": row.ak2_index, + "set_control_number": row.set_control_number, + "set_accept_reject_code": row.set_accept_reject_code, + "linked_at": linked_iso, + "linked_by": row.linked_by, + "claim_state": claim_state, + } \ No newline at end of file diff --git a/backend/tests/test_acks.py b/backend/tests/test_acks.py index 858cb1c..3d1d560 100644 --- a/backend/tests/test_acks.py +++ b/backend/tests/test_acks.py @@ -51,7 +51,7 @@ def test_migration_0002_creates_acks_table(): def test_migration_latest_idempotent_on_fresh_db(): """Re-running the migration on the same DB must be a no-op (PRAGMA - user_version already at the latest version — currently 16 after + user_version already at the latest version — currently 18 after 0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007 providers/payers/clearhouse, SP10's 0008 payer_rejected, SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged, @@ -59,15 +59,16 @@ def test_migration_latest_idempotent_on_fresh_db(): SP-auth's 0013 users + sessions, SP-audit's 0014 audit_log.user_id, SP22's 0015 drop_claims_unique_constraint, SP27-Task 11's 0016 claims.matched_remittance_id index, SP27-Task 17's 0017 - claim.patient_control_number backfill UPDATE).""" + claim.patient_control_number backfill UPDATE, SP28's 0018 + claim_acks join table).""" with db.engine().begin() as c: v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v1 == 17 + assert v1 == 18 # A second run should not raise and should not bump the version. db_migrate.run(db.engine()) with db.engine().begin() as c: v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v2 == 17 + assert v2 == 18 def test_add_ack_persists_row(): diff --git a/backend/tests/test_db_migrate.py b/backend/tests/test_db_migrate.py index 2ef44ca..995b0be 100644 --- a/backend/tests/test_db_migrate.py +++ b/backend/tests/test_db_migrate.py @@ -123,14 +123,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None: index (drift-check perf). SP27 Task 17 bumped it to 17 with the patient_control_number backfill UPDATE (the migration runner now applies DML too). + SP28 bumped it to 18 with the claim_acks join table. """ engine = _fresh_engine(tmp_path) db_migrate.run(engine) v_after_first = _user_version(engine) - assert v_after_first == 17, f"expected head=17, got {v_after_first}" + assert v_after_first == 18, f"expected head=18, got {v_after_first}" db_migrate.run(engine) - assert _user_version(engine) == 17, "second run should not bump version" + assert _user_version(engine) == 18, "second run should not bump version" def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -156,7 +157,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py engine = _fresh_engine(tmp_path) db_migrate.run(engine) - assert _user_version(engine) == 17, f"expected head=17, got {_user_version(engine)}" + assert _user_version(engine) == 18, f"expected head=18, got {_user_version(engine)}" # Two claims in one batch with the same patient_control_number # must be insertable. If 0015's table recreation re-introduced a