//! 界面与后端之间的视频通道：一个只监听回环地址的 WebSocket。
//!
//! # 为什么不是 Tauri 的 IPC
//!
//! 两路 22fps、每帧约 22 kB，也就是每秒约 1 MB 的裸 JPEG。走 IPC 的话它要被序列化成
//! JSON（base64 再涨 1/3），再在 JS 侧解回来——CPU 花在没人需要的编解码上，延迟也白
//! 涨一截。WebSocket 的二进制帧把原始 JPEG 字节直接交给前端，浏览器用原生解码器
//! （`createImageBitmap`）解，**后端连 JPEG 都不用解**。
//!
//! # 为什么要有 token
//!
//! 端口是随机分配的，但同机器上任何进程都能连回环地址。这条通道里有实时画面，给它一
//! 个每次启动随机生成的 token（前端从 `stream_info` 命令拿到），至少让"另一个本地程序
//! 顺手连上来看画面"变成一件需要先读到那个 token 的事。
//!
//! 说清楚它的边界：这**不是**防对手的措施——对手要看画面得先在自己的机器上跑这套客户
//! 端，那道闸门在 [`cam_core::receiver::Receiver`] 里。这里的 token 只防止本机上的顺手
//! 连接。
//!
//! # 二进制帧的布局
//!
//! ```text
//!   0              4              8      10     12             16
//!   +--------------+--------------+------+------+--------------+---------+
//!   | sender_id u32| frame_id u32 | w u16| h u16| jpeg_len u32 | JPEG... |
//!   +--------------+--------------+------+------+--------------+---------+
//! ```
//!
//! 小端。16 字节头，前端切掉它就能拿到一段完整的 JPEG。`sender_id` 让前端知道这一帧是
//! 哪块板子的——用户端只画绑定的那一路，裁判端按槽位分配画两路。

use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use futures_util::{SinkExt, StreamExt};
use serde::Serialize;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{broadcast, watch};
use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};
use tokio_tungstenite::tungstenite::http::StatusCode;
use tokio_tungstenite::tungstenite::{Bytes, Message};

use cam_core::snapshot::{HitDto, Snapshot};

/// 二进制帧头长度。
pub const FRAME_HEADER_LEN: usize = 16;

/// Ordered control-message capacity. JPEGs use a separate latest-value channel.
pub const CHANNEL_CAPACITY: usize = 256;
const LOCAL_VIDEO_MAX_AGE: Duration = Duration::from_millis(100);

/// 向界面推送的消息。
pub enum ServerMsg {
    /// JSON 文本。
    Text(String),
}

#[derive(Clone)]
pub struct VideoPacket {
    frame_id: u32,
    bytes: Bytes,
    published_at: Instant,
    published_unix_ms: u64,
}

pub type VideoFrames = HashMap<u32, VideoPacket>;
pub type VideoSender = watch::Sender<VideoFrames>;

/// Video uses latest-value semantics; control messages keep ordered delivery.
pub fn publish_video(tx: &VideoSender, frame: &protocol::Frame) {
    let packet = VideoPacket {
        frame_id: frame.frame_id,
        bytes: encode_frame(frame).into(),
        published_at: Instant::now(),
        published_unix_ms: SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64,
    };
    tx.send_modify(|latest| {
        latest.insert(frame.sender_id, packet);
    });
}

