#!/usr/bin/env python3
"""Upgrade prover — does a real app upgrade keep the customer's data, and can it be undone?

R-449. `SPIKE-app-update-2026-09-01.md` §7 measured exactly ONE upgrade, by hand, on one app. A
one-off measurement that nothing repeats decays into a claim, and this project has paid for that
before. This is the thing that repeats it.

Method, per EDGE (one app, one FROM image set, one TO image set):

  1. render the catalog template with the FROM images and `docker compose up -d`
  2. SEED through the app's OWN INTERFACE — its HTTP API or its own CLI inside the container
  3. VERIFY the seed reads back  ← control C1. A fixture that cannot prove itself first proves
     nothing after.
  4. re-render with the TO images, `up -d`, settle
  5. VERIFY the seed reads back AGAIN  ← THE RESULT
  6. ABORT: put the FROM images back, `up -d`, and record what happens

WHAT SUCCESS IS, AND THE RULE THAT DOES *NOT* CARRY OVER FROM survive2.py.
`survive2.py` calls a file survived only if sha256 AND inode both match. That is right for a
redeploy and WRONG for an upgrade: a migration is SUPPOSED to rewrite files, so that rule fails
every correct upgrade. Success here is an APPLICATION-LEVEL READBACK — ask the app for the value.

THE RULE THAT DOES CARRY OVER, verbatim from survive2.py: "Nothing is ever seeded into a volume by
hand." R-156's evidence shows a root-written canary making an empty volume read as populated. Every
seed goes in through the app's own interface. An app with no non-browser route is recorded
`inconclusive`, WITH what was tried — that is a result, not a licence to plant a file.

"the container started" IS NOT A PASS. The spike measured an app that was HTTP 200
"update completed" and crash-looping at the same time.

The undo is an ABORT, never a "rollback". The word is struck — see
felhom.eu/documentation/architecture/09-update-architecture.md §4: once a migration has run, the old
image refuses to start on the migrated data, so there is no rollback to speak of.

Usage:  python3 upgrade-test.py <edge-id> [<edge-id> …]      (see EDGES)
        python3 upgrade-test.py --list
Layout: templates under /opt/upg/templates, evidence under /opt/upg/evidence
"""
import importlib.util, json, os, re, shutil, subprocess, sys, time
from datetime import datetime, timezone
from pathlib import Path

HARNESS_VERSION = 1
ROOT = Path("/opt/upg")
TEMPLATES = ROOT / "templates"
EVIDENCE = ROOT / "evidence"

_spec = importlib.util.spec_from_file_location("cvp", str(ROOT / "check-volume-persistence.py"))
cvp = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(cvp)

_fspec = importlib.util.spec_from_file_location("fx", str(ROOT / "upgrade_fixtures.py"))
fx = importlib.util.module_from_spec(_fspec)
_fspec.loader.exec_module(fx)


