#!/usr/bin/env python3
"""End-to-end correctness scenarios for the hot-reload runner.

Drives the real example with the built kan-hot-dev and asserts on its output:
nested const and function-body reloads, inlined-body correctness, replay
fallback, content dedupe, in-flight streaming across a reload, runtime-change
restart, and builder panics keeping the previous generation.
"""
import os, re, subprocess, sys, threading, time

ROOT = "/Users/oscar/.worktree/hot-reload/labs/hot-reload"
CLI = "/Users/oscar/.worktree/hot-reload/target/debug/kan-hot-dev"
COMMON = os.path.join(ROOT, "api-common/src/lib.rs")
API = os.path.join(ROOT, "api/src/lib.rs")
SHARED = os.path.join(ROOT, "shared/src/lib.rs")
RECORD = os.path.join(ROOT, "target/kan-hot/debug/rustc")

originals = {p: open(p).read() for p in (COMMON, API, SHARED)}
failures = []


def check(name, condition, detail=""):
    print(("PASS " if condition else "FAIL ") + name + (f"  [{detail}]" if detail and not condition else ""))
    if not condition:
        failures.append(name)


env = dict(os.environ, RUST_LOG="info", KAN_HOT_TIMING="1",
           KAN_HOT_EXAMPLE_INTERVAL_MS="100", KAN_HOT_EXAMPLE_SLOW="1")
process = subprocess.Popen([CLI, "run", "--bin", "server", "--lib", "api"], cwd=ROOT,
                           stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                           stderr=subprocess.STDOUT, env=env, text=True, bufsize=1)
lines = []
lock = threading.Lock()


def reader():
    for line in process.stdout:
        with lock:
            lines.append((time.monotonic(), line))
        sys.stdout.write(line)
        sys.stdout.flush()


threading.Thread(target=reader, daemon=True).start()


def snapshot():
    with lock:
        return list(lines)


def wait(pred, timeout=30.0, since=0.0):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        for stamp, line in reversed(snapshot()):
            if stamp >= since and pred(line):
                return stamp, line
        time.sleep(0.002)
    raise TimeoutError("timed out")


def write(path, content):
    with open(path, "w") as handle:
        handle.write(content)


def count(pred, since):
    return sum(1 for stamp, line in snapshot() if stamp >= since and pred(line))


try:
    wait(lambda l: "gen 1 hello" in l)

    # 1. Nested const reload (the benchmark's edit).
    t0 = time.monotonic()
    write(COMMON, originals[COMMON].replace("nested-v1", "const-probe"))
    wait(lambda l: '"nested":"const-probe"' in l, since=t0)
    check("nested const reload", True)
    check("in-memory loader used (not dlopen fallback)",
          any(re.search(r"loaded generation \d+ in (\d+\.\d+)ms", l) and
              float(re.search(r"in (\d+\.\d+)ms", l).group(1)) < 50 for _, l in snapshot()))
    check("state preserved across reload (calls kept counting)",
          any('"calls":' in l and '"nested":"const-probe"' in l for _, l in snapshot()))

    # 2. Changed function body in the nested crate, inlined into `api`.
    body = originals[COMMON].replace('format!("hello {name} ({BUILD_LABEL})")',
                                     'format!("body-probe {name} ({BUILD_LABEL})")')
    t0 = time.monotonic()
    write(COMMON, body.replace("nested-v1", "const-probe"))
    wait(lambda l: '"message":"body-probe world (const-probe)"' in l, since=t0)
    check("nested function-body reload (inlined callee)", True)

    # 3. Replay fallback: remove a recording, edit, expect a Cargo build.
    os.rename(os.path.join(RECORD, "api_common.json"), os.path.join(RECORD, "api_common.json.off"))
    t0 = time.monotonic()
    write(COMMON, body.replace("nested-v1", "fallback-probe"))
    wait(lambda l: '"nested":"fallback-probe"' in l, since=t0)
    fallback_build = [l for s, l in snapshot() if s >= t0 and "cargo hot build" in l]
    check("replay fallback reloads", True)
    check("replay fallback used Cargo (no 'rustc replay')",
          any("rustc replay" not in l for l in fallback_build), str(fallback_build))
    os.rename(os.path.join(RECORD, "api_common.json.off"), os.path.join(RECORD, "api_common.json"))

    # 4. Content dedupe: writing identical bytes must not schedule a build.
    t0 = time.monotonic()
    write(COMMON, body.replace("nested-v1", "fallback-probe"))
    time.sleep(0.8)
    check("content dedupe suppresses identical write",
          count(lambda l: "change detected" in l, t0) == 0)

    # 5. In-flight streaming survives a reload (slow stream started before it).
    t0 = time.monotonic()
    write(COMMON, body.replace("nested-v1", "stream-probe"))
    wait(lambda l: '"nested":"stream-probe"' in l, since=t0)
    slow = None
    deadline = time.monotonic() + 12
    while time.monotonic() < deadline:
        started = [l for s, l in snapshot() if s >= t0 and "slow stream started" in l]
        for line in started:
            m = re.search(r"started at (\d+) finished at (\d+) with (\d+) ticks", line)
            if m and int(m.group(2)) > int(m.group(1)):
                slow = (int(m.group(1)), int(m.group(2)), int(m.group(3)))
        if slow:
            break
        time.sleep(0.05)
    check("in-flight stream spans a reload with all ticks",
          slow is not None and slow[2] == 10, str(slow))

    # 6. Builder panic keeps the previous generation serving.
    panic_source = originals[API].replace(
        "pub fn build_router() -> Router<Json, AppState> {",
        "pub fn build_router() -> Router<Json, AppState> {\n    if std::env::var_os(\"NOPE\").is_none() { panic!(\"probe panic\") }")
    t0 = time.monotonic()
    write(API, panic_source)
    wait(lambda l: "reload failed" in l, since=t0, timeout=30)
    check("builder panic is reported as a failed reload",
          any("previous generation is still active" in l for s, l in snapshot() if s >= t0))
    wait(lambda l: '"nested":"stream-probe"' in l, since=t0, timeout=20)
    check("builder panic keeps previous generation serving", True)
    t0 = time.monotonic()
    write(API, originals[API])
    wait(lambda l: "loaded generation" in l, since=t0, timeout=30)
    check("recovering from a builder panic reloads again", True)

    # 7. Runtime crate edit restarts the host (state resets).
    t0 = time.monotonic()
    write(SHARED, originals[SHARED] + "\n// restart probe\n")
    wait(lambda l: "rebuilding host" in l, since=t0, timeout=60)
    wait(lambda l: "generation: 1" in l, since=t0, timeout=120)
    restarted = [l for s, l in snapshot() if s >= t0 and "kan-hot-dev: host exited" not in l]
    check("runtime change restarts the host", True)
finally:
    for path, content in originals.items():
        write(path, content)
    time.sleep(0.5)
    try:
        process.stdin.write("q\n")
        process.stdin.flush()
    except BrokenPipeError:
        pass
    try:
        process.wait(timeout=10)
    except subprocess.TimeoutExpired:
        process.kill()

print()
if failures:
    print("FAILED:", failures)
    sys.exit(1)
print("all scenarios passed")