#!/usr/bin/env python3
"""Controlled SQLite runtime/index experiment; creates only disposable fixtures."""
import argparse
from contextlib import closing
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import platform
import shutil
import sqlite3
import tempfile


def require(ok, message):
    if not ok:
        raise RuntimeError(message)


def sha(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def compare(a, b):
    return (a > b) - (a < b)


def register(db, kind, version):
    if version == "missing":
        return
    if kind == "expression":
        offset = 0 if version == "original" else 100
        db.create_function("score", 1, lambda x: x + offset, deterministic=True)
    elif kind == "collation":
        sign = 1 if version == "original" else -1
        db.create_collation("APP_ORDER", lambda a, b: sign * compare(a, b))
    else:
        callback = compare if version == "original" else lambda a, b: compare(a.lower(), b.lower())
        db.create_collation("APP_ORDER", callback)


def execute(db, sql, params=()):
    try:
        return {"rows": [list(row) for row in db.execute(sql, params)]}
    except sqlite3.Error as error:
        return {"error": str(error), "errorCode": getattr(error, "sqlite_errorcode", None),
                "errorName": getattr(error, "sqlite_errorname", None)}


def create_fixture(path, kind):
    with closing(sqlite3.connect(path)) as db:
        register(db, kind, "original")
        require(db.execute("PRAGMA journal_mode=DELETE").fetchone()[0] == "delete", "journal")
        db.execute("CREATE TABLE sample(id INTEGER PRIMARY KEY, value)")
        values = [1, 2, 3] if kind == "expression" else ["ant", "bee", "cat"]
        if kind == "unique":
            values = ["A", "a"]
        expected = [[i + 1, value] for i, value in enumerate(values)]
        db.executemany("INSERT INTO sample VALUES (?, ?)", expected)
        index = "score(value)" if kind == "expression" else "value COLLATE APP_ORDER"
        db.execute("CREATE " + ("UNIQUE " if kind == "unique" else "") + "INDEX idx ON sample(" + index + ")")
        db.commit()
        require(execute(db, "PRAGMA integrity_check") == {"rows": [["ok"]]}, "fixture integrity")
    return expected


def observe(path, kind, version, expected):
    before = sha(path)
    with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)) as db:
        register(db, kind, version)
        predicate = "score(value)=?" if kind == "expression" else "value COLLATE APP_ORDER >= ?"
        parameter = (2 if version == "original" else 102) if kind == "expression" else "bee"
        indexed_sql = "SELECT id, value FROM sample INDEXED BY idx WHERE " + predicate + " ORDER BY id"
        scan_sql = "SELECT id, value FROM sample NOT INDEXED WHERE " + predicate + " ORDER BY id"
        raw = execute(db, "SELECT id, value FROM sample NOT INDEXED ORDER BY id")
        require(raw == {"rows": expected}, "base rows changed")
        indexed, scanned = execute(db, indexed_sql, (parameter,)), execute(db, scan_sql, (parameter,))
        result = {
            "kind": kind, "runtime": version, "sha256Before": before, "baseRows": raw["rows"],
            "indexedSql": indexed_sql, "scanSql": scan_sql, "parameter": parameter,
            "indexed": indexed, "scan": scanned,
            "indexedPlan": execute(db, "EXPLAIN QUERY PLAN " + indexed_sql, (parameter,)),
            "scanPlan": execute(db, "EXPLAIN QUERY PLAN " + scan_sql, (parameter,)),
            "quickCheck": execute(db, "PRAGMA quick_check"),
            "integrityCheck": execute(db, "PRAGMA integrity_check"),
            "resultsAgree": indexed["rows"] == scanned["rows"] if "rows" in indexed and "rows" in scanned else None,
        }
    result["sha256After"] = sha(path)
    require(before == result["sha256After"], "read-only observation changed bytes")
    return result


def valid_hash(value):
    return isinstance(value, str) and len(value) == 64 and all(c in "0123456789abcdef" for c in value)


def expect_error(outcome):
    require(set(outcome) == {"error", "errorCode", "errorName"}, "error outcome shape")
    require(isinstance(outcome["error"], str) and bool(outcome["error"]), "error message")
    require(type(outcome["errorCode"]) is int and outcome["errorCode"] != 0, "error code")
    require(isinstance(outcome["errorName"], str) and outcome["errorName"].startswith("SQLITE_"), "error name")


def expect_diagnostic(outcome, fragment):
    rows = outcome.get("rows")
    require(set(outcome) == {"rows"} and isinstance(rows, list) and bool(rows), "diagnostic rows")
    require(all(isinstance(row, list) and len(row) == 1 and isinstance(row[0], str)
                and fragment in row[0] for row in rows), "diagnostic content")


