diff --git a/backend/src/cyclone/migrations/0022_submission_dedup.sql b/backend/src/cyclone/migrations/0022_submission_dedup.sql new file mode 100644 index 0000000..53a740c --- /dev/null +++ b/backend/src/cyclone/migrations/0022_submission_dedup.sql @@ -0,0 +1,15 @@ +-- version: 22 +-- SP41: claim-id dedup at SFTP pre-flight. +-- +-- The 30-day dedup window is enforced by cyclone.store.submission_dedup. +-- This table records (claim_id, submitted_at) for every push that passed +-- the pre-flight check. Past the window, the prior record is ignored +-- (the guard lets through re-submits older than DEFAULT_WINDOW_DAYS). + +CREATE TABLE submission_dedup ( + claim_id TEXT PRIMARY KEY, + submitted_at DATETIME NOT NULL +); + +CREATE INDEX submission_dedup_submitted_at_idx + ON submission_dedup (submitted_at); diff --git a/backend/src/cyclone/store/submission_dedup.py b/backend/src/cyclone/store/submission_dedup.py new file mode 100644 index 0000000..95b9c4b --- /dev/null +++ b/backend/src/cyclone/store/submission_dedup.py @@ -0,0 +1,74 @@ +"""Claim-id dedup at the SFTP pre-flight stage. + +Schema: a ``submission_dedup`` table records (claim_id, submitted_at). A +re-submission of the same claim_id within ``DEFAULT_WINDOW_DAYS`` (30) +is rejected with ``DuplicateClaimError``. Past the window, the guard +lets it through — the operator may be retrying a 999-rejected file +from >30 days ago, which is a legitimate rebuild path. +""" +from __future__ import annotations + +from datetime import datetime, timedelta + +from sqlalchemy import Column, DateTime, String, create_engine, insert, select +from sqlalchemy.orm import declarative_base + +DEFAULT_WINDOW_DAYS = 30 + +_Base = declarative_base() + + +class _SubmissionRecord(_Base): + __tablename__ = "submission_dedup" + claim_id = Column(String(64), primary_key=True) + submitted_at = Column(DateTime, nullable=False, index=True) + + +class DuplicateClaimError(Exception): + def __init__(self, claim_id: str, original_submission_at: datetime): + super().__init__( + f"claim_id {claim_id!r} was already submitted at " + f"{original_submission_at.isoformat()}; within {DEFAULT_WINDOW_DAYS}-day window" + ) + self.claim_id = claim_id + self.original_submission_at = original_submission_at + + +def _engine(db_url: str): + return create_engine(db_url, future=True) + + +def init_submission_dedup(db_url: str) -> None: + """Create the table if it doesn't exist. Idempotent.""" + _Base.metadata.create_all(_engine(db_url)) + + +def record_submission(claim_id: str, submitted_at: datetime, db_url: str) -> None: + init_submission_dedup(db_url) + eng = _engine(db_url) + with eng.begin() as conn: + conn.execute(insert(_SubmissionRecord).values( + claim_id=claim_id, submitted_at=submitted_at, + ).prefix_with("OR REPLACE")) + + +def check_duplicate( + claim_id: str, + db_url: str, + now: datetime | None = None, + window_days: int = DEFAULT_WINDOW_DAYS, +) -> None: + """Raise DuplicateClaimError if claim_id was submitted within the window.""" + init_submission_dedup(db_url) + if now is None: + now = datetime.utcnow() + cutoff = now - timedelta(days=window_days) + eng = _engine(db_url) + with eng.connect() as conn: + stmt = select(_SubmissionRecord).where( + _SubmissionRecord.claim_id == claim_id, + _SubmissionRecord.submitted_at >= cutoff, + ) + row = conn.execute(stmt).first() + if row is not None: + raise DuplicateClaimError(claim_id, row.submitted_at) diff --git a/backend/tests/test_acks.py b/backend/tests/test_acks.py index 5eae3dd..9c740d3 100644 --- a/backend/tests/test_acks.py +++ b/backend/tests/test_acks.py @@ -51,7 +51,7 @@ def test_migration_0002_creates_acks_table(): def test_migration_latest_idempotent_on_fresh_db(): """Re-running the migration on the same DB must be a no-op (PRAGMA - user_version already at the latest version — currently 21 after + user_version already at the latest version — currently 22 after 0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007 providers/payers/clearhouse, SP10's 0008 payer_rejected, SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged, @@ -62,16 +62,16 @@ def test_migration_latest_idempotent_on_fresh_db(): claim.patient_control_number backfill UPDATE, SP28's 0018 claim_acks join table, SP32's 0019 rendering_provider_npi + service_provider_npi, SP37's 0020 - transaction_set_control_number, and SP39's 0021 resubmissions - audit table).""" + transaction_set_control_number, SP39's 0021 resubmissions + audit table, and SP41's 0022 submission_dedup).""" with db.engine().begin() as c: v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v1 == 21 + assert v1 == 22 # A second run should not raise and should not bump the version. db_migrate.run(db.engine()) with db.engine().begin() as c: v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v2 == 21 + assert v2 == 22 def test_add_ack_persists_row(): diff --git a/backend/tests/test_db_migrate.py b/backend/tests/test_db_migrate.py index 931debf..bc96ca2 100644 --- a/backend/tests/test_db_migrate.py +++ b/backend/tests/test_db_migrate.py @@ -128,14 +128,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None: service_provider_npi on claims and remittances. SP37 bumped it to 20 with batch transaction_set_control_number. SP39 bumped it to 21 with the resubmissions audit table. + SP41 bumped it to 22 with submission_dedup. """ engine = _fresh_engine(tmp_path) db_migrate.run(engine) v_after_first = _user_version(engine) - assert v_after_first == 21, f"expected head=21, got {v_after_first}" + assert v_after_first == 22, f"expected head=22, got {v_after_first}" db_migrate.run(engine) - assert _user_version(engine) == 21, "second run should not bump version" + assert _user_version(engine) == 22, "second run should not bump version" def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -161,7 +162,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py engine = _fresh_engine(tmp_path) db_migrate.run(engine) - assert _user_version(engine) == 21, f"expected head=21, got {_user_version(engine)}" + assert _user_version(engine) == 22, f"expected head=22, got {_user_version(engine)}" # Two claims in one batch with the same patient_control_number # must be insertable. If 0015's table recreation re-introduced a diff --git a/backend/tests/test_migration_0020.py b/backend/tests/test_migration_0020.py index b001a02..e6f4415 100644 --- a/backend/tests/test_migration_0020.py +++ b/backend/tests/test_migration_0020.py @@ -102,10 +102,10 @@ def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): engine = _fresh_engine(tmp_path / "mig0020.db") db_migrate.run(engine) - # Confirm head is 21 (every migration applied, including SP39's 0021). + # Confirm head is 22 (every migration applied, including SP41's 0022). with engine.connect() as conn: v = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v == 21, f"expected migration head=21, got {v}" + assert v == 22, f"expected migration head=22, got {v}" yield engine engine.dispose() diff --git a/backend/tests/test_submission_dedup.py b/backend/tests/test_submission_dedup.py new file mode 100644 index 0000000..8b5ad3c --- /dev/null +++ b/backend/tests/test_submission_dedup.py @@ -0,0 +1,38 @@ +"""Claim-id dedup at the SFTP pre-flight stage.""" +from datetime import datetime +from cyclone.store.submission_dedup import ( + DuplicateClaimError, + record_submission, + check_duplicate, + DEFAULT_WINDOW_DAYS, +) + + +def test_check_duplicate_returns_none_for_first_submission(tmp_path): + db = tmp_path / "test.db" + assert check_duplicate("ORIG-001", db_url=f"sqlite:///{db}") is None + + +def test_check_duplicate_raises_within_window(tmp_path): + db = tmp_path / "test.db" + record_submission("ORIG-002", submitted_at=datetime(2026, 6, 1), db_url=f"sqlite:///{db}") + try: + check_duplicate("ORIG-002", db_url=f"sqlite:///{db}", + now=datetime(2026, 6, 15)) + except DuplicateClaimError as e: + assert e.claim_id == "ORIG-002" + assert e.original_submission_at == datetime(2026, 6, 1) + else: + raise AssertionError("expected DuplicateClaimError") + + +def test_check_duplicate_passes_after_window(tmp_path): + db = tmp_path / "test.db" + record_submission("ORIG-003", submitted_at=datetime(2026, 1, 1), db_url=f"sqlite:///{db}") + # 100 days later — past the 30-day default window + assert check_duplicate("ORIG-003", db_url=f"sqlite:///{db}", + now=datetime(2026, 4, 11)) is None + + +def test_default_window_is_30_days(): + assert DEFAULT_WINDOW_DAYS == 30 \ No newline at end of file