#!/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 io
import json
import os
import socket
import ssl
import struct
import subprocess
import sys
import tarfile
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 docker_cp_root_cert() -> bytes:
    """`docker cp <container>:/data/caddy/pki/authorities/local/root.crt -`
    streams a TAR archive on stdout, not raw PEM (T7 fix). Parse the
    tar stream + extract the root.crt member's bytes."""
    r = subprocess.run(
        ["docker", "cp",
         f"{CONTAINER_NAME}:/data/caddy/pki/authorities/local/root.crt", "-"],
        capture_output=True, timeout=10,
    )
    if r.returncode != 0 or not r.stdout:
        fail(f"docker cp root.crt failed: rc={r.returncode}, stderr={r.stderr.decode('utf-8', 'replace')[-500:]}")
    # The tar archive's member is named "root.crt" (the basename of the
    # source path). Walk the stream + extract that member.
    with tarfile.open(fileobj=io.BytesIO(r.stdout)) as tar:
        for member in tar.getmembers():
            if member.name == "root.crt" or member.name.endswith("/root.crt"):
                f = tar.extractfile(member)
                if f is None:
                    fail(f"docker cp tar stream: root.crt member unreadable")
                return f.read()
        fail(f"docker cp tar stream: no root.crt member found (members: {[m.name for m in tar.getmembers()]})")
    return b""  # unreachable


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 (via docker_cp_root_cert's tar-extract path)."""
    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:
            return docker_cp_root_cert()
        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)
    # T7: docker cp streams a tar archive, not raw PEM; use the shared
    # docker_cp_root_cert() helper's tarfile extraction path.
    root_cert_pem = docker_cp_root_cert()
    ctx.load_verify_locations(cadata=root_cert_pem.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)
