"""Measure 'view -> counter' time while background traffic runs, like hitting refresh on your post analytics.

For each rate: start the producer (background views over Kafka), and every few seconds send one probe view to a
fresh probe post, then poll its counter (the materialized view) until it shows 1. Records seconds from send to visible,
the time of each refresh query, and sent vs counted totals. Output: results.json
Run: EH_NS=... EH_CONN=... python measure.py
"""
import json, os, statistics as st, subprocess, sys, time, uuid
from datetime import datetime, timezone
from confluent_kafka import Producer
from kql import run

PY = sys.executable
RATES = [int(x) for x in os.environ.get("RATES", "100,1000,5000").split(",")]
SECONDS = 90
probe = Producer({"bootstrap.servers": f"{os.environ['EH_NS']}.servicebus.windows.net:9093", "security.protocol": "SASL_SSL",
                  "sasl.mechanism": "PLAIN", "sasl.username": "$ConnectionString", "sasl.password": os.environ["EH_CONN"], "linger.ms": 0})


def probe_once(tag):
    post = f"probe-{tag}-{uuid.uuid4().hex[:6]}"
    ev = {"post_id": post, "viewer_id": "probe", "run": f"probe-{tag}", "sent_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds")}
    t0 = time.time()
    probe.produce("post-views", json.dumps(ev).encode(), key=post.encode()); probe.flush(10)
    refresh_ms = []
    while time.time() - t0 < 120:
        rows, ms = run(f"PostViewCounts | where post_id == '{post}' | project view_count")
        refresh_ms.append(ms)
        if rows and rows[0]["view_count"] >= 1:
            return time.time() - t0, refresh_ms
        time.sleep(0.25)
    return None, refresh_ms


results = []
for rate in RATES:
    tag = f"r{rate}-{int(time.time())}"  # unique per run, so earlier attempts never mix in
    load = subprocess.Popen([PY, "producer.py", "--rate", str(rate), "--seconds", str(SECONDS), "--run", tag], stdout=subprocess.PIPE, text=True)
    time.sleep(10)  # let traffic ramp up
    lat, refresh = [], []
    while load.poll() is None:
        s, r = probe_once(tag)
        if s is not None:
            lat.append(s)
        refresh += r
        time.sleep(2)
    sent = json.loads(load.stdout.read().strip().splitlines()[-1])
    time.sleep(45)  # let the last views land before counting
    counted = run(f"PostViews | where run == '{tag}' | count")[0][0]["Count"]
    ingest = run(f"PostViews | where run == '{tag}' | extend s = (ingestion_time() - sent_at) / 1ms "
                 f"| summarize p50 = percentile(s, 50), p95 = percentile(s, 95), p99 = percentile(s, 99)")[0][0]
    ingest = {k: (v if v is not None else -1) for k, v in ingest.items()}
    res = {"rate": rate, "sent": sent, "counted": counted,
           "view_to_counter_s": {"n": len(lat), "p50": round(st.median(lat), 2) if lat else None,
                                 "p95": round(sorted(lat)[int(0.95 * (len(lat) - 1))], 2) if lat else None, "max": round(max(lat), 2) if lat else None},
           "refresh_query_ms": {"n": len(refresh), "p50": round(st.median(refresh)), "p95": round(sorted(refresh)[int(0.95 * (len(refresh) - 1))])},
           "send_to_queryable_ms": {k: round(v) for k, v in ingest.items()}}
    results.append(res)
    print(json.dumps(res), flush=True)
json.dump({"run_at": datetime.now(timezone.utc).isoformat(), "results": results}, open("results.json", "w"), indent=1)