#!/usr/bin/env python3
"""deploy/smoke/allinone_smoke.py — T1 all-in-one TLS smoke (deploy-B §9).

Boots `rutster-allinone` with RUTSTER_LOCAL_CERTS=true (Caddy internal CA —
no ACME in CI per spec §9), boots, extracts Caddy's internal-CA root cert
from /data, opens a wss:// WS to /twilio/media-stream, sends Twilio's
connected + start handshake frames, streams 200 PCM-zeroed frames at 20ms
cadence, and asserts the engine replies with at least one text frame
through the real edge->FOB WS path end-to-end through TLS.

Python stdlib ONLY — no pip install required (the CI runner image ships
Python 3 + ssl + socket + json + time). The WS handshake is hand-rolled
for the same reason (websocket-client/websockets would require network
during CI).

Exit code 0 = smoke passed. Non-zero = failed (CI fails the smoke job).

Reusable from: crates/rutster/tests/trunk_sim_e2e.rs (the #[ignore]'d TODO
body sketch — this script implements the same handshake pattern against a
real containerized binary rather than an in-process
MockTwilioMediaStreamsServer) + crates/rutster-trunk/tests/ws_ping.rs (the
connected/start frame sequence — same JSON shapes verbatim).
"""

import json
import os
import socket
import ssl
import struct
import subprocess
import sys
import tempfile
import time
from base64 import b64encode

IMAGE = os.environ.get("RUTSTER_ALLINONE_IMAGE", "rutster-allinone:smoke")
CONTAINER_NAME = "rutster-allinone-smoke"
HOST_PORT = int(os.environ.get("RUTSTER_ALLINONE_PORT", "18443"))
DATA_VOLUME = "rutster-smoke-caddy-data"

# 20ms of 8kHz u-law silence = 160 bytes; base64-encoded in a Twilio
# Media event envelope (matches the slice-A trunk_sim_e2e.rs TODO pattern
# + crates/rutster-trunk/tests/ws_ping.rs' frame shape).
PCM_FRAME_BYTES = b"\x00" * 160
FRAMES_TO_SEND = 200
CADENCE_MS = 20
RECV_DEADLINE_S = 10.0


def log(msg: str) -> None:
    print(f"[allinone_smoke] {msg}", flush=True)


def fail(msg: str) -> "NoReturn":  # noqa: F821 — Python 3.11+ syntax
    print(f"[allinone_smoke] FAIL: {msg}", file=sys.stderr, flush=True)
    subprocess.run(
        ["docker", "logs", CONTAINER_NAME], capture_output=False, timeout=10
    )
    sys.exit(1)


def run(cmd, check=True, timeout=30.0) -> str:
    """Run a command; return stdout. Fail the smoke on non-zero exit (if check)."""
    log(f"$ {' '.join(cmd)}")
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if check and r.returncode != 0:
        print(f"  stdout: {r.stdout[-2000:]}", file=sys.stderr)
        print(f"  stderr: {r.stderr[-2000:]}", file=sys.stderr)
        fail(f"command exited {r.returncode}: {' '.join(cmd)}")
    return r.stdout


def bootstrap_container() -> None:
    """Boot rutster-allinone with RUTSTER_LOCAL_CERTS=true."""
    subprocess.run(["docker", "rm", "-f", CONTAINER_NAME],
                   capture_output=True, timeout=10)
    subprocess.run(["docker", "volume", "rm", "-f", DATA_VOLUME],
                   capture_output=True, timeout=10)
    env = [
        "-e", "RUTSTER_DOMAIN=localhost",
        "-e", "RUTSTER_ACME_EMAIL=smoke@example.com",
        # CRITICAL: RUTSTER_LOCAL_CERTS=true makes Caddy use its internal CA
        # — no ACME in CI (spec §9).
        "-e", "RUTSTER_LOCAL_CERTS=true",
        "-e", "RUTSTER_HTTP_BIND=0.0.0.0:8080",
        "-e", "RUTSTER_MEDIA_BIND_IP=127.0.0.1",
        "-e", "RUTSTER_MEDIA_PORT_RANGE=49160-49170",
        "-e", "RUTSTER_DRAIN_DEADLINE_SECS=10",
        "-e", "RUTSTER_TAP_URL=ws://127.0.0.1:8082",
        # Twilio creds unset — WebRTC + media-stream WS smoke only.
        "-p", f"{HOST_PORT}:443",
    ]
    run(["docker", "run", "-d", "--name", CONTAINER_NAME,
         "-v", f"{DATA_VOLUME}:/data",
         *env, IMAGE], timeout=120.0)
    root_cert = wait_for_root_cert()
    log(f"root cert extracted ({len(root_cert)} bytes)")
    wait_for_healthz(root_cert)


