#!/usr/bin/env python3
"""Reproduce a synthetic invoice-feed migration with Python 3 and SQLite."""
import argparse
import csv
import hashlib
import io
import json
import os
import sqlite3
from decimal import Decimal, InvalidOperation
from pathlib import Path
from tempfile import NamedTemporaryFile

EVENTS = """position,changed_at,invoice,revision,account,currency,amount,operation
101,2026-10-01T09:00:00Z,A,1,C1,USD,10.00,upsert
102,2026-10-01T09:00:00Z,B,1,C1,USD,15.00,upsert
103,2026-10-01T09:01:00Z,A,2,C1,USD,,delete
104,2026-10-01T09:02:00Z,C,1,UNKNOWN,USD,9.00,upsert
105,2026-09-30T23:59:00Z,B,2,C1,USD,17.50,upsert
106,2026-10-01T09:03:00Z,B,1,C1,USD,15.00,upsert
107,2026-10-01T09:04:00Z,C,2,C2,USD,9.00,upsert
"""
ACCOUNTS = {"C1": "North", "C2": "North"}
SCHEMA = """
CREATE TABLE state (feed TEXT PRIMARY KEY, position INTEGER NOT NULL);
INSERT INTO state VALUES ('invoices', 100);
CREATE TABLE batch (
    high INTEGER PRIMARY KEY, low INTEGER NOT NULL, payload TEXT NOT NULL,
    digest TEXT NOT NULL, result TEXT
);
CREATE TABLE invoice (
    id TEXT PRIMARY KEY, revision INTEGER NOT NULL, company TEXT,
    currency TEXT, cents INTEGER, deleted INTEGER NOT NULL
);
CREATE TABLE disposition (
    position INTEGER PRIMARY KEY, invoice TEXT NOT NULL, status TEXT NOT NULL
);
"""
# An independently written oracle for this lab's fixture, not a measured legacy run.
EXPECTED = {
    106: {"low": 100, "high": 106, "status": {
        "upserted": 3, "deleted": 1, "unknown_account": 1, "stale": 1,
    }, "invoices": [["A", 2, "North", "USD", None, 1],
                    ["B", 2, "North", "USD", 1750, 0]],
          "active_totals": [["North", "USD", 1750]]},
    107: {"low": 106, "high": 107, "status": {"upserted": 1},
          "invoices": [["A", 2, "North", "USD", None, 1],
                       ["B", 2, "North", "USD", 1750, 0],
                       ["C", 2, "North", "USD", 900, 0]],
          "active_totals": [["North", "USD", 2650]]},
}


def json_text(value):
    return json.dumps(value, indent=2, sort_keys=True) + "\n"


def atomic_write(path, contents):
    if path.exists():
        if path.read_text() != contents:
            raise ValueError(f"Published receipt differs: {path.name}")
        return
    with NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as tmp:
        tmp.write(contents)
        tmp.flush()
        os.fsync(tmp.fileno())
        temp_path = Path(tmp.name)
    try:
        temp_path.replace(path)
    finally:
        temp_path.unlink(missing_ok=True)


def initialize(root):
    root.mkdir(parents=True, exist_ok=False)
    (root / "source.csv").write_text(EVENTS)
    (root / "accounts.json").write_text(json_text(ACCOUNTS))
    (root / "receipts").mkdir()
    (root / "expected").mkdir()
    with sqlite3.connect(root / "target.sqlite") as db:
        db.executescript(SCHEMA)
    for high, snapshot in EXPECTED.items():
        (root / "expected" / f"{high}.json").write_text(json_text(snapshot))


def checkpoint(db):
    return db.execute("SELECT position FROM state WHERE feed = 'invoices'").fetchone()[0]


def prepare(db, root, high):
    db.execute("BEGIN IMMEDIATE")
    with db:
        if db.execute("SELECT 1 FROM batch WHERE high = ?", (high,)).fetchone():
            return
        low = checkpoint(db)
        if high <= low:
            raise ValueError(f"Window must advance beyond {low}")
        if db.execute("SELECT 1 FROM batch WHERE low = ? AND result IS NULL", (low,)).fetchone():
            raise ValueError(f"An uncommitted manifest starts at {low}; resume it first")
        rows = sorted(
            (row for row in csv.DictReader(io.StringIO((root / "source.csv").read_text()))
             if low < int(row["position"]) <= high),
            key=lambda row: int(row["position"]),
        )
        # This synthetic source promises contiguous positions. Real CDC cursors can be sparse.
        if len(rows) != high - low or any(
            int(row["position"]) != position
            for position, row in enumerate(rows, start=low + 1)
        ):
            raise ValueError("Expected positions are missing or duplicated")
        payload = json_text({"rows": rows, "accounts": json.loads((root / "accounts.json").read_text())})
        digest = hashlib.sha256(payload.encode()).hexdigest()
        db.execute("INSERT INTO batch VALUES (?, ?, ?, ?, NULL)", (high, low, payload, digest))


