#!/usr/bin/env python3
"""Offline high-cardinality invalidation replay + HTTP load-driver smoke test.

The replay models bookkeeping cardinality, not a running RustProxy heap. The HTTP
requests target a Python test peer solely to validate the staging load harness.
"""
from __future__ import annotations

import argparse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import os
from pathlib import Path
import sys
import threading

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "dev"))
from memory_load import run_load  # noqa: E402


def purge_replay(operations: int = 300_000, windows: int = 6) -> list[tuple[int, int]]:
    old_generations = {}
    new_generations = {}
    counts = []
    assert operations % windows == 0
    for i in range(operations):
        key = i.to_bytes(32, "big")
        # Legacy: each already-removed key keeps a generation slot forever.
        old_generations[key] = 1
        # Same-version fix: the global generation barrier cancels all older
        # tickets, so only the current purge's per-key metadata is needed.
        new_generations.clear()
        new_generations[key] = 1
        if (i + 1) % (operations // windows) == 0:
            counts.append((len(old_generations), len(new_generations)))
    assert counts == [(operations // windows * i, 1) for i in range(1, windows + 1)]
    return counts


class TestPeer(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    def do_GET(self) -> None:
        body = b"HTTP-load-driver-ok"
        self.send_response(200)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *_: object) -> None:
        pass


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--http-requests", type=int, default=2_000, help="HTTP requests against the local Python test peer (default: 2000)")
    args = parser.parse_args()
    if not 1 <= args.http_requests <= 1_000_000:
        parser.error("--http-requests must be 1..1000000")
    checkpoints = purge_replay()
    print("PASS: 300,000 simulated distinct PURGEs; cumulative vs reset per-key generation slots:", checkpoints)
    peer = ThreadingHTTPServer(("127.0.0.1", 0), TestPeer)
    runner = threading.Thread(target=peer.serve_forever, daemon=True)
    runner.start()
    try:
        url = f"http://127.0.0.1:{peer.server_port}/"
        report = run_load(url=url, requests=args.http_requests, concurrency=24, unique_paths=min(1000, args.http_requests),
                          timeout=5, pid=os.getpid(), sample_interval=0.05, cooldown=0.15)
        assert report["requests_sent"] == args.http_requests, report
        assert report["statuses"] == {"200": args.http_requests}, report
        assert report["response_bytes_read"] == args.http_requests * len(b"HTTP-load-driver-ok"), report
        assert report["rss_peak_mib"] is not None, report
        print(f"PASS: load tool generated {args.http_requests:,} real HTTP requests with 24 connections and {min(1000, args.http_requests):,} paths against a Python test peer")
        print("NOTE: peer/load-tool RSS is NOT RustProxy RSS; live RustProxy profiling remains mandatory")
    finally:
        peer.shutdown()
        peer.server_close()
        runner.join(timeout=5)


if __name__ == "__main__":
    main()