def wait_for_root_cert() -> bytes:
    """Poll until Caddy wrote /data/caddy/pki/authorities/local/root.crt,
    then docker cp it out. Caddy creates this on first boot with local_certs."""
    deadline = time.time() + 30.0
    while time.time() < deadline:
        r = subprocess.run(
            ["docker", "exec", CONTAINER_NAME, "test", "-f",
             "/data/caddy/pki/authorities/local/root.crt"],
            capture_output=True, timeout=5,
        )
        if r.returncode == 0:
            r = subprocess.run(
                ["docker", "cp",
                 f"{CONTAINER_NAME}:/data/caddy/pki/authorities/local/root.crt", "-"],
                capture_output=True, timeout=10,
            )
            if r.returncode == 0 and r.stdout:
                return r.stdout
        time.sleep(0.5)
    fail("Caddy internal CA root cert did not appear within 30s")
    return b""  # unreachable


def wait_for_healthz(root_cert_pem: bytes) -> None:
    """Curl /healthz through TLS using the extracted root cert."""
    with tempfile.NamedTemporaryFile(suffix=".pem", delete=False) as f:
        f.write(root_cert_pem)
        ca_file = f.name
    try:
        deadline = time.time() + 30.0
        r = subprocess.run(["true"])  # for the warning-less r init
        while time.time() < deadline:
            r = subprocess.run(
                ["curl", "--cacert", ca_file, "-fsS",
                 f"https://localhost:{HOST_PORT}/healthz"],
                capture_output=True, text=True, timeout=5,
            )
            if r.returncode == 0 and r.stdout.strip() == "ok":
                log("/healthz ok")
                return
            time.sleep(0.5)
        fail(f"/healthz never returned ok within 30s (last stderr: "
             f"{getattr(r, 'stderr', '')[-500:]})")
    finally:
        os.unlink(ca_file)


# --- hand-rolled WS frame encode/decode (RFC 6455 §5) ---

def make_ws_frame(payload: bytes, opcode: int = 0x1) -> bytes:
    """Encode a client->server WS frame. opcode 0x1=text, 0x2=binary.
    Client frames MUST be masked (RFC 6455 §5.3)."""
    mask = os.urandom(4)
    masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
    header = bytearray([0x80 | opcode])  # FIN + opcode
    n = len(payload)
    if n < 126:
        header.append(0x80 | n)
    elif n < 65536:
        header.append(0x80 | 126)
        header.extend(struct.pack(">H", n))
    else:
        header.append(0x80 | 127)
        header.extend(struct.pack(">Q", n))
    header.extend(mask)
    return bytes(header) + masked


def read_ws_frame(sock) -> tuple:
    """Read one WS frame from the server. Returns (opcode, payload).
    Server frames are NOT masked (RFC 6455 §5.1).
    Returns (-1, b"") on EOF."""
    h1 = sock.recv(1)
    if not h1:
        return -1, b""
    fin_opcode = h1[0]
    h2 = sock.recv(1)[0]
    masked = bool(h2 & 0x80)
    length = h2 & 0x7F
    if length == 126:
        length = struct.unpack(">H", sock.recv(2))[0]
    elif length == 127:
        length = struct.unpack(">Q", sock.recv(8))[0]
    if masked:
        mask = sock.recv(4)
    payload = b""
    while len(payload) < length:
        chunk = sock.recv(length - len(payload))
        if not chunk:
            break
        payload += chunk
    if masked:
        payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
    return fin_opcode & 0x0F, payload


def open_wss():
    """Open the wss:// WS handshake to /twilio/media-stream inside the
    containerized all-in-one."""
    sock = socket.create_connection(("localhost", HOST_PORT), timeout=5)
    ctx = ssl.create_default_context(cafile=None)
    r = subprocess.run(
        ["docker", "cp",
         f"{CONTAINER_NAME}:/data/caddy/pki/authorities/local/root.crt", "-"],
        capture_output=True, timeout=10,
    )
    ctx.load_verify_locations(cadata=r.stdout.decode("ascii"))
    ctx.check_hostname = False  # the cert is for "localhost" — Caddy's local_certs issues it
    ssl_sock = ctx.wrap_socket(sock, server_hostname="localhost")
    log(f"TLS established (cipher={ssl_sock.cipher()})")

    key = os.urandom(16).hex()
    handshake = (
        f"GET /twilio/media-stream HTTP/1.1\r\n"
        f"Host: localhost:{HOST_PORT}\r\n"
        f"Upgrade: websocket\r\n"
        f"Connection: Upgrade\r\n"
        f"Sec-WebSocket-Key: {key}\r\n"
        f"Sec-WebSocket-Version: 13\r\n"
        f"\r\n"
    )
    ssl_sock.sendall(handshake.encode("ascii"))
    buf = b""
    while b"\r\n\r\n" not in buf:
        chunk = ssl_sock.recv(4096)
        if not chunk:
            fail("WS upgrade response closed prematurely")
        buf += chunk
    if b"101 Switching Protocols" not in buf:
        fail(f"WS upgrade did not return 101: {buf[:300]!r}")
    log("WS upgraded (HTTP/1.1 101 Switching Protocols)")
    return ssl_sock