def validate(result):
    require(result["schemaVersion"] == 1, "schema version")
    require(valid_hash(result["harnessSha256"]), "harness hash format")
    require(len(result["cases"]) == 8, "scenario count")
    for kind in ("expression", "collation"):
        group = {r["runtime"]: r for r in result["cases"] if r["kind"] == kind}
        require(set(group) == {"original", "missing", "changed", "reindexed"}, "scenario identities")
        base = [[1, 1], [2, 2], [3, 3]] if kind == "expression" else [[1, "ant"], [2, "bee"], [3, "cat"]]
        original = [[2, 2]] if kind == "expression" else base[1:]
        changed = [[2, 2]] if kind == "expression" else base[:2]
        stale = [] if kind == "expression" else base
        predicate = "score(value)=?" if kind == "expression" else "value COLLATE APP_ORDER >= ?"
        for version, row in group.items():
            require(valid_hash(row["sha256Before"]) and valid_hash(row["sha256After"]), "database hash format")
            require(row["sha256Before"] == row["sha256After"], "read-only hash")
            require(row["baseRows"] == base, "fixture rows")
            require(row["sameBytesAsSource"] is (version != "reindexed"), "source byte identity")
            parameter = (2 if version == "original" else 102) if kind == "expression" else "bee"
            require(row["parameter"] == parameter, "query parameter")
            for field, clause in (("indexedSql", "INDEXED BY idx"), ("scanSql", "NOT INDEXED")):
                require(row[field] == "SELECT id, value FROM sample " + clause + " WHERE " + predicate + " ORDER BY id", "query contract")
            if version == "missing":
                for field in ("indexed", "scan", "indexedPlan", "scanPlan", "integrityCheck"):
                    expect_error(row[field])
                if kind == "collation":
                    expect_error(row["quickCheck"])
                else:
                    require(row["quickCheck"] == {"rows": [["ok"]]}, "missing function quick check")
                require(row["resultsAgree"] is None, "failed queries are not comparable")
                continue
            expected = original if version == "original" else changed
            require(row["scan"] == {"rows": expected}, "independent scan oracle")
            require(row["indexed"] == {"rows": stale if version == "changed" else expected}, "indexed row oracle")
            require(row["resultsAgree"] is (version != "changed"), "query agreement")
            require(row["quickCheck"] == {"rows": [["ok"]]}, "quick-check observation")
            if version == "changed":
                expect_diagnostic(row["integrityCheck"], "missing from index idx")
            else:
                require(row["integrityCheck"] == {"rows": [["ok"]]}, "positive integrity control")
            # EQP wording can vary by version; validate the access path, not node IDs.
            for field in ("indexedPlan", "scanPlan"):
                plan = row[field]
                rows = plan.get("rows")
                require(set(plan) == {"rows"} and isinstance(rows, list) and bool(rows), "query plan rows")
                require(all(isinstance(r, list) and len(r) == 4 and isinstance(r[3], str) for r in rows), "query plan shape")
                details = [r[3] for r in rows]
                if field == "indexedPlan":
                    require(any("USING " in d and "INDEX idx" in d for d in details), "forced index plan")
                else:
                    require(any("SCAN sample" in d for d in details) and not any("USING " in d for d in details), "forced table scan")
        source_hash = group["original"]["sha256Before"]
        require(all(group[v]["sha256Before"] == source_hash for v in ("missing", "changed")), "identical bytes")
        require(group["reindexed"]["sha256Before"] != source_hash, "rebuilt bytes changed")
    collision = result["uniqueCollision"]
    expect_error(collision["reindex"])
    require(collision["reindex"]["errorName"] == "SQLITE_CONSTRAINT_UNIQUE"
            and collision["reindex"]["errorCode"] == 2067, "unique collision")
    require(collision["quickCheck"] == {"rows": [["ok"]]}, "collision quick check")
    expect_diagnostic(collision["integrityCheck"], "non-unique entry in index idx")
    require(collision["baseRows"] == [[1, "A"], [2, "a"]], "collision preserves rows")
    require(valid_hash(collision["sha256Before"]) and valid_hash(collision["sha256AfterRollback"]), "collision hash format")
    require(collision["sha256Before"] == collision["sha256AfterRollback"], "rollback byte preservation")