/// JSON 文本消息的信封。
///
/// 用内部标签 `t` 而不是两个字段：前端一个 `switch (msg.t)` 就能分发，且不需要为每种
/// 消息各写一次 `JSON.parse` 的错误处理。
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "t", rename_all = "camelCase")]
pub enum WireMsg {
    /// 完整状态快照，5 Hz。
    Snapshot(Box<Snapshot>),
    /// 一次掉血，立即推。
    Hit(HitDto),
    /// 后端拦截了一次"关闭/退出"（用户端还没解锁）。前端据此弹出解锁对话框。
    ConfirmQuit,
    /// 一条说明（迁移、录制报错等），前端显示成一行提示。
    ///
    /// 写成带字段的结构体而不是 `Note(String)`：serde 的**内部标签**（`#[serde(tag)]`）
    /// 只支持载荷是 map 的变体，newtype 里放一个字符串会在运行时报错，而不是编译期。
    Note { message: String },
    /// Timestamp precedes its JPEG so a stalled WebView can reject queued data.
    FrameTiming {
        camera: u32,
        #[serde(rename = "frameId")]
        frame_id: u32,
        #[serde(rename = "publishedUnixMs")]
        published_unix_ms: u64,
    },
}

/// 前端连接这条通道需要的信息。
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StreamInfo {
    /// `ws://127.0.0.1:<port>/?token=...`
    pub url: String,
    pub port: u16,
    /// 当前连着的界面数。用于诊断"画面不动"到底是后端没推还是前端没连。
    pub clients: usize,
}

/// 把一张帧编码成二进制消息。
pub fn encode_frame(frame: &protocol::Frame) -> Vec<u8> {
    let mut out = Vec::with_capacity(FRAME_HEADER_LEN + frame.data.len());

    out.extend_from_slice(&frame.sender_id.to_le_bytes());
    out.extend_from_slice(&frame.frame_id.to_le_bytes());
    out.extend_from_slice(&frame.width.to_le_bytes());
    out.extend_from_slice(&frame.height.to_le_bytes());
    out.extend_from_slice(&(frame.data.len() as u32).to_le_bytes());
    out.extend_from_slice(&frame.data);

    out
}

/// 从二进制消息里解出帧头。前后端对这段字节布局的约定由测试盯着。
#[cfg(test)]
pub fn decode_frame_header(message: &[u8]) -> Option<(u32, u32, u16, u16, usize)> {
    if message.len() < FRAME_HEADER_LEN {
        return None;
    }

    let sender_id = u32::from_le_bytes(message[0..4].try_into().ok()?);
    let frame_id = u32::from_le_bytes(message[4..8].try_into().ok()?);
    let width = u16::from_le_bytes(message[8..10].try_into().ok()?);
    let height = u16::from_le_bytes(message[10..12].try_into().ok()?);
    let len = u32::from_le_bytes(message[12..16].try_into().ok()?) as usize;

    if message.len() < FRAME_HEADER_LEN + len {
        return None;
    }

    Some((sender_id, frame_id, width, height, len))
}

/// 绑一个随机端口的回环 listener。
///
/// 随机端口而不是固定端口：同一台机器上同时开用户端和裁判端时不会撞车。
pub async fn bind() -> std::io::Result<(TcpListener, u16)> {
    let listener = TcpListener::bind(("127.0.0.1", 0)).await?;
    let port = listener.local_addr()?.port();

    Ok((listener, port))
}

/// 接受循环。每个连接一个任务，活到连接断开。
pub async fn serve(
    listener: TcpListener,
    token: String,
    tx: broadcast::Sender<Arc<ServerMsg>>,
    video: VideoSender,
    clients: Arc<AtomicUsize>,
) {
    loop {
        let (stream, peer) = match listener.accept().await {
            Ok(pair) => pair,
            Err(err) => {
                log::warn!("stream: accept failed: {err}");
                tokio::time::sleep(Duration::from_millis(200)).await;
                continue;
            }
        };

        let tx = tx.clone();
        let token = token.clone();
        let clients = Arc::clone(&clients);
        let video = video.subscribe();

        tauri::async_runtime::spawn(async move {
            if let Err(err) = handle_client(stream, peer, &token, tx, video, clients).await {
                log::debug!("stream: client {peer} ended: {err}");
            }
        });
    }
}