def send_twilio_handshake(sock) -> None:
    """Send the Twilio Media Streams connected + start frames (the same
    JSON sequence crates/rutster-trunk/tests/ws_ping.rs uses against the
    in-process router — here against the real containerized edge->FOB path)."""
    connected = json.dumps({
        "event": "connected",
        "protocol": "twilio-media-stream",
        "version": "1.0.0",
    })
    start = json.dumps({
        "event": "start",
        "start": {"streamSid": "MZsmoke", "callSid": "CAsmoke"},
    })
    sock.sendall(make_ws_frame(connected.encode("utf-8")))
    sock.sendall(make_ws_frame(start.encode("utf-8")))
    log("sent Twilio connected + start handshake frames")


def stream_pcm_frames(sock) -> int:
    """Stream 200 PCM-zeroed frames at 20ms cadence as Twilio Media events.
    Returns the count of text frames received back from the engine."""
    received = 0
    deadline_global = time.time() + (FRAMES_TO_SEND * CADENCE_MS / 1000.0) + RECV_DEADLINE_S
    for _ in range(FRAMES_TO_SEND):
        media_event = json.dumps({
            "event": "media",
            "media": {"payload": b64encode(PCM_FRAME_BYTES).decode("ascii")},
        })
        sock.sendall(make_ws_frame(media_event.encode("utf-8")))
        # Drain non-blocking: read anything available without blocking the
        # 20ms cadence.
        sock.setblocking(False)
        try:
            while True:
                op, _ = read_ws_frame(sock)
                if op == -1:
                    break
                if op == 0x1:  # text frame — engine-originated event
                    received += 1
        except (BlockingIOError, ssl.SSLWantReadError):
            pass
        finally:
            sock.setblocking(True)
        time.sleep(CADENCE_MS / 1000.0)
    sock.settimeout(RECV_DEADLINE_S)
    try:
        while time.time() < deadline_global:
            op, _ = read_ws_frame(sock)
            if op == -1:
                break
            if op == 0x1:
                received += 1
    except (socket.timeout, ssl.SSLWantReadError):
        pass
    return received


def main() -> int:
    log(f"image={IMAGE}, host port={HOST_PORT}")
    bootstrap_container()
    sock = open_wss()
    send_twilio_handshake(sock)
    received = stream_pcm_frames(sock)
    log(f"sent {FRAMES_TO_SEND} frames; received {received} text frames back")
    # TODO slice-A: re-enable engine-reply assertion.
    # The `received >= 1` assertion depends on Plan A's hygiene wash being
    # merged first: TCP_NODELAY (serve.rs serve_with_nodelay) so 20ms WS frames
    # aren't Nagle-coalesced behind the Caddy edge, app-level WS pings so
    # neither Twilio's nor Telnyx's idle timers can kill a quiet mid-call WS,
    # config::twilio_credentials startup validation, RUTSTER_TWILIO_WEBHOOK_BASE
    # derivation into the TwiML Stream URL, and RUTSTER_TRUSTED_PROXIES
    # trusted X-Forwarded-Proto/Host reconstruction. The 200-frame streaming
    # cycle is the verification surface pre-A (image built + WS handshake
    # succeeds + frames flow through Caddy TLS) — that's what we assert until
    # Plan A merges, then enable the engine-reply fail() assertion.
    ASSERT_ENGINE_REPLY = True  # Set True once Plan A lands.
    if received < 1:
        if ASSERT_ENGINE_REPLY:
            fail("expected at least one engine-originated text frame (mark/outbound)")
        log("INFO: received 0 engine-originated frames — assertion deferred to slice A")
    else:
        log("PASS: real edge->FOB WS path through Caddy TLS verified — engine replied")
    log("PASS: warm-cache image build + WS handshake + 200-frame stream cycle")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    finally:
        # Always clean up the container + volume — even on FAIL.
        subprocess.run(["docker", "rm", "-f", CONTAINER_NAME],
                       capture_output=True, timeout=10)
        subprocess.run(["docker", "volume", "rm", "-f", DATA_VOLUME],
                       capture_output=True, timeout=10)