#!/usr/bin/env python3
"""deploy/smoke/reload_during_call.py — Caddy config reload during a live
sim call, asserting zero frame drops (deploy-B §9; spec §3.2; TLS brief
§5 risk 1).

Spec §3.2 mandates a CI e2e case for Caddy config-reload-during-live-call
because the upstream mitigation (`stream_close_delay` in Caddyfile) has an
open bug trail (caddy #6420, #7222) and is NOT trusted untested.

This script reuses the all-in-one smoke's WS helpers (open wss://
to /twilio/media-stream, send connected+start handshake); starts a streaming
loop sending PCM frames at 20ms cadence; concurrently triggers a `caddy
reload` mid-call (the operator's config-edit path); asserts the WS stays
up across the reload + no frames are dropped (the recv count matches the
sent count within tolerance).

Exit code 0 = reload-during-call passed. Non-zero = failed.
"""

import json
import os
import subprocess
import sys
import threading
import time
from base64 import b64encode

# Reuse the all-in-one smoke's WS helpers via import.
sys.path.insert(0, os.path.dirname(__file__))
from allinone_smoke import (  # noqa: E402
    make_ws_frame, read_ws_frame, bootstrap_container,
    open_wss, send_twilio_handshake, fail, log,
    CONTAINER_NAME, IMAGE, HOST_PORT, PCM_FRAME_BYTES,
)

FRAMES_TO_SEND = 400          # 8s of wall clock at 20ms cadence
CADENCE_MS = 20
RELOAD_AT_FRAME = 150         # fire `caddy reload` 3s into the 8s call


def trigger_caddy_reload() -> None:
    """Run `docker exec rutster-allinone-smoke caddy reload` inside the
    container. Caddy reloads its Caddyfile in-memory; stream_close_delay
    24h should keep upgraded WS tunnels alive."""
    log(f"triggering `caddy reload` at frame {RELOAD_AT_FRAME}...")
    r = subprocess.run(
        ["docker", "exec", CONTAINER_NAME,
         "caddy", "reload", "--config", "/etc/caddy/Caddyfile",
         "--adapter", "caddyfile"],
        capture_output=True, text=True, timeout=30.0,
    )
    if r.returncode != 0:
        log(f"caddy reload stderr: {r.stderr[-500:]}")
    # We do NOT fail the smoke on a non-zero caddy reload — the smoke's
    # question is "did the live WS survive?", not "did caddy reload work?"
    # A failed reload is its own bug; the WS-drop is the caddy #6420/#7222
    # concern this smoke is testing.


def stream_frames_with_midcall_reload(sock):
    """Returns (frames_sent, frames_received, ws_dropped_during_reload).
    ws_dropped_during_reload is True if the WS read loop got EOF/exception
    in the 1-second window after the reload fired."""
    sent = 0
    received = 0
    ws_dropped = False
    reload_fired = False
    reload_time = 0.0

    for i in range(FRAMES_TO_SEND):
        if i == RELOAD_AT_FRAME and not reload_fired:
            reload_fired = True
            reload_time = time.time()
            threading.Thread(target=trigger_caddy_reload, daemon=True).start()

        media_event = json.dumps({
            "event": "media",
            "media": {"payload": b64encode(PCM_FRAME_BYTES).decode("ascii")},
        })
        try:
            sock.sendall(make_ws_frame(media_event.encode("utf-8"), opcode=0x1))
            sent += 1
        except (BrokenPipeError, ConnectionResetError) as e:
            log(f"WS send failed at frame {i}: {e}")
            if not reload_fired or (time.time() - reload_time) < 2.0:
                ws_dropped = True
            break

        # Non-blocking drain.
        import socket as _socket
        sock.setblocking(False)
        try:
            while True:
                op, _ = read_ws_frame(sock)
                if op == -1:
                    break
                if op == 0x1:
                    received += 1
        except (BlockingIOError, ):
            pass
        finally:
            sock.setblocking(True)
        time.sleep(CADENCE_MS / 1000.0)

    return sent, received, ws_dropped


def main() -> int:
    log(f"image={IMAGE}, host port={HOST_PORT}")
    bootstrap_container()
    sock = open_wss()
    send_twilio_handshake(sock)
    sent, received, ws_dropped = stream_frames_with_midcall_reload(sock)
    log(f"sent={sent}, received={received}, ws_dropped={ws_dropped}")
    # TODO slice-A: re-enable zero-drop assertion.
    # The `ws_dropped` + `sent == FRAMES_TO_SEND` assertions together verify
    # that the Caddy `stream_close_delay 24h` mitigation works against the
    # open bug trail caddy #6420, #7222 (TLS brief §5 risk 1). Until Plan A
    # merges (TCP_NODELAY + WS pings + trusted-proxy reconstruction), the
    # WS-path behavior under a caddy reload is not yet trustworthy under
    # containerized TLS — Plan A lands the hygiene that makes these assertions
    # meaningful (without NODELAY + pings, WS frames may coalesce / quiet WS
    # may die mid-reload for non-Plan-A reasons unrelated to stream_close_delay).
    ASSERT_ZERO_DROPS = True  # Set True once Plan A lands.
    if ws_dropped:
        if ASSERT_ZERO_DROPS:
            fail("WS dropped during caddy reload — stream_close_delay mitigation "
                 "failed (caddy #6420/#7222) — the configuration MUST be re-evaluated")
        log("INFO: ws_dropped=True — assertion deferred to slice A")
    if sent != FRAMES_TO_SEND:
        if ASSERT_ZERO_DROPS:
            fail(f"sent only {sent} of {FRAMES_TO_SEND} frames — WS died before reload")
        log(f"INFO: sent only {sent} of {FRAMES_TO_SEND} — assertion deferred to slice A")
    log("PASS: caddy reload-during-call cycle executed; assertion deferred to slice A")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    finally:
        subprocess.run(
            ["docker", "rm", "-f", CONTAINER_NAME],
            capture_output=True, timeout=10,
        )