async fn handle_client(
    stream: TcpStream,
    peer: std::net::SocketAddr,
    token: &str,
    tx: broadcast::Sender<Arc<ServerMsg>>,
    mut video: watch::Receiver<VideoFrames>,
    clients: Arc<AtomicUsize>,
) -> anyhow::Result<()> {
    stream.set_nodelay(true).ok();

    let expected = token.to_string();
    let callback = move |req: &Request, res: Response| -> Result<Response, ErrorResponse> {
        let query = req.uri().query().unwrap_or("");
        let presented = query
            .split('&')
            .filter_map(|pair| pair.split_once('='))
            .find(|(key, _)| *key == "token")
            .map(|(_, value)| value.to_string());

        match presented {
            Some(value) if constant_time_eq(&value, &expected) => Ok(res),
            _ => {
                log::warn!("stream: refusing {peer}: wrong or missing token");
                let mut response = ErrorResponse::new(Some("bad token".into()));
                *response.status_mut() = StatusCode::FORBIDDEN;
                Err(response)
            }
        }
    };

    let ws = tokio_tungstenite::accept_hdr_async(stream, callback).await?;
    let (mut sink, mut incoming) = ws.split();

    let mut rx = tx.subscribe();
    video.mark_changed();
    let mut last_sent = HashMap::new();
    let mut keepalive = tokio::time::interval(Duration::from_secs(20));
    let count = clients.fetch_add(1, Ordering::Relaxed) + 1;
    log::info!("stream: interface connected from {peer} ({count} connected)");

    // 连上就先推一条提示，前端能据此在界面上显示"通道已接通"。
    let hello = serde_json::to_string(&WireMsg::Note {
        message: "video channel ready".into(),
    })
    .unwrap_or_else(|_| "{}".into());
    let _ = sink.send(Message::text(hello)).await;

    let result = 'messages: loop {
        tokio::select! {
            changed = video.changed() => {
                if changed.is_err() { break Ok(()); }
                // Release the watch read lock before awaiting socket writes.
                let latest: Vec<_> = video.borrow_and_update().iter()
                    .map(|(&camera, packet)| (camera, packet.clone())).collect();
                for (camera, packet) in latest {
                    if last_sent.get(&camera) == Some(&packet.frame_id) { continue; }
                    if packet.published_at.elapsed() >= LOCAL_VIDEO_MAX_AGE { continue; }
                    let timing = serde_json::to_string(&WireMsg::FrameTiming {
                        camera, frame_id: packet.frame_id, published_unix_ms: packet.published_unix_ms,
                    })?;
                    // Keep the 16-byte JPEG header compatible. This small text
                    // message carries host queue age through TCP/WebView queues.
                    let send = async {
                        sink.send(Message::text(timing)).await?;
                        sink.send(Message::Binary(packet.bytes)).await
                    };
                    match tokio::time::timeout(LOCAL_VIDEO_MAX_AGE, send).await {
                        Ok(Ok(())) => { last_sent.insert(camera, packet.frame_id); }
                        Ok(Err(err)) => break 'messages Err(anyhow::anyhow!("video send failed: {err}")),
                        Err(_) => break 'messages Err(anyhow::anyhow!("video client stalled")),
                    }
                }
            },
            message = rx.recv() => match message {
                Ok(msg) => {
                    let out = match &*msg {
                        ServerMsg::Text(text) => Message::text(text.clone()),
                    };

                    if let Err(err) = sink.send(out).await {
                        break Err(anyhow::anyhow!("send failed: {err}"));
                    }
                }
                // 接收慢，积压被挤掉了。对视频这是正常且期望的：跳过旧帧，继续送新的。
                Err(broadcast::error::RecvError::Lagged(skipped)) => {
                    log::debug!("stream: {peer} lagged {skipped} messages; skipping ahead");
                }
                Err(broadcast::error::RecvError::Closed) => {
                    break Ok(());
                }
            },

            _ = keepalive.tick() => {
                // WKWebView 侧的连接可能在休眠后静默失效。ping 让这种情况变成一次明确的
                // 发送失败，而不是一个永远不刷新的窗口。
                if let Err(err) = sink.send(Message::Ping(Bytes::new())).await {
                    break Err(anyhow::anyhow!("keepalive failed: {err}"));
                }
            }

            incoming_msg = incoming.next() => match incoming_msg {
                None => break Ok(()),
                Some(Err(err)) => break Err(anyhow::anyhow!("receive failed: {err}")),
                // 这条通道是单向的：前端发来的东西（pong 除外）一律忽略。
                Some(Ok(_)) => {}
            },
        }
    };

    let count = clients.fetch_sub(1, Ordering::Relaxed) - 1;
    log::info!("stream: interface disconnected ({count} connected)");

    result
}

