feat(backend): add PRAGMA user_version migration runner + 0001_initial

backend/src/cyclone/db_migrate.py

@@ -1,13 +1,73 @@
-"""Database migration runner.
+"""PRAGMA user_version-based migration runner.
 
-Stub. Task 3 replaces this with a real migration implementation
+Each migration is a `.sql` file under `backend/src/cyclone/migrations/`
-that tracks applied migrations in a ``schema_migrations`` table.
+with a leading comment of the form `-- version: N`. The runner reads
+the current `PRAGMA user_version`, sorts migrations by filename, and
+applies any whose version is greater than the current value. Each
+migration runs in its own transaction; a failure rolls back the
+version bump so the next startup retries.
+
+This is intentionally NOT Alembic. One operator, one machine, one
+schema-version counter.
 """
 
 from __future__ import annotations
 
+import re
+from pathlib import Path
+
 import sqlalchemy as sa
 
+MIGRATIONS_DIR = Path(__file__).parent / "migrations"
+
+_VERSION_RE = re.compile(r"^--\s*version:\s*(\d+)", re.MULTILINE)
+
+
+def _migration_version(sql: str, filename: str) -> int:
+    m = _VERSION_RE.search(sql)
+    if not m:
+        raise RuntimeError(
+            f"{filename}: missing '-- version: N' header comment"
+        )
+    return int(m.group(1))
+
+
+def _strip_comments(sql: str) -> str:
+    """Remove `--` line comments; preserve everything else."""
+    lines = []
+    for line in sql.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("--"):
+            continue
+        lines.append(line)
+    return "\n".join(lines)
+
 
 def run(engine: sa.Engine) -> None:
-    """Apply pending migrations. No-op until Task 3."""
+    """Apply any pending migrations. Idempotent."""
+    migrations_dir = MIGRATIONS_DIR
+    if not migrations_dir.exists():
+        return  # no migrations directory yet — fresh checkout, no-op
+
+    migration_files = sorted(migrations_dir.glob("*.sql"))
+    if not migration_files:
+        return
+
+    with engine.begin() as conn:
+        current = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
+
+    for path in migration_files:
+        sql = path.read_text()
+        version = _migration_version(sql, path.name)
+        if version <= current:
+            continue
+
+        statements_sql = _strip_comments(sql)
+        statements = [s.strip() for s in statements_sql.split(";") if s.strip()]
+
+        with engine.begin() as conn:
+            for stmt in statements:
+                conn.exec_driver_sql(stmt)
+            conn.exec_driver_sql(f"PRAGMA user_version = {version}")
+
+        current = version

backend/src/cyclone/migrations/0001_initial.sql

@@ -0,0 +1,87 @@
+-- version: 1
+-- Initial schema: 6 tables for Cyclone DB + reconciliation.
+
+CREATE TABLE batches (
+    id TEXT PRIMARY KEY,
+    kind TEXT NOT NULL,
+    input_filename TEXT NOT NULL,
+    parsed_at DATETIME NOT NULL,
+    totals_json TEXT,
+    validation_json TEXT,
+    raw_result_json TEXT
+);
+CREATE INDEX ix_batches_parsed_at ON batches(parsed_at DESC);
+
+CREATE TABLE claims (
+    id TEXT PRIMARY KEY,
+    batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,
+    patient_control_number TEXT NOT NULL,
+    service_date_from DATE,
+    service_date_to DATE,
+    charge_amount NUMERIC(12, 2) NOT NULL DEFAULT 0,
+    provider_npi TEXT,
+    payer_id TEXT,
+    state TEXT NOT NULL DEFAULT 'submitted',
+    state_before_reversal TEXT,
+    matched_remittance_id TEXT REFERENCES remittances(id),
+    raw_json TEXT,
+    UNIQUE (batch_id, patient_control_number)
+);
+CREATE INDEX ix_claims_state ON claims(state);
+CREATE INDEX ix_claims_patient_control_number ON claims(patient_control_number);
+CREATE INDEX ix_claims_service_date_from ON claims(service_date_from);
+
+CREATE TABLE remittances (
+    id TEXT PRIMARY KEY,
+    batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,
+    payer_claim_control_number TEXT NOT NULL,
+    claim_id TEXT REFERENCES claims(id),
+    status_code TEXT NOT NULL,
+    status_label TEXT,
+    total_charge NUMERIC(12, 2) NOT NULL DEFAULT 0,
+    total_paid NUMERIC(12, 2) NOT NULL DEFAULT 0,
+    patient_responsibility NUMERIC(12, 2),
+    adjustment_amount NUMERIC(12, 2) NOT NULL DEFAULT 0,
+    received_at DATETIME NOT NULL,
+    service_date DATE,
+    is_reversal INTEGER NOT NULL DEFAULT 0,
+    raw_json TEXT,
+    UNIQUE (batch_id, payer_claim_control_number)
+);
+CREATE INDEX ix_remittances_claim_id ON remittances(claim_id);
+CREATE INDEX ix_remittances_payer_claim_control_number ON remittances(payer_claim_control_number);
+CREATE INDEX ix_remittances_status_code ON remittances(status_code);
+
+CREATE TABLE cas_adjustments (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    remittance_id TEXT NOT NULL REFERENCES remittances(id) ON DELETE CASCADE,
+    group_code TEXT NOT NULL,
+    reason_code TEXT NOT NULL,
+    amount NUMERIC(12, 2) NOT NULL,
+    quantity NUMERIC(10, 2)
+);
+CREATE INDEX ix_cas_adjustments_remittance_id ON cas_adjustments(remittance_id);
+
+CREATE TABLE matches (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    claim_id TEXT NOT NULL UNIQUE REFERENCES claims(id) ON DELETE CASCADE,
+    remittance_id TEXT NOT NULL REFERENCES remittances(id) ON DELETE CASCADE,
+    strategy TEXT NOT NULL,
+    matched_at DATETIME NOT NULL,
+    prior_claim_state TEXT,
+    is_reversal INTEGER NOT NULL DEFAULT 0
+);
+CREATE INDEX ix_matches_remittance_id ON matches(remittance_id);
+CREATE INDEX ix_matches_matched_at ON matches(matched_at);
+
+CREATE TABLE activity_events (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    ts DATETIME NOT NULL,
+    kind TEXT NOT NULL,
+    batch_id TEXT,
+    claim_id TEXT,
+    remittance_id TEXT,
+    payload_json TEXT
+);
+CREATE INDEX ix_activity_events_ts ON activity_events(ts DESC);
+CREATE INDEX ix_activity_events_kind ON activity_events(kind);

backend/tests/test_db_migrate.py

@@ -0,0 +1,67 @@
+"""Tests for the PRAGMA user_version migration runner."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+import sqlalchemy as sa
+
+from cyclone import db_migrate
+
+
+def _fresh_engine(tmp_path: Path) -> sa.Engine:
+    return sa.create_engine(f"sqlite:///{tmp_path}/m.db", future=True)
+
+
+def _user_version(engine: sa.Engine) -> int:
+    with engine.connect() as c:
+        return c.exec_driver_sql("PRAGMA user_version").scalar() or 0
+
+
+def test_run_on_empty_db_bumps_to_latest(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
+    (tmp_path / "0001_initial.sql").write_text("-- version: 1\nCREATE TABLE t (id INTEGER);\n")
+
+    engine = _fresh_engine(tmp_path)
+    db_migrate.run(engine)
+
+    assert _user_version(engine) == 1
+    with engine.connect() as c:
+        rows = c.exec_driver_sql("SELECT name FROM sqlite_master WHERE type='table' AND name='t'").all()
+    assert len(rows) == 1
+
+
+def test_run_is_idempotent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
+    (tmp_path / "0001_initial.sql").write_text("-- version: 1\nCREATE TABLE t (id INTEGER);\n")
+
+    engine = _fresh_engine(tmp_path)
+    db_migrate.run(engine)
+    db_migrate.run(engine)  # second call should be a no-op
+
+    assert _user_version(engine) == 1
+
+
+def test_run_skips_already_applied_migrations(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
+    (tmp_path / "0001_initial.sql").write_text("-- version: 1\nCREATE TABLE t1 (id INTEGER);\n")
+    (tmp_path / "0002_add_col.sql").write_text("-- version: 2\nALTER TABLE t1 ADD COLUMN name TEXT;\n")
+
+    engine = _fresh_engine(tmp_path)
+    db_migrate.run(engine)
+    assert _user_version(engine) == 2
+
+    # Modify 0002 (a no-op for SQLite ALTER, but we want to verify the
+    # runner doesn't re-run already-applied migrations).
+    db_migrate.run(engine)
+    assert _user_version(engine) == 2
+
+
+def test_run_raises_on_missing_version_header(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
+    (tmp_path / "0001_bad.sql").write_text("CREATE TABLE t (id INTEGER);\n")  # no header
+
+    engine = _fresh_engine(tmp_path)
+    with pytest.raises(RuntimeError, match="missing '-- version"):
+        db_migrate.run(engine)