#!/usr/bin/env python3
"""Observe SWMR visibility, not crash durability. Only creates private temp files."""
import argparse
from datetime import datetime, timezone
import hashlib
import json
import multiprocessing as mp
import os
from pathlib import Path
import platform
import subprocess
import tempfile

import h5py
import numpy as np

TIMEOUT = 20


def receive(pipe):
    if not pipe.poll(TIMEOUT):
        raise TimeoutError("Reader did not reply within the bounded wait")
    message = pipe.recv()
    if "error" in message:
        raise RuntimeError(message["error"])
    return message


def select_prefix(n, x, y):
    if n < 0 or n > min(len(x), len(y)):
        raise ValueError("Publication watermark exceeds available paired rows")
    return [[int(a), int(b)] for a, b in zip(x[:n], y[:n])]


def reader(path, pipe):
    try:
        with h5py.File(path, "r", libver="latest", swmr=True) as f:
            x, y, mark = f["x"], f["y"], f["published"]
            pipe.send({"ready": True})
            while True:
                if not pipe.poll(TIMEOUT):
                    raise TimeoutError("Writer did not send the next command")
                phase = pipe.recv()
                if phase == "stop":
                    break
                # Read the watermark first, then refresh payload datasets.
                mark.refresh()
                n = int(mark[0])
                x.refresh()
                y.refresh()
                a, b = x[:].tolist(), y[:].tolist()
                try:
                    prefix = select_prefix(n, a, b)
                    rejected = False
                except ValueError:
                    prefix, rejected = None, True
                pipe.send({
                    "phase": phase, "lengths": [len(a), len(b)],
                    "watermark": n, "x": a, "y": b,
                    "selectedPrefix": prefix, "rejected": rejected,
                })
    except BaseException as exc:
        pipe.send({"error": type(exc).__name__ + ": " + str(exc)})
        raise
    finally:
        pipe.close()


def validate_observations(rows):
    """Reject unexpected measurements without relying on removable assertions."""
    expected = [
        ("initial", [], [], 0, []),
        ("x1-flushed", [10], [], 0, []),
        ("y1-flushed-unpublished", [10], [100], 0, []),
        ("published-1", [10], [100], 1, [[10, 100]]),
        ("x2-flushed", [10, 20], [100], 1, [[10, 100]]),
        ("y2-flushed-unpublished", [10, 20], [100, 200], 1, [[10, 100]]),
        ("published-2", [10, 20], [100, 200], 2, [[10, 100], [20, 200]]),
        ("invalid-watermark", [10, 20], [100, 200], 3, None),
        ("mutated-published-prefix", [999, 20], [100, 200], 2, [[999, 100], [20, 200]]),
    ]
    if not isinstance(rows, list) or len(rows) != len(expected):
        raise RuntimeError("Expected exactly nine ordered observations")
    for row, (phase, x, y, watermark, prefix) in zip(rows, expected, strict=True):
        wanted = {
            "phase": phase, "lengths": [len(x), len(y)], "watermark": watermark,
            "x": x, "y": y, "selectedPrefix": prefix, "rejected": prefix is None,
        }
        if row != wanted:
            raise RuntimeError(f"Unexpected observation at {phase}: {row!r}")


def trial(directory):
    ctx = mp.get_context("spawn")
    parent, child = ctx.Pipe()
    process = None
    rows = []
    path = Path(directory) / "observation.h5"
    with h5py.File(path, "w", libver="latest", driver="sec2") as f:
        x = f.create_dataset("x", (0,), maxshape=(None,), chunks=(1,), dtype="i8")
        y = f.create_dataset("y", (0,), maxshape=(None,), chunks=(1,), dtype="i8")
        mark = f.create_dataset("published", data=np.array([0], dtype="u8"), chunks=(1,))
        f.swmr_mode = True
        process = ctx.Process(target=reader, args=(str(path), child))
        process.start()
        child.close()
        try:
            ready = receive(parent)
            if ready != {"ready": True}:
                raise RuntimeError(f"Unexpected reader handshake: {ready!r}")

            def observe(phase):
                parent.send(phase)
                row = receive(parent)
                if row.get("phase") != phase:
                    raise RuntimeError(f"Out-of-order reader response: {row!r}")
                rows.append(row)

            def append(dataset, value):
                n = len(dataset)
                dataset.resize((n + 1,))
                dataset[n] = value
                dataset.flush()

            observe("initial")
            append(x, 10)
            observe("x1-flushed")
            append(y, 100)
            observe("y1-flushed-unpublished")
            mark[0] = 1
            mark.flush()
            observe("published-1")
            append(x, 20)
            observe("x2-flushed")
            append(y, 200)
            observe("y2-flushed-unpublished")
            mark[0] = 2
            mark.flush()
            observe("published-2")
            # Deliberate protocol violations, not proposed production behavior.
            mark[0] = 3
            mark.flush()
            observe("invalid-watermark")
            mark[0] = 2
            mark.flush()
            x[0] = 999
            x.flush()
            observe("mutated-published-prefix")
            parent.send("stop")
            process.join(TIMEOUT)
            if process.is_alive() or process.exitcode != 0:
                raise RuntimeError("Reader did not exit successfully")
        finally:
            if process.is_alive():
                process.terminate()
                process.join(TIMEOUT)
            parent.close()
    validate_observations(rows)
    return rows


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repetitions", type=int, default=3)
    parser.add_argument("--temp-dir")
    args = parser.parse_args()
    if not 1 <= args.repetitions <= 100:
        parser.error("--repetitions must be between 1 and 100")
    with tempfile.TemporaryDirectory(prefix="hesela-swmr-", dir=args.temp_dir) as directory:
        filesystem = subprocess.check_output(
            ["stat", "-f", "-c", "%T", directory], text=True
        ).strip()
        runs = []
        for _ in range(args.repetitions):
            with tempfile.TemporaryDirectory(dir=directory) as run_dir:
                runs.append(trial(run_dir))
    result = {
        "schemaVersion": 1,
        "experiment": "hdf5-swmr-visibility",
        "evidenceType": "measured-api-behavior",
        "generatedAt": datetime.now(timezone.utc).isoformat(),
        "repetitions": args.repetitions, "checkpointsPerRepetition": 9,
        "environment": {
            "python": platform.python_version(), "kernel": platform.release(),
            "machine": platform.machine(), "h5py": h5py.__version__,
            "hdf5": h5py.version.hdf5_version, "numpy": np.__version__,
            "filesystem": filesystem, "driver": "sec2", "libver": "latest",
            "multiprocessing": "spawn",
            "HDF5_USE_FILE_LOCKING": os.environ.get("HDF5_USE_FILE_LOCKING"),
            "HDF5_IGNORE_DISABLED_FILE_LOCKS": os.environ.get("HDF5_IGNORE_DISABLED_FILE_LOCKS"),
        },
        "sourceSha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        "method": "One live writer and one spawned reader; pipe handshakes pause writes at each observation; all datasets created before SWMR; fixed-size integers, one-element chunks, no compression.",
        "limits": "One host. Synthetic values are test inputs, not sensor measurements. No power loss, process crash, fsync audit, network filesystem, competing writers or performance test. Watermark ordering is an application demonstration, not a crash-safe transaction protocol.",
        "observations": runs,
    }
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()