#!/usr/bin/env python3
"""Deterministic cache-fill model, not a distributed cache or load benchmark.

Run with Python 3.10+: python3 experiment.py
Writes results.json, results.csv and trace.json beside this script.
All requests in a wave arrive before any origin operation completes.
Processes are labels; no network, threads, leases or wall-clock timings exist.
"""
import csv
import json
from pathlib import Path


class Model:
    def __init__(self, policy):
        self.policy = policy
        self.cache = {}
        self.pending = {}
        self.trace = []

    def wave(self, case, keys, fail=False):
        requests = []
        fills = []
        joins = 0
        for key in keys:
            for i in range(12):
                process = i % 3
                request = {"key": key, "process": process}
                requests.append(request)
                if key in self.cache:
                    request["value"] = self.cache[key]
                    self.trace.append({"case": case, "event": "hit", "key": key})
                    continue
                token = (key, len(requests)) if self.policy == "none" else (
                    (key, process) if self.policy == "per-process" else (key,)
                )
                flight = self.pending.get(token)
                if flight is None:
                    flight = {"key": key, "token": token, "waiters": []}
                    self.pending[token] = flight
                    fills.append(flight)
                    self.trace.append({"case": case, "event": "start", "key": key,
                                       "process": process})
                else:
                    joins += 1
                    self.trace.append({"case": case, "event": "join", "key": key,
                                       "process": process})
                flight["waiters"].append(request)

        # The explicit completion barrier makes overlap reproducible.
        for flight in fills:
            key = flight["key"]
            try:
                if fail:
                    for request in flight["waiters"]:
                        request["error"] = "origin-unavailable"
                else:
                    value = "value-for-" + key
                    self.cache[key] = value
                    for request in flight["waiters"]:
                        request["value"] = value
            finally:
                del self.pending[flight["token"]]
            self.trace.append({"case": case, "event": "fail" if fail else "publish",
                               "key": key, "waiters": len(flight["waiters"])})

        assert not self.pending, "Completed flights must leave no pending entry"
        for request in requests:
            if "value" in request:
                assert request["value"] == "value-for-" + request["key"]
        return {"case": case, "policy": self.policy, "requests": len(requests),
                "origin_fills": len(fills), "joined_waiters": joins,
                "successes": sum("value" in r for r in requests),
                "errors": sum("error" in r for r in requests)}


def main():
    rows = []
    trace = []
    key_a = "tenant-a:catalog:v1"
    key_b = "tenant-b:catalog:v1"
    for keys in ([key_a], [key_a, key_b]):
        for policy, per_key in [("none", 12), ("per-process", 3), ("shared-model", 1)]:
            model = Model(policy)
            row = model.wave("cold-" + str(len(keys)) + "-key", keys)
            assert row["origin_fills"] == len(keys) * per_key
            assert row["successes"] == len(keys) * 12 and row["errors"] == 0
            rows.append(row)
            trace.extend(dict(policy=policy, **event) for event in model.trace)

    model = Model("shared-model")
    failed = model.wave("failed-fill", [key_a], fail=True)
    assert failed["origin_fills"] == 1 and failed["errors"] == 12
    assert key_a not in model.cache
    retry = model.wave("next-wave-retry", [key_a])
    assert retry["origin_fills"] == 1 and retry["successes"] == 12
    warm = model.wave("warm-cache", [key_a])
    assert warm["origin_fills"] == 0 and warm["successes"] == 12
    rows.extend([failed, retry, warm])
    trace.extend(dict(policy=model.policy, **event) for event in model.trace)
    report = {"scope": "Single-process deterministic model with three process labels",
              "overlap": "Every request arrives before origin completion",
              "not_tested": ["distributed leases", "process failure", "latency",
                             "waiter cancellation", "cache eviction", "network partitions"],
              "cases": rows, "assertions": "passed"}
    root = Path(__file__).resolve().parent
    (root / "results.json").write_text(json.dumps(report, indent=2) + "\n")
    (root / "trace.json").write_text(json.dumps(trace, indent=2) + "\n")
    with (root / "results.csv").open("w", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()