diff --git a/README.md b/README.md index ccdbf04..670bd56 100644 --- a/README.md +++ b/README.md @@ -155,6 +155,7 @@ parses and rejects claims, the inbox reflects the new | POST | `/api/inbox/candidates/{remit_id}/match` | Manual match. `409` if the claim state moved out from under us. | | POST | `/api/inbox/candidates/dismiss` | `{pairs: [{claim_id, remit_id}]}`. Session-scoped. | | POST | `/api/inbox/rejected/resubmit` | `{claim_ids: [...]}`. `200` with `conflicts` for non-rejected. | +| POST | `/api/inbox/payer-rejected/acknowledge` | `{claim_ids: [...], actor: "..."}`. Bulk-acknowledge payer rejections. Idempotent. `200` with `transitioned` / `already_acked` / `not_found` / `not_rejected` counts. Writes an audit event per transition; never overwrites the underlying 277CA rejection. | | GET | `/api/inbox/export.csv?lane=` | Streams CSV of the lane's rows. | ## Outbound 837 Serializer @@ -321,8 +322,9 @@ both be true for the same claim. The `audit_log` table is the canonical record of every state transition the system has ever observed — claim lifecycle, reconciliation -decisions, config reloads, SFTP submissions, 277CA rejects. SP11 made -it tamper-evident: every row carries a SHA-256 hash of +decisions, config reloads, SFTP submissions, 277CA rejects, Payer-Rejected +acknowledgements, SQLCipher key rotations. SP11 made it tamper-evident: +every row carries a SHA-256 hash of `(prev_hash || row_payload)`, forming a chain back to a genesis row. Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable in a single walk. @@ -357,6 +359,40 @@ operator has created the Keychain entry on first run. See for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv) mapping. +### Key rotation + +The DB encryption key is rotated in place via `POST /api/admin/db/rotate-key`. +The handler: + +1. Generates a fresh 256-bit CSPRNG key (`db_crypto.generate_db_key()`). +2. Persists the new key to the Keychain under the same + `cyclone.db.key` account (overwriting the old one). +3. Disposes the SQLAlchemy engine so pooled connections release the + file (SQLCipher refuses to `PRAGMA rekey` while another connection + holds the DB). +4. Opens with the old key, issues `PRAGMA rekey`, and verifies the + schema survived (table-count sanity check). +5. Rebuilds the engine with the new key. +6. Writes a `db.key_rotated` audit event carrying the SHA-256 + fingerprints of the old and new keys (first 8 hex chars) plus the + post-rotation `table_count`. + +When SQLCipher is enabled the engine uses SQLAlchemy's `NullPool` +instead of the default `QueuePool`. `QueuePool` returns connections to +a shared queue that any thread can pull from, which breaks SQLCipher's +thread affinity. `NullPool` trades connection reuse for thread safety +— the only correct behavior under FastAPI's per-request threadpool. + +The rotation endpoint is serialized with a module-level +`threading.Lock` (one rotation in flight at a time), returns `409` if +a rotation is already running, `400` if encryption is not enabled, +and `503` with a `reason` on `PRAGMA rekey` or Keychain failure so the +operator can take the next step without parsing the traceback. + +| Method | Path | Notes | +| ------ | --------------------------------- | -------------------------------------------------------------- | +| POST | `/api/admin/db/rotate-key` | Rotate the SQLCipher key in place. Audit-logged. | + ## SFTP Wire-Up (paramiko) The `clearhouse.submit` endpoint uses `paramiko` to push a batch of @@ -382,6 +418,58 @@ was a one-file change (`sftp_paramiko.py` replacing `sftp_stub.py`). the `paramiko` extras aren't installed so the test suite stays green on Linux dev boxes. +## Inbound MFT Scheduler (SP16) + +The dzinesco SFTP server accepts inbound files (835 remits, 999/TA1 +acks, 271 eligibility responses, 277/277CA claim acknowledgments) that +need to be pulled on a cadence, parsed, and persisted. SP16 ships the +**operator-facing control surface** for the polling loop, even though +the actual background scheduler is still in flight: + +| Method | Path | Notes | +| ------ | --------------------------------------------- | ---------------------------------------------------------------- | +| POST | `/api/admin/scheduler/start` | Start the background MFT polling loop (idempotent). | +| POST | `/api/admin/scheduler/stop` | Stop the background loop. Waits for the current tick to finish. | +| GET | `/api/admin/scheduler/status` | Running state + last tick summary. | +| POST | `/api/admin/scheduler/tick` | Force a single poll cycle. Coalesces with an in-flight tick. | +| GET | `/api/admin/scheduler/processed-files` | Recently processed inbound files. `?status=ok|error|skipped|pending`, `?limit=` (1–500, default 50). | + +The endpoints read from a `cyclone.scheduler` module and a +`cyclone.db.ProcessedInboundFile` ORM model that are being landed +alongside SP16 — the admin router's import surface is in place, the +implementation is in progress. The control surface is forward-compatible +with the eventual scheduler: starting / stopping / ticking / listing +processed files will all work as soon as the loop lands. + +The `paramiko` SFTP client from [SP13](#sftp-wire-up-paramiko) is the +underlying transport; the scheduler reuses it to list + download files +under the inbound SFTP path, then dispatches each downloaded file to +the matching parser based on the `` token in the filename +(see [SP13 file-naming template](#sftp-wire-up-paramiko)). + +## Batches, Reconciliation, and Activity + +These read/write endpoints are the core operator surface for browsing +the parsed record and acting on matches. They predate the per-SP +endpoint reference sections in the **Roadmap** and are listed here in +one place so the route inventory stays discoverable. + +| Method | Path | Notes | +| ------ | --------------------------------------------- | ---------------------------------------------------------------- | +| GET | `/api/batches` | All parsed batches, newest first. `?limit=` (1–1000, default 100). | +| GET | `/api/batches/{batch_id}` | One batch detail (envelope, claims / remits, validation summary). | +| GET | `/api/batch-diff?a=&b=` | Side-by-side diff of two batches: `added` / `removed` / `changed` claims + envelope metadata for each. Both query params required. | +| GET | `/api/reconciliation/unmatched` | `{"claims": [...], "remittances": [...]}` — every Claim with no paired Remittance and vice versa. The two lists are always present (empty list, never absent) so the UI can index unconditionally. | +| POST | `/api/reconciliation/match` | Body `{claim_id, remit_id}`. Manually pair a claim with a remit. `400` on missing ids, `404` on unknown id, `409` on already-matched or terminal-state claim. | +| POST | `/api/reconciliation/unmatch` | Body `{claim_id}`. Remove the current match and reset the claim to `submitted`. `404` on unknown id, `409` on no current match. | +| GET | `/api/providers` | Distinct providers across parsed claims. `?npi=`, `?state=`, `?limit=`, `?offset=`. Distinct from `/api/config/providers/{npi}` (SP9 config table) — this endpoint surfaces providers derived from the parsed claim stream. | +| GET | `/api/activity` | Recent activity events. `?kind=`, `?since=`, `?limit=` (1–500, default 200). Powers the Activity page; the streaming counterpart `/api/activity/stream` is documented under **Live updates**. | + +The per-SP endpoint blocks at the bottom of the **Roadmap** cover the +SP-specific routes (parse, 999/TA1/277CA, Inbox, claim drawer, line +reconciliation, outbound 837, multi-payer config, audit log, 277CA, +key rotation, payer-rejected acknowledge). + ## Persistence Parsed batches, claims, remittances, matches, 277CA rejections, @@ -415,8 +503,31 @@ backup API). . ├── backend/ │ ├── src/cyclone/ -│ │ ├── api.py # FastAPI app, GET + parse routes, /api/{resource}/stream +│ │ ├── api.py # FastAPI app + middleware; mounts api_routers/* sub-apps │ │ ├── api_helpers.py # NDJSON / content-negotiation / live-tail helpers +│ │ ├── api_routers/ # FastAPI APIRouters extracted from api.py +│ │ │ ├── health.py # GET /api/health +│ │ │ ├── acks.py # GET /api/acks, /api/acks/{id} +│ │ │ ├── ta1_acks.py # GET /api/ta1-acks, /api/ta1-acks/{id} +│ │ │ ├── activity.py # GET /api/activity, /api/activity/stream +│ │ │ ├── batches.py # GET /api/batches, /api/batches/{id} +│ │ │ ├── claims.py # GET /api/claims, /api/claims/stream, /api/claims/{id}, +│ │ │ │ # /api/claims/{id}/serialize-837, /api/claims/{id}/line-reconciliation +│ │ │ ├── remittances.py # GET /api/remittances, /api/remittances/stream, /api/remittances/{id} +│ │ │ ├── providers.py # GET /api/providers +│ │ │ ├── inbox.py # GET /api/inbox/lanes, POST /api/inbox/* (match/dismiss/acknowledge/resubmit), +│ │ │ │ # GET /api/inbox/export.csv +│ │ │ ├── reconciliation.py # GET /api/reconciliation/unmatched, /api/batch-diff, +│ │ │ │ # POST /api/reconciliation/match, /api/reconciliation/unmatch +│ │ │ ├── eligibility.py # POST /api/eligibility/request, /api/eligibility/parse-271 +│ │ │ ├── clearhouse.py # GET /api/clearhouse, POST /api/clearhouse/submit +│ │ │ ├── config.py # /api/config/providers[/...], /api/config/payers[/...], /api/admin/reload-config +│ │ │ ├── parse_835.py # POST /api/parse-835 +│ │ │ ├── parse_837.py # POST /api/parse-837 +│ │ │ ├── parse_999.py # POST /api/parse-999 +│ │ │ ├── parse_ta1.py # POST /api/parse-ta1 +│ │ │ └── admin.py # /api/admin/audit-log[/verify], /api/admin/db/rotate-key, +│ │ │ # /api/admin/scheduler/* (SP16) │ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out) │ │ ├── store.py # CycloneStore, mappers, publish-on-write │ │ ├── db.py # SQLAlchemy engine, session factory, ORM models @@ -473,7 +584,7 @@ backup API). ## Roadmap -Sub-projects 2 through 13 are **shipped**. See the [completeness +Sub-projects 2 through 15 are **shipped**. See the [completeness review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for the honest gap analysis against the industry definition of a HIPAA clearinghouse — the short version is that the local-only, @@ -484,6 +595,19 @@ scope. Shipped sub-projects (most recent first): +- **Sub-project 15 (shipped) — SQLCipher key rotation.** In-place + rotation via `PRAGMA rekey`, serialized through a module-level + `threading.Lock` and a SQLAlchemy `NullPool` to keep SQLCipher + thread-affine under FastAPI's per-request threadpool. Writes a + `db.key_rotated` audit event with old + new key fingerprints and + post-rotation `table_count`. See + [Encryption at Rest — Key rotation](#key-rotation). +- **Sub-project 14 (shipped) — 5-lane Inbox UI.** The Payer-Rejected + lane is now rendered in the Inbox alongside Rejected / Candidates / + Unmatched / Done today. New bulk action + `POST /api/inbox/payer-rejected/acknowledge` drops claims from the + working surface without erasing the original 277CA rejection event + (audit log stays intact, SP11). - **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed `SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint actually pushes to @@ -752,6 +876,34 @@ the one-time setup recipe. `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`. SFTP credentials are fetched from the macOS Keychain at call time. +### SP14 endpoints (5-lane Inbox UI + acknowledge) + +- `GET /api/inbox/lanes` — same shape as SP6; the `payer_rejected` + lane payload now also includes + `payer_rejected_acknowledged_at` + `payer_rejected_acknowledged_actor` + per row so the UI can badge acknowledged claims (forward-compat for + a future "Recently acknowledged" view). +- `POST /api/inbox/payer-rejected/acknowledge` — bulk-acknowledge + Payer-Rejected claims. Body: `{claim_ids: [...], actor: "..."}`. + Idempotent. Response: `200` with + `{transitioned, already_acked, not_found, not_rejected}`. Writes an + `inbox.payer_rejected_acknowledged` audit event per transition. + Acknowledgement hides a claim from the lane but never overwrites + `payer_rejected_at` / `payer_rejected_reason` / + `payer_rejected_by_277ca_id` — the original 277CA evidence stays + intact in the audit log. + +### SP15 endpoints (SQLCipher key rotation) + +- `POST /api/admin/db/rotate-key` — rotate the SQLCipher DB key in + place. Generates a fresh 256-bit key, writes it to the Keychain + (overwriting `cyclone.db.key`), disposes the engine, issues + `PRAGMA rekey`, verifies the schema, rebuilds the engine. Writes + a `db.key_rotated` audit event with old + new key fingerprints + and `table_count`. Returns `409` when a rotation is already in + flight, `400` when encryption is not enabled, `503` with a + `reason` on `PRAGMA rekey` or Keychain failure. + ## License No license file yet; this is internal-use software. Add a `LICENSE` file diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index a602897..dd6253c 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -15,8 +15,6 @@ plus GET/POST with any header. from __future__ import annotations -import csv -import io import json import logging import uuid @@ -26,23 +24,16 @@ from typing import Any, AsyncIterator from fastapi import FastAPI, File, HTTPException, Query, Request, UploadFile from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import JSONResponse, Response, StreamingResponse +from fastapi.responses import JSONResponse, StreamingResponse from pydantic import ValidationError from cyclone import __version__, db from cyclone.db import Claim, ClaimState, Remittance from cyclone.inbox_state import apply_999_rejections from cyclone.inbox_state_277ca import apply_277ca_rejections -from cyclone.audit_log import AuditEvent, append_event, verify_chain +from cyclone.audit_log import AuditEvent, append_event from cyclone.parsers.exceptions import CycloneParseError -from cyclone.parsers.models import BatchSummary, ClaimOutput, Envelope, ParseResult -from cyclone.parsers.models_270 import ( - EligibilityBenefitInquiry, - InformationReceiver270, - InformationSource270, - ParseResult270, - Subscriber270, -) +from cyclone.parsers.models import BatchSummary, Envelope from cyclone.parsers.models_271 import ParseResult271 from cyclone.parsers.models_835 import ParseResult835 from cyclone.parsers.payer import PayerConfig, PayerConfig835 @@ -53,9 +44,8 @@ from cyclone.parsers.parse_837 import parse from cyclone.parsers.parse_835 import parse as parse_835 from cyclone.parsers.parse_999 import parse_999_text from cyclone.parsers.parse_ta1 import parse_ta1_text -from cyclone.parsers.serialize_270 import serialize_270 from cyclone.parsers.serialize_999 import serialize_999 -from cyclone.parsers.serialize_837 import SerializeError as SerializeError837, serialize_837, serialize_837_for_resubmit + from cyclone.parsers.batch_ack_builder import build_ack_for_batch from cyclone.parsers.validator_835 import validate as validate_835 from cyclone.pubsub import EventBus @@ -141,11 +131,63 @@ app.add_middleware( # and are wired in here. (Kept as a top-level package rather than nested # under `cyclone.api` so the existing ``cyclone.api`` module path keeps # working — Python prefers packages over same-named modules.) -from cyclone.api_routers import acks, health, ta1_acks # noqa: E402 +from cyclone.api_routers import ( # noqa: E402 + acks, + activity, + admin, + batches, + claims, + clearhouse, + config, + eligibility, + health, + inbox, + providers, + reconciliation, + remittances, + ta1_acks, +) app.include_router(health.router) app.include_router(acks.router) app.include_router(ta1_acks.router) +app.include_router(providers.router) +app.include_router(clearhouse.router) +app.include_router(config.router) +app.include_router(activity.router) +app.include_router(batches.router) +app.include_router(claims.router) +app.include_router(remittances.router) +app.include_router(inbox.router) +app.include_router(reconciliation.router) +app.include_router(eligibility.router) +app.include_router(admin.router) + +# Backwards-compat re-exports: test_api_stream_live.py imports these +# names directly from ``cyclone.api`` to invoke the endpoint coroutines +# in-process. Re-pointing at the router keeps the public surface stable +# while the bodies live in the routers package. +activity_stream = activity.activity_stream_endpoint +claims_stream = claims.claims_stream_endpoint +remittances_stream = remittances.remittances_stream_endpoint + +# Backwards-compat lock re-export: test_api_rotate_key.py does +# ``api_mod._db_rotate_lock.acquire()`` to simulate a concurrent +# rotation request. The lock itself lives in the admin router (it +# guards that router's rotate-key endpoint) but we expose the same +# object here so the test holds the same lock the endpoint does. +_db_rotate_lock = admin._db_rotate_lock + +# Backwards-compat module re-exports: test_api_rotate_key.py does +# ``monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", ...)`` +# and ``monkeypatch.setattr("cyclone.api._secrets.set_secret", ...)`` +# to stub crypto/keychain calls. Re-exporting here keeps the +# ``cyclone.api._db_crypto`` / ``cyclone.api._secrets`` paths alive +# so the monkeypatch targets the same module objects the admin +# endpoint actually uses (both are the same ``cyclone.db_crypto`` / +# ``cyclone.secrets`` module instances). +_db_crypto = admin.db_crypto +_secrets = admin.secrets def _resolve_payer(name: str) -> PayerConfig: @@ -875,1026 +917,17 @@ def _serialize_ta1_from_row(row: db.Ta1Ack) -> str: # --------------------------------------------------------------------------- # -@app.get("/api/inbox/lanes") -def inbox_lanes(): - """Return all Inbox lanes in one call.""" - dismissed_pairs = getattr(app.state, "dismissed_pairs", set()) - with db.SessionLocal()() as session: - from cyclone.inbox_lanes import compute_lanes - lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs) - return { - "rejected": lanes.rejected, - # SP10: payer-rejected lane (277CA STC A4/A6/A7). Distinct from - # the 999 envelope rejection in ``rejected`` above. - "payer_rejected": lanes.payer_rejected, - "candidates": lanes.candidates, - "unmatched": lanes.unmatched, - "done_today": lanes.done_today, - } +# --- /api/inbox/* lives in cyclone.api_routers.inbox --- +# --- /api/reconciliation/* + /api/batch-diff live in cyclone.api_routers.reconciliation --- -@app.post("/api/inbox/candidates/{remit_id}/match") -def inbox_match_candidate(remit_id: str, body: dict): - """Manually link a remit to a claim.""" - claim_id = body.get("claim_id") - if not claim_id: - raise HTTPException(400, "claim_id required") - with db.SessionLocal()() as s: - claim = s.get(Claim, claim_id) - remit = s.get(Remittance, remit_id) - if claim is None or remit is None: - raise HTTPException(404, "claim or remit not found") - if claim.matched_remittance_id and claim.matched_remittance_id != remit_id: - raise HTTPException( - 409, - detail={ - "error": "claim_already_matched", - "current_state": ( - claim.state.value if hasattr(claim.state, "value") - else str(claim.state) - ), - "matched_remittance_id": claim.matched_remittance_id, - }, - ) - claim.matched_remittance_id = remit_id - remit.claim_id = claim_id - s.commit() - return {"ok": True, "claim_id": claim_id, "remit_id": remit_id} -@app.post("/api/inbox/candidates/dismiss") -def inbox_dismiss_candidates(body: dict): - """Add candidate pairs to the session-scoped dismissed set.""" - pairs = body.get("pairs") or [] - if not hasattr(app.state, "dismissed_pairs"): - app.state.dismissed_pairs = set() - for p in pairs: - cid = p.get("claim_id") - rid = p.get("remit_id") - if cid and rid: - app.state.dismissed_pairs.add(frozenset({cid, rid})) - return {"ok": True, "dismissed_count": len(pairs)} +# --- /api/remittances lives in cyclone.api_routers.remittances --- -# --------------------------------------------------------------------------- # -# SP14: Payer-Rejected acknowledge -# -# Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear -# the claim from the working surface. We don't delete the rejection -# (the original payer_rejected_* fields stay for SP11 audit), we just -# set payer_rejected_acknowledged_at so the lane query filters it out. -# -# Idempotent: re-acknowledging an already-acknowledged claim is a noop -# (the timestamp is not bumped). Returns the count actually transitioned -# so the UI can show "3 of 5 were already acknowledged". -# --------------------------------------------------------------------------- # -@app.post("/api/inbox/payer-rejected/acknowledge") -def inbox_acknowledge_payer_rejected(body: dict): - """Mark Payer-Rejected claims as acknowledged by the operator.""" - claim_ids = body.get("claim_ids") or [] - actor = body.get("actor") or "operator" - if not isinstance(claim_ids, list) or not claim_ids: - raise HTTPException(400, "claim_ids must be a non-empty list") - if not all(isinstance(c, str) for c in claim_ids): - raise HTTPException(400, "claim_ids must be a list of strings") - with db.SessionLocal()() as session: - from cyclone.db import Claim - now = datetime.now(timezone.utc) - transitioned = 0 - already_acked = 0 - not_found = 0 - not_rejected = 0 - for cid in claim_ids: - claim = session.get(Claim, cid) - if claim is None: - not_found += 1 - continue - if claim.payer_rejected_at is None: - not_rejected += 1 - continue - if claim.payer_rejected_acknowledged_at is not None: - already_acked += 1 - continue - claim.payer_rejected_acknowledged_at = now - claim.payer_rejected_acknowledged_actor = actor - transitioned += 1 - # SP11: audit event for the acknowledge action. - try: - from cyclone.audit_log import append_event, AuditEvent - append_event(session, AuditEvent( - event_type="claim.payer_rejected_acknowledged", - entity_type="claim", - entity_id=claim.id, - actor=actor, - payload={ - "payer_rejected_status_code": claim.payer_rejected_status_code, - "payer_rejected_by_277ca_id": claim.payer_rejected_by_277ca_id, - }, - )) - except Exception: # noqa: BLE001 - # Audit append is best-effort; don't block the operator's - # acknowledge action on an audit-log failure. - pass - if transitioned: - session.commit() - return { - "ok": True, - "transitioned": transitioned, - "already_acked": already_acked, - "not_found": not_found, - "not_rejected": not_rejected, - } - - -@app.post("/api/inbox/rejected/resubmit") -def inbox_resubmit_rejected( - request: Request, - body: dict, - download: bool = Query(False, description="When true, return a ZIP of regenerated 837 files for the resubmitted claims (instead of JSON)."), -): - """Bulk move REJECTED claims back to SUBMITTED. - - With ``?download=true``, the response is a ``application/zip`` archive - containing one ``claim-{id}.x12`` per successfully resubmitted claim - (regenerated via ``serialize_837_for_resubmit`` so each file gets a - unique interchange/group control number). Conflicts are omitted from - the ZIP — they remain visible to the caller via the JSON shape of the - non-download path. Empty resubmit + download → 200 with an empty zip - so the UI can still hand the user a downloadable artifact. - """ - ids = body.get("claim_ids") or [] - if not ids: - raise HTTPException(400, "claim_ids required") - accepted: list[str] = [] - conflicts: list[dict] = [] - # Track which claims are about to be resubmitted (and their index in - # the bundle) so the download path can serialize them with unique - # control numbers — back-to-back resubmits in the same file would - # otherwise all share ISA13/GS06 = "000000001". - accepted_with_rows: list[tuple[str, "Claim"]] = [] - with db.SessionLocal()() as s: - for cid in ids: - c = s.get(Claim, cid) - if c is None: - continue - if c.state != ClaimState.REJECTED: - conflicts.append({ - "claim_id": cid, - "current_state": ( - c.state.value if hasattr(c.state, "value") - else str(c.state) - ), - }) - continue - c.state = ClaimState.SUBMITTED - c.state_changed_at = datetime.now(timezone.utc) - c.rejection_reason = None - c.rejected_at = None - c.resubmit_count = (c.resubmit_count or 0) + 1 - accepted.append(cid) - accepted_with_rows.append((cid, c)) - s.commit() - - if not download: - return {"ok": True, "resubmitted": accepted, "conflicts": conflicts} - - # Build a ZIP of regenerated 837s for the accepted claims. Conflicts - # and missing ids are deliberately excluded — the user already saw - # them in the JSON response on prior actions; the download is the - # "give me the files I asked for" payload. - import zipfile - buf = io.BytesIO() - serialize_errors: list[dict] = [] - with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf: - for idx, (cid, c) in enumerate(accepted_with_rows, start=1): - if not c.raw_json: - serialize_errors.append({"claim_id": cid, "reason": "no raw_json"}) - continue - try: - claim_obj = ClaimOutput.model_validate(c.raw_json) - except Exception as exc: - serialize_errors.append({"claim_id": cid, "reason": f"raw_json invalid: {exc}"}) - continue - try: - text = serialize_837_for_resubmit(claim_obj, interchange_index=idx) - except SerializeError837 as exc: - serialize_errors.append({"claim_id": cid, "reason": str(exc)}) - continue - zf.writestr(f"claim-{cid}.x12", text) - buf.seek(0) - headers = { - "Content-Disposition": ( - f'attachment; filename="resubmit-{len(accepted)}-claims.zip"' - ), - } - # Surface per-claim serialization failures as a custom response header - # so the UI can show "10 resubmitted, 2 couldn't be regenerated" without - # parsing the binary. The header value is JSON-encoded; the UI is - # expected to JSON.parse it after a fetch with response.ok. - if serialize_errors: - headers["X-Cyclone-Serialize-Errors"] = json.dumps(serialize_errors) - return Response( - content=buf.getvalue(), - media_type="application/zip", - headers=headers, - ) - - -@app.get("/api/inbox/export.csv") -def inbox_export_csv(lane: str): - """Stream a CSV for a single lane.""" - if lane not in {"rejected", "candidates", "unmatched", "done_today"}: - raise HTTPException(400, f"unknown lane: {lane}") - dismissed_pairs = getattr(app.state, "dismissed_pairs", set()) - with db.SessionLocal()() as session: - from cyclone.inbox_lanes import compute_lanes - lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs) - rows = getattr(lanes, lane) - - buf = io.StringIO() - writer = csv.writer(buf) - writer.writerow([ - "id", "kind", "patient_control_number", "charge_amount", - "payer_id", "provider_npi", "state", "rejection_reason", - "service_date", "score", - ]) - for r in rows: - writer.writerow([ - r.get("id") or r.get("payer_claim_control_number"), - r.get("kind"), - r.get("patient_control_number"), - r.get("charge_amount"), - r.get("payer_id"), - r.get("provider_npi") or r.get("rendering_provider_npi"), - r.get("state"), - r.get("rejection_reason"), - r.get("service_date_from") or r.get("service_date"), - r.get("score"), - ]) - buf.seek(0) - return StreamingResponse( - iter([buf.getvalue()]), - media_type="text/csv", - headers={"Content-Disposition": f'attachment; filename="inbox-{lane}.csv"'}, - ) - - -# --------------------------------------------------------------------------- # -# GET endpoints (read views over the in-memory store) -# --------------------------------------------------------------------------- # - - -def _batch_summary_claim_count(rec: BatchRecord) -> int: - """Return the number of claims on a batch, handling both 837P and 835.""" - if rec.kind == "837p": - return len(rec.result.claims) # type: ignore[attr-defined] - if rec.kind == "835": - return len(rec.result.claims) # type: ignore[attr-defined] - return 0 - - -@app.get("/api/batches") -def list_batches( - request: Request, - limit: int = Query(100, ge=1, le=1000), -) -> Any: - """Summary of all parsed batches, newest first.""" - records = store.list(limit=limit) - items = [ - { - "id": r.id, - "kind": r.kind, - "inputFilename": r.input_filename, - "parsedAt": r.parsed_at.isoformat().replace("+00:00", "Z"), - "claimCount": _batch_summary_claim_count(r), - } - for r in records - ] - all_records = store.all() - total = len(all_records) - returned = len(items) - has_more = total > returned - if _wants_ndjson(request): - return StreamingResponse( - _ndjson_stream_list(items, total, returned, has_more), - media_type="application/x-ndjson", - ) - return { - "items": items, - "total": total, - "returned": returned, - "has_more": has_more, - } - - -@app.get("/api/batches/{batch_id}") -def get_batch(batch_id: str) -> Any: - rec = store.get(batch_id) - if rec is None: - raise HTTPException( - status_code=404, - detail={"error": "Not found", "detail": f"Batch {batch_id} not found"}, - ) - return json.loads(rec.result.model_dump_json()) - - -@app.get("/api/claims") -def list_claims( - request: Request, - batch_id: str | None = Query(None), - status: str | None = Query(None), - provider_npi: str | None = Query(None), - payer: str | None = Query(None), - date_from: str | None = Query(None), - date_to: str | None = Query(None), - sort: str | None = Query(None), - order: str = Query("desc"), - limit: int = Query(100, ge=1, le=1000), - offset: int = Query(0, ge=0), -) -> Any: - common = dict( - batch_id=batch_id, - status=status, - provider_npi=provider_npi, - payer=payer, - date_from=date_from, - date_to=date_to, - ) - items = list(store.iter_claims( - sort=sort, order=order, limit=limit, offset=offset, **common, - )) - total = len(list(store.iter_claims(**common))) - returned = len(items) - has_more = total > offset + returned - if _wants_ndjson(request): - return StreamingResponse( - _ndjson_stream_list(items, total, returned, has_more), - media_type="application/x-ndjson", - ) - return { - "items": items, - "total": total, - "returned": returned, - "has_more": has_more, - } - - -# --------------------------------------------------------------------------- # -# Live-tail NDJSON streaming endpoints (Phase 3 — SP5) -# --------------------------------------------------------------------------- # - - -@app.get("/api/claims/stream") -async def claims_stream( - request: Request, - status: str | None = Query(None), - provider_npi: str | None = Query(None), - payer: str | None = Query(None), - date_from: str | None = Query(None), - date_to: str | None = Query(None), - sort: str | None = Query(None), - order: str = Query("desc"), - limit: int = Query(100, ge=1, le=1000), -) -> StreamingResponse: - """Stream Claims as NDJSON: snapshot first, then live events. - - Wire format: - * ``{"type":"item","data":}`` per snapshot row, then per - new ``claim_written`` event - * ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot - * ``{"type":"heartbeat","data":{"ts":}}`` every - ``CYCLONE_TAIL_HEARTBEAT_S`` seconds when idle - - Query params mirror :func:`list_claims` so a frontend can swap a - one-shot fetch for a tail with no URL surgery. - - NOTE: registered before ``/api/claims/{claim_id}`` so the literal - ``stream`` path segment doesn't get matched as a claim id. - """ - bus: EventBus = request.app.state.event_bus - - async def gen() -> AsyncIterator[bytes]: - # 1. Snapshot (eager — iter_claims returns a list already). - rows = store.iter_claims( - status=status, provider_npi=provider_npi, payer=payer, - date_from=date_from, date_to=date_to, - sort=sort or "-submission_date", order=order, limit=limit, - ) - for row in rows: - yield _ndjson_line({"type": "item", "data": row}) - yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) - - # 2. Subscribe + heartbeats. - async for chunk in _tail_events(request, bus, ["claim_written"]): - yield chunk - - return StreamingResponse(gen(), media_type="application/x-ndjson") - - -@app.get("/api/claims/{claim_id}") -def get_claim_detail_endpoint(claim_id: str) -> dict: - """Return one claim with full drawer context (SP4). - - Body shape is produced by :meth:`CycloneStore.get_claim_detail`: - header, state, service lines, diagnoses, parties, validation, - raw segments, ``stateHistory`` (most-recent-first, capped at 50), - and a populated ``matchedRemittance`` block when paired. - - Path param is ``claim_id`` (matches the SP3 ``/api/acks/{ack_id}`` - convention). Returns 404 — never 500 — on a missing claim so the - UI can distinguish "doesn't exist" from a transient fetch error. - """ - body = store.get_claim_detail(claim_id) - if body is None: - raise HTTPException( - status_code=404, - detail={ - "error": "Not found", - "detail": f"Claim {claim_id} not found", - }, - ) - return body - - -@app.get("/api/claims/{claim_id}/serialize-837") -def serialize_claim_as_837(claim_id: str): - """Return the claim as a regenerated X12 837P file (SP8). - - Loads the ClaimOutput from the persisted ``raw_json`` and runs the - outbound serializer. Returns 404 if the claim doesn't exist, 422 if - the stored payload has no parseable ClaimOutput (data integrity - issue, not a transient failure). - """ - with db.SessionLocal()() as s: - row = s.get(Claim, claim_id) - if row is None: - return JSONResponse( - {"error": "Not found", "detail": f"Claim {claim_id} not found"}, - status_code=404, - ) - if not row.raw_json: - return JSONResponse( - { - "error": "Unprocessable", - "detail": f"Claim {claim_id} has no raw_json; cannot serialize", - }, - status_code=422, - ) - try: - claim_obj = ClaimOutput.model_validate(row.raw_json) - except Exception as exc: - return JSONResponse( - { - "error": "Unprocessable", - "detail": f"Claim {claim_id} raw_json is malformed: {exc}", - }, - status_code=422, - ) - - try: - text = serialize_837(claim_obj) - except SerializeError837 as exc: - return JSONResponse( - {"error": "Unprocessable", "detail": str(exc)}, - status_code=422, - ) - - return Response( - content=text, - media_type="text/x12", - headers={ - "Content-Disposition": f'attachment; filename="claim-{claim_id}.x12"' - }, - ) - - -@app.get("/api/claims/{claim_id}/line-reconciliation") -def get_claim_line_reconciliation(claim_id: str) -> dict: - """Per-line reconciliation view for the ClaimDrawer tab. - - Spec §5.1. Returns the 837 service lines and 835 SVC composites - side-by-side, with per-line CAS adjustments and a summary block. - - Architecture note: 837 service lines live in ``Claim.raw_json`` - (not a separate ORM table), so the 837-side rows are read from the - JSON blob; the 835-side rows come from ``ServiceLinePayment`` ORM. - ``LineReconciliation.claim_service_line_number`` stores the 1-based - line number to join them. - """ - from sqlalchemy import select - from cyclone.db import ( - LineReconciliation, ServiceLinePayment, CasAdjustment, - ) - import json as _json - from decimal import Decimal - - with db.SessionLocal()() as s: - claim = s.get(db.Claim, claim_id) - if claim is None: - raise HTTPException( - status_code=404, - detail={"error": "Not found", "detail": f"Claim {claim_id} not found"}, - ) - - # 837 service lines: from raw_json. - raw = claim.raw_json or {} - claim_lines_raw = raw.get("service_lines") or [] - # Normalize to dicts for the response. - claim_lines = [_claim_line_dict(d) for d in claim_lines_raw] - - # 835 service payments: ORM rows from the matched remit. - remits = list( - s.execute( - select(db.Remittance).where(db.Remittance.claim_id == claim_id) - ).scalars().all() - ) - svc_payments: list[dict] = [] - svc_ids: list[int] = [] - if remits: - svc_rows = list( - s.execute( - select(ServiceLinePayment).where( - ServiceLinePayment.remittance_id.in_([r.id for r in remits]) - ).order_by(ServiceLinePayment.line_number) - ).scalars().all() - ) - for svc in svc_rows: - d = _svc_to_dict(svc) - svc_payments.append(d) - svc_ids.append(svc.id) - - # LineReconciliation rows. - lrs = list( - s.execute( - select(LineReconciliation).where(LineReconciliation.claim_id == claim_id) - ).scalars().all() - ) - # Index by claim_service_line_number and service_line_payment_id. - lr_by_claim_num: dict[int, LineReconciliation] = { - lr.claim_service_line_number: lr for lr in lrs if lr.claim_service_line_number is not None - } - lr_by_svc: dict[int, LineReconciliation] = { - lr.service_line_payment_id: lr for lr in lrs if lr.service_line_payment_id is not None - } - - # CAS rows grouped by svc id. - cas_by_svc: dict[int, list[CasAdjustment]] = {} - if svc_ids: - cas_rows = list( - s.execute( - select(CasAdjustment).where(CasAdjustment.service_line_payment_id.in_(svc_ids)) - ).scalars().all() - ) - for c in cas_rows: - cas_by_svc.setdefault(c.service_line_payment_id, []).append(c) - - # Build output lines array, preserving 837 order then 835-only. - svc_by_id: dict[int, dict] = {d["id"]: d for d in svc_payments} - lines_out: list[dict] = [] - billed_total = Decimal("0") - paid_total = Decimal("0") - adjustment_total = Decimal("0") - matched_count = 0 - used_svc_ids: set[int] = set() - - for cl in claim_lines: - billed_total += Decimal(str(cl["charge"])) - lr = lr_by_claim_num.get(cl["line_number"]) - if lr is None: - lines_out.append({ - "claim_service_line": cl, - "service_line_payment": None, - "status": "unmatched_837_only", - "adjustments": [], - }) - continue - svc_id = lr.service_line_payment_id - svc = svc_by_id.get(svc_id) if svc_id else None - if svc_id is not None: - used_svc_ids.add(svc_id) - cas_list = cas_by_svc.get(svc_id, []) if svc_id is not None else [] - cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0")) - if svc: - paid_total += Decimal(str(svc["payment"])) - adjustment_total += cas_total - if lr.status == "matched": - matched_count += 1 - lines_out.append({ - "claim_service_line": cl, - "service_line_payment": svc, - "status": lr.status, - "adjustments": [ - {"group_code": c.group_code, "reason_code": c.reason_code, - "amount": str(Decimal(str(c.amount)))} - for c in cas_list - ], - }) - - # 835-only lines (no claim match). - for lr in lrs: - if lr.claim_service_line_number is not None: - continue - svc_id = lr.service_line_payment_id - if svc_id is None: - continue - if svc_id in used_svc_ids: - continue - svc = svc_by_id.get(svc_id) - cas_list = cas_by_svc.get(svc_id, []) - cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0")) - if svc: - paid_total += Decimal(str(svc["payment"])) - adjustment_total += cas_total - lines_out.append({ - "claim_service_line": None, - "service_line_payment": svc, - "status": lr.status, - "adjustments": [ - {"group_code": c.group_code, "reason_code": c.reason_code, - "amount": str(Decimal(str(c.amount)))} - for c in cas_list - ], - }) - - return { - "claim_id": claim_id, - "summary": { - "billed_total": str(billed_total), - "paid_total": str(paid_total), - "adjustment_total": str(adjustment_total), - "matched_lines": matched_count, - "total_lines": len(claim_lines), - }, - "lines": lines_out, - } - - -def _claim_line_dict(d: dict) -> dict: - """Project an 837 service-line dict from ``Claim.raw_json`` to wire shape.""" - from decimal import Decimal - proc = d.get("procedure") or {} - charge = d.get("charge") - units = d.get("units") - return { - "line_number": d.get("line_number"), - "procedure_qualifier": proc.get("qualifier", "HC"), - "procedure_code": proc.get("code", ""), - "modifiers": proc.get("modifiers") or [], - "charge": str(Decimal(str(charge))) if charge is not None else "0", - "units": str(Decimal(str(units))) if units is not None else None, - "unit_type": d.get("unit_type"), - "service_date": d.get("service_date"), - } - - -def _svc_to_dict(svc) -> dict: - """Project an ORM ``ServiceLinePayment`` to wire shape.""" - import json as _json - from decimal import Decimal - return { - "id": svc.id, - "line_number": svc.line_number, - "procedure_qualifier": svc.procedure_qualifier, - "procedure_code": svc.procedure_code, - "modifiers": _json.loads(svc.modifiers_json or "[]"), - "charge": str(Decimal(str(svc.charge))), - "payment": str(Decimal(str(svc.payment))), - "units": str(Decimal(str(svc.units))) if svc.units is not None else None, - "unit_type": svc.unit_type, - "service_date": svc.service_date.isoformat() if svc.service_date else None, - } - - -@app.get("/api/reconciliation/unmatched") -def get_reconciliation_unmatched() -> dict: - """Return unmatched Claims (left) and unmatched Remittances (right). - - Powers the reconciliation review surface: every Claim with no - paired Remittance appears on the left, every Remittance with no - paired Claim appears on the right. The two lists are always present - (empty list, never absent) so the UI can index unconditionally. - """ - return store.list_unmatched(kind="both") - - -# --------------------------------------------------------------------------- # -# Side-by-side diff between two batches (SP3 P4 / T18) -# --------------------------------------------------------------------------- # - - -@app.get("/api/batch-diff") -def get_batch_diff( - a: str | None = Query(None), - b: str | None = Query(None), -) -> dict: - """Return a side-by-side diff of two batches identified by id. - - Query params: ``a=``, ``b=`` (both required). - - Response body (snake_case keys, see :mod:`cyclone.batch_diff` for the - projector shapes): - - ``a`` / ``b`` — small metadata blocks (id, kind, parsedAt, - inputFilename, claimCount) - - ``added`` — claims present in B but not A - - ``removed`` — claims present in A but not B - - ``changed`` — claims present in both, with field deltas - - ``summary`` — precomputed counts - - Errors: - - 400 — missing ``a`` or ``b`` - - 404 — either batch id is unknown - - Pure read endpoint — never mutates the store. Both 837P and 835 - batches are accepted (mixed-kind diffs are valid: comparing the - submitted claims against the matching remittances). - """ - if not a or not b: - raise HTTPException( - status_code=400, - detail={"error": "Missing param", "detail": "Both ?a= and ?b= are required."}, - ) - try: - a_rec, b_rec = store.load_two_for_diff(a, b) - except LookupError as exc: - raise HTTPException( - status_code=404, - detail={"error": "Not found", "detail": str(exc)}, - ) - - # Lazy import — keeps the module's import surface small until the - # endpoint is actually hit. Mirrors the same pattern used by other - # endpoint-local helpers (e.g. reconciler). - from cyclone.batch_diff import diff_batches_to_wire - - return diff_batches_to_wire(a_rec, b_rec) - - -@app.post("/api/reconciliation/match") -def post_reconciliation_match(body: dict) -> dict: - """Manually pair a Claim with a Remittance (operator override). - - Body: ``{"claim_id": ..., "remit_id": ...}``. Returns - ``{"claim": , "match": }`` on success. Errors: - - 400: missing ``claim_id`` or ``remit_id`` - - 404: claim or remittance not found - - 409: claim already matched, or apply_* returned a noop - (claim in terminal state) — detail echoes ``current_state`` - and ``activity_kind`` so the UI can render a precise message. - """ - claim_id = body.get("claim_id") - remit_id = body.get("remit_id") - if not claim_id or not remit_id: - raise HTTPException( - status_code=400, - detail="claim_id and remit_id required", - ) - try: - return store.manual_match(claim_id, remit_id) - except AlreadyMatchedError as e: - raise HTTPException( - status_code=409, - detail={"error": "already_matched", "message": str(e)}, - ) - except InvalidStateError as e: - raise HTTPException( - status_code=409, - detail={ - "error": "invalid_state", - "current_state": e.current_state, - "activity_kind": e.activity_kind, - }, - ) - except LookupError: - # manual_match raises LookupError when the claim or remittance - # row is missing (we catch the parent class so any future - # KeyError subclasses in the store get the same treatment). - raise HTTPException( - status_code=404, - detail="claim_or_remit_not_found", - ) - - -@app.post("/api/reconciliation/unmatch") -def post_reconciliation_unmatch(body: dict) -> dict: - """Remove the current match for a Claim; reset Claim to submitted. - - Body: ``{"claim_id": ...}``. Returns - ``{"claim": , "deletedMatches": }``. Errors: - - 400: missing ``claim_id`` - - 404: claim not found - - 409: claim has no current match (NotMatchedError is mapped - by the store; we surface 409 to match the manual_match contract) - """ - from cyclone.store import NotMatchedError - claim_id = body.get("claim_id") - if not claim_id: - raise HTTPException( - status_code=400, - detail="claim_id required", - ) - try: - return store.manual_unmatch(claim_id) - except NotMatchedError as e: - raise HTTPException( - status_code=409, - detail={"error": "not_matched", "message": str(e)}, - ) - except LookupError: - raise HTTPException( - status_code=404, - detail="claim_not_found", - ) - - -@app.get("/api/remittances") -def list_remittances( - request: Request, - batch_id: str | None = Query(None), - payer: str | None = Query(None), - claim_id: str | None = Query(None), - date_from: str | None = Query(None), - date_to: str | None = Query(None), - sort: str | None = Query(None), - order: str = Query("desc"), - limit: int = Query(100, ge=1, le=1000), - offset: int = Query(0, ge=0), -) -> Any: - common = dict( - batch_id=batch_id, - payer=payer, - claim_id=claim_id, - date_from=date_from, - date_to=date_to, - ) - items = list(store.iter_remittances( - sort=sort, order=order, limit=limit, offset=offset, **common, - )) - total = len(list(store.iter_remittances(**common))) - returned = len(items) - has_more = total > offset + returned - if _wants_ndjson(request): - return StreamingResponse( - _ndjson_stream_list(items, total, returned, has_more), - media_type="application/x-ndjson", - ) - return { - "items": items, - "total": total, - "returned": returned, - "has_more": has_more, - } - - -@app.get("/api/remittances/stream") -async def remittances_stream( - request: Request, - payer: str | None = Query(None), - claim_id: str | None = Query(None), - date_from: str | None = Query(None), - date_to: str | None = Query(None), - sort: str | None = Query(None), - order: str = Query("desc"), - limit: int = Query(100, ge=1, le=1000), -) -> StreamingResponse: - """Stream Remittances as NDJSON: snapshot first, then live events. - - Subscribes to ``remittance_written``. Default sort is - ``-received_date`` (newest-first), matching the list endpoint's - most common sort. - - NOTE: registered before ``/api/remittances/{remittance_id}`` so - the literal ``stream`` path segment doesn't get matched as a - remittance id. - """ - bus: EventBus = request.app.state.event_bus - - async def gen() -> AsyncIterator[bytes]: - rows = store.iter_remittances( - payer=payer, claim_id=claim_id, - date_from=date_from, date_to=date_to, - sort=sort or "-received_date", order=order, limit=limit, - ) - for row in rows: - yield _ndjson_line({"type": "item", "data": row}) - yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) - - async for chunk in _tail_events(request, bus, ["remittance_written"]): - yield chunk - - return StreamingResponse(gen(), media_type="application/x-ndjson") - - -@app.get("/api/remittances/{remittance_id}") -def get_remittance(remittance_id: str) -> dict: - """Return one remittance with its labeled CAS ``adjustments`` array. - - Path param is ``remittance_id`` (not ``id``) to avoid shadowing - FastAPI's internal ``id`` name and to keep OpenAPI docs self- - describing. Returns 404 when the remittance is missing — never 500. - """ - body = store.get_remittance(remittance_id) - if body is None: - raise HTTPException( - status_code=404, - detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"}, - ) - return body - - -@app.get("/api/providers") -def list_providers( - request: Request, - npi: str | None = Query(None), - state: str | None = Query(None), - limit: int = Query(100, ge=1, le=1000), - offset: int = Query(0, ge=0), -) -> Any: - items = store.distinct_providers() - if npi is not None: - items = [p for p in items if p["npi"] == npi] - if state is not None: - items = [p for p in items if p.get("state") == state] - paged = items[offset:offset + limit] - total = len(items) - returned = len(paged) - has_more = total > offset + returned - if _wants_ndjson(request): - return StreamingResponse( - _ndjson_stream_list(paged, total, returned, has_more), - media_type="application/x-ndjson", - ) - return { - "items": paged, - "total": total, - "returned": returned, - "has_more": has_more, - } - - -@app.get("/api/activity") -def list_activity( - request: Request, - kind: str | None = Query(None), - since: str | None = Query(None), - limit: int = Query(200, ge=1, le=500), -) -> Any: - events = store.recent_activity(limit=limit) - if kind is not None: - events = [e for e in events if e["kind"] == kind] - if since is not None: - events = [e for e in events if e["timestamp"] >= since] - total = len(events) - has_more = False - if _wants_ndjson(request): - return StreamingResponse( - _ndjson_stream_list(events, total, total, has_more), - media_type="application/x-ndjson", - ) - return { - "items": events, - "total": total, - "returned": total, - "has_more": has_more, - } - - -@app.get("/api/activity/stream") -async def activity_stream( - request: Request, - kind: str | None = Query(None), - since: str | None = Query(None), - limit: int = Query(50, ge=1, le=500), -) -> StreamingResponse: - """Stream Activity events as NDJSON: snapshot first, then live events. - - Subscribes to ``activity_recorded``. Default ``limit`` is 50 - (smaller than the list endpoint's 200) because activity is - high-volume — callers usually want the most recent handful, not a - full replay. - """ - bus: EventBus = request.app.state.event_bus - - async def gen() -> AsyncIterator[bytes]: - # Snapshot reuses the same in-memory filter as ``list_activity`` - # so the two endpoints are interchangeable for the snapshot - # half. - events = store.recent_activity(limit=limit) - if kind is not None: - events = [e for e in events if e["kind"] == kind] - if since is not None: - events = [e for e in events if e["timestamp"] >= since] - for ev in events: - yield _ndjson_line({"type": "item", "data": ev}) - yield _ndjson_line({ - "type": "snapshot_end", "data": {"count": len(events)}, - }) - - async for chunk in _tail_events(request, bus, ["activity_recorded"]): - yield chunk - - return StreamingResponse(gen(), media_type="application/x-ndjson") # --------------------------------------------------------------------------- # @@ -1907,568 +940,10 @@ async def activity_stream( # --------------------------------------------------------------------------- # -def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]: - """Build a :class:`ParseResult270` from a request body dict. +# --- /api/eligibility/* lives in cyclone.api_routers.eligibility --- - The body shape is the minimum surface needed to build a valid 270 - inquiry (per spec section 3.4 — operator-driven, ephemeral): +# --- /api/admin/* lives in cyclone.api_routers.admin --- - :: - - { - "subscriber": {first_name, last_name, member_id, dob}, - "provider": {npi, name}, - "payer": {id, name}, - "service_type_code": "1" - } - - Returns ``(ParseResult270, service_type_code)``. Raises - :class:`HTTPException` (400) when the body is missing required - fields. - """ - subscriber_in = body.get("subscriber") or {} - provider_in = body.get("provider") or {} - payer_in = body.get("payer") or {} - service_type_code = (body.get("service_type_code") or "").strip() - - # Required-field checks. We surface a single 400 with the first - # missing field name to match the rest of the API's error contract. - if not service_type_code: - raise HTTPException( - status_code=400, - detail={"error": "Bad request", "detail": "service_type_code is required"}, - ) - if not subscriber_in.get("member_id"): - raise HTTPException( - status_code=400, - detail={"error": "Bad request", "detail": "subscriber.member_id is required"}, - ) - if not provider_in.get("npi"): - raise HTTPException( - status_code=400, - detail={"error": "Bad request", "detail": "provider.npi is required"}, - ) - if not payer_in.get("name"): - raise HTTPException( - status_code=400, - detail={"error": "Bad request", "detail": "payer.name is required"}, - ) - - # Build the Pydantic models. The serializer handles all envelope - # generation (sender_id/receiver_id/control_number/transaction_date - # are filled in by the serializer with sensible defaults). - from datetime import date as _date - - subscriber_dob_raw = subscriber_in.get("dob") - subscriber_dob: _date | None = None - if subscriber_dob_raw: - try: - subscriber_dob = _date.fromisoformat(subscriber_dob_raw) - except (TypeError, ValueError) as exc: - raise HTTPException( - status_code=400, - detail={ - "error": "Bad request", - "detail": f"subscriber.dob must be YYYY-MM-DD: {exc}", - }, - ) from exc - - result = ParseResult270( - envelope=Envelope( - sender_id="SUBMITTERID", - receiver_id=str(payer_in.get("id") or "RECEIVERID"), - control_number="000000001", - transaction_date=_date.today(), - implementation_guide="005010X279A1", - ), - information_source=InformationSource270( - name=str(payer_in["name"]), - id=str(payer_in.get("id") or "") or None, - ), - information_receiver=InformationReceiver270( - name=str(provider_in.get("name") or ""), - npi=str(provider_in["npi"]), - ), - subscriber=Subscriber270( - member_id=str(subscriber_in["member_id"]), - first_name=str(subscriber_in.get("first_name") or "") or None, - last_name=str(subscriber_in.get("last_name") or "") or None, - dob=subscriber_dob, - ), - inquiries=[EligibilityBenefitInquiry(service_type_code=service_type_code)], - summary=BatchSummary( - input_file="eligibility_request", - control_number="000000001", - transaction_date=_date.today(), - total_claims=1, - passed=1, - failed=0, - ), - ) - return result, service_type_code - - -@app.post("/api/eligibility/request") -def post_eligibility_request(body: dict) -> Any: - """Build a 270 eligibility inquiry from a small JSON body. - - Returns ``{"raw_270_text": , "parsed": }`` - so the operator can either download the raw text (paste into a - payer portal) or render the parsed fields directly. Per spec - section 3.4, nothing is persisted to the DB. - """ - try: - result, _ = _validate_eligibility_request(body) - except HTTPException: - raise - except (KeyError, TypeError, ValueError) as exc: - raise HTTPException( - status_code=400, - detail={"error": "Bad request", "detail": f"Malformed body: {exc}"}, - ) from exc - - raw_270_text = serialize_270(result) - return { - "raw_270_text": raw_270_text, - "parsed": json.loads(result.model_dump_json()), - } - - -@app.post("/api/eligibility/parse-271") -async def post_eligibility_parse_271( - file: UploadFile = File(...), -) -> Any: - """Parse a 271 eligibility response and return the structured summary. - - Accepts the raw 271 text as a file upload (multipart/form-data), - mirrors the ``/api/parse-999`` contract. Per spec section 3.4 the - result is NOT persisted — the operator re-pastes the 271 each - time they need a fresh read. - - The response body is a JSON object with three top-level keys: - ``coverage_benefits``, ``subscriber``, and ``summary``. 400 is - returned on empty / undecodable / malformed EDI; 200 on success. - """ - raw = await file.read() - if not raw: - return JSONResponse( - status_code=400, - content={"error": "Empty file", "detail": "Uploaded file contained no bytes."}, - ) - try: - text = raw.decode("utf-8") - except UnicodeDecodeError as exc: - return JSONResponse( - status_code=400, - content={"error": "Encoding error", "detail": str(exc)}, - ) - - try: - result = parse_271_text(text, input_file=file.filename or "") - except CycloneParseError as exc: - return JSONResponse( - status_code=400, - content={"error": "Parse error", "detail": str(exc)}, - ) - except Exception as exc: # pragma: no cover - safety net - log.exception("Unexpected parser failure on 271") - return JSONResponse( - status_code=500, - content={"error": "Internal server error", "detail": str(exc)}, - ) - - return { - "coverage_benefits": [ - json.loads(cb.model_dump_json()) for cb in result.coverage_benefits - ], - "subscriber": json.loads(result.subscriber.model_dump_json()), - "summary": json.loads(result.summary.model_dump_json()), - "envelope": json.loads(result.envelope.model_dump_json()), - "information_source": json.loads(result.information_source.model_dump_json()), - "information_receiver": json.loads(result.information_receiver.model_dump_json()), - } - - -# --------------------------------------------------------------------------- -# SP9: providers / payers / clearhouse endpoints -# --------------------------------------------------------------------------- - - -@app.get("/api/clearhouse") -def get_clearhouse(): - """Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block).""" - ch = store.get_clearhouse() - if ch is None: - raise HTTPException(status_code=404, detail="clearhouse not seeded") - return json.loads(ch.model_dump_json()) - - -@app.post("/api/clearhouse/submit") -def submit_to_clearhouse(body: dict): - """Submit a batch of claims to the clearhouse (SFTP). SP9: stub. - - Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}`` - - Stub behavior: serializes each claim via the SP7 serializer, builds - an HCPF-compliant outbound filename, and copies the result to - ``{staging_dir}/{outbound_path}/{filename}`` instead of opening a - real SFTP connection. Returns a receipt per claim. - """ - from cyclone.clearhouse import make_client - from cyclone.edi.filenames import build_outbound_filename - - claim_ids = body.get("claim_ids", []) - payer_id = body.get("payer_id") - if not claim_ids: - raise HTTPException(status_code=400, detail="claim_ids required") - if not payer_id: - raise HTTPException(status_code=400, detail="payer_id required") - - ch = store.get_clearhouse() - if ch is None: - raise HTTPException(status_code=500, detail="clearhouse not seeded") - - client = make_client(ch.sftp_block) - results = [] - for cid in claim_ids: - try: - x12_text = _serialize_claim_for_submit(cid) - except Exception as exc: # noqa: BLE001 - results.append({"claim_id": cid, "ok": False, "error": str(exc)}) - continue - filename = build_outbound_filename(ch.tpid, "837P") - remote = f"{ch.sftp_block.paths['outbound']}/{filename}" - staging_path = client.write_file(remote, x12_text.encode("utf-8")) - results.append({ - "claim_id": cid, - "ok": True, - "filename": filename, - "staging_path": str(staging_path), - "remote_path": remote, - }) - # SP11: audit trail for each successful clearhouse submission. - with db.SessionLocal()() as audit_s: - append_event(audit_s, AuditEvent( - event_type="clearhouse.submitted", - entity_type="claim", - entity_id=cid, - payload={ - "filename": filename, - "remote_path": remote, - "tpid": ch.tpid, - "stub": ch.sftp_block.stub, - }, - actor="clearhouse-submit", - )) - audit_s.commit() - return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub} - - -def _serialize_claim_for_submit(claim_id: str) -> str: - """Serialize a claim to X12 for SFTP submission. Lazy import of the - serializer to avoid pulling FastAPI machinery at module import time.""" - from cyclone.parsers.serialize_837 import serialize_837 - from cyclone import db - with db.SessionLocal()() as s: - row = s.get(db.Claim, claim_id) - if row is None: - raise ValueError(f"claim {claim_id!r} not found") - # Re-parse the stored raw_json to get a ClaimOutput - from cyclone.parsers.models import ClaimOutput, Envelope, Subscriber, Payer, BillingProvider - from cyclone.parsers.parse_837 import parse - raw = row.raw_json or {} - # Reconstruct minimal ClaimOutput from raw_json; this is best-effort. - return _serialize_claim_from_raw(row, raw) - - -def _serialize_claim_from_raw(claim_row, raw: dict) -> str: - """Best-effort serializer that uses the stored raw_json to emit a fresh 837. - - For SP9 this delegates to the existing serialize_837 helper if the - claim has a complete raw_segments array. Otherwise it returns a - minimal placeholder. - """ - from cyclone.parsers.serialize_837 import serialize_837 - from cyclone.parsers.parse_837 import parse - - # Re-parse the original batch text (need to re-derive from store). - # SP9 stub: if the claim has a `raw_json` with `x12_text`, use that. - if isinstance(raw, dict) and raw.get("x12_text"): - result = parse(raw["x12_text"]) - if result.claims: - return serialize_837(result.claims[0]) - # Fallback: raise so the caller sees an error. - raise RuntimeError( - f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text" - ) - - -@app.get("/api/config/providers") -def list_configured_providers(is_active: bool | None = Query(default=True)): - """List the configured provider rows (3 NPIs for SP9).""" - return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)] - - -# --------------------------------------------------------------------------- # -# SP11: tamper-evident audit log (admin) -# --------------------------------------------------------------------------- # - - -@app.get("/api/admin/audit-log") -def list_audit_log_endpoint( - entity_type: str | None = Query(default=None), - entity_id: str | None = Query(default=None), - event_type: str | None = Query(default=None), - limit: int = Query(default=100, ge=1, le=1000), -) -> Any: - """List audit-log rows, newest first, with optional filters. - - Filters match the (entity_type, entity_id) pair (typical use: - "show me everything that happened to claim C-123") or a single - event_type (typical use: "show me all clearhouse.submitted - events today"). - """ - with db.SessionLocal()() as s: - q = s.query(db.AuditLog) - if entity_type: - q = q.filter(db.AuditLog.entity_type == entity_type) - if entity_id: - q = q.filter(db.AuditLog.entity_id == entity_id) - if event_type: - q = q.filter(db.AuditLog.event_type == event_type) - rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all() - return { - "total": len(rows), - "items": [ - { - "id": r.id, - "event_type": r.event_type, - "entity_type": r.entity_type, - "entity_id": r.entity_id, - "actor": r.actor, - "payload": json.loads(r.payload_json) if r.payload_json else None, - "created_at": r.created_at.isoformat() if r.created_at else None, - "prev_hash": r.prev_hash, - "hash": r.hash, - } - for r in rows - ], - } - - -@app.get("/api/admin/audit-log/verify") -def verify_audit_log_endpoint() -> Any: - """Walk the audit-log chain and verify every row's hash. - - Returns ``{"ok": true, "checked": N}`` for a clean chain, or - ``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}`` - for a broken chain. This is the operator's "did anyone tamper?" - endpoint; run it on demand or via a nightly cron job. - """ - with db.SessionLocal()() as s: - result = verify_chain(s) - return { - "ok": result.ok, - "checked": result.checked, - "first_bad_id": result.first_bad_id, - "reason": result.reason, - } - - -# --------------------------------------------------------------------------- -# SP15: SQLCipher key rotation -# -# Re-encrypts the DB in place with a fresh key, then updates the -# Keychain so subsequent connections open with the new key. This is -# a 1-time operation per rotation; for routine read/write the rest -# of the API is unchanged. -# -# Concurrency: the rotation holds a module-level lock so two -# concurrent requests can't race and end up with mismatched Keychain -# + DB. The lock is a simple threading.Lock; a process restart -# resets it (intentional — the operator's next start-up opens with -# whatever key is in the Keychain). -# --------------------------------------------------------------------------- -import threading as _threading -from cyclone import db_crypto as _db_crypto -from cyclone import secrets as _secrets - -_db_rotate_lock = _threading.Lock() - - -@app.post("/api/admin/db/rotate-key") -def rotate_db_key_endpoint(body: dict | None = None) -> Any: - """Generate a fresh DB key, re-encrypt the DB, update the Keychain. - - Request body (optional): - actor: who initiated the rotation. Defaults to "operator". - reason: human-readable reason. Written to the audit log. - - Returns: - ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}`` - on success. On failure (DB not encrypted, rekey failed, - Keychain update failed) returns the same shape with - ``ok=false`` and a ``reason``. HTTP 503 is returned if the - rekey fails or encryption is not enabled. - - The Keychain write happens *after* the rekey succeeds. If the - Keychain write fails, the DB has the new key but the Keychain - still has the old one — the endpoint returns 503 with a - "keychain update failed" reason and the operator must restore - the old key manually (``cyclone db restore-key ``) to - avoid being locked out. - """ - body = body or {} - actor = body.get("actor") or "operator" - reason = body.get("reason") or "" - - if not _db_crypto.is_encryption_enabled(): - raise HTTPException( - status_code=400, - detail="encryption not enabled (sqlcipher3 missing or no Keychain key)", - ) - - # Acquire the lock; non-blocking so a stuck rotation doesn't - # silently hold up other requests. - if not _db_rotate_lock.acquire(blocking=False): - raise HTTPException( - status_code=409, - detail="another key rotation is in progress", - ) - try: - url = db._resolve_url() - old_key = _db_crypto.get_db_key() - if not old_key: - raise HTTPException( - status_code=400, - detail="no DB key in Keychain; cannot rotate", - ) - - new_key = _db_crypto.generate_db_key() - result = _db_crypto.rotate_db_key( - url=url, old_key=old_key, new_key=new_key, - ) - if not result.ok: - # Rekey failed. The DB still has the old key. The - # Keychain is unchanged. Caller should NOT retry with - # the same new key (it's lost); generate a fresh one. - log.error("SQLCipher rotate failed: %s", result.reason) - raise HTTPException( - status_code=503, - detail={ - "ok": False, - "old_fingerprint": result.old_fingerprint, - "new_fingerprint": result.new_fingerprint, - "rotated_at": result.rotated_at, - "reason": result.reason, - }, - ) - - # Rekey succeeded. Now update the Keychain. If this fails - # the DB is locked behind the new key — operator must - # restore the old key manually. - if not _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT, new_key): - log.error("Keychain update failed after successful rekey!") - raise HTTPException( - status_code=503, - detail={ - "ok": False, - "old_fingerprint": result.old_fingerprint, - "new_fingerprint": result.new_fingerprint, - "rotated_at": result.rotated_at, - "reason": ( - "rekey succeeded but Keychain update failed — " - "the DB is now encrypted with the new key but " - "the Keychain still has the old one. " - "Restore the old key to the Keychain to recover." - ), - }, - ) - - # Store the old key in the "previous" account for a grace - # period so the operator can roll back if they discover the - # new key is broken (e.g. the Keychain entry got truncated). - _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key) - - # Rebuild the engine so subsequent connections use the new - # key. dispose_engine() closes every pooled connection that - # was using the old key; init_db() opens new ones with the - # new key from the (now-updated) Keychain. - db.reinit_engine() - - # Audit log the rotation. We do this after the engine is - # rebuilt so the audit event is written with the new key — - # proving that the new key works for new writes. - try: - from cyclone.audit_log import append_event, AuditEvent - with db.SessionLocal()() as s: - append_event(s, AuditEvent( - event_type="db.key_rotated", - entity_type="database", - entity_id="cyclone.db", - actor=actor, - payload={ - "old_fingerprint": result.old_fingerprint, - "new_fingerprint": result.new_fingerprint, - "table_count": result.table_count, - "reason": reason, - }, - )) - s.commit() - except Exception as exc: # noqa: BLE001 - # Audit append is best-effort; rotation already succeeded. - log.warning("could not write audit event for rotation: %s", exc) - - return { - "ok": True, - "old_fingerprint": result.old_fingerprint, - "new_fingerprint": result.new_fingerprint, - "rotated_at": result.rotated_at, - "table_count": result.table_count, - } - finally: - _db_rotate_lock.release() - - -@app.get("/api/config/providers/{npi}") -def get_configured_provider(npi: str): - p = store.get_provider(npi) - if p is None: - raise HTTPException(status_code=404, detail=f"provider {npi!r} not found") - return json.loads(p.model_dump_json()) - - -@app.get("/api/config/payers") -def list_configured_payers(is_active: bool | None = Query(default=True)): - return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)] - - -@app.get("/api/config/payers/{payer_id}/configs") -def list_payer_configs(payer_id: str): - """List all (transaction_type, config_json) blocks for a payer.""" - from cyclone import payers as payer_loader - configs = [ - {"transaction_type": tx, "config_json": block, "source": "yaml"} - for (pid, tx), block in payer_loader.all_configs().items() - if pid == payer_id - ] - # Also check the DB for runtime-overridden configs - for tx in ("837P", "835", "277CA", "999", "TA1"): - live = store.get_payer_config(payer_id, tx) - if live is not None: - configs.append({"transaction_type": tx, "config_json": live, "source": "db"}) - return configs - - -@app.post("/api/admin/reload-config") -def reload_config(): - """Re-read ``config/payers.yaml`` and revalidate. Returns counts.""" - from cyclone import payers as payer_loader - try: - configs = payer_loader.load_payer_configs() - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - return {"ok": True, "loaded": len(configs), "errors": []} __all__ = ["app"] diff --git a/backend/src/cyclone/api_routers/activity.py b/backend/src/cyclone/api_routers/activity.py new file mode 100644 index 0000000..6ca63fe --- /dev/null +++ b/backend/src/cyclone/api_routers/activity.py @@ -0,0 +1,97 @@ +"""``/api/activity`` — list & live-tail of recorded Activity events. + +The Activity log is the cross-resource audit feed: every parser write, +inbox state change, match/dismiss/resubmit, clearhouse submission, etc. +records one row keyed by ``kind``. The list endpoint returns the most +recent ``limit`` events with optional ``kind`` / ``since`` filters; the +stream endpoint emits a snapshot followed by live updates from the +``EventBus``. + +The default ``limit`` on the list endpoint is 200 (matches the prior +inline behavior); the stream endpoint defaults to 50 because activity is +high-volume — callers usually want the most recent handful, not a full +replay. +""" +from __future__ import annotations + +from typing import Any, AsyncIterator + +from fastapi import APIRouter, Query, Request +from fastapi.responses import StreamingResponse + +from cyclone.api_helpers import ( + ndjson_line, + ndjson_stream_list, + tail_events, + wants_ndjson, +) +from cyclone.pubsub import EventBus +from cyclone.store import store + +router = APIRouter() + + +@router.get("/api/activity") +def list_activity_endpoint( + request: Request, + kind: str | None = Query(None), + since: str | None = Query(None), + limit: int = Query(200, ge=1, le=500), +) -> Any: + """Return the recent Activity log, newest first. + + Optional ``kind`` filters by ``event["kind"]`` (e.g. ``claim_submitted``). + Optional ``since`` is an inclusive ISO timestamp lower bound applied + to ``event["timestamp"]``. + """ + events = store.recent_activity(limit=limit) + if kind is not None: + events = [e for e in events if e["kind"] == kind] + if since is not None: + events = [e for e in events if e["timestamp"] >= since] + total = len(events) + has_more = False + if wants_ndjson(request): + return StreamingResponse( + ndjson_stream_list(events, total, total, has_more), + media_type="application/x-ndjson", + ) + return { + "items": events, + "total": total, + "returned": total, + "has_more": has_more, + } + + +@router.get("/api/activity/stream") +async def activity_stream_endpoint( + request: Request, + kind: str | None = Query(None), + since: str | None = Query(None), + limit: int = Query(50, ge=1, le=500), +) -> StreamingResponse: + """Stream Activity events as NDJSON: snapshot first, then live events. + + Subscribes to ``activity_recorded``. The snapshot half reuses the + same in-memory filter as ``list_activity`` so the two endpoints are + interchangeable for the snapshot half. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + events = store.recent_activity(limit=limit) + if kind is not None: + events = [e for e in events if e["kind"] == kind] + if since is not None: + events = [e for e in events if e["timestamp"] >= since] + for ev in events: + yield ndjson_line({"type": "item", "data": ev}) + yield ndjson_line({ + "type": "snapshot_end", "data": {"count": len(events)}, + }) + + async for chunk in tail_events(request, bus, ["activity_recorded"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") diff --git a/backend/src/cyclone/api_routers/admin.py b/backend/src/cyclone/api_routers/admin.py new file mode 100644 index 0000000..7d978cc --- /dev/null +++ b/backend/src/cyclone/api_routers/admin.py @@ -0,0 +1,332 @@ +"""``/api/admin/*`` — operator-facing admin surface (SP11 + SP15). + +Four endpoints: + +* ``GET /api/admin/audit-log`` — list audit-log rows with optional + ``entity_type`` / ``entity_id`` / ``event_type`` filters (SP11). +* ``GET /api/admin/audit-log/verify`` — walk the audit-log hash chain + and verify every row (SP11). +* ``POST /api/admin/db/rotate-key`` — SP15 SQLCipher key rotation. + Re-encrypts the DB in place with a fresh key, updates the Keychain. + Protected by a module-level ``threading.Lock`` so two concurrent + rotations can't race (a single process restart resets the lock — + intentional, the next start-up opens with whatever key is in the + Keychain). +* ``POST /api/admin/reload-config`` — re-read ``config/payers.yaml`` + and revalidate; returns the number of configs loaded. +""" +from __future__ import annotations + +import json +import logging +import threading +from typing import Any + +from fastapi import APIRouter, HTTPException, Query + +from cyclone import db, db_crypto, secrets +from cyclone.audit_log import AuditEvent, append_event, verify_chain + +log = logging.getLogger(__name__) + +router = APIRouter() + +# Module-level lock: protects against concurrent key rotations within +# the same process. A process restart resets it (intentional — the +# operator's next start-up opens with whatever key is in the Keychain). +_db_rotate_lock = threading.Lock() + + +@router.get("/api/admin/audit-log") +def list_audit_log_endpoint( + entity_type: str | None = Query(default=None), + entity_id: str | None = Query(default=None), + event_type: str | None = Query(default=None), + limit: int = Query(default=100, ge=1, le=1000), +) -> Any: + """List audit-log rows, newest first, with optional filters. + + Filters match the (entity_type, entity_id) pair (typical use: + "show me everything that happened to claim C-123") or a single + event_type (typical use: "show me all clearhouse.submitted + events today"). + """ + with db.SessionLocal()() as s: + q = s.query(db.AuditLog) + if entity_type: + q = q.filter(db.AuditLog.entity_type == entity_type) + if entity_id: + q = q.filter(db.AuditLog.entity_id == entity_id) + if event_type: + q = q.filter(db.AuditLog.event_type == event_type) + rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all() + return { + "total": len(rows), + "items": [ + { + "id": r.id, + "event_type": r.event_type, + "entity_type": r.entity_type, + "entity_id": r.entity_id, + "actor": r.actor, + "payload": json.loads(r.payload_json) if r.payload_json else None, + "created_at": r.created_at.isoformat() if r.created_at else None, + "prev_hash": r.prev_hash, + "hash": r.hash, + } + for r in rows + ], + } + + +@router.get("/api/admin/audit-log/verify") +def verify_audit_log_endpoint() -> Any: + """Walk the audit-log chain and verify every row's hash. + + Returns ``{"ok": true, "checked": N}`` for a clean chain, or + ``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}`` + for a broken chain. This is the operator's "did anyone tamper?" + endpoint; run it on demand or via a nightly cron job. + """ + with db.SessionLocal()() as s: + result = verify_chain(s) + return { + "ok": result.ok, + "checked": result.checked, + "first_bad_id": result.first_bad_id, + "reason": result.reason, + } + + +@router.post("/api/admin/db/rotate-key") +def rotate_db_key_endpoint(body: dict | None = None) -> Any: + """Generate a fresh DB key, re-encrypt the DB, update the Keychain. + + Request body (optional): + actor: who initiated the rotation. Defaults to "operator". + reason: human-readable reason. Written to the audit log. + + Returns: + ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}`` + on success. On failure (DB not encrypted, rekey failed, + Keychain update failed) returns the same shape with + ``ok=false`` and a ``reason``. HTTP 503 is returned if the + rekey fails or encryption is not enabled. + + The Keychain write happens *after* the rekey succeeds. If the + Keychain write fails, the DB has the new key but the Keychain + still has the old one — the endpoint returns 503 with a + "keychain update failed" reason and the operator must restore + the old key manually (``cyclone db restore-key ``) to + avoid being locked out. + """ + body = body or {} + actor = body.get("actor") or "operator" + reason = body.get("reason") or "" + + if not db_crypto.is_encryption_enabled(): + raise HTTPException( + status_code=400, + detail="encryption not enabled (sqlcipher3 missing or no Keychain key)", + ) + + # Acquire the lock; non-blocking so a stuck rotation doesn't + # silently hold up other requests. + if not _db_rotate_lock.acquire(blocking=False): + raise HTTPException( + status_code=409, + detail="another key rotation is in progress", + ) + try: + url = db._resolve_url() + old_key = db_crypto.get_db_key() + if not old_key: + raise HTTPException( + status_code=400, + detail="no DB key in Keychain; cannot rotate", + ) + + new_key = db_crypto.generate_db_key() + result = db_crypto.rotate_db_key( + url=url, old_key=old_key, new_key=new_key, + ) + if not result.ok: + # Rekey failed. The DB still has the old key. The + # Keychain is unchanged. Caller should NOT retry with + # the same new key (it's lost); generate a fresh one. + log.error("SQLCipher rotate failed: %s", result.reason) + raise HTTPException( + status_code=503, + detail={ + "ok": False, + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "rotated_at": result.rotated_at, + "reason": result.reason, + }, + ) + + # Rekey succeeded. Now update the Keychain. If this fails + # the DB is locked behind the new key — operator must + # restore the old key manually. + if not secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT, new_key): + log.error("Keychain update failed after successful rekey!") + raise HTTPException( + status_code=503, + detail={ + "ok": False, + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "rotated_at": result.rotated_at, + "reason": ( + "rekey succeeded but Keychain update failed — " + "the DB is now encrypted with the new key but " + "the Keychain still has the old one. " + "Restore the old key to the Keychain to recover." + ), + }, + ) + + # Store the old key in the "previous" account for a grace + # period so the operator can roll back if they discover the + # new key is broken (e.g. the Keychain entry got truncated). + secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key) + + # Rebuild the engine so subsequent connections use the new + # key. dispose_engine() closes every pooled connection that + # was using the old key; init_db() opens new ones with the + # new key from the (now-updated) Keychain. + db.reinit_engine() + + # Audit log the rotation. We do this after the engine is + # rebuilt so the audit event is written with the new key — + # proving that the new key works for new writes. + try: + with db.SessionLocal()() as s: + append_event(s, AuditEvent( + event_type="db.key_rotated", + entity_type="database", + entity_id="cyclone.db", + actor=actor, + payload={ + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "table_count": result.table_count, + "reason": reason, + }, + )) + s.commit() + except Exception as exc: # noqa: BLE001 + # Audit append is best-effort; rotation already succeeded. + log.warning("could not write audit event for rotation: %s", exc) + + return { + "ok": True, + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "rotated_at": result.rotated_at, + "table_count": result.table_count, + } + finally: + _db_rotate_lock.release() + + +@router.post("/api/admin/reload-config") +def reload_config(): + """Re-read ``config/payers.yaml`` and revalidate. Returns counts.""" + from cyclone import payers as payer_loader + try: + configs = payer_loader.load_payer_configs() + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + return {"ok": True, "loaded": len(configs), "errors": []} + + +# --------------------------------------------------------------------------- +# SP16: inbound MFT scheduler +# --------------------------------------------------------------------------- + + +@router.post("/api/admin/scheduler/start") +async def scheduler_start(): + """Start the background MFT polling loop (idempotent).""" + from cyclone import scheduler as sched_mod + sched = sched_mod.get_scheduler() + await sched.start() + return {"ok": True, "status": sched.status().as_dict()} + + +@router.post("/api/admin/scheduler/stop") +async def scheduler_stop(): + """Stop the background MFT polling loop. Waits for the current tick.""" + from cyclone import scheduler as sched_mod + sched = sched_mod.get_scheduler() + await sched.stop() + return {"ok": True, "status": sched.status().as_dict()} + + +@router.get("/api/admin/scheduler/status") +def scheduler_status(): + """Return the scheduler's running state + last tick summary.""" + from cyclone import scheduler as sched_mod + sched = sched_mod.get_scheduler() + return sched.status().as_dict() + + +@router.post("/api/admin/scheduler/tick") +async def scheduler_tick(): + """Force a single poll cycle. Returns the TickResult. + + Useful for the operator's "run now" button or for testing. If a + scheduled tick is already in flight, this call coalesces onto it + (waits for it to finish, returns its result). + """ + from cyclone import scheduler as sched_mod + sched = sched_mod.get_scheduler() + result = await sched.tick() + return {"ok": True, "tick": result.as_dict()} + + +@router.get("/api/admin/scheduler/processed-files") +def scheduler_processed_files( + limit: int = 50, + status: str | None = None, +): + """List recently-processed inbound files (newest first). + + The operator uses this to confirm the scheduler is keeping up + with MFT traffic. ``status`` filters to one of: ok, error, + skipped, pending. + """ + from sqlalchemy import desc + from cyclone import db + from cyclone.db import ProcessedInboundFile + with db.SessionLocal()() as session: + q = session.query(ProcessedInboundFile).order_by( + desc(ProcessedInboundFile.processed_at), + ) + if status: + q = q.filter(ProcessedInboundFile.status == status) + q = q.limit(min(limit, 500)) + rows = q.all() + return { + "count": len(rows), + "files": [ + { + "id": r.id, + "name": r.name, + "size": r.size, + "file_type": r.file_type, + "processed_at": r.processed_at.isoformat() if r.processed_at else None, + "parser_used": r.parser_used, + "claim_count": r.claim_count, + "status": r.status, + "error_message": ( + r.error_message[:200] + "..." + if r.error_message and len(r.error_message) > 200 + else r.error_message + ), + } + for r in rows + ], + } diff --git a/backend/src/cyclone/api_routers/batches.py b/backend/src/cyclone/api_routers/batches.py new file mode 100644 index 0000000..958ca6d --- /dev/null +++ b/backend/src/cyclone/api_routers/batches.py @@ -0,0 +1,75 @@ +"""``/api/batches`` — list & detail endpoints for parsed batches. + +Each ``store.add(...)`` call (837P, 835, 999, TA1, 277CA) produces a +``BatchRecord`` kept in memory by :class:`CycloneStore`. The list +endpoint returns a lightweight summary (id, kind, filename, parsedAt, +claim count) for the recent batches; the detail endpoint returns the +full parsed payload for one batch. +""" +from __future__ import annotations + +import json +from typing import Any + +from fastapi import APIRouter, HTTPException, Query, Request +from fastapi.responses import StreamingResponse + +from cyclone.api_helpers import ndjson_stream_list, wants_ndjson +from cyclone.store import BatchRecord, store + +router = APIRouter() + + +def _batch_summary_claim_count(rec: BatchRecord) -> int: + """Return the number of claims on a batch, handling both 837P and 835.""" + if rec.kind == "837p": + return len(rec.result.claims) # type: ignore[attr-defined] + if rec.kind == "835": + return len(rec.result.claims) # type: ignore[attr-defined] + return 0 + + +@router.get("/api/batches") +def list_batches_endpoint( + request: Request, + limit: int = Query(100, ge=1, le=1000), +) -> Any: + """Summary of all parsed batches, newest first.""" + records = store.list(limit=limit) + items = [ + { + "id": r.id, + "kind": r.kind, + "inputFilename": r.input_filename, + "parsedAt": r.parsed_at.isoformat().replace("+00:00", "Z"), + "claimCount": _batch_summary_claim_count(r), + } + for r in records + ] + all_records = store.all() + total = len(all_records) + returned = len(items) + has_more = total > returned + if wants_ndjson(request): + return StreamingResponse( + ndjson_stream_list(items, total, returned, has_more), + media_type="application/x-ndjson", + ) + return { + "items": items, + "total": total, + "returned": returned, + "has_more": has_more, + } + + +@router.get("/api/batches/{batch_id}") +def get_batch_endpoint(batch_id: str) -> Any: + """Return the full parsed payload for one batch.""" + rec = store.get(batch_id) + if rec is None: + raise HTTPException( + status_code=404, + detail={"error": "Not found", "detail": f"Batch {batch_id} not found"}, + ) + return json.loads(rec.result.model_dump_json()) diff --git a/backend/src/cyclone/api_routers/claims.py b/backend/src/cyclone/api_routers/claims.py new file mode 100644 index 0000000..71cb612 --- /dev/null +++ b/backend/src/cyclone/api_routers/claims.py @@ -0,0 +1,411 @@ +"""``/api/claims`` — list, live-tail, detail, X12 regenerate, line-reconciliation. + +This is the largest single-resource router. Five endpoints: + +* ``GET /api/claims`` — paginated list with filters + (status / batch_id / provider_npi / payer / date range / sort). +* ``GET /api/claims/stream`` — NDJSON snapshot + live tail via the + ``claim_written`` EventBus kind. +* ``GET /api/claims/{claim_id}`` — drawer detail + (header / state / service lines / diagnoses / parties / validation / + raw segments / stateHistory / matchedRemittance). +* ``GET /api/claims/{claim_id}/serialize-837`` — regenerate X12 for + download (SP8). +* ``GET /api/claims/{claim_id}/line-reconciliation`` — per-line + reconciliation view for the ClaimDrawer tab (spec §5.1). + +**Route ordering.** ``/stream`` is declared **before** ``/{claim_id}`` so +FastAPI's first-match routing doesn't swallow the literal ``stream`` +segment as a claim id. Same order as the prior inline code. + +The two projection helpers ``_claim_line_dict`` / ``_svc_to_dict`` are +private to this router; both only used by the line-reconciliation +endpoint. +""" +from __future__ import annotations + +from decimal import Decimal +from typing import Any, AsyncIterator + +from fastapi import APIRouter, HTTPException, Query, Request +from fastapi.responses import JSONResponse, Response, StreamingResponse + +from cyclone import db +from cyclone.api_helpers import ( + ndjson_line, + ndjson_stream_list, + tail_events, + wants_ndjson, +) +from cyclone.db import Claim +from cyclone.parsers.models import ClaimOutput +from cyclone.parsers.serialize_837 import SerializeError as SerializeError837, serialize_837 +from cyclone.pubsub import EventBus +from cyclone.store import store + +router = APIRouter() + + +@router.get("/api/claims") +def list_claims_endpoint( + request: Request, + batch_id: str | None = Query(None), + status: str | None = Query(None), + provider_npi: str | None = Query(None), + payer: str | None = Query(None), + date_from: str | None = Query(None), + date_to: str | None = Query(None), + sort: str | None = Query(None), + order: str = Query("desc"), + limit: int = Query(100, ge=1, le=1000), + offset: int = Query(0, ge=0), +) -> Any: + """Paginated claims list with filters; supports NDJSON streaming. + + ``has_more`` is computed as ``total > offset + returned`` so paginating + forward keeps returning true until the last page is reached. + """ + common = dict( + batch_id=batch_id, + status=status, + provider_npi=provider_npi, + payer=payer, + date_from=date_from, + date_to=date_to, + ) + items = list(store.iter_claims( + sort=sort, order=order, limit=limit, offset=offset, **common, + )) + total = len(list(store.iter_claims(**common))) + returned = len(items) + has_more = total > offset + returned + if wants_ndjson(request): + return StreamingResponse( + ndjson_stream_list(items, total, returned, has_more), + media_type="application/x-ndjson", + ) + return { + "items": items, + "total": total, + "returned": returned, + "has_more": has_more, + } + + +@router.get("/api/claims/stream") +async def claims_stream_endpoint( + request: Request, + status: str | None = Query(None), + provider_npi: str | None = Query(None), + payer: str | None = Query(None), + date_from: str | None = Query(None), + date_to: str | None = Query(None), + sort: str | None = Query(None), + order: str = Query("desc"), + limit: int = Query(100, ge=1, le=1000), +) -> StreamingResponse: + """Stream Claims as NDJSON: snapshot first, then live events. + + Wire format: + * ``{"type":"item","data":}`` per snapshot row, then per + new ``claim_written`` event + * ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot + * ``{"type":"heartbeat","data":{"ts":}}`` every + ``CYCLONE_TAIL_HEARTBEAT_S`` seconds when idle + + Query params mirror :func:`list_claims_endpoint` so a frontend can + swap a one-shot fetch for a tail with no URL surgery. + + NOTE: registered before ``/api/claims/{claim_id}`` so the literal + ``stream`` path segment doesn't get matched as a claim id. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + # 1. Snapshot (eager — iter_claims returns a list already). + rows = store.iter_claims( + status=status, provider_npi=provider_npi, payer=payer, + date_from=date_from, date_to=date_to, + sort=sort or "-submission_date", order=order, limit=limit, + ) + for row in rows: + yield ndjson_line({"type": "item", "data": row}) + yield ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) + + # 2. Subscribe + heartbeats. + async for chunk in tail_events(request, bus, ["claim_written"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") + + +@router.get("/api/claims/{claim_id}") +def get_claim_detail_endpoint(claim_id: str) -> dict: + """Return one claim with full drawer context (SP4). + + Body shape is produced by :meth:`CycloneStore.get_claim_detail`: + header, state, service lines, diagnoses, parties, validation, + raw segments, ``stateHistory`` (most-recent-first, capped at 50), + and a populated ``matchedRemittance`` block when paired. + + Returns 404 — never 500 — on a missing claim so the UI can + distinguish "doesn't exist" from a transient fetch error. + """ + body = store.get_claim_detail(claim_id) + if body is None: + raise HTTPException( + status_code=404, + detail={ + "error": "Not found", + "detail": f"Claim {claim_id} not found", + }, + ) + return body + + +@router.get("/api/claims/{claim_id}/serialize-837") +def serialize_claim_as_837(claim_id: str): + """Return the claim as a regenerated X12 837P file (SP8). + + Loads the ClaimOutput from the persisted ``raw_json`` and runs the + outbound serializer. Returns 404 if the claim doesn't exist, 422 if + the stored payload has no parseable ClaimOutput (data integrity + issue, not a transient failure). + """ + with db.SessionLocal()() as s: + row = s.get(Claim, claim_id) + if row is None: + return JSONResponse( + {"error": "Not found", "detail": f"Claim {claim_id} not found"}, + status_code=404, + ) + if not row.raw_json: + return JSONResponse( + { + "error": "Unprocessable", + "detail": f"Claim {claim_id} has no raw_json; cannot serialize", + }, + status_code=422, + ) + try: + claim_obj = ClaimOutput.model_validate(row.raw_json) + except Exception as exc: + return JSONResponse( + { + "error": "Unprocessable", + "detail": f"Claim {claim_id} raw_json is malformed: {exc}", + }, + status_code=422, + ) + + try: + text = serialize_837(claim_obj) + except SerializeError837 as exc: + return JSONResponse( + {"error": "Unprocessable", "detail": str(exc)}, + status_code=422, + ) + + return Response( + content=text, + media_type="text/x12", + headers={ + "Content-Disposition": f'attachment; filename="claim-{claim_id}.x12"' + }, + ) + + +@router.get("/api/claims/{claim_id}/line-reconciliation") +def get_claim_line_reconciliation(claim_id: str) -> dict: + """Per-line reconciliation view for the ClaimDrawer tab. + + Spec §5.1. Returns the 837 service lines and 835 SVC composites + side-by-side, with per-line CAS adjustments and a summary block. + + Architecture note: 837 service lines live in ``Claim.raw_json`` + (not a separate ORM table), so the 837-side rows are read from the + JSON blob; the 835-side rows come from ``ServiceLinePayment`` ORM. + ``LineReconciliation.claim_service_line_number`` stores the 1-based + line number to join them. + """ + from sqlalchemy import select + + from cyclone.db import ( + CasAdjustment, + LineReconciliation, + ServiceLinePayment, + ) + + with db.SessionLocal()() as s: + claim = s.get(db.Claim, claim_id) + if claim is None: + raise HTTPException( + status_code=404, + detail={"error": "Not found", "detail": f"Claim {claim_id} not found"}, + ) + + # 837 service lines: from raw_json. + raw = claim.raw_json or {} + claim_lines_raw = raw.get("service_lines") or [] + # Normalize to dicts for the response. + claim_lines = [_claim_line_dict(d) for d in claim_lines_raw] + + # 835 service payments: ORM rows from the matched remit. + remits = list( + s.execute( + select(db.Remittance).where(db.Remittance.claim_id == claim_id) + ).scalars().all() + ) + svc_payments: list[dict] = [] + svc_ids: list[int] = [] + if remits: + svc_rows = list( + s.execute( + select(ServiceLinePayment).where( + ServiceLinePayment.remittance_id.in_([r.id for r in remits]) + ).order_by(ServiceLinePayment.line_number) + ).scalars().all() + ) + for svc in svc_rows: + d = _svc_to_dict(svc) + svc_payments.append(d) + svc_ids.append(svc.id) + + # LineReconciliation rows. + lrs = list( + s.execute( + select(LineReconciliation).where(LineReconciliation.claim_id == claim_id) + ).scalars().all() + ) + # Index by claim_service_line_number and service_line_payment_id. + lr_by_claim_num: dict[int, LineReconciliation] = { + lr.claim_service_line_number: lr for lr in lrs if lr.claim_service_line_number is not None + } + lr_by_svc: dict[int, LineReconciliation] = { + lr.service_line_payment_id: lr for lr in lrs if lr.service_line_payment_id is not None + } + + # CAS rows grouped by svc id. + cas_by_svc: dict[int, list[CasAdjustment]] = {} + if svc_ids: + cas_rows = list( + s.execute( + select(CasAdjustment).where(CasAdjustment.service_line_payment_id.in_(svc_ids)) + ).scalars().all() + ) + for c in cas_rows: + cas_by_svc.setdefault(c.service_line_payment_id, []).append(c) + + # Build output lines array, preserving 837 order then 835-only. + svc_by_id: dict[int, dict] = {d["id"]: d for d in svc_payments} + lines_out: list[dict] = [] + billed_total = Decimal("0") + paid_total = Decimal("0") + adjustment_total = Decimal("0") + matched_count = 0 + used_svc_ids: set[int] = set() + + for cl in claim_lines: + billed_total += Decimal(str(cl["charge"])) + lr = lr_by_claim_num.get(cl["line_number"]) + if lr is None: + lines_out.append({ + "claim_service_line": cl, + "service_line_payment": None, + "status": "unmatched_837_only", + "adjustments": [], + }) + continue + svc_id = lr.service_line_payment_id + svc = svc_by_id.get(svc_id) if svc_id else None + if svc_id is not None: + used_svc_ids.add(svc_id) + cas_list = cas_by_svc.get(svc_id, []) if svc_id is not None else [] + cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0")) + if svc: + paid_total += Decimal(str(svc["payment"])) + adjustment_total += cas_total + if lr.status == "matched": + matched_count += 1 + lines_out.append({ + "claim_service_line": cl, + "service_line_payment": svc, + "status": lr.status, + "adjustments": [ + {"group_code": c.group_code, "reason_code": c.reason_code, + "amount": str(Decimal(str(c.amount)))} + for c in cas_list + ], + }) + + # 835-only lines (no claim match). + for lr in lrs: + if lr.claim_service_line_number is not None: + continue + svc_id = lr.service_line_payment_id + if svc_id is None: + continue + if svc_id in used_svc_ids: + continue + svc = svc_by_id.get(svc_id) + cas_list = cas_by_svc.get(svc_id, []) + cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0")) + if svc: + paid_total += Decimal(str(svc["payment"])) + adjustment_total += cas_total + lines_out.append({ + "claim_service_line": None, + "service_line_payment": svc, + "status": lr.status, + "adjustments": [ + {"group_code": c.group_code, "reason_code": c.reason_code, + "amount": str(Decimal(str(c.amount)))} + for c in cas_list + ], + }) + + return { + "claim_id": claim_id, + "summary": { + "billed_total": str(billed_total), + "paid_total": str(paid_total), + "adjustment_total": str(adjustment_total), + "matched_lines": matched_count, + "total_lines": len(claim_lines), + }, + "lines": lines_out, + } + + +def _claim_line_dict(d: dict) -> dict: + """Project an 837 service-line dict from ``Claim.raw_json`` to wire shape.""" + proc = d.get("procedure") or {} + charge = d.get("charge") + units = d.get("units") + return { + "line_number": d.get("line_number"), + "procedure_qualifier": proc.get("qualifier", "HC"), + "procedure_code": proc.get("code", ""), + "modifiers": proc.get("modifiers") or [], + "charge": str(Decimal(str(charge))) if charge is not None else "0", + "units": str(Decimal(str(units))) if units is not None else None, + "unit_type": d.get("unit_type"), + "service_date": d.get("service_date"), + } + + +def _svc_to_dict(svc) -> dict: + """Project an ORM ``ServiceLinePayment`` to wire shape.""" + import json as _json + return { + "id": svc.id, + "line_number": svc.line_number, + "procedure_qualifier": svc.procedure_qualifier, + "procedure_code": svc.procedure_code, + "modifiers": _json.loads(svc.modifiers_json or "[]"), + "charge": str(Decimal(str(svc.charge))), + "payment": str(Decimal(str(svc.payment))), + "units": str(Decimal(str(svc.units))) if svc.units is not None else None, + "unit_type": svc.unit_type, + "service_date": svc.service_date.isoformat() if svc.service_date else None, + } diff --git a/backend/src/cyclone/api_routers/clearhouse.py b/backend/src/cyclone/api_routers/clearhouse.py new file mode 100644 index 0000000..4ad72a1 --- /dev/null +++ b/backend/src/cyclone/api_routers/clearhouse.py @@ -0,0 +1,126 @@ +"""``/api/clearhouse`` — SP9 single-payer SFTP submit surface. + +Two endpoints: +- ``GET /api/clearhouse`` — return the singleton clearhouse config + (dzinesco's identity, SFTP block, filename block). 404 when the + config hasn't been seeded yet. +- ``POST /api/clearhouse/submit`` — submit a batch of claims to the + clearhouse (SFTP). SP9 ships a stub that copies to ``staging_dir`` + instead of opening a real SFTP connection; SP13 wires the real + paramiko-backed client behind the same ``make_client()`` factory. + +For each claim the endpoint serializes a fresh 837 via +:func:`_serialize_claim_for_submit`, builds an HCPF-compliant outbound +filename, and records an audit-log row on success. Failures are +captured per-claim so the operator sees exactly which claim_ids +didn't go through. +""" +from __future__ import annotations + +import json + +from fastapi import APIRouter, HTTPException + +from cyclone import db +from cyclone.audit_log import AuditEvent, append_event +from cyclone.parsers.parse_837 import parse +from cyclone.parsers.serialize_837 import serialize_837 +from cyclone.store import store + +router = APIRouter() + + +@router.get("/api/clearhouse") +def get_clearhouse(): + """Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block).""" + ch = store.get_clearhouse() + if ch is None: + raise HTTPException(status_code=404, detail="clearhouse not seeded") + return json.loads(ch.model_dump_json()) + + +@router.post("/api/clearhouse/submit") +def submit_to_clearhouse(body: dict): + """Submit a batch of claims to the clearhouse (SFTP). SP9: stub. + + Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}`` + + Stub behavior: serializes each claim via the SP7 serializer, builds + an HCPF-compliant outbound filename, and copies the result to + ``{staging_dir}/{outbound_path}/{filename}`` instead of opening a + real SFTP connection. Returns a receipt per claim. + """ + from cyclone.clearhouse import make_client + from cyclone.edi.filenames import build_outbound_filename + + claim_ids = body.get("claim_ids", []) + payer_id = body.get("payer_id") + if not claim_ids: + raise HTTPException(status_code=400, detail="claim_ids required") + if not payer_id: + raise HTTPException(status_code=400, detail="payer_id required") + + ch = store.get_clearhouse() + if ch is None: + raise HTTPException(status_code=500, detail="clearhouse not seeded") + + client = make_client(ch.sftp_block) + results = [] + for cid in claim_ids: + try: + x12_text = _serialize_claim_for_submit(cid) + except Exception as exc: # noqa: BLE001 + results.append({"claim_id": cid, "ok": False, "error": str(exc)}) + continue + filename = build_outbound_filename(ch.tpid, "837P") + remote = f"{ch.sftp_block.paths['outbound']}/{filename}" + staging_path = client.write_file(remote, x12_text.encode("utf-8")) + results.append({ + "claim_id": cid, + "ok": True, + "filename": filename, + "staging_path": str(staging_path), + "remote_path": remote, + }) + # SP11: audit trail for each successful clearhouse submission. + with db.SessionLocal()() as audit_s: + append_event(audit_s, AuditEvent( + event_type="clearhouse.submitted", + entity_type="claim", + entity_id=cid, + payload={ + "filename": filename, + "remote_path": remote, + "tpid": ch.tpid, + "stub": ch.sftp_block.stub, + }, + actor="clearhouse-submit", + )) + audit_s.commit() + return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub} + + +def _serialize_claim_for_submit(claim_id: str) -> str: + """Serialize a claim to X12 for SFTP submission.""" + with db.SessionLocal()() as s: + row = s.get(db.Claim, claim_id) + if row is None: + raise ValueError(f"claim {claim_id!r} not found") + raw = row.raw_json or {} + return _serialize_claim_from_raw(row, raw) + + +def _serialize_claim_from_raw(claim_row, raw: dict) -> str: + """Best-effort serializer that uses the stored raw_json to emit a fresh 837. + + For SP9 this delegates to the existing serialize_837 helper if the + claim has a complete raw_segments array. Otherwise it raises. + """ + # SP9 stub: if the claim has a `raw_json` with `x12_text`, use that. + if isinstance(raw, dict) and raw.get("x12_text"): + result = parse(raw["x12_text"]) + if result.claims: + return serialize_837(result.claims[0]) + raise RuntimeError( + f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text" + ) diff --git a/backend/src/cyclone/api_routers/config.py b/backend/src/cyclone/api_routers/config.py new file mode 100644 index 0000000..9ed5f9d --- /dev/null +++ b/backend/src/cyclone/api_routers/config.py @@ -0,0 +1,56 @@ +"""``/api/config/*`` — configured-provider / configured-payer lookups. + +The ``config`` prefix is the operator-facing read-only window into the +``config/payers.yaml`` static config and the seeded providers table. +Used by the frontend Settings page to show what's available. + +The payers endpoint at ``/api/config/payers/{payer_id}/configs`` is +slightly different: it returns both the YAML-loaded config and any +DB-overridden config (so a runtime override wins over the static +fallback). 404 / 200 / array shape match the rest of the API. +""" +from __future__ import annotations + +import json + +from fastapi import APIRouter, HTTPException, Query + +from cyclone import payers as payer_loader +from cyclone.store import store + +router = APIRouter() + + +@router.get("/api/config/providers") +def list_configured_providers(is_active: bool | None = Query(default=True)): + """List the configured provider rows (3 NPIs for SP9).""" + return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)] + + +@router.get("/api/config/providers/{npi}") +def get_configured_provider(npi: str): + p = store.get_provider(npi) + if p is None: + raise HTTPException(status_code=404, detail=f"provider {npi!r} not found") + return json.loads(p.model_dump_json()) + + +@router.get("/api/config/payers") +def list_configured_payers(is_active: bool | None = Query(default=True)): + return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)] + + +@router.get("/api/config/payers/{payer_id}/configs") +def list_payer_configs(payer_id: str): + """List all (transaction_type, config_json) blocks for a payer.""" + configs = [ + {"transaction_type": tx, "config_json": block, "source": "yaml"} + for (pid, tx), block in payer_loader.all_configs().items() + if pid == payer_id + ] + # Also check the DB for runtime-overridden configs + for tx in ("837P", "835", "277CA", "999", "TA1"): + live = store.get_payer_config(payer_id, tx) + if live is not None: + configs.append({"transaction_type": tx, "config_json": live, "source": "db"}) + return configs diff --git a/backend/src/cyclone/api_routers/eligibility.py b/backend/src/cyclone/api_routers/eligibility.py new file mode 100644 index 0000000..e75ad7d --- /dev/null +++ b/backend/src/cyclone/api_routers/eligibility.py @@ -0,0 +1,220 @@ +"""``/api/eligibility/*`` — 270/271 eligibility surface (SP3 P4 T23–T24). + +API-only — no DB persistence per spec section 3.4. The operator builds +a 270 inquiry from a small JSON body, downloads the raw text to paste +into a payer portal, then re-pastes the 271 response and parses it +back into structured fields on demand. + +Two endpoints: +* ``POST /api/eligibility/request`` — build a 270 inquiry from a JSON + body and return ``{raw_270_text, parsed}``. +* ``POST /api/eligibility/parse-271`` — accept a raw 271 file upload + and return the parsed ``coverage_benefits`` / ``subscriber`` / + ``summary`` / envelope triples. + +The ``_validate_eligibility_request`` helper is private to this router +— it builds the Pydantic ``ParseResult270`` and is only used by the +``/request`` endpoint. +""" +from __future__ import annotations + +import json +import logging +from datetime import date as _date +from typing import Any + +from fastapi import APIRouter, File, HTTPException, UploadFile +from fastapi.responses import JSONResponse + +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.models import BatchSummary, Envelope +from cyclone.parsers.models_270 import ( + EligibilityBenefitInquiry, + InformationReceiver270, + InformationSource270, + ParseResult270, + Subscriber270, +) +from cyclone.parsers.parse_271 import parse as parse_271_text +from cyclone.parsers.serialize_270 import serialize_270 + +log = logging.getLogger(__name__) + +router = APIRouter() + + +def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]: + """Build a :class:`ParseResult270` from a request body dict. + + The body shape is the minimum surface needed to build a valid 270 + inquiry (per spec section 3.4 — operator-driven, ephemeral): + + :: + + { + "subscriber": {first_name, last_name, member_id, dob}, + "provider": {npi, name}, + "payer": {id, name}, + "service_type_code": "1" + } + + Returns ``(ParseResult270, service_type_code)``. Raises + :class:`HTTPException` (400) when the body is missing required + fields. + """ + subscriber_in = body.get("subscriber") or {} + provider_in = body.get("provider") or {} + payer_in = body.get("payer") or {} + service_type_code = (body.get("service_type_code") or "").strip() + + # Required-field checks. We surface a single 400 with the first + # missing field name to match the rest of the API's error contract. + if not service_type_code: + raise HTTPException( + status_code=400, + detail={"error": "Bad request", "detail": "service_type_code is required"}, + ) + if not subscriber_in.get("member_id"): + raise HTTPException( + status_code=400, + detail={"error": "Bad request", "detail": "subscriber.member_id is required"}, + ) + if not provider_in.get("npi"): + raise HTTPException( + status_code=400, + detail={"error": "Bad request", "detail": "provider.npi is required"}, + ) + if not payer_in.get("name"): + raise HTTPException( + status_code=400, + detail={"error": "Bad request", "detail": "payer.name is required"}, + ) + + subscriber_dob_raw = subscriber_in.get("dob") + subscriber_dob: _date | None = None + if subscriber_dob_raw: + try: + subscriber_dob = _date.fromisoformat(subscriber_dob_raw) + except (TypeError, ValueError) as exc: + raise HTTPException( + status_code=400, + detail={ + "error": "Bad request", + "detail": f"subscriber.dob must be YYYY-MM-DD: {exc}", + }, + ) from exc + + result = ParseResult270( + envelope=Envelope( + sender_id="SUBMITTERID", + receiver_id=str(payer_in.get("id") or "RECEIVERID"), + control_number="000000001", + transaction_date=_date.today(), + implementation_guide="005010X279A1", + ), + information_source=InformationSource270( + name=str(payer_in["name"]), + id=str(payer_in.get("id") or "") or None, + ), + information_receiver=InformationReceiver270( + name=str(provider_in.get("name") or ""), + npi=str(provider_in["npi"]), + ), + subscriber=Subscriber270( + member_id=str(subscriber_in["member_id"]), + first_name=str(subscriber_in.get("first_name") or "") or None, + last_name=str(subscriber_in.get("last_name") or "") or None, + dob=subscriber_dob, + ), + inquiries=[EligibilityBenefitInquiry(service_type_code=service_type_code)], + summary=BatchSummary( + input_file="eligibility_request", + control_number="000000001", + transaction_date=_date.today(), + total_claims=1, + passed=1, + failed=0, + ), + ) + return result, service_type_code + + +@router.post("/api/eligibility/request") +def post_eligibility_request(body: dict) -> Any: + """Build a 270 eligibility inquiry from a small JSON body. + + Returns ``{"raw_270_text": , "parsed": }`` + so the operator can either download the raw text (paste into a + payer portal) or render the parsed fields directly. Per spec + section 3.4, nothing is persisted to the DB. + """ + try: + result, _ = _validate_eligibility_request(body) + except HTTPException: + raise + except (KeyError, TypeError, ValueError) as exc: + raise HTTPException( + status_code=400, + detail={"error": "Bad request", "detail": f"Malformed body: {exc}"}, + ) from exc + + raw_270_text = serialize_270(result) + return { + "raw_270_text": raw_270_text, + "parsed": json.loads(result.model_dump_json()), + } + + +@router.post("/api/eligibility/parse-271") +async def post_eligibility_parse_271( + file: UploadFile = File(...), +) -> Any: + """Parse a 271 eligibility response and return the structured summary. + + Accepts the raw 271 text as a file upload (multipart/form-data), + mirrors the ``/api/parse-999`` contract. Per spec section 3.4 the + result is NOT persisted — the operator re-pastes the 271 each + time they need a fresh read. + + The response body is a JSON object with three top-level keys: + ``coverage_benefits``, ``subscriber``, and ``summary``. 400 is + returned on empty / undecodable / malformed EDI; 200 on success. + """ + raw = await file.read() + if not raw: + return JSONResponse( + status_code=400, + content={"error": "Empty file", "detail": "Uploaded file contained no bytes."}, + ) + try: + text = raw.decode("utf-8") + except UnicodeDecodeError as exc: + return JSONResponse( + status_code=400, + content={"error": "Encoding error", "detail": str(exc)}, + ) + + try: + result = parse_271_text(text, input_file=file.filename or "") + except CycloneParseError as exc: + return JSONResponse( + status_code=400, + content={"error": "Parse error", "detail": str(exc)}, + ) + except Exception as exc: # pragma: no cover - safety net + log.exception("Unexpected parser failure on 271") + return JSONResponse( + status_code=500, + content={"error": "Internal server error", "detail": str(exc)}, + ) + + return { + "coverage_benefits": [ + json.loads(cb.model_dump_json()) for cb in result.coverage_benefits + ], + "subscriber": json.loads(result.subscriber.model_dump_json()), + "summary": json.loads(result.summary.model_dump_json()), + "envelope": json.loads(result.envelope.model_dump_json()), + "information_source": json.loads(result.information_source.model_dump_json()), + "information_receiver": json.loads(result.information_receiver.model_dump_json()), + } diff --git a/backend/src/cyclone/api_routers/inbox.py b/backend/src/cyclone/api_routers/inbox.py new file mode 100644 index 0000000..2dd2c7e --- /dev/null +++ b/backend/src/cyclone/api_routers/inbox.py @@ -0,0 +1,309 @@ +"""``/api/inbox/*`` — operator-facing inbox surface. + +Six endpoints powering the Inbox page (lanes, candidates, payer-rejected +acknowledge, rejected resubmit, CSV export). All endpoints run inside a +``db.SessionLocal()`` scope; the lanes / dismiss endpoints also read +``app.state.dismissed_pairs`` (a session-scoped set so dismissed items +don't re-appear until process restart). + +Endpoints: +* ``GET /api/inbox/lanes`` — return all four lanes in one call + (rejected / payer_rejected / candidates / unmatched / done_today). +* ``POST /api/inbox/candidates/{remit_id}/match`` — manually link a + remit to a claim from the candidates lane. +* ``POST /api/inbox/candidates/dismiss`` — add candidate pairs to the + dismissed set so they stop appearing in the candidates lane. +* ``POST /api/inbox/payer-rejected/acknowledge`` — clear SP14 + payer-rejected claims from the working surface without deleting the + rejection (idempotent: re-acknowledging is a noop). +* ``POST /api/inbox/rejected/resubmit`` — bulk move REJECTED claims + back to SUBMITTED; with ``?download=true`` returns a ZIP of + regenerated 837s with unique interchange control numbers. +* ``GET /api/inbox/export.csv?lane=…`` — stream a CSV for a single lane. +""" +from __future__ import annotations + +import csv +import io +import json +import zipfile +from datetime import datetime, timezone + +from fastapi import APIRouter, HTTPException, Query, Request, Response +from fastapi.responses import StreamingResponse + +from cyclone import db +from cyclone.audit_log import AuditEvent, append_event +from cyclone.db import Claim, ClaimState, Remittance +from cyclone.parsers.models import ClaimOutput +from cyclone.parsers.serialize_837 import ( + SerializeError as SerializeError837, + serialize_837_for_resubmit, +) + +router = APIRouter() + + +@router.get("/api/inbox/lanes") +def inbox_lanes(request: Request) -> dict: + """Return all Inbox lanes in one call. + + Reads the session-scoped ``dismissed_pairs`` set so the + candidates lane excludes pairs the operator already dismissed in + this process lifetime. + """ + dismissed_pairs = getattr(request.app.state, "dismissed_pairs", set()) + with db.SessionLocal()() as session: + from cyclone.inbox_lanes import compute_lanes + lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs) + return { + "rejected": lanes.rejected, + # SP10: payer-rejected lane (277CA STC A4/A6/A7). Distinct from + # the 999 envelope rejection in ``rejected`` above. + "payer_rejected": lanes.payer_rejected, + "candidates": lanes.candidates, + "unmatched": lanes.unmatched, + "done_today": lanes.done_today, + } + + +@router.post("/api/inbox/candidates/{remit_id}/match") +def inbox_match_candidate(remit_id: str, body: dict): + """Manually link a remit to a claim.""" + claim_id = body.get("claim_id") + if not claim_id: + raise HTTPException(400, "claim_id required") + with db.SessionLocal()() as s: + claim = s.get(Claim, claim_id) + remit = s.get(Remittance, remit_id) + if claim is None or remit is None: + raise HTTPException(404, "claim or remit not found") + if claim.matched_remittance_id and claim.matched_remittance_id != remit_id: + raise HTTPException( + 409, + detail={ + "error": "claim_already_matched", + "current_state": ( + claim.state.value if hasattr(claim.state, "value") + else str(claim.state) + ), + "matched_remittance_id": claim.matched_remittance_id, + }, + ) + claim.matched_remittance_id = remit_id + remit.claim_id = claim_id + s.commit() + return {"ok": True, "claim_id": claim_id, "remit_id": remit_id} + + +@router.post("/api/inbox/candidates/dismiss") +def inbox_dismiss_candidates(request: Request, body: dict): + """Add candidate pairs to the session-scoped dismissed set.""" + pairs = body.get("pairs") or [] + if not hasattr(request.app.state, "dismissed_pairs"): + request.app.state.dismissed_pairs = set() + for p in pairs: + cid = p.get("claim_id") + rid = p.get("remit_id") + if cid and rid: + request.app.state.dismissed_pairs.add(frozenset({cid, rid})) + return {"ok": True, "dismissed_count": len(pairs)} + + +@router.post("/api/inbox/payer-rejected/acknowledge") +def inbox_acknowledge_payer_rejected(body: dict): + """Mark Payer-Rejected claims as acknowledged by the operator. + + Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear + the claim from the working surface. We don't delete the rejection + (the original payer_rejected_* fields stay for SP11 audit), we just + set ``payer_rejected_acknowledged_at`` so the lane query filters it + out. + + Idempotent: re-acknowledging an already-acknowledged claim is a noop + (the timestamp is not bumped). Returns the count actually transitioned + so the UI can show "3 of 5 were already acknowledged". + """ + claim_ids = body.get("claim_ids") or [] + actor = body.get("actor") or "operator" + if not isinstance(claim_ids, list) or not claim_ids: + raise HTTPException(400, "claim_ids must be a non-empty list") + if not all(isinstance(c, str) for c in claim_ids): + raise HTTPException(400, "claim_ids must be a list of strings") + + with db.SessionLocal()() as session: + now = datetime.now(timezone.utc) + transitioned = 0 + already_acked = 0 + not_found = 0 + not_rejected = 0 + for cid in claim_ids: + claim = session.get(Claim, cid) + if claim is None: + not_found += 1 + continue + if claim.payer_rejected_at is None: + not_rejected += 1 + continue + if claim.payer_rejected_acknowledged_at is not None: + already_acked += 1 + continue + claim.payer_rejected_acknowledged_at = now + claim.payer_rejected_acknowledged_actor = actor + transitioned += 1 + # SP11: audit event for the acknowledge action. Best-effort: + # an audit-log failure must not block the operator's action. + try: + append_event(session, AuditEvent( + event_type="claim.payer_rejected_acknowledged", + entity_type="claim", + entity_id=claim.id, + actor=actor, + payload={ + "payer_rejected_status_code": claim.payer_rejected_status_code, + "payer_rejected_by_277ca_id": claim.payer_rejected_by_277ca_id, + }, + )) + except Exception: # noqa: BLE001 + pass + if transitioned: + session.commit() + + return { + "ok": True, + "transitioned": transitioned, + "already_acked": already_acked, + "not_found": not_found, + "not_rejected": not_rejected, + } + + +@router.post("/api/inbox/rejected/resubmit") +def inbox_resubmit_rejected( + body: dict, + download: bool = Query(False, description="When true, return a ZIP of regenerated 837 files for the resubmitted claims (instead of JSON)."), +): + """Bulk move REJECTED claims back to SUBMITTED. + + With ``?download=true``, the response is a ``application/zip`` archive + containing one ``claim-{id}.x12`` per successfully resubmitted claim + (regenerated via ``serialize_837_for_resubmit`` so each file gets a + unique interchange/group control number). Conflicts are omitted from + the ZIP — they remain visible to the caller via the JSON shape of the + non-download path. Empty resubmit + download → 200 with an empty zip + so the UI can still hand the user a downloadable artifact. + """ + ids = body.get("claim_ids") or [] + if not ids: + raise HTTPException(400, "claim_ids required") + accepted: list[str] = [] + conflicts: list[dict] = [] + # Track which claims are about to be resubmitted (and their index in + # the bundle) so the download path can serialize them with unique + # control numbers — back-to-back resubmits in the same file would + # otherwise all share ISA13/GS06 = "000000001". + accepted_with_rows: list[tuple[str, Claim]] = [] + with db.SessionLocal()() as s: + for cid in ids: + c = s.get(Claim, cid) + if c is None: + continue + if c.state != ClaimState.REJECTED: + conflicts.append({ + "claim_id": cid, + "current_state": ( + c.state.value if hasattr(c.state, "value") + else str(c.state) + ), + }) + continue + c.state = ClaimState.SUBMITTED + c.state_changed_at = datetime.now(timezone.utc) + c.rejection_reason = None + c.rejected_at = None + c.resubmit_count = (c.resubmit_count or 0) + 1 + accepted.append(cid) + accepted_with_rows.append((cid, c)) + s.commit() + + if not download: + return {"ok": True, "resubmitted": accepted, "conflicts": conflicts} + + # Build a ZIP of regenerated 837s for the accepted claims. Conflicts + # and missing ids are deliberately excluded — the user already saw + # them in the JSON response on prior actions; the download is the + # "give me the files I asked for" payload. + buf = io.BytesIO() + serialize_errors: list[dict] = [] + with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf: + for idx, (cid, c) in enumerate(accepted_with_rows, start=1): + if not c.raw_json: + serialize_errors.append({"claim_id": cid, "reason": "no raw_json"}) + continue + try: + claim_obj = ClaimOutput.model_validate(c.raw_json) + except Exception as exc: + serialize_errors.append({"claim_id": cid, "reason": f"raw_json invalid: {exc}"}) + continue + try: + text = serialize_837_for_resubmit(claim_obj, interchange_index=idx) + except SerializeError837 as exc: + serialize_errors.append({"claim_id": cid, "reason": str(exc)}) + continue + zf.writestr(f"claim-{cid}.x12", text) + buf.seek(0) + headers = { + "Content-Disposition": ( + f'attachment; filename="resubmit-{len(accepted)}-claims.zip"' + ), + } + # Surface per-claim serialization failures as a custom response header + # so the UI can show "10 resubmitted, 2 couldn't be regenerated" without + # parsing the binary. The header value is JSON-encoded; the UI is + # expected to JSON.parse it after a fetch with response.ok. + if serialize_errors: + headers["X-Cyclone-Serialize-Errors"] = json.dumps(serialize_errors) + return Response( + content=buf.getvalue(), + media_type="application/zip", + headers=headers, + ) + + +@router.get("/api/inbox/export.csv") +def inbox_export_csv(request: Request, lane: str): + """Stream a CSV for a single lane.""" + if lane not in {"rejected", "candidates", "unmatched", "done_today"}: + raise HTTPException(400, f"unknown lane: {lane}") + dismissed_pairs = getattr(request.app.state, "dismissed_pairs", set()) + with db.SessionLocal()() as session: + from cyclone.inbox_lanes import compute_lanes + lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs) + rows = getattr(lanes, lane) + + buf = io.StringIO() + writer = csv.writer(buf) + writer.writerow([ + "id", "kind", "patient_control_number", "charge_amount", + "payer_id", "provider_npi", "state", "rejection_reason", + "service_date", "score", + ]) + for r in rows: + writer.writerow([ + r.get("id") or r.get("payer_claim_control_number"), + r.get("kind"), + r.get("patient_control_number"), + r.get("charge_amount"), + r.get("payer_id"), + r.get("provider_npi") or r.get("rendering_provider_npi"), + r.get("state"), + r.get("rejection_reason"), + r.get("service_date_from") or r.get("service_date"), + r.get("score"), + ]) + buf.seek(0) + return StreamingResponse( + iter([buf.getvalue()]), + media_type="text/csv", + headers={"Content-Disposition": f'attachment; filename="inbox-{lane}.csv"'}, + ) diff --git a/backend/src/cyclone/api_routers/providers.py b/backend/src/cyclone/api_routers/providers.py new file mode 100644 index 0000000..62eea9a --- /dev/null +++ b/backend/src/cyclone/api_routers/providers.py @@ -0,0 +1,49 @@ +"""``GET /api/providers`` — distinct providers observed across claims. + +The list is built by ``store.distinct_providers()`` which scans claim +rows and returns one entry per unique ``billing_provider_npi``. The +``npi`` and ``state`` query params filter client-side. ``limit`` / +``offset`` paginate. Accepts ``application/x-ndjson`` like the other +list endpoints — see ``api_helpers.ndjson_stream_list``. +""" +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Query, Request +from fastapi.responses import StreamingResponse + +from cyclone.api_helpers import ndjson_stream_list, wants_ndjson +from cyclone.store import store + +router = APIRouter() + + +@router.get("/api/providers") +def list_providers( + request: Request, + npi: str | None = Query(None), + state: str | None = Query(None), + limit: int = Query(100, ge=1, le=1000), + offset: int = Query(0, ge=0), +) -> Any: + items = store.distinct_providers() + if npi is not None: + items = [p for p in items if p["npi"] == npi] + if state is not None: + items = [p for p in items if p.get("state") == state] + paged = items[offset:offset + limit] + total = len(items) + returned = len(paged) + has_more = total > offset + returned + if wants_ndjson(request): + return StreamingResponse( + ndjson_stream_list(paged, total, returned, has_more), + media_type="application/x-ndjson", + ) + return { + "items": paged, + "total": total, + "returned": returned, + "has_more": has_more, + } diff --git a/backend/src/cyclone/api_routers/reconciliation.py b/backend/src/cyclone/api_routers/reconciliation.py new file mode 100644 index 0000000..e069be5 --- /dev/null +++ b/backend/src/cyclone/api_routers/reconciliation.py @@ -0,0 +1,164 @@ +"""``/api/reconciliation/*`` + ``/api/batch-diff`` — operator review surface. + +Four endpoints powering the reconciliation page: + +* ``GET /api/reconciliation/unmatched`` — return unmatched Claims + (left) and unmatched Remittances (right). +* ``POST /api/reconciliation/match`` — manually pair a Claim with a + Remittance (operator override). +* ``POST /api/reconciliation/unmatch`` — remove the current match and + reset the Claim to submitted. +* ``GET /api/batch-diff`` — side-by-side diff of two batches (SP3 P4 / + T18). + +The two POST endpoints delegate to ``store.manual_match`` / +``store.manual_unmatch``; the router is responsible for translating the +store's exception hierarchy (``AlreadyMatchedError``, ``InvalidStateError``, +``NotMatchedError``, ``LookupError``) into the HTTP error contract. +""" +from __future__ import annotations + +from fastapi import APIRouter, HTTPException, Query + +from cyclone.store import ( + AlreadyMatchedError, + InvalidStateError, + store, +) + +router = APIRouter() + + +@router.get("/api/reconciliation/unmatched") +def get_reconciliation_unmatched() -> dict: + """Return unmatched Claims (left) and unmatched Remittances (right). + + Powers the reconciliation review surface: every Claim with no + paired Remittance appears on the left, every Remittance with no + paired Claim appears on the right. The two lists are always present + (empty list, never absent) so the UI can index unconditionally. + """ + return store.list_unmatched(kind="both") + + +@router.get("/api/batch-diff") +def get_batch_diff( + a: str | None = Query(None), + b: str | None = Query(None), +) -> dict: + """Return a side-by-side diff of two batches identified by id. + + Query params: ``a=``, ``b=`` (both required). + + Response body (snake_case keys, see :mod:`cyclone.batch_diff` for the + projector shapes): + - ``a`` / ``b`` — small metadata blocks (id, kind, parsedAt, + inputFilename, claimCount) + - ``added`` — claims present in B but not A + - ``removed`` — claims present in A but not B + - ``changed`` — claims present in both, with field deltas + - ``summary`` — precomputed counts + + Errors: + - 400 — missing ``a`` or ``b`` + - 404 — either batch id is unknown + + Pure read endpoint — never mutates the store. Both 837P and 835 + batches are accepted (mixed-kind diffs are valid: comparing the + submitted claims against the matching remittances). + """ + if not a or not b: + raise HTTPException( + status_code=400, + detail={"error": "Missing param", "detail": "Both ?a= and ?b= are required."}, + ) + try: + a_rec, b_rec = store.load_two_for_diff(a, b) + except LookupError as exc: + raise HTTPException( + status_code=404, + detail={"error": "Not found", "detail": str(exc)}, + ) + + # Lazy import — keeps the module's import surface small until the + # endpoint is actually hit. + from cyclone.batch_diff import diff_batches_to_wire + + return diff_batches_to_wire(a_rec, b_rec) + + +@router.post("/api/reconciliation/match") +def post_reconciliation_match(body: dict) -> dict: + """Manually pair a Claim with a Remittance (operator override). + + Body: ``{"claim_id": ..., "remit_id": ...}``. Returns + ``{"claim": , "match": }`` on success. Errors: + - 400: missing ``claim_id`` or ``remit_id`` + - 404: claim or remittance not found + - 409: claim already matched, or apply_* returned a noop + (claim in terminal state) — detail echoes ``current_state`` + and ``activity_kind`` so the UI can render a precise message. + """ + claim_id = body.get("claim_id") + remit_id = body.get("remit_id") + if not claim_id or not remit_id: + raise HTTPException( + status_code=400, + detail="claim_id and remit_id required", + ) + try: + return store.manual_match(claim_id, remit_id) + except AlreadyMatchedError as e: + raise HTTPException( + status_code=409, + detail={"error": "already_matched", "message": str(e)}, + ) + except InvalidStateError as e: + raise HTTPException( + status_code=409, + detail={ + "error": "invalid_state", + "current_state": e.current_state, + "activity_kind": e.activity_kind, + }, + ) + except LookupError: + # manual_match raises LookupError when the claim or remittance + # row is missing (we catch the parent class so any future + # KeyError subclasses in the store get the same treatment). + raise HTTPException( + status_code=404, + detail="claim_or_remit_not_found", + ) + + +@router.post("/api/reconciliation/unmatch") +def post_reconciliation_unmatch(body: dict) -> dict: + """Remove the current match for a Claim; reset Claim to submitted. + + Body: ``{"claim_id": ...}``. Returns + ``{"claim": , "deletedMatches": }``. Errors: + - 400: missing ``claim_id`` + - 404: claim not found + - 409: claim has no current match (NotMatchedError is mapped + by the store; we surface 409 to match the manual_match contract) + """ + from cyclone.store import NotMatchedError + claim_id = body.get("claim_id") + if not claim_id: + raise HTTPException( + status_code=400, + detail="claim_id required", + ) + try: + return store.manual_unmatch(claim_id) + except NotMatchedError as e: + raise HTTPException( + status_code=409, + detail={"error": "not_matched", "message": str(e)}, + ) + except LookupError: + raise HTTPException( + status_code=404, + detail="claim_not_found", + ) diff --git a/backend/src/cyclone/api_routers/remittances.py b/backend/src/cyclone/api_routers/remittances.py new file mode 100644 index 0000000..ffd6a1a --- /dev/null +++ b/backend/src/cyclone/api_routers/remittances.py @@ -0,0 +1,130 @@ +"""``/api/remittances`` — list, live-tail, detail endpoints for 835 ERAs. + +Mirror of the ``claims`` router for the 835 / Remittance resource: + +* ``GET /api/remittances`` — paginated list with filters + (batch_id / payer / claim_id / date range / sort). +* ``GET /api/remittances/stream`` — NDJSON snapshot + live tail via the + ``remittance_written`` EventBus kind. +* ``GET /api/remittances/{remittance_id}`` — detail with the labeled + CAS ``adjustments`` array. + +**Route ordering.** ``/stream`` is declared **before** ``/{remittance_id}`` +so FastAPI's first-match routing doesn't swallow the literal ``stream`` +segment as a remittance id. Same order as the prior inline code. +""" +from __future__ import annotations + +from typing import Any, AsyncIterator + +from fastapi import APIRouter, HTTPException, Query, Request +from fastapi.responses import StreamingResponse + +from cyclone.api_helpers import ( + ndjson_line, + ndjson_stream_list, + tail_events, + wants_ndjson, +) +from cyclone.pubsub import EventBus +from cyclone.store import store + +router = APIRouter() + + +@router.get("/api/remittances") +def list_remittances_endpoint( + request: Request, + batch_id: str | None = Query(None), + payer: str | None = Query(None), + claim_id: str | None = Query(None), + date_from: str | None = Query(None), + date_to: str | None = Query(None), + sort: str | None = Query(None), + order: str = Query("desc"), + limit: int = Query(100, ge=1, le=1000), + offset: int = Query(0, ge=0), +) -> Any: + """Paginated remittances list with filters; supports NDJSON streaming. + + ``has_more`` is computed as ``total > offset + returned`` so paginating + forward keeps returning true until the last page is reached. + """ + common = dict( + batch_id=batch_id, + payer=payer, + claim_id=claim_id, + date_from=date_from, + date_to=date_to, + ) + items = list(store.iter_remittances( + sort=sort, order=order, limit=limit, offset=offset, **common, + )) + total = len(list(store.iter_remittances(**common))) + returned = len(items) + has_more = total > offset + returned + if wants_ndjson(request): + return StreamingResponse( + ndjson_stream_list(items, total, returned, has_more), + media_type="application/x-ndjson", + ) + return { + "items": items, + "total": total, + "returned": returned, + "has_more": has_more, + } + + +@router.get("/api/remittances/stream") +async def remittances_stream_endpoint( + request: Request, + payer: str | None = Query(None), + claim_id: str | None = Query(None), + date_from: str | None = Query(None), + date_to: str | None = Query(None), + sort: str | None = Query(None), + order: str = Query("desc"), + limit: int = Query(100, ge=1, le=1000), +) -> StreamingResponse: + """Stream Remittances as NDJSON: snapshot first, then live events. + + Subscribes to ``remittance_written``. Default sort is + ``-received_date`` (newest-first), matching the list endpoint's + most common sort. + + NOTE: registered before ``/api/remittances/{remittance_id}`` so + the literal ``stream`` path segment doesn't get matched as a + remittance id. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + rows = store.iter_remittances( + payer=payer, claim_id=claim_id, + date_from=date_from, date_to=date_to, + sort=sort or "-received_date", order=order, limit=limit, + ) + for row in rows: + yield ndjson_line({"type": "item", "data": row}) + yield ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) + + async for chunk in tail_events(request, bus, ["remittance_written"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") + + +@router.get("/api/remittances/{remittance_id}") +def get_remittance_endpoint(remittance_id: str) -> dict: + """Return one remittance with its labeled CAS ``adjustments`` array. + + Returns 404 when the remittance is missing — never 500. + """ + body = store.get_remittance(remittance_id) + if body is None: + raise HTTPException( + status_code=404, + detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"}, + ) + return body