#!/usr/bin/env python3
"""Turn nats bench text logs + latency JSON into markdown."""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path


def fmt_int(s: str | None) -> str:
    if not s:
        return "—"
    return f"{int(s):,}"


def parse_bench(text: str) -> dict[str, str]:
    out: dict[str, str] = {"kind": "throughput"}
    def _rate(pattern: str, msgs_key: str, mb_key: str) -> None:
        m = re.search(pattern, text)
        if not m:
            return
        out[msgs_key] = m.group(1).replace(",", "")
        n = float(m.group(2))
        unit = m.group(3).upper()
        if unit == "KB":
            n = n / 1024.0
        elif unit == "GB":
            n = n * 1024.0
        out[mb_key] = f"{n:.2f}"

    _rate(
        r"(?m)^\s*Pub stats:\s*([0-9,]+)\s*msgs/sec\s*~\s*([0-9.]+)\s*(KB|MB|GB)/sec",
        "pub_msgs",
        "pub_mb",
    )
    _rate(
        r"(?m)^\s*Sub stats:\s*([0-9,]+)\s*msgs/sec\s*~\s*([0-9.]+)\s*(KB|MB|GB)/sec",
        "sub_msgs",
        "sub_mb",
    )
    _rate(
        r"NATS Pub/Sub stats:\s*([0-9,]+)\s*msgs/sec\s*~\s*([0-9.]+)\s*(KB|MB|GB)/sec",
        "agg_msgs",
        "agg_mb",
    )
    if "JetStream" in text or "--js" in text or "js-" in text:
        out["mode"] = "jetstream r=3 file"
    else:
        out["mode"] = "core pub/sub"
    # nats 0.1.6 prints min/avg/max as msgs/sec across publishers, not µs delay
    m = re.search(
        r"min\s+([0-9,]+)\s*\|\s*avg\s+([0-9,]+)\s*\|\s*max\s+([0-9,]+)\s*\|\s*stddev\s+([0-9,]+)\s*msgs",
        text,
    )
    if m:
        out["pub_spread"] = f"{m.group(1)}–{m.group(3)} (avg {m.group(2)})"
    return out


def parse_lat(text: str) -> dict[str, str] | None:
    for line in text.splitlines():
        line = line.strip()
        if line.startswith("{") and "p99_us" in line:
            d = json.loads(line)
            mode = d.get("mode") or ""
            return {
                "kind": "latency",
                "count": str(d.get("count", "")),
                "pubs": str(d.get("pubs", "")),
                "size": str(d.get("size", "")),
                "mode": str(mode),
                "min": d.get("min", ""),
                "avg": d.get("avg", ""),
                "p50": d.get("p50", ""),
                "p90": d.get("p90", ""),
                "p99": d.get("p99", ""),
                "max": d.get("max", ""),
            }
    return None


def lat_mode(run: str, recorded: str) -> str:
    if recorded in ("ping", "flood"):
        return recorded
    if "ping" in run:
        return "ping"
    return "flood"


def thru_sort(p: dict[str, str]) -> tuple:
    return (0 if p.get("mode", "").startswith("core") else 1, p.get("run", ""))


def lat_sort(p: dict[str, str]) -> tuple:
    mode = lat_mode(p.get("run", ""), p.get("mode", ""))
    return (0 if mode == "ping" else 1, int(p.get("count") or 0), p.get("run", ""))


