"""Cold->hot seam handoff contract.

Reproduces the 2026-08-08 production wedge: a cursor subscriber that
connects while the hot tail is empty (post-restart, before ingest dials)
drains the archive backlog on the cold path and must then FOLLOW THE LIVE
TIP once ingest starts delivering — not trail block-seal cadence.

Sequence:
  1. stream boots against a dead upstream port: serving up, hot tail empty.
  2. subscriber connects with cursor=1, drains the sample archive cold.
  3. the upstream simulator starts on that port; ingest dials and appends.
  4. CONTRACT: the subscriber receives a newly ingested event within
     DELIVERY_DEADLINE_S of the upstream seq first advancing. A wedged
     subscriber receives nothing until a 4096-event block seals (minutes at
     simulator rates), so the deadline cleanly separates the two behaviors.

Upstream reference: the hot/cold boundary is transparent to Tail.ReadFrom
(internal/subscribe/doc.go) and a non-advancing read is a contract
violation (handler.go) — a subscriber can never be durably wedged.
"""

import base64
import hashlib
import json
import os
import signal
import socket
import subprocess
import tempfile
import time
import urllib.request

HOST = "127.0.0.1"
PORT = int(os.environ["STREAM_SEAM_PORT"])
SIM_PORT = int(os.environ["STREAM_SEAM_SIM_PORT"])
UPSTREAM_DIR = os.environ["STREAM_SEAM_UPSTREAM_DIR"]
DELIVERY_DEADLINE_S = 20.0


def handshake(path):
    connection = socket.create_connection((HOST, PORT), timeout=5)
    connection.settimeout(2)
    key = base64.b64encode(os.urandom(16)).decode()
    connection.sendall(
        (
            f"GET {path} HTTP/1.1\r\n"
            f"Host: {HOST}:{PORT}\r\n"
            "Upgrade: websocket\r\n"
            "Connection: Upgrade\r\n"
            f"Sec-WebSocket-Key: {key}\r\n"
            "Sec-WebSocket-Version: 13\r\n\r\n"
        ).encode()
    )
    response = b""
    while b"\r\n\r\n" not in response:
        response += connection.recv(4096)
    head, buffered = response.split(b"\r\n\r\n", 1)
    status = head.decode().split("\r\n")[0]
    assert status.startswith("HTTP/1.1 101"), status
    return connection, buffered


def read_frame(connection, buffered):
    def read_exact(buf, length):
        while len(buf) < length:
            buf += connection.recv(length - len(buf))
        return buf[:length], buf[length:]

    header, buffered = read_exact(buffered, 2)
    opcode = header[0] & 0x0F
    length = header[1] & 0x7F
    if length == 126:
        raw, buffered = read_exact(buffered, 2)
        length = int.from_bytes(raw, "big")
    elif length == 127:
        raw, buffered = read_exact(buffered, 8)
        length = int.from_bytes(raw, "big")
    payload, buffered = read_exact(buffered, length)
    return opcode, payload, buffered


def metric(name):
    body = urllib.request.urlopen(f"http://{HOST}:{PORT}/metrics", timeout=5).read().decode()
    for line in body.splitlines():
        if line.startswith(name + " "):
            return float(line.split()[1])
    return None


# 1+2: subscribe while the tail is empty, drain the archive backlog cold.
connection, buffered = handshake("/subscribe?cursor=1")
drained = 0
last_seq_seen = 0
while True:
    try:
        opcode, payload, buffered = read_frame(connection, buffered)
    except socket.timeout:
        break  # >2s of silence: backlog drained, subscriber is at the seam
    if opcode == 1:
        drained += 1
        event = json.loads(payload)
        last_seq_seen = max(last_seq_seen, event.get("time_us", 0))
assert drained > 0, "sample archive replay delivered nothing"
print(f"drained {drained} backlog events; subscriber now at the seam")

# 3: start the simulator on the port stream's --upstream already points at.
# own data dir (the shared ./data/simulator belongs to the standing :7777
# world — a --reset here wiped its WAL mid-rename, 2026-08-08) and own
# process group (go run's compiled child outlives terminate() otherwise)
sim_data = tempfile.mkdtemp(prefix="seam-sim-")
sim = subprocess.Popen(
    [
        "go", "run", "./cmd/simulator", "serve",
        "--reset", "--accounts=20", "--commits-per-sec=20",
        f"--addr=:{SIM_PORT}", f"--data-dir={sim_data}",
    ],
    cwd=UPSTREAM_DIR,
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL,
    start_new_session=True,
)
try:
    # wait for ingest to dial and the upstream seq to advance
    deadline = time.monotonic() + 120
    fed_at = None
    while time.monotonic() < deadline:
        seq = metric("stream_upstream_seq")
        if seq and seq > 0:
            fed_at = time.monotonic()
            break
        time.sleep(1)
    assert fed_at is not None, "ingest never dialed the simulator"
    print("ingest dialed; live events flowing")

    # 4: the contract — a live event must reach the wedged subscriber fast.
    connection.settimeout(1)
    got_live = None
    while time.monotonic() - fed_at < DELIVERY_DEADLINE_S:
        try:
            opcode, payload, buffered = read_frame(connection, buffered)
        except socket.timeout:
            continue
        if opcode == 1:
            got_live = time.monotonic() - fed_at
            break
    assert got_live is not None, (
        f"seam wedge: no live event reached the subscriber within "
        f"{DELIVERY_DEADLINE_S}s of ingest starting (cold path never hands "
        f"off to the hot tail)"
    )
    print(f"live event delivered {got_live:.1f}s after ingest started")
finally:
    os.killpg(os.getpgid(sim.pid), signal.SIGTERM)
    sim.wait(timeout=10)

print("seam handoff contract: PASS")