"""Fake 'member viewed a post' events, sent over the Kafka protocol to Azure Event Hubs.

10,000 posts with Zipf-like popularity (a few posts get most views, like a real feed).
Each event carries sent_at, so the store can measure how long a view takes to become countable.
Run: EH_NS=... EH_CONN='Endpoint=sb://...' python producer.py --rate 1000 --seconds 60 --run r1000
"""
import argparse, json, os, random, time, uuid
from datetime import datetime, timezone
from confluent_kafka import Producer

ap = argparse.ArgumentParser()
ap.add_argument("--rate", type=int, default=1000, help="views per second")
ap.add_argument("--seconds", type=int, default=60)
ap.add_argument("--run", default="test", help="label stored with every event")
ap.add_argument("--viewers", type=int, default=0, help="member pool size; 0 = every view is a new viewer")
a = ap.parse_args()

p = 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": 5, "batch.num.messages": 1000,
})
posts = [f"post-{i:05d}" for i in range(10_000)]
weights = [1 / (i + 1) ** 1.1 for i in range(len(posts))]
members = [f"m{i:06d}" for i in range(a.viewers)]
import itertools
mcum = list(itertools.accumulate(1 / (i + 1) ** 0.6 for i in range(a.viewers)))  # some members scroll far more than others
pcum = list(itertools.accumulate(weights))
sent, t0 = 0, time.time()
while time.time() - t0 < a.seconds:
    tick = time.time()
    batch = random.choices(posts, cum_weights=pcum, k=a.rate // 10)
    who = random.choices(members, cum_weights=mcum, k=len(batch)) if members else [uuid.uuid4().hex[:12] for _ in batch]
    for post, viewer in zip(batch, who):
        ev = {"post_id": post, "viewer_id": viewer, "run": a.run,
              "sent_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds")}
        p.produce("post-views", json.dumps(ev).encode(), key=post.encode())
        sent += 1
    p.poll(0)
    time.sleep(max(0, 0.1 - (time.time() - tick)))
p.flush(30)
print(json.dumps({"run": a.run, "target_rate": a.rate, "sent": sent, "seconds": round(time.time() - t0, 1),
                  "actual_rate": round(sent / (time.time() - t0))}))