"""PostHog-loop playbook, step 9: write the band to PostHog, without the held-out users.

A PostHog destination writes `tessera_band` (and the score, slot and policy
fingerprint) as person properties. A backfill delivers every scored user
once; afterwards each band change is delivered as it happens. The destination
writer withholds every held-out user before it builds a request.

On the local stack PostHog is a twin read through the control origin, and the
writer runs one pass per control call, with its clock set eight days back: the
follow-up outcomes step 10 reads mature one label horizon (7 days) after the
delivery, and an outcome can only be stored once it has happened. The check at
the end recomputes the holdout from the protocol's salt with the documented
assignment rule:
sha256(salt || "{project_id}#{type}#{id}"), first 8 bytes as an unsigned
integer, divided by 2**64, below the holdout fraction.
"""

import hashlib
import os
import sys
from pathlib import Path

import httpx

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # playbooks/playbook_api.py
from playbook_api import Api, State

api, state = Api(), State("posthog-loop")

CONTROL = os.environ["TESSERA_CONTROL_BASE"]
POSTHOG_KEY = "phc_fixture_project_key"
httpx.post(f"{CONTROL}/fakes/posthog/keys", json={"api_key": POSTHOG_KEY}, timeout=30.0)
destination = api.post(
    "/destinations",
    {
        "vendor": "posthog",
        "slot_id": state["slot_id"],
        "config": {"host": "https://posthog.fake.tessera.test", "project_id": 4242},
        "credentials": {"api_key": POSTHOG_KEY},
    },
    key="posthog-loop-destination",
)
properties = ", ".join(sorted(destination["declared_property_names"]))
print(f"destination: {destination['status']}, properties: {properties}")
api.post(f"/destinations/{destination['destination_id']}:backfill", key="posthog-loop-backfill")
drained = httpx.post(
    f"{CONTROL}/ticks/destination-writer", json={"advance_seconds": -8 * 86_400}, timeout=600.0
).json()
print(f"writer delivered batches: {drained.get('batches', 0) > 0}")
receipts = api.get(f"/destinations/{destination['destination_id']}/deliveries").get("data") or []
withheld = sum(int(receipt.get("withheld_count") or 0) for receipt in receipts)
applied = {
    experiment
    for receipt in receipts
    for experiment in (receipt.get("suppression_manifest") or {}).get("experiments_applied") or []
}
print(f"deliveries succeeded: {all(receipt['status'] == 'succeeded' for receipt in receipts)}")
print(f"held-out users withheld: {withheld > 0}")
print(f"experiment applied: {state['experiment_id'] in applied}")

persons = httpx.get(f"{CONTROL}/fakes/posthog", timeout=30.0).json().get("persons") or {}
project_id = os.environ["TESSERA_PROJECT_ID"]


# The salt declared with the protocol (step 4) and the experiment's fraction
# (step 8), as stored: the check never restates them.
salt = str(state["holdout_salt"]).encode("utf-8")
numerator, denominator = float(state["holdout_fraction"]).as_integer_ratio()


def held_out(user: str) -> bool:
    key = f"{project_id}#user#{user}".encode()
    bucket = int.from_bytes(hashlib.sha256(salt + key).digest()[:8], "big")
    # bucket / 2**64 < fraction, in exact integer arithmetic
    return bucket * denominator < numerator * (1 << 64)


banded = sorted(user for user, properties in persons.items() if "tessera_band" in properties)
leaked = [user for user in banded if held_out(user)]
print(f"users with a tessera_band in PostHog: {len(banded) > 0}")
print(f"held-out users in PostHog: {len(leaked)}")
if leaked:
    sys.exit(f"held-out users reached PostHog: {leaked[:10]}")
state.update(destination_id=destination["destination_id"])