"""The counter goes down: does the stream keep the views, and how long does catching up take?

1. Stop the store cluster (the 'counting service' is down).
2. Keep sending views for 180 s at 1,000/sec. They wait in Event Hubs (24 h retention).
3. Start the cluster again, then poll until every view sent while it was down has been counted.
Output: recovery_results.json
"""
import json, subprocess, sys, time
from kql import run, CLUSTER

NAME = CLUSTER.split("//")[1].split(".")[0]
az = lambda *a: subprocess.run(["az", "kusto", "cluster", *a, "-n", NAME, "-g", "rg-impr2", "-o", "none"], capture_output=True, text=True)
tag = f"down-{int(time.time())}"
log = {"run": tag}

t = time.time(); r = az("stop"); log["stop_s"] = round(time.time() - t); log["stop_err"] = r.stderr[-300:] if r.returncode else ""
out = subprocess.run([sys.executable, "producer.py", "--rate", "1000", "--seconds", "180", "--run", tag], capture_output=True, text=True).stdout
log["sent"] = json.loads(out.strip().splitlines()[-1])
t = time.time(); r = az("start"); log["start_s"] = round(time.time() - t); log["start_err"] = r.stderr[-300:] if r.returncode else ""

up = time.time(); seen = []
while time.time() - up < 1800:
    try:
        n = run(f"PostViews | where run == '{tag}' | count")[0][0]["Count"]
    except Exception as e:
        n = None
    seen.append((round(time.time() - up), n))
    if n is not None and n >= log["sent"]["sent"]:
        break
    time.sleep(10)
log["catch_up_s_after_start"] = seen[-1][0]
log["counted"] = seen[-1][1]
log["duplicates_or_missing"] = (seen[-1][1] or 0) - log["sent"]["sent"]
log["progress"] = seen
json.dump(log, open("recovery_results.json", "w"), indent=1)
print(json.dumps({k: v for k, v in log.items() if k != "progress"}, indent=1))