"""Run with Python 3. Synthetic Glue migration contract using local SQLite."""

import hashlib
import json
import sqlite3
from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory


def load(db, manifest, rows):
    digest = hashlib.sha256(json.dumps(rows).encode()).hexdigest()
    with db:
        receipt = db.execute(
            "SELECT digest FROM receipts WHERE manifest = ?", (manifest,)
        ).fetchone()
        if receipt:
            if receipt != (digest,):
                raise ValueError("Manifest contents changed")
            return "receipt reused"

        seen = {}
        for key, version, operation, name in rows:
            identity = (key, version)
            payload = (operation, name)
            if seen.setdefault(identity, payload) != payload:
                raise ValueError("Different payloads within one manifest")

        # Check every staged version against the pre-merge target.
        for key, version, operation, name in rows:
            previous = db.execute(
                "SELECT version, operation, name FROM customers WHERE key = ?", (key,)
            ).fetchone()
            if previous and version == previous[0] and (operation, name) != previous[1:]:
                raise ValueError("Different payload at the same customer version")

        for key, version, operation, name in rows:
            db.execute(
                """
                INSERT INTO customers VALUES (?, ?, ?, ?)
                ON CONFLICT(key) DO UPDATE SET
                  version = excluded.version, operation = excluded.operation,
                  name = excluded.name
                WHERE excluded.version > customers.version
                """,
                (key, version, operation, name),
            )

        db.execute("INSERT INTO receipts VALUES (?, ?)", (manifest, digest))
    return "committed"


def main():
    # Expected snapshots represent a fixture oracle, independent of the loader.
    expected = [("C1", 3, "DELETE", None), ("C2", 1, "UPSERT", "Dev")]
    delete = [("C1", 3, "DELETE", None)]
    with TemporaryDirectory() as directory:
        run_lab(Path(directory) / "target.sqlite", expected, delete)


def run_lab(path, expected, delete):
    with closing(sqlite3.connect(path)) as db:
        db.executescript(
            """
            CREATE TABLE customers (
              key TEXT PRIMARY KEY, version INTEGER NOT NULL,
              operation TEXT NOT NULL CHECK (operation IN ('UPSERT', 'DELETE')),
              name TEXT, CHECK ((operation = 'DELETE') = (name IS NULL))
            );
            CREATE TABLE receipts (manifest TEXT PRIMARY KEY, digest TEXT NOT NULL);
            """
        )
        load(db, "s3/baseline.csv#v1", [
            ("C1", 1, "UPSERT", "Ana"), ("C2", 1, "UPSERT", "Dev"),
        ])
        load(db, "s3/delete.csv#v1", delete)

    # Reopen after simulating a lost acknowledgment of the committed sink write.
    with closing(sqlite3.connect(path)) as db:
        assert load(db, "s3/delete.csv#v1", delete) == "receipt reused"
        load(db, "s3/late.csv#v1", [("C1", 2, "UPSERT", "Ana Lee")])
        try:
            load(db, "s3/conflict.csv#v1", [
                ("C2", 2, "UPSERT", "Dev Patel"),
                ("C1", 3, "UPSERT", "Ana Lee"),
            ])
        except ValueError as error:
            assert str(error) == "Different payload at the same customer version"
        else:
            raise AssertionError("The conflicting manifest should be held")
        try:
            load(db, "s3/stale-conflict.csv#v1", [
                ("C1", 2, "UPSERT", "Ana"),
                ("C1", 2, "UPSERT", "Ana Lee"),
            ])
        except ValueError as error:
            assert str(error) == "Different payloads within one manifest"
        else:
            raise AssertionError("Conflicting stale versions should be held")

        assert db.execute("SELECT * FROM customers ORDER BY key").fetchall() == expected
        receipts = db.execute("SELECT manifest FROM receipts ORDER BY manifest").fetchall()
        assert receipts == [
            ("s3/baseline.csv#v1",), ("s3/delete.csv#v1",), ("s3/late.csv#v1",),
        ]

    print("Parity passed: C1 stays deleted at v3 and C2 stays active at v1.")
    print("Replay reused one delete receipt. The same-version conflict rolled back.")


if __name__ == "__main__":
    main()