#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
import queue
import re
import signal
import subprocess
import sys
import threading
import time
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import TextIO


@dataclass(frozen=True)
class Scenario:
    id: str
    name: str


SCENARIOS = (
    Scenario("1", "equal versions, both unlocked, both manifest-v5"),
    Scenario("2", "unequal data versions, both unlocked, both manifest-v5"),
    Scenario("3", "equal versions, one manifest-v5 and one manifest-v3"),
    Scenario("4", "unequal data versions, one manifest-v5 and one manifest-v3"),
    Scenario("9", "manifest negotiation: no peer sends SYNC_DEVICE_MANIFEST unilaterally"),
    Scenario("13", "vault id mismatch stops cleanly"),
    Scenario("5", "browser locked (spool active), android unlocked, unequal versions"),
    Scenario("6", "android locked (listener running), browser unlocked"),
    Scenario("8", "both sides locked"),
    Scenario("10", "process death mid-session, next round completes cleanly"),
    Scenario("12", "manifest with a device vouched too deeply is rejected"),
    Scenario("14", "manifest issued outside the freshness window is rejected"),
    Scenario("15", "10,000-entry vault with padded frames"),
    Scenario("16.3.1", "stale issuedAt → rejected"),
    Scenario("16.3.2", "future issuedAt → rejected"),
    Scenario("16.3.3", "vouchDepth above the absolute bound → rejected"),
    Scenario("16.3.4", "device count above the bound → rejected"),
    Scenario("16.3.5", "malformed device key → rejected"),
    Scenario("16.3.6", "oversized issuer id → rejected"),
    Scenario("16.3.7", "signature from an untrusted key → rejected"),
    Scenario("16.3.8", "valid manifest (control) → accepted"),
)

COVERAGE_EXCLUSIONS = (
    {
        "id": "7",
        "disposition": "external_evidence_required",
        "reason": "cold-locked Android requires unit and real-device evidence",
    },
    {
        "id": "11",
        "disposition": "external_evidence_required",
        "reason": "revocation then re-vouch requires storage-adapter evidence",
    },
)

RESULT_PATTERN = re.compile(r"^\s*✓\s+(\S+)\s+(.+?)\s+\(([0-9.]+)s\)\s*$")


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def emit_json(event: dict[str, object]) -> None:
    print(json.dumps(event, separators=(",", ":")), flush=True)


