from __future__ import annotations

import argparse
import asyncio
import csv
import json
import math
import statistics
import time
from collections import Counter
from pathlib import Path
from typing import Any

import httpx


PROFILE_DEFAULTS = {
    "baseline_100_100": {"voice": 100, "digital": 100, "ramp_seconds": 15, "hold_seconds": 60},
    "step_250_250": {"voice": 250, "digital": 250, "ramp_seconds": 30, "hold_seconds": 120},
    "target_500_500": {"voice": 500, "digital": 500, "ramp_seconds": 60, "hold_seconds": 180},
}
DIGITAL_DISTRIBUTION = (("telegram", 0.4), ("webchat", 0.3), ("email", 0.3))
READ_TRAFFIC_INTERVAL_SECONDS = 1.0
MAX_ERROR_SAMPLES = 200
WORKER_PAUSE_SECONDS = 1.0


def resolve_profile(
    name: str,
    ramp_seconds: int | None = None,
    hold_seconds: int | None = None,
    voice: int | None = None,
    digital: int | None = None,
) -> dict[str, int | str]:
    if name not in PROFILE_DEFAULTS:
        raise ValueError(f"Unsupported profile: {name}")
    profile = dict(PROFILE_DEFAULTS[name])
    profile["profile"] = name
    if ramp_seconds is not None:
        profile["ramp_seconds"] = ramp_seconds
    if hold_seconds is not None:
        profile["hold_seconds"] = hold_seconds
    if voice is not None:
        profile["voice"] = voice
    if digital is not None:
        profile["digital"] = digital
    return profile


def split_digital_mix(total: int) -> dict[str, int]:
    remaining = total
    result: dict[str, int] = {}
    for idx, (name, ratio) in enumerate(DIGITAL_DISTRIBUTION):
        if idx == len(DIGITAL_DISTRIBUTION) - 1:
            count = remaining
        else:
            count = int(round(total * ratio))
            remaining -= count
        result[name] = count
    return result


def build_mix_profile(profile: dict[str, int | str], include_read_traffic: bool) -> dict[str, Any]:
    voice = int(profile["voice"])
    digital = int(profile["digital"])
    digital_mix = split_digital_mix(digital)
    payload = {
        "profile": profile["profile"],
        "voice_workers": voice,
        "digital_workers": digital,
        "ramp_seconds": int(profile["ramp_seconds"]),
        "hold_seconds": int(profile["hold_seconds"]),
        "channels": {"voice": voice, **digital_mix},
        "include_read_traffic": include_read_traffic,
        "read_endpoints": [
            "/proxy/supervisor/supervisor/realtime",
            "/proxy/reporting/reports/kpi",
        ]
        if include_read_traffic
        else [],
    }
    return payload


def _build_voice_payload(idx: int) -> tuple[str, str, dict[str, Any]]:
    return (
        "POST",
        "/proxy/interaction/interactions",
        {
            "channel": "voice",
            "subject": f"Track7 voice load #{idx}",
            "customer_id": None,
            "queue_id": "q_load_voice",
            "priority": 3,
        },
    )


def _build_telegram_payload(idx: int) -> tuple[str, str, dict[str, Any]]:
    return (
        "POST",
        "/proxy/telegram/integrations/telegram/webhook",
        {
            "chat_id": f"load_chat_{idx}",
            "text": f"Track7 telegram load #{idx}",
            "customer_external_id": None,
            "payload": {"source": "track7-load"},
        },
    )


def _build_webchat_payload(idx: int) -> tuple[str, str, dict[str, Any]]:
    return (
        "POST",
        "/proxy/webchat/integrations/webchat/messages",
        {
            "session_id": f"load_session_{idx}",
            "text": f"Track7 webchat load #{idx}",
            "visitor_name": "Load Test Visitor",
            "customer_external_id": None,
            "queue_id": "q_load_webchat",
            "priority": 3,
            "payload": {"source": "track7-load"},
        },
    )