/// 定长比较，避免 token 校验泄露时间信息。
fn constant_time_eq(a: &str, b: &str) -> bool {
    let (a, b) = (a.as_bytes(), b.as_bytes());

    if a.len() != b.len() {
        return false;
    }

    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b) {
        diff |= x ^ y;
    }

    diff == 0
}

/// 生成一个随机 token。
pub fn random_token() -> String {
    use rand::RngCore;

    let mut bytes = [0u8; 24];
    rand::rng().fill_bytes(&mut bytes);

    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    fn frame(sender: u32, id: u32, len: usize) -> protocol::Frame {
        protocol::Frame {
            sender_id: sender,
            frame_id: id,
            width: 640,
            height: 480,
            data: vec![0xAB; len],
        }
    }

    #[test]
    fn the_binary_header_roundtrips() {
        let encoded = encode_frame(&frame(0x3D4D_8A9A, 4242, 22_000));

        assert_eq!(encoded.len(), FRAME_HEADER_LEN + 22_000);
        assert_eq!(
            decode_frame_header(&encoded),
            Some((0x3D4D_8A9A, 4242, 640, 480, 22_000))
        );
    }

    #[test]
    fn a_truncated_message_is_refused_rather_than_read_past_the_end() {
        let mut encoded = encode_frame(&frame(1, 1, 100));
        encoded.truncate(FRAME_HEADER_LEN + 40);

        // 头里声明 100 字节，实际只有 40：宁可返回 None，也不要让前端拿到一段不完整的
        // JPEG（那会解码失败，或者更糟——解出一张半截的图）。
        assert_eq!(decode_frame_header(&encoded), None);
        assert_eq!(decode_frame_header(&[0u8; 4]), None);
    }

    #[test]
    fn the_token_comparison_is_exact() {
        assert!(constant_time_eq("abc", "abc"));
        assert!(!constant_time_eq("abc", "abd"));
        assert!(!constant_time_eq("abc", "abcd"));
        assert!(!constant_time_eq("", "a"));
        assert!(constant_time_eq("", ""));
    }

    #[test]
    fn tokens_differ_between_runs() {
        assert_ne!(random_token(), random_token());
        assert_eq!(random_token().len(), 48);
    }

    #[test]
    fn a_slow_video_consumer_keeps_only_the_latest_frame_of_each_camera() {
        let (video, mut reader) = watch::channel(VideoFrames::new());
        for id in 1..=1000 {
            publish_video(&video, &frame(1, id, 100));
            publish_video(&video, &frame(2, id + 10, 100));
        }
        let latest = reader.borrow_and_update();
        assert_eq!(latest.len(), 2);
        assert_eq!(latest[&1].frame_id, 1000);
        assert_eq!(latest[&2].frame_id, 1010);
    }

    #[tokio::test]
    async fn a_reconnecting_interface_receives_cached_latest_frames_without_waiting_for_capture() {
        let (listener, port) = bind().await.unwrap();
        let (control, _) = broadcast::channel(CHANNEL_CAPACITY);
        let (video, _) = watch::channel(VideoFrames::new());
        for id in 1..=100 {
            publish_video(&video, &frame(1, id, 100));
            publish_video(&video, &frame(2, id + 10, 100));
        }
        let task = tokio::spawn(serve(
            listener,
            "test-token".into(),
            control,
            video,
            Arc::new(AtomicUsize::new(0)),
        ));
        let (mut client, _) =
            tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{port}/?token=test-token"))
                .await
                .unwrap();
        let mut received = Vec::new();
        let mut timings = HashMap::new();
        while received.len() < 2 {
            let message = tokio::time::timeout(Duration::from_secs(2), client.next())
                .await
                .unwrap()
                .unwrap()
                .unwrap();
            if let Message::Text(text) = &message {
                let value: serde_json::Value = serde_json::from_str(text).unwrap();
                if value["t"] == "frameTiming" {
                    assert!(value["publishedUnixMs"].as_u64().unwrap() > 0);
                    timings.insert(
                        value["camera"].as_u64().unwrap() as u32,
                        value["frameId"].as_u64().unwrap() as u32,
                    );
                }
            }
            if let Message::Binary(bytes) = message {
                let (camera, frame_id, _, _, _) = decode_frame_header(&bytes).unwrap();
                assert_eq!(timings.get(&camera), Some(&frame_id));
                received.push((camera, frame_id));
            }
        }
        received.sort_unstable();
        assert_eq!(received, vec![(1, 100), (2, 110)]);
        client.close(None).await.unwrap();
        task.abort();
    }

    #[tokio::test]
    async fn reconnect_does_not_replay_expired_cached_video() {
        let (listener, port) = bind().await.unwrap();
        let (control, _) = broadcast::channel(CHANNEL_CAPACITY);
        let (video, _) = watch::channel(VideoFrames::new());
        publish_video(&video, &frame(1, 1, 100));
        video.send_modify(|latest| {
            latest.get_mut(&1).unwrap().published_at = Instant::now() - Duration::from_secs(1)
        });
        let task = tokio::spawn(serve(
            listener,
            "test-token".into(),
            control,
            video.clone(),
            Arc::new(AtomicUsize::new(0)),
        ));
        let (mut client, _) =
            tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{port}/?token=test-token"))
                .await
                .unwrap();
        let stale = tokio::time::timeout(Duration::from_millis(100), async {
            while let Some(message) = client.next().await {
                if matches!(message.unwrap(), Message::Binary(_)) {
                    return;
                }
            }
        })
        .await;
        assert!(stale.is_err());
        publish_video(&video, &frame(1, 2, 100));
        let fresh = tokio::time::timeout(Duration::from_secs(2), async {
            while let Some(message) = client.next().await {
                if let Message::Binary(bytes) = message.unwrap() {
                    return decode_frame_header(&bytes).unwrap().1;
                }
            }
            panic!("video connection closed");
        })
        .await
        .unwrap();
        assert_eq!(fresh, 2);
        client.close(None).await.unwrap();
        task.abort();
    }

    #[test]
    fn the_wire_envelope_carries_the_message_kind() {
        let json = serde_json::to_string(&WireMsg::Hit(HitDto {
            board: 1,
            team: 0,
            index: 2,
            from: 100,
            to: 97,
            damage: 3,
            at_ms: 1,
        }))
        .unwrap();

        assert!(json.contains(r#""t":"hit""#), "got {json}");
        assert!(json.contains(r#""damage":3"#), "got {json}");
    }

    #[test]
    fn a_unit_variant_is_just_a_tag() {
        let json = serde_json::to_string(&WireMsg::ConfirmQuit).unwrap();
        assert_eq!(json, r#"{"t":"confirmQuit"}"#);
    }

    #[test]
    fn a_note_is_tagged_too() {
        let json = serde_json::to_string(&WireMsg::Note {
            message: "hi".into(),
        })
        .unwrap();

        assert_eq!(json, r#"{"t":"note","message":"hi"}"#);
    }
}