def terminate_process_group(process: subprocess.Popen[str]) -> None:
    if process.poll() is not None:
        return
    try:
        os.killpg(process.pid, signal.SIGTERM)
        process.wait(timeout=0.2)
    except ProcessLookupError:
        return
    except subprocess.TimeoutExpired:
        try:
            os.killpg(process.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        process.wait()


def write_summary(path: Path, summary: dict[str, object]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    temporary.write_text(f"{json.dumps(summary, indent=2)}\n", encoding="utf-8")
    os.replace(temporary, path)


def run_gate(
    command: Sequence[str],
    summary_path: Path,
    *,
    scenario_timeout_seconds: float,
    total_timeout_seconds: float,
    emit: Callable[[dict[str, object]], None] = emit_json,
    raw_output: TextIO = sys.stderr,
) -> int:
    started_at = utc_now()
    gate_started = time.monotonic()
    records: list[dict[str, object]] = [
        {
            "id": scenario.id,
            "name": scenario.name,
            "status": "pending",
            "startedAt": None,
            "finishedAt": None,
        }
        for scenario in SCENARIOS
    ]
    current_index = 0
    current_started = gate_started
    runner_status = "start_failed"
    runner_exit_code: int | None = None

    def start_scenario(index: int) -> None:
        nonlocal current_started
        current_started = time.monotonic()
        records[index]["startedAt"] = utc_now()
        emit(
            {
                "event": "scenario_start",
                "id": SCENARIOS[index].id,
                "name": SCENARIOS[index].name,
                "timeoutSeconds": scenario_timeout_seconds,
            }
        )

    def finish_scenario(
        index: int,
        status: str,
        reason: str | None = None,
        runner_seconds: float | None = None,
    ) -> None:
        record = records[index]
        record["status"] = status
        record["finishedAt"] = utc_now()
        if record["startedAt"] is not None:
            record["elapsedSeconds"] = round(time.monotonic() - current_started, 3)
        if reason is not None:
            record["reason"] = reason
        if runner_seconds is not None:
            record["runnerReportedSeconds"] = runner_seconds
        event: dict[str, object] = {
            "event": "scenario_finish",
            "id": record["id"],
            "name": record["name"],
            "status": status,
            "started": record["startedAt"] is not None,
        }
        if reason is not None:
            event["reason"] = reason
        emit(event)

    def fail_unreported(reason: str) -> None:
        for index in range(current_index, len(records)):
            if records[index]["status"] == "pending":
                finish_scenario(index, "fail", reason)

    process: subprocess.Popen[str] | None = None
    reader_thread: threading.Thread | None = None
    try:
        process = subprocess.Popen(
            list(command),
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
            start_new_session=True,
        )
        if process.stdout is None:
            raise RuntimeError("runner stdout pipe was not created")

        output_queue: queue.Queue[str | None] = queue.Queue()

        def read_output() -> None:
            for line in process.stdout:
                output_queue.put(line)
            output_queue.put(None)

        reader_thread = threading.Thread(target=read_output, daemon=True)
        reader_thread.start()
        runner_status = "running"
        start_scenario(current_index)

        while True:
            now = time.monotonic()
            total_remaining = total_timeout_seconds - (now - gate_started)
            scenario_remaining = (
                scenario_timeout_seconds - (now - current_started)
                if current_index < len(records)
                else total_remaining
            )
            wait_seconds = min(total_remaining, scenario_remaining)
            if wait_seconds <= 0:
                if current_index < len(records):
                    reason = "total_timeout" if total_remaining <= 0 else "scenario_timeout"
                    finish_scenario(current_index, "timeout", reason)
                    current_index += 1
                    fail_unreported(f"not_started_after_{reason}")
                runner_status = "timeout"
                terminate_process_group(process)
                break

            try:
                line = output_queue.get(timeout=wait_seconds)
            except queue.Empty:
                continue

            if line is None:
                runner_exit_code = process.wait()
                runner_status = "pass" if runner_exit_code == 0 else f"exit_{runner_exit_code}"
                if current_index < len(records):
                    fail_unreported(
                        "runner_exited_before_result"
                        if runner_exit_code == 0
                        else f"runner_exit_{runner_exit_code}"
                    )
                break

            raw_output.write(line)
            raw_output.flush()
            match = RESULT_PATTERN.match(line)
            if match is None:
                continue

            reported_id, reported_name, reported_seconds = match.groups()
            if current_index >= len(SCENARIOS):
                runner_status = "unexpected_result"
                terminate_process_group(process)
                break

            expected = SCENARIOS[current_index]
            if reported_id != expected.id or reported_name != expected.name:
                finish_scenario(
                    current_index,
                    "fail",
                    f"unexpected_result:{reported_id}:{reported_name}",
                )
                current_index += 1
                fail_unreported("not_started_after_unexpected_result")
                runner_status = "unexpected_result"
                terminate_process_group(process)
                break

            finish_scenario(current_index, "pass", runner_seconds=float(reported_seconds))
            current_index += 1
            if current_index < len(SCENARIOS):
                start_scenario(current_index)
    except OSError as error:
        fail_unreported(f"runner_start_error:{error.__class__.__name__}")
        runner_status = "start_failed"
    finally:
        if process is not None and process.poll() is None:
            terminate_process_group(process)
        if reader_thread is not None:
            reader_thread.join(timeout=0.2)
        if process is not None and process.stdout is not None:
            process.stdout.close()

    counts = {
        status: sum(record["status"] == status for record in records)
        for status in ("pass", "fail", "timeout")
    }
    overall_status = (
        "pass"
        if runner_status == "pass" and counts == {"pass": len(records), "fail": 0, "timeout": 0}
        else "fail"
    )
    summary: dict[str, object] = {
        "schemaVersion": 1,
        "startedAt": started_at,
        "finishedAt": utc_now(),
        "overallStatus": overall_status,
        "runnerStatus": runner_status,
        "runnerExitCode": runner_exit_code,
        "scenarioTimeoutSeconds": scenario_timeout_seconds,
        "totalTimeoutSeconds": total_timeout_seconds,
        "counts": counts,
        "scenarios": records,
        "coverageExclusions": list(COVERAGE_EXCLUSIONS),
    }
    write_summary(summary_path, summary)
    emit(
        {
            "event": "summary",
            "overallStatus": overall_status,
            "runnerStatus": runner_status,
            "counts": counts,
            "path": str(summary_path),
        }
    )
    return 0 if overall_status == "pass" else 1


def positive_seconds(value: str) -> float:
    seconds = float(value)
    if seconds <= 0:
        raise argparse.ArgumentTypeError("must be greater than zero")
    return seconds


def main() -> int:
    script_dir = Path(__file__).resolve().parent
    parser = argparse.ArgumentParser(description="Run the bounded cross-client sync matrix")
    parser.add_argument(
        "--scenario-timeout-seconds",
        type=positive_seconds,
        default=300.0,
        help="deadline for each process scenario (default: 300)",
    )
    parser.add_argument(
        "--total-timeout-seconds",
        type=positive_seconds,
        default=1200.0,
        help="deadline for the complete process matrix (default: 1200)",
    )
    parser.add_argument(
        "--summary",
        type=Path,
        default=script_dir.parent / "build/reports/cross-client-sync-summary.json",
        help="machine-readable result path",
    )
    arguments = parser.parse_args()
    runner = script_dir.parent.parent / "peach-browser/scripts/check-cross-client-sync.mjs"
    return run_gate(
        ["node", str(runner)],
        arguments.summary,
        scenario_timeout_seconds=arguments.scenario_timeout_seconds,
        total_timeout_seconds=arguments.total_timeout_seconds,
    )


if __name__ == "__main__":
    raise SystemExit(main())