def amount_cents(text):
    try:
        amount = Decimal(text) * 100
        if amount.is_finite() and amount == amount.to_integral_value() and abs(amount) <= 10**15:
            return int(amount)
    except InvalidOperation:
        pass
    return None


def decide(db, row, accounts):
    revision = int(row["revision"])
    invoice_id = row["invoice"]
    previous = db.execute(
        "SELECT revision, company, currency, cents, deleted FROM invoice WHERE id = ?",
        (invoice_id,),
    ).fetchone()
    if previous and revision < previous[0]:
        return "stale"
    company = accounts.get(row["account"])
    if company is None:
        return "unknown_account"
    operation = row["operation"]
    if operation not in {"upsert", "delete"}:
        return "unknown_operation"
    cents = amount_cents(row["amount"]) if operation == "upsert" else None
    if operation == "upsert" and cents is None:
        return "invalid_amount"
    current = (revision, company, row["currency"], cents, int(operation == "delete"))
    if previous and revision == previous[0]:
        return "duplicate" if previous == current else "revision_conflict"
    db.execute("""
        INSERT INTO invoice VALUES (?, ?, ?, ?, ?, ?)
        ON CONFLICT(id) DO UPDATE SET
            revision = excluded.revision, company = excluded.company,
            currency = excluded.currency, cents = excluded.cents, deleted = excluded.deleted
    """, (invoice_id, *current))
    return "deleted" if operation == "delete" else "upserted"


def apply(db, high, fail_before_commit):
    db.execute("BEGIN IMMEDIATE")
    with db:
        low, payload, digest, result = db.execute(
            "SELECT low, payload, digest, result FROM batch WHERE high = ?", (high,),
        ).fetchone()
        if result is not None:
            return result
        if checkpoint(db) != low or hashlib.sha256(payload.encode()).hexdigest() != digest:
            raise ValueError("Checkpoint or manifest changed after preparation")
        snapshot = json.loads(payload)
        for row in snapshot["rows"]:
            disposition = decide(db, row, snapshot["accounts"])
            db.execute("INSERT INTO disposition VALUES (?, ?, ?)",
                       (int(row["position"]), row["invoice"], disposition))
        if fail_before_commit:
            raise RuntimeError("Injected stop before checkpoint commit")
        statuses = dict(db.execute("""
            SELECT status, COUNT(*) FROM disposition
            WHERE position > ? AND position <= ? GROUP BY status
        """, (low, high)))
        records = db.execute("SELECT * FROM invoice ORDER BY id").fetchall()
        totals = db.execute("""
            SELECT company, currency, SUM(cents) FROM invoice WHERE deleted = 0
            GROUP BY company, currency ORDER BY company, currency
        """).fetchall()
        result = json_text({"low": low, "high": high, "status": statuses,
                            "invoices": records, "active_totals": totals})
        db.execute("UPDATE state SET position = ? WHERE feed = 'invoices'", (high,))
        db.execute("UPDATE batch SET result = ? WHERE high = ?", (result, high))
        return result


def inspect(db):
    rows = db.execute("SELECT high, result IS NOT NULL FROM batch ORDER BY high").fetchall()
    return {
        "checkpoint": checkpoint(db),
        "batches": {high: "committed" if committed else "staged" for high, committed in rows},
        "dispositions": db.execute("SELECT * FROM disposition ORDER BY position").fetchall(),
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("lab_dir", type=Path)
    subcommands = parser.add_subparsers(dest="action", required=True)
    subcommands.add_parser("init", help="Create the fixture in a new directory")
    run = subcommands.add_parser("run", help="Prepare, apply and publish one window")
    run.add_argument("--through", type=int, required=True)
    run.add_argument("--fail", choices=["before-commit", "after-commit"])
    subcommands.add_parser("inspect", help="Read the checkpoint and row outcomes")
    compare = subcommands.add_parser("compare", help="Compare a receipt with the fixture oracle")
    compare.add_argument("--through", type=int, required=True)
    args = parser.parse_args()
    root = args.lab_dir

    try:
        if args.action == "init":
            initialize(root)
            return
        receipt = root / "receipts" / f"{getattr(args, 'through', None)}.json"
        if args.action == "compare":
            actual = json.loads(receipt.read_text())
            expected = json.loads((root / "expected" / receipt.name).read_text())
            if actual != expected:
                raise ValueError(f"Parity mismatch at position {args.through}\n{json_text(actual)}")
            print(f"Parity matched at position {args.through}")
            return
        with sqlite3.connect((root / "target.sqlite").resolve().as_uri() + "?mode=rw", uri=True) as db:
            if args.action == "inspect":
                print(json_text(inspect(db)), end="")
                return
            prepare(db, root, args.through)
            result = apply(db, args.through, args.fail == "before-commit")
        if args.fail == "after-commit":
            raise RuntimeError("Injected stop after checkpoint commit and before receipt")
        atomic_write(receipt, result)
        print(f"Published {receipt}")
    except (OSError, sqlite3.Error, ValueError, RuntimeError) as error:
        parser.exit(1, f"{error}\n")


if __name__ == "__main__":
    main()