"""B1-O: interleaved fresh/reused Redis admission calls on the evaluation host.

Uses the real shared minute budget (24 calls by default), never resets its keys.
Captures phase/type only, not credentials, command arguments, or exception bodies.
"""

from __future__ import annotations

import argparse
import asyncio
import hashlib
import inspect
import json
import platform
import sys
import time
from collections import deque
from pathlib import Path
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from redis.asyncio import Redis  # noqa: E402
from src.config import get_settings  # noqa: E402
from src.workers.tasks.project_discovery import reserve_discovery_call  # noqa: E402

GATE_SHA256 = "a041b6d43366d65f6196c82c15d27e2088fc3fb2e82b976ef44125b8842d79b0"


def latency_distribution(samples):
    ordered = sorted(samples)
    if not ordered:
        return {"p50": None, "p95": None, "n": 0}

    def percentile(fraction):
        rank = fraction * (len(ordered) - 1)
        lower = int(rank)
        return round(
            ordered[lower]
            + (ordered[min(lower + 1, len(ordered) - 1)] - ordered[lower]) * (rank - lower),
            6,
        )

    return {"p50": percentile(0.5), "p95": percentile(0.95), "n": len(ordered)}


def preflight():
    settings = get_settings()
    fingerprint = hashlib.sha256(inspect.getsource(reserve_discovery_call).encode()).hexdigest()
    if fingerprint != GATE_SHA256:
        raise ValueError("Admission implementation differs from the reviewed B1-S baseline")
    if settings.project_intake_calls_per_minute != 60 or not settings.redis_url:
        raise ValueError("Expected configured Redis and unchanged 60/min admission policy")
    return {
        "preflight": "passed",
        "working_directory": str(ROOT),
        "gate_source_sha256": fingerprint,
        "admission_calls_per_minute": 60,
        "script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        "provider_calls": 0,
        "required_environment_names": ["REDIS_URL"],
        "production_host_attestation": "Operator must run inside the actual worker environment",
    }


def verify(output: Path):
    result = json.loads(output.read_text(encoding="utf-8"))
    assert result["measurement_kind"] == "production_admission_gated"
    assert result["admission_only"] is True and result["provider_calls"] == 0
    assert result["gate_source_sha256"] == GATE_SHA256
    assert len(result["rows"]) == result["samples_per_mode"] * 2
    assert result["policy_changed"] is False
    return {
        "verification": "passed",
        "environment": result["environment"],
        "receipt_sha256": hashlib.sha256(output.read_bytes()).hexdigest(),
        "summary": result["summary"],
        "operation_errors": [r for r in result["rows"] if r["error_type"]],
    }


