diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index cac5e4c..dd6253c 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -31,16 +31,9 @@ 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, Envelope -from cyclone.parsers.models_270 import ( - EligibilityBenefitInquiry, - InformationReceiver270, - InformationSource270, - ParseResult270, - Subscriber270, -) from cyclone.parsers.models_271 import ParseResult271 from cyclone.parsers.models_835 import ParseResult835 from cyclone.parsers.payer import PayerConfig, PayerConfig835 @@ -51,7 +44,6 @@ 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.batch_ack_builder import build_ack_for_batch @@ -142,10 +134,12 @@ app.add_middleware( from cyclone.api_routers import ( # noqa: E402 acks, activity, + admin, batches, claims, clearhouse, config, + eligibility, health, inbox, providers, @@ -166,6 +160,8 @@ 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 @@ -175,6 +171,24 @@ 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: if name not in PAYER_FACTORIES: @@ -926,424 +940,10 @@ def _serialize_ta1_from_row(row: db.Ta1Ack) -> str: # --------------------------------------------------------------------------- # -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()), - } - - - - - -# --------------------------------------------------------------------------- # -# 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.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/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/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()), + }