def run(root):
    cases = []
    for kind in ("expression", "collation"):
        source = root / (kind + "-source.db")
        expected = create_fixture(source, kind)
        original_hash = sha(source)
        for version in ("original", "missing", "changed"):
            # The source is closed and uses DELETE journals; no live database is copied.
            target = root / (kind + "-" + version + ".db")
            shutil.copyfile(source, target)
            require(sha(target) == original_hash, "copy mismatch")
            row = observe(target, kind, version, expected)
            row["sameBytesAsSource"] = True
            cases.append(row)
        rebuilt = root / (kind + "-rebuilt.db")
        shutil.copyfile(source, rebuilt)
        with closing(sqlite3.connect(rebuilt)) as db:
            register(db, kind, "changed")
            db.execute("REINDEX main.idx")
            db.commit()
        row = observe(rebuilt, kind, "changed", expected)
        row["runtime"] = "reindexed"
        row["sameBytesAsSource"] = sha(rebuilt) == original_hash
        cases.append(row)
    path = root / "unique.db"
    create_fixture(path, "unique")
    before = sha(path)
    with closing(sqlite3.connect(path)) as db:
        register(db, "unique", "changed")
        quick, integrity = execute(db, "PRAGMA quick_check"), execute(db, "PRAGMA integrity_check")
        db.execute("BEGIN IMMEDIATE")
        reindex = execute(db, "REINDEX main.idx")
        db.rollback()
        rows = execute(db, "SELECT id, value FROM sample NOT INDEXED ORDER BY id")
    collision = {
        "baseRows": rows.get("rows"), "originalSemantics": "ASCII case-sensitive order",
        "changedSemantics": "ASCII lower-case comparison: A and a compare equal",
        "quickCheck": quick, "integrityCheck": integrity, "reindex": reindex,
        "sha256Before": before, "sha256AfterRollback": sha(path),
    }
    with closing(sqlite3.connect(":memory:")) as db:
        environment = {
            "python": platform.python_version(), "sqlite": sqlite3.sqlite_version,
            "sqliteSourceId": db.execute("SELECT sqlite_source_id()").fetchone()[0],
            "compileOptions": [r[0] for r in db.execute("PRAGMA compile_options")],
            "os": platform.system(), "kernel": platform.release(), "machine": platform.machine(),
        }
    result = {
        "schemaVersion": 1, "generatedAt": datetime.now(timezone.utc).isoformat(),
        "evidenceType": "Controlled software experiment with constructed data",
        "environment": environment, "harnessSha256": sha(__file__),
        "method": {
            "primaryScenarios": 8, "rowsPerPrimaryFixture": 3,
            "additionalUniqueCollisionFixtures": 1, "rowsPerUniqueFixture": 2,
            "journalMode": "DELETE", "concurrentWriters": False,
            "functionVersions": {"original": "score(x)=x", "changed": "score(x)=x+100"},
            "collationVersions": {"original": "ASCII ascending", "changed": "ASCII descending"},
            "querySelection": "Force INDEXED BY idx and NOT INDEXED; order result rows by id",
            "copyMethod": "Byte copy of closed fixtures; original/missing/changed share identical SHA-256",
        },
        "sources": [
            "https://sqlite.org/c3ref/create_collation.html", "https://sqlite.org/deterministic.html",
            "https://sqlite.org/expridx.html", "https://sqlite.org/staleexpridx.html",
            "https://www.sqlite.org/pragma.html#pragma_quick_check", "https://www.sqlite.org/lang_reindex.html",
        ],
        "cases": cases, "uniqueCollision": collision,
        "limits": [
            "Changed callback semantics deliberately violate existing index assumptions; no SQLite defect claimed.",
            "One SQLite build, tiny designed fixtures; not an incident rate or performance measurement.",
            "No backup API, live-file copy, WAL, crash, power loss, ICU upgrade or floating-point drift tested.",
            "Forced access paths isolate the mechanism, not production planner choices.",
            "Targeted REINDEX does not validate application meaning or repair arbitrary storage corruption.",
        ],
    }
    validate(result)
    return result


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--work-root", type=Path, default=Path("."))
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    require(args.work_root.is_dir(), "work root must already exist")
    require(not args.output.exists(), "refusing existing output")
    with tempfile.TemporaryDirectory(prefix="hesela-sqlite-runtime-", dir=args.work_root) as root:
        result = run(Path(root))
    with args.output.open("x", encoding="utf-8") as handle:
        json.dump(result, handle, indent=2, ensure_ascii=True)
        handle.write("\n")
    print("Validated 8 runtime/index scenarios and 1 UNIQUE collision control.")


if __name__ == "__main__":
    main()