def _build_email_payload(idx: int) -> tuple[str, str, dict[str, Any]]:
    return (
        "POST",
        "/proxy/email/integrations/email/messages",
        {
            "from_email": f"load{idx}@example.test",
            "subject": f"Track7 email load #{idx}",
            "body": "Synthetic email for scale validation.",
            "customer_external_id": None,
            "queue_id": "q_load_email",
            "priority": 3,
            "payload": {"source": "track7-load"},
        },
    )


def build_request(target: str, idx: int) -> tuple[str, str, dict[str, Any]]:
    if target == "voice":
        return _build_voice_payload(idx)
    if target == "telegram":
        return _build_telegram_payload(idx)
    if target == "webchat":
        return _build_webchat_payload(idx)
    if target == "email":
        return _build_email_payload(idx)
    raise ValueError(f"Unsupported target: {target}")


def build_worker_targets(profile: dict[str, int | str]) -> list[str]:
    targets = ["voice"] * int(profile["voice"])
    digital_mix = split_digital_mix(int(profile["digital"]))
    for name, count in digital_mix.items():
        targets.extend([name] * count)
    return targets


def percentile(samples: list[float], ratio: float) -> float:
    if not samples:
        return 0.0
    sorted_samples = sorted(samples)
    index = max(0, math.ceil(len(sorted_samples) * ratio) - 1)
    return sorted_samples[index]


def ensure_report_dir(report_dir: str | None = None) -> Path:
    if report_dir:
        target = Path(report_dir)
    else:
        timestamp = time.strftime("%Y%m%d_%H%M%S")
        target = Path(".artifacts") / "track7" / timestamp
    target.mkdir(parents=True, exist_ok=True)
    return target


async def build_auth_headers(client: httpx.AsyncClient, auth_mode: str) -> dict[str, str]:
    if auth_mode == "legacy_headers":
        return {"X-User": "admin", "X-Role": "admin"}
    if auth_mode != "bearer":
        raise ValueError(f"Unsupported auth mode: {auth_mode}")

    response = await client.post(
        "/proxy/auth/auth/login",
        json={"username": "admin", "password": "admin123"},
        timeout=10,
    )
    response.raise_for_status()
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}


def _record_result(
    results: list[dict[str, Any]],
    errors: list[dict[str, Any]],
    target: str,
    started_at: float,
    elapsed: float,
    status_code: int,
    ok: bool,
    error_message: str | None = None,
) -> None:
    results.append(
        {
            "ts": round(started_at, 6),
            "target": target,
            "status_code": status_code,
            "latency_seconds": round(elapsed, 6),
            "ok": ok,
        }
    )
    if (not ok or status_code >= 500 or error_message) and len(errors) < MAX_ERROR_SAMPLES:
        errors.append(
            {
                "ts": round(started_at, 6),
                "target": target,
                "status_code": status_code,
                "latency_seconds": round(elapsed, 6),
                "error": error_message,
            }
        )


async def _run_worker(
    client: httpx.AsyncClient,
    headers: dict[str, str],
    target: str,
    idx: int,
    stop_event: asyncio.Event,
    results: list[dict[str, Any]],
    errors: list[dict[str, Any]],
) -> None:
    method, path, payload = build_request(target, idx)
    while not stop_event.is_set():
        started = time.perf_counter()
        try:
            response = await client.request(method, path, headers=headers, json=payload, timeout=20)
            elapsed = time.perf_counter() - started
            ok = response.status_code < 500 and response.status_code < 400
            _record_result(results, errors, target, started, elapsed, response.status_code, ok)
        except Exception as exc:
            elapsed = time.perf_counter() - started
            _record_result(results, errors, target, started, elapsed, 0, False, str(exc))
        await asyncio.sleep(WORKER_PAUSE_SECONDS)


async def _run_read_probe(
    client: httpx.AsyncClient,
    headers: dict[str, str],
    path: str,
    stop_event: asyncio.Event,
    results: list[dict[str, Any]],
    errors: list[dict[str, Any]],
) -> None:
    target = f"read:{path.rsplit('/', 1)[-1]}"
    while not stop_event.is_set():
        started = time.perf_counter()
        try:
            response = await client.get(path, headers=headers, timeout=20)
            elapsed = time.perf_counter() - started
            ok = response.status_code < 500 and response.status_code < 400
            _record_result(results, errors, target, started, elapsed, response.status_code, ok)
        except Exception as exc:
            elapsed = time.perf_counter() - started
            _record_result(results, errors, target, started, elapsed, 0, False, str(exc))
        await asyncio.sleep(READ_TRAFFIC_INTERVAL_SECONDS)


def build_summary(
    results: list[dict[str, Any]],
    mix_profile: dict[str, Any],
    thresholds: dict[str, float | None],
    started_at: str,
    completed_at: str,
) -> dict[str, Any]:
    latencies = [item["latency_seconds"] for item in results]
    total = len(results)
    success = sum(1 for item in results if item["ok"])
    failures = total - success
    five_xx = sum(1 for item in results if int(item["status_code"]) >= 500 or int(item["status_code"]) == 0)
    by_target = Counter(item["target"] for item in results)
    summary = {
        "profile": mix_profile["profile"],
        "started_at": started_at,
        "completed_at": completed_at,
        "traffic": mix_profile,
        "results": {
            "total_requests": total,
            "success": success,
            "failed": failures,
            "five_xx_or_transport": five_xx,
            "success_rate": round((success / total) * 100, 2) if total else 0.0,
            "five_xx_rate": round((five_xx / total) * 100, 2) if total else 0.0,
            "avg_seconds": round(statistics.mean(latencies), 4) if latencies else 0.0,
            "p95_seconds": round(percentile(latencies, 0.95), 4) if latencies else 0.0,
            "p99_seconds": round(percentile(latencies, 0.99), 4) if latencies else 0.0,
            "by_target": dict(by_target),
        },
        "thresholds": thresholds,
    }
    summary["passed"] = evaluate_thresholds(summary)
    return summary


def evaluate_thresholds(summary: dict[str, Any]) -> bool:
    results = summary["results"]
    thresholds = summary["thresholds"]
    if thresholds.get("require_success_rate") is not None and results["success_rate"] < thresholds["require_success_rate"]:
        return False
    if thresholds.get("require_p95_seconds") is not None and results["p95_seconds"] > thresholds["require_p95_seconds"]:
        return False
    if thresholds.get("require_p99_seconds") is not None and results["p99_seconds"] > thresholds["require_p99_seconds"]:
        return False
    return True


def write_report(report_dir: Path, summary: dict[str, Any], results: list[dict[str, Any]], errors: list[dict[str, Any]]) -> None:
    (report_dir / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
    (report_dir / "mix_profile.json").write_text(json.dumps(summary["traffic"], indent=2), encoding="utf-8")
    (report_dir / "error_samples.json").write_text(json.dumps(errors, indent=2), encoding="utf-8")
    with (report_dir / "latency_samples.csv").open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=["ts", "target", "status_code", "latency_seconds", "ok"])
        writer.writeheader()
        writer.writerows(results)


async def run_profile(
    *,
    base_url: str,
    profile_name: str,
    ramp_seconds: int | None,
    hold_seconds: int | None,
    voice: int | None,
    digital: int | None,
    include_read_traffic: bool,
    auth_mode: str,
    report_dir: str | None,
    require_success_rate: float | None,
    require_p95_seconds: float | None,
    require_p99_seconds: float | None,
) -> tuple[dict[str, Any], Path]:
    profile = resolve_profile(
        profile_name,
        ramp_seconds=ramp_seconds,
        hold_seconds=hold_seconds,
        voice=voice,
        digital=digital,
    )
    mix_profile = build_mix_profile(profile, include_read_traffic)
    out_dir = ensure_report_dir(report_dir)
    started_at_iso = time.strftime("%Y-%m-%dT%H:%M:%S")
    results: list[dict[str, Any]] = []
    errors: list[dict[str, Any]] = []
    stop_event = asyncio.Event()
    worker_targets = build_worker_targets(profile)

    async with httpx.AsyncClient(base_url=base_url) as client:
        headers = await build_auth_headers(client, auth_mode)
        workers: list[asyncio.Task[None]] = []
        ramp_window = max(1, int(profile["ramp_seconds"]))
        spacing = ramp_window / max(1, len(worker_targets))
        for idx, target in enumerate(worker_targets):
            workers.append(asyncio.create_task(_run_worker(client, headers, target, idx, stop_event, results, errors)))
            if spacing > 0:
                await asyncio.sleep(spacing)

        read_tasks: list[asyncio.Task[None]] = []
        if include_read_traffic:
            read_tasks = [
                asyncio.create_task(
                    _run_read_probe(client, headers, "/proxy/supervisor/supervisor/realtime", stop_event, results, errors)
                ),
                asyncio.create_task(
                    _run_read_probe(client, headers, "/proxy/reporting/reports/kpi", stop_event, results, errors)
                ),
            ]

        await asyncio.sleep(int(profile["hold_seconds"]))
        stop_event.set()
        await asyncio.gather(*workers, *read_tasks, return_exceptions=True)

    completed_at_iso = time.strftime("%Y-%m-%dT%H:%M:%S")
    thresholds = {
        "require_success_rate": require_success_rate,
        "require_p95_seconds": require_p95_seconds,
        "require_p99_seconds": require_p99_seconds,
    }
    summary = build_summary(results, mix_profile, thresholds, started_at_iso, completed_at_iso)
    write_report(out_dir, summary, results, errors)
    return summary, out_dir


async def main() -> None:
    parser = argparse.ArgumentParser(description="Track 7 mixed workload load harness through the gateway.")
    parser.add_argument("--base-url", default="http://localhost:8080", help="Gateway URL")
    parser.add_argument(
        "--profile",
        default="baseline_100_100",
        choices=sorted(PROFILE_DEFAULTS),
        help="Named load profile",
    )
    parser.add_argument("--ramp-seconds", type=int, default=None, help="Override ramp duration")
    parser.add_argument("--hold-seconds", type=int, default=None, help="Override hold duration")
    parser.add_argument("--voice", type=int, default=None, help="Backward-compatible voice worker override")
    parser.add_argument("--digital", type=int, default=None, help="Backward-compatible digital worker override")
    parser.add_argument("--report-dir", default=None, help="Directory for JSON/CSV reports")
    parser.add_argument("--require-success-rate", type=float, default=None, help="Optional success-rate threshold")
    parser.add_argument("--require-p95-seconds", type=float, default=None, help="Optional p95 threshold")
    parser.add_argument("--require-p99-seconds", type=float, default=None, help="Optional p99 threshold")
    parser.add_argument(
        "--include-read-traffic",
        type=int,
        choices=[0, 1],
        default=1,
        help="Include background supervisor/reporting reads during the hold window",
    )
    parser.add_argument(
        "--auth-mode",
        choices=["legacy_headers", "bearer"],
        default="legacy_headers",
        help="Authentication mode for load traffic",
    )
    args = parser.parse_args()

    summary, out_dir = await run_profile(
        base_url=args.base_url,
        profile_name=args.profile,
        ramp_seconds=args.ramp_seconds,
        hold_seconds=args.hold_seconds,
        voice=args.voice,
        digital=args.digital,
        include_read_traffic=bool(args.include_read_traffic),
        auth_mode=args.auth_mode,
        report_dir=args.report_dir,
        require_success_rate=args.require_success_rate,
        require_p95_seconds=args.require_p95_seconds,
        require_p99_seconds=args.require_p99_seconds,
    )
    print("Track 7 load test result:")
    print(json.dumps(summary, indent=2))
    print(f"Report dir: {out_dir}")
    if not summary["passed"]:
        raise SystemExit(1)


if __name__ == "__main__":
    asyncio.run(main())