async def diagnose(samples: int, output: Path, environment: str = "local"):
    preflight()
    if not 1 <= samples <= 12:
        raise ValueError("Use 1 to 12 samples per mode to bound shared-budget consumption")
    if output.exists():
        raise ValueError("Select a new diagnostic receipt")
    settings = get_settings()
    original_factory = Redis.from_url
    events = []
    current = {}

    def event(phase, started, error=None):
        events.append(
            {
                **current,
                "phase": phase,
                "seconds": time.perf_counter() - started,
                "error_type": type(error).__name__ if error else None,
            }
        )

    def client():
        redis = original_factory(settings.redis_url, socket_timeout=2, socket_connect_timeout=2)
        base = redis.connection_pool.connection_class

        class ObservedConnection(base):
            async def _connect(self):
                started = time.perf_counter()
                loop = asyncio.get_running_loop()
                getaddrinfo = loop.getaddrinfo

                async def dns(*args, **kwargs):
                    at = time.perf_counter()
                    try:
                        return await getaddrinfo(*args, **kwargs)
                    finally:
                        event("dns", at)

                try:
                    with patch.object(loop, "getaddrinfo", dns):
                        await super()._connect()
                except Exception as exc:
                    event("connect_including_dns_tls", started, exc)
                    raise
                else:
                    event("connect_including_dns_tls", started)

            async def on_connect(self):
                started = time.perf_counter()
                self.pending = deque()
                try:
                    await super().on_connect()
                finally:
                    event("authentication_and_setup", started)

            async def send_command(self, *args, **kwargs):
                if not hasattr(self, "pending"):
                    self.pending = deque()
                self.pending.append((str(args[0]), time.perf_counter()))
                return await super().send_command(*args, **kwargs)

            async def read_response(self, *args, **kwargs):
                command, started = (
                    self.pending.popleft() if self.pending else ("unknown", time.perf_counter())
                )
                try:
                    result = await super().read_response(*args, **kwargs)
                except Exception as exc:
                    event(f"command:{command}", started, exc)
                    raise
                event(f"command:{command}", started)
                return result

        redis.connection_pool.connection_class = ObservedConnection
        return redis

    pooled = client()
    rows = []

    class Borrowed:
        async def __aenter__(self):
            return pooled

        async def __aexit__(self, *args):
            return False

    async def info():
        try:
            return await pooled.info("errorstats")
        except Exception as exc:
            return {"unavailable": type(exc).__name__}

    # Warm-up is explicit and excluded from the reused steady-state comparison.
    current.update(mode="warmup", sample=-1)
    started = time.perf_counter()
    await pooled.ping()
    warmup = time.perf_counter() - started
    info_before = await info()
    for sample in range(samples):
        for mode in ("fresh", "reused") if sample % 2 == 0 else ("reused", "fresh"):
            current.update(mode=mode, sample=sample)
            owned = client() if mode == "fresh" else None

            class Owned:
                async def __aenter__(self):
                    return owned

                async def __aexit__(self, *args):
                    at = time.perf_counter()
                    try:
                        await owned.aclose()
                    finally:
                        event("close", at)

            started = time.perf_counter()
            error = None
            try:
                with patch.object(Redis, "from_url", return_value=Owned() if owned else Borrowed()):
                    await reserve_discovery_call(settings)
            except Exception as exc:
                error = type(exc).__name__
            rows.append({**current, "seconds": time.perf_counter() - started, "error_type": error})
    current.update(mode="ping", sample=-1)
    ping = []
    for _ in range(5):
        at = time.perf_counter()
        await pooled.ping()
        ping.append(time.perf_counter() - at)
    info_after = await info()
    await pooled.aclose()
    result = {
        "measurement_kind": "production_admission_gated",
        "admission_only": True,
        "environment": environment,
        "provider_calls": 0,
        "gate_source_sha256": GATE_SHA256,
        "host_environment": platform.system(),
        "samples_per_mode": samples,
        "run_id": output.stem,
        "redis_host_ping": (
            "Not measured here; run on Redis host separately if operator access permits"
        ),
        "policy_changed": False,
        "shared_budget_calls": samples * 2,
        "warmup_seconds": warmup,
        "ping_application_host": latency_distribution(ping),
        "errorstats_before": info_before,
        "errorstats_after": info_after,
        "summary": {
            mode: latency_distribution([r["seconds"] for r in rows if r["mode"] == mode])
            for mode in ("fresh", "reused")
        },
        "rows": rows,
        "events": events,
        "timing_note": (
            "Nested timings overlap. TCP/TLS remainder = connect minus DNS; "
            "setup includes AUTH/CLIENT/SELECT. Command time includes network "
            "and scheduling, not server-only execution."
        ),
    }
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(result["summary"], indent=2))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--samples", type=int, default=12)
    parser.add_argument("--out", type=Path)
    parser.add_argument("--environment", choices=("local", "production"), default="local")
    parser.add_argument("--preflight", action="store_true")
    parser.add_argument("--verify", type=Path)
    args = parser.parse_args()
    try:
        if args.preflight:
            print(json.dumps(preflight(), indent=2))
        elif args.verify:
            print(json.dumps(verify(args.verify), indent=2))
        elif args.out:
            asyncio.run(diagnose(args.samples, args.out, args.environment))
        else:
            parser.error("Supply --preflight, --verify RECEIPT, or --out NEW_RECEIPT")
    except Exception as exc:
        # Never print settings, Redis URLs, command arguments, or exception bodies.
        print(json.dumps({"diagnostic_failed": type(exc).__name__}), file=sys.stderr)
        raise SystemExit(1) from None