# --- the edges ---------------------------------------------------------------------------------
#
# Every FROM/TO pair below is a transition THE CATALOG ITSELF MADE, read from its own git history,
# except where the comment says otherwise. C3's `alpine:3.20` is a real image that pulls cleanly and
# exits immediately — measured in SPIKE-app-update-2026-09-01 §4.
EDGES = {
    "C2": dict(app="privatebin", note="no-op control: a version to ITSELF",
               frm={"privatebin": "privatebin/pdo:2.0.5"},
               to={"privatebin": "privatebin/pdo:2.0.5"}),
    "C3": dict(app="privatebin", note="NEGATIVE control: the TO image starts and exits immediately",
               frm={"privatebin": "privatebin/pdo:2.0.5"},
               to={"privatebin": "alpine:3.20"}),
    "E1": dict(app="privatebin", note="catalog transition cf8b645, a major",
               frm={"privatebin": "privatebin/pdo:1.7.5"},
               to={"privatebin": "privatebin/pdo:2.0.5"}),
    "E2": dict(app="docmost", note="catalog transition a2115b2; PostgreSQL constant across it",
               frm={"docmost": "docmost/docmost:0.25.3"},
               to={"docmost": "docmost/docmost:0.95.0"}),
    "E3": dict(app="bookstack", note="catalog transition 0b73e5e: app AND engine together",
               frm={"bookstack": "lscr.io/linuxserver/bookstack:25.02.2", "bookstack-db": "mariadb:11.6"},
               to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:12.3"}),
    "E3a": dict(app="bookstack", note="AUTHORED step (the catalog never carried it): app half alone",
                frm={"bookstack": "lscr.io/linuxserver/bookstack:25.02.2", "bookstack-db": "mariadb:11.6"},
                to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:11.6"}),
    "E3b": dict(app="bookstack", note="AUTHORED step: engine half alone",
                frm={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:11.6"},
                to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:12.3"}),

    # --- added 2026-09-21 by the update night (R-462's widening) -------------------------------
    # Every one of these was run BOX-SIDE FIRST, through the product's own guarded Update on guest
    # 9202, with the data read back through the app's own front door before and after. They are
    # here so the same edge also gets its ABORT step, which the box deliberately does not offer
    # (`09` §6.1: the box never puts the old version back by itself, because whether the old image
    # starts on migrated data is per-app and cannot be predicted).
    #
    # These are REAL UPSTREAM MOVES that existed on 2026-09-21 and that the catalog has NOT made.
    # They are candidates the operator may promote; the harness is where the abort answer for each
    # of them comes from.
    "U1": dict(app="privatebin", note="real upstream move 2.0.5 -> 2.0.6; file-backed, no database",
               frm={"privatebin": "privatebin/pdo:2.0.5"},
               to={"privatebin": "privatebin/pdo:2.0.6"}),
    "U2": dict(app="docmost", note="real upstream move 0.95.0 -> 0.96.0; PostgreSQL CONSTANT across it",
               frm={"docmost": "docmost/docmost:0.95.0", "docmost-postgres": "postgres:16-alpine"},
               to={"docmost": "docmost/docmost:0.96.0", "docmost-postgres": "postgres:16-alpine"}),
    "U3": dict(app="bookstack", note="real upstream move 26.05.2 -> 26.05.5; MariaDB CONSTANT across it",
               frm={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:12.3"},
               to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.5", "bookstack-db": "mariadb:12.3"}),
    "U4": dict(app="actualbudget", note="real upstream move 26.7.0 -> 26.9.0; SQLite in its own volume",
               frm={"actualbudget": "actualbudget/actual-server:26.7.0"},
               to={"actualbudget": "actualbudget/actual-server:26.9.0"}),
    "U5": dict(app="navidrome", note="real upstream move 0.63.2 -> 0.64.0; DATABASE HALF ONLY",
               frm={"navidrome": "deluan/navidrome:0.63.2"},
               to={"navidrome": "deluan/navidrome:0.64.0"}),
    "U6": dict(app="audiobookshelf", note="real upstream move 2.35.1 -> 2.36.1; DATABASE HALF ONLY",
               frm={"audiobookshelf": "ghcr.io/advplyr/audiobookshelf:2.35.1"},
               to={"audiobookshelf": "ghcr.io/advplyr/audiobookshelf:2.36.1"}),
    "U7": dict(app="vikunja", note="real upstream move 2.3.0 -> 2.6.0; the app migrates its own SQLite",
               frm={"vikunja": "vikunja/vikunja:2.3.0"},
               to={"vikunja": "vikunja/vikunja:2.6.0"}),
}


# --- compose plumbing --------------------------------------------------------------------------

def render(app: str, images: dict, workdir: Path, env: dict) -> Path:
    """Write the catalog template into workdir with `images` substituted per SERVICE.

    Substitution is per service and only on that service's own `image:` line — never a blind
    string replace, which would also rewrite an image name that appears in a comment or an env var.
    """
    src = (TEMPLATES / app / "docker-compose.yml").read_text()
    out, cur = [], None
    for line in src.splitlines():
        m = re.match(r"^  ([A-Za-z0-9_-]+):\s*$", line)
        if m:
            cur = m.group(1)
        mi = re.match(r"^(\s+image:\s*)(\S+)\s*$", line)
        if mi and cur in images:
            line = mi.group(1) + images[cur]
        out.append(line)
    workdir.mkdir(parents=True, exist_ok=True)
    (workdir / "docker-compose.yml").write_text("\n".join(out) + "\n")
    (workdir / ".env").write_text("".join(f"{k}={v}\n" for k, v in env.items()))
    return workdir / "docker-compose.yml"


def compose(workdir: Path, project: str, *args, timeout=1800):
    return cvp._sh(["docker", "compose", "-p", project, "-f", str(workdir / "docker-compose.yml"),
                    "--env-file", str(workdir / ".env")] + list(args), timeout=timeout)


def container_ip(name: str) -> str:
    r = cvp._sh(["docker", "inspect", "-f",
                 "{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}", name])
    return (r.stdout or "").split()[0] if r.stdout.strip() else ""


def settle(project: str, workdir: Path, wait=420):
    """Wait until every container is running AND every one that declares a healthcheck is healthy.

    Returns (ok, seconds, per-container state). A container that has NO healthcheck counts as
    settled once it is running — but `running` is never reported as the RESULT, only as a
    precondition for asking the app itself (see the module docstring).
    """
    t0 = time.time()
    deadline = t0 + wait
    states = {}
    while time.time() < deadline:
        r = compose(workdir, project, "ps", "-aq", timeout=120)
        cids = [c for c in r.stdout.split() if c]
        if not cids:
            time.sleep(3)
            continue
        states, pending = {}, False
        for cid in cids:
            info = cvp._inspect(cid)
            if not info:
                pending = True
                continue
            name = info["Name"].lstrip("/")
            st = info.get("State", {})
            health = (st.get("Health") or {}).get("Status")
            states[name] = {"status": st.get("Status"), "health": health,
                            "restarts": st.get("RestartCount", 0), "exit": st.get("ExitCode")}
            if st.get("Status") != "running":
                pending = True
            elif health in ("starting", "unhealthy"):
                pending = True
        if not pending:
            return True, round(time.time() - t0, 1), states
        time.sleep(5)
    return False, round(time.time() - t0, 1), states


# --- the ENGINE's own view of itself (R-459) ------------------------------------------------------
#
# WHY THIS EXISTS. On edge E3b this harness returned `proven` — correctly: the app's data survived,
# which is what it asked. But MariaDB was at that moment logging that the datadir conversion it
# requires had been SKIPPED, and nothing here could see it. The harness watched the app and the
# migration log; neither looks at engine state.
#
# IT IS REPORTED BESIDE THE VERDICT, NEVER FOLDED INTO IT. An unconverted datadir is not known to be
# a failure — SPIKE-r459-mariadb-upgrade-2026-09-06 measured 5 of 5 restarts with no degradation — so
# a verdict that called it `failed` would encode an unproven judgement, which is worse than reporting
# a fact and letting a person read both.
#
# The two engines fail differently and the field carries both: MariaDB starts anyway and skips the
# conversion quietly; PostgreSQL REFUSES to start on a datadir from an older major. So "the engine
# would not start" is as much an engine-state observation as "the engine says it needs a check".
ENGINE_PROBES = {
    # image-name fragment -> (probe command inside the container, what the answer means)
    "mariadb": (
        "cat /var/lib/mysql/mariadb_upgrade_info 2>&1; echo '|'; "
        "mariadb-upgrade --check-if-upgrade-is-needed --user=root "
        "--password=$MYSQL_ROOT_PASSWORD 2>&1; echo \"[exit=$?]\"",
        "datadir version | the engine's own upgrade verdict",
    ),
    "postgres": (
        "cat /var/lib/postgresql/data/PG_VERSION 2>&1", "datadir major version",
    ),
}


def engine_state(images: dict):
    """Ask every database engine in this stack what it thinks of its own datadir.

    Best-effort and never fatal: a probe that cannot run records why, because "we could not ask" and
    "the engine is content" are different facts and only one of them is about the engine.
    """
    out = {}
    for svc, ref in images.items():
        for frag, (cmd, meaning) in ENGINE_PROBES.items():
            if frag not in ref:
                continue
            r = cvp._sh(["docker", "exec", svc, "sh", "-c", cmd], timeout=120)
            out[svc] = {"image": ref, "probe": meaning,
                        "answer": " ".join(((r.stdout or "") + (r.stderr or "")).split())[:600],
                        "probe_rc": r.returncode}
    return out


MIGRATION_RE = re.compile(
    r"migrat|upgrad|schema|alter table|CREATE TABLE|InnoDB: Upgrad|mysql_upgrade|"
    r"mariadb-upgrade|Running .* migration|Applying|db:migrate",
    re.I)


def migration_lines(project: str, workdir: Path, since_iso: str, limit=6):
    """Verbatim log lines that SAY a migration ran. Never inferred from timing — the finding is the
    sentence the app printed, exactly as the Nextcloud refusal was."""
    r = compose(workdir, project, "logs", "--since", since_iso, "--no-color", timeout=180)
    hits = [ln.strip() for ln in (r.stdout + r.stderr).splitlines() if MIGRATION_RE.search(ln)]
    return hits[:limit]


# --- one edge ----------------------------------------------------------------------------------

def run_edge(edge_id: str) -> dict:
    e = EDGES[edge_id]
    app = e["app"]
    project = f"upg{edge_id.lower()}"
    workdir = ROOT / "work" / edge_id
    ev = EVIDENCE / edge_id
    ev.mkdir(parents=True, exist_ok=True)
    shutil.rmtree(workdir, ignore_errors=True)

    felhom = (TEMPLATES / app / ".felhom.yml").read_text()
    compose_text = (TEMPLATES / app / "docker-compose.yml").read_text()
    env = cvp.build_env(app, felhom, compose_text)

    rec = {"harness_version": HARNESS_VERSION, "edge": edge_id, "app": app, "note": e["note"],
           "from": e["frm"], "to": e["to"], "verdict": "inconclusive",
           "seed_read_before": False, "seed_read_after": False, "healthy_after": False,
           "migration_observed": None, "abort": "not-attempted", "abort_detail": None,
           # engine_state is a REPORT, not a judgement — see ENGINE_PROBES.
           "engine_state_after": None,
           "duration_s": 0, "measured_at": None, "evidence": f"evidence/{edge_id}"}
    t0 = time.time()
    log = []

    def say(msg):
        line = f"[{datetime.now(timezone.utc).strftime('%H:%M:%S')}] {msg}"
        print(line, flush=True)
        log.append(line)

    try:
        # --- 1. FROM ---
        say(f"{edge_id}: deploying {app} at FROM {e['frm']}")
        render(app, e["frm"], workdir, env)
        up = compose(workdir, project, "up", "-d")
        if up.returncode != 0:
            rec["abort_detail"] = f"FROM deploy failed rc={up.returncode}: {up.stderr[-800:]}"
            say("FROM deploy FAILED — inconclusive, the edge was never reached")
            return rec
        ok, secs, states = settle(project, workdir)
        say(f"FROM settled={ok} in {secs}s :: {json.dumps(states)}")
        if not ok:
            rec["abort_detail"] = f"FROM never settled: {json.dumps(states)}"
            say("FROM never became healthy — inconclusive, not a verdict about the upgrade")
            return rec

        # --- 2/3. seed + C1 ---
        fixture = fx.FIXTURES.get(app)
        if fixture is None:
            rec["abort_detail"] = "no fixture"
            say("no fixture for this app — inconclusive")
            return rec
        seeded = fixture.seed(container_ip, say)
        if seeded is None:
            rec["verdict"] = "inconclusive"
            rec["abort_detail"] = "no non-browser seed route"
            say("INCONCLUSIVE — no non-browser seed route. Nothing was planted by hand.")
            return rec
        rec["seed_read_before"] = bool(fixture.verify(container_ip, seeded, say))
        say(f"C1 (seed reads back BEFORE): {rec['seed_read_before']}")
        if not rec["seed_read_before"]:
            rec["abort_detail"] = "C1 failed: the fixture could not prove itself before the upgrade"
            say("C1 FAILED — a fixture that cannot prove itself first proves nothing after")
            return rec

        # --- 4. TO ---
        swap_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        say(f"{edge_id}: swapping to TO {e['to']}")
        render(app, e["to"], workdir, env)
        up2 = compose(workdir, project, "up", "-d")
        say(f"TO up -d rc={up2.returncode}")
        ok2, secs2, states2 = settle(project, workdir)
        rec["healthy_after"] = ok2
        rec["duration_s"] = secs2
        say(f"TO settled={ok2} in {secs2}s :: {json.dumps(states2)}")
        (ev / "to-states.json").write_text(json.dumps(states2, indent=2))

        # CAPTURE THE WHOLE TO-STEP LOG *NOW*, not at the end.
        # `docker compose logs` only shows the CONTAINERS THAT EXIST, and the abort below replaces
        # them — so the TO images' own output is GONE from any capture taken afterwards. Measured on
        # E3, where the single most important line of the run ("MariaDB upgrade … required, but
        # skipped") survived only because it had already been extracted. Same class as R-320: the
        # intermediate teardown is the one that loses the evidence.
        full_to = compose(workdir, project, "logs", "--no-color", timeout=180)
        (ev / "to-full.log").write_text((full_to.stdout + full_to.stderr)[-400000:])

        rec["engine_state_after"] = engine_state(e["to"]) or None
        if rec["engine_state_after"]:
            for svc, st in rec["engine_state_after"].items():
                say(f"engine state {svc}: {st['answer'][:180]}")
        (ev / "engine-state.json").write_text(json.dumps(rec["engine_state_after"], indent=2))

        mig = migration_lines(project, workdir, swap_at)
        rec["migration_observed"] = mig[0] if mig else None
        (ev / "migration-lines.txt").write_text("\n".join(mig))
        say(f"migration lines observed: {len(mig)}")

        # --- 5. THE RESULT ---
        rec["seed_read_after"] = bool(fixture.verify(container_ip, seeded, say))
        say(f"RESULT (seed reads back AFTER): {rec['seed_read_after']}")
        rec["verdict"] = "proven" if (ok2 and rec["seed_read_after"]) else "failed"

        # --- 6. the ABORT ---
        say(f"{edge_id}: ABORT — putting the FROM images back")
        render(app, e["frm"], workdir, env)
        up3 = compose(workdir, project, "up", "-d")
        ok3, secs3, states3 = settle(project, workdir, wait=180)
        (ev / "abort-states.json").write_text(json.dumps(states3, indent=2))
        if not ok3:
            rec["abort"] = "refuses"
            lg = compose(workdir, project, "logs", "--tail", "40", "--no-color", timeout=120)
            tail = (lg.stdout + lg.stderr).strip()
            (ev / "abort-refusal.txt").write_text(tail)
            rec["abort_detail"] = tail[-900:]
            say(f"ABORT: the app did NOT come back (rc={up3.returncode}, {secs3}s)")
        else:
            back = bool(fixture.verify(container_ip, seeded, say))
            rec["abort"] = "starts-and-serves" if back else "starts-data-gone"
            rec["abort_detail"] = None if back else "the app started but the seeded data was gone"
            say(f"ABORT: app came back in {secs3}s; data present={back}")
        return rec
    finally:
        rec["measured_at"] = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        rec["duration_s"] = rec["duration_s"] or round(time.time() - t0, 1)
        rec["total_s"] = round(time.time() - t0, 1)
        # EVIDENCE FIRST, TEARDOWN SECOND (R-320): the intermediate teardown is the one that gets
        # forgotten, so everything is written before a single container is removed.
        (ev / "run.log").write_text("\n".join(log))
        lg = compose(workdir, project, "logs", "--no-color", timeout=180)
        (ev / "compose-final.log").write_text((lg.stdout + lg.stderr)[-400000:])  # post-abort state only — see to-full.log
        (ev / "verdict.json").write_text(json.dumps(rec, indent=2))
        compose(workdir, project, "down", "-v", "--remove-orphans", timeout=900)


def main(argv):
    if not argv or argv[0] == "--list":
        for k, v in EDGES.items():
            print(f"{k:5s} {v['app']:12s} {v['note']}")
        return 0
    EVIDENCE.mkdir(parents=True, exist_ok=True)
    results = []
    for edge_id in argv:
        if edge_id not in EDGES:
            print(f"unknown edge {edge_id}", file=sys.stderr)
            return 2
        rec = run_edge(edge_id)
        results.append(rec)
        print(json.dumps(rec, indent=2), flush=True)
    (EVIDENCE / "summary.json").write_text(json.dumps(results, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))