def main() -> int:
    folder = Path(sys.argv[1] if len(sys.argv) > 1 else ".")
    thru: list[dict[str, str]] = []
    lats: list[dict[str, str]] = []
    for f in sorted(folder.glob("*.txt")):
        text = f.read_text(encoding="utf-8", errors="replace")
        lat = parse_lat(text)
        if lat:
            lat["run"] = f.stem
            lats.append(lat)
            continue
        p = parse_bench(text)
        if p.get("pub_msgs") or p.get("agg_msgs"):
            p["run"] = f.stem
            thru.append(p)

    stamp = folder.name if re.fullmatch(r"\d{8}T\d{6}Z", folder.name) else ""

    print("# NATS cluster message speed")
    print()
    if stamp:
        print(f"Run **`{stamp}`** (UTC). ", end="")
    print(
        "Client: LXC **510** `verae-px-worker` (`10.10.10.20`), not a nats-* server. "
        "Servers: `nats-a/b/c` on `10.10.10.21–23` (`vmbr1` only)."
    )
    print()
    print("Client URL:")
    print()
    print("```text")
    print("nats://10.10.10.21:4222,nats://10.10.10.22:4222,nats://10.10.10.23:4222")
    print("```")
    print()
    print("## Method")
    print()
    print("- **Core NATS** is fire-and-forget pub/sub (`nats bench`). No disk, no replica ack.")
    print(
        "- **JetStream** uses **file** storage and **replicas=3** (same as product streams). "
        "The unique stream `benchstream` is deleted between JS loads."
    )
    print("- Throughput is **msgs/sec** from nats CLI **0.1.6** (`--no-progress --csv`). Its min/avg/max are publisher **rate spread**, not delay.")
    print(
        "- **Ping** delay: one publisher, sequential publish-then-wait. This is one-message round-trip through the cluster."
    )
    print(
        "- **Flood** delay: N publishers dump the whole batch, then the subscriber drains. "
        "This is **queueing under burst**, not wire RTT."
    )
    print("- Probe: `scripts/latency.mjs` (two connections, header timestamp).")
    print()
    print("## Throughput")
    print()
    print("| Run | Mode | Aggregate msgs/s | Pub msgs/s | Pub MB/s | Sub msgs/s | Sub MB/s |")
    print("|-----|------|------------------|------------|----------|------------|----------|")
    for p in sorted(thru, key=thru_sort):
        print(
            f"| `{p['run']}` | {p.get('mode', '')} | {fmt_int(p.get('agg_msgs'))} | "
            f"{fmt_int(p.get('pub_msgs'))} | {p.get('pub_mb') or '—'} | "
            f"{fmt_int(p.get('sub_msgs'))} | {p.get('sub_mb') or '—'} |"
        )
    print()
    print("## Round-trip delay")
    print()
    print("| Run | Kind | Count | Pubs | Size | min | avg | p50 | p90 | p99 | max |")
    print("|-----|------|-------|------|------|-----|-----|-----|-----|-----|-----|")
    for p in sorted(lats, key=lat_sort):
        kind = lat_mode(p.get("run", ""), p.get("mode", ""))
        label = "ping (sequential RTT)" if kind == "ping" else "flood (burst queueing)"
        print(
            f"| `{p['run']}` | {label} | {p.get('count', '')} | {p.get('pubs', '')} | "
            f"{p.get('size', '')} B | {p.get('min', '')} | {p.get('avg', '')} | "
            f"{p.get('p50', '')} | {p.get('p90', '')} | {p.get('p99', '')} | {p.get('max', '')} |"
        )
    print()
    print("## What the numbers mean")
    print()
    print(
        "Product job/event/archive traffic is **JetStream r=3 file**. On this three-LXC stand that is about "
        "**16k durable 128 B pubs/s** (about **13k** at 1 KiB). Core NATS is an upper bound for "
        "non-durable fan-out: about **0.7–2.0M msgs/s** aggregate at 128 B, or **~630k msgs/s (~616 MB/s)** at 1 KiB with 4 publishers."
    )
    print()
    print(
        "A quiet request-reply is **~0.3 ms** average, **p99 < 1 ms**. Flood rows in the **150–500 ms** band "
        "are the subscriber catching up after a burst, which is what a job-events mailbox sees if publishers outrun consumers."
    )
    print()
    print("Re-run on NS1: `bash scripts/bench.sh`. Raw logs/CSVs are under `results/<utc>/`.")
    return 0


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