//! Lark WebSocket long-connection event client.
//!
//! [`WsClient`] connects to the Lark WebSocket gateway, handles protobuf-framed
//! messages, reassembles fragmented payloads, and dispatches events to an
//! [`EventDispatcher`].
//!
//! # Example
//!
//! ```rust,no_run
//! use larksuite_oapi_sdk_rs::{Client, EventDispatcher};
//!
//! # #[tokio::main]
//! # async fn main() {
//! let client = Client::builder("APP_ID", "APP_SECRET").build().unwrap();
//! let dispatcher = EventDispatcher::new("VERIFICATION_TOKEN", "ENCRYPT_KEY");
//! client.ws_client(dispatcher).start().await.unwrap();
//! # }
//! ```

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};

use futures_util::{SinkExt as _, StreamExt as _};
use prost::Message as ProstMessage;
use tokio::sync::Mutex;
use tokio::time::sleep;
use tokio_tungstenite::connect_async;
use tokio_tungstenite::tungstenite::Message;

use crate::config::Config;
use crate::constants::HEADER_X_TARGET_SERVICE;
use crate::error::LarkError;
use crate::event::EventDispatcher;

// ── Generated protobuf types ──

pub mod proto {
    include!(concat!(env!("OUT_DIR"), "/larksuite.ws.rs"));
}

// Frame method constants (matches Go SDK FrameType)
const METHOD_CONTROL: i32 = 0;
const METHOD_DATA: i32 = 1;

// Message type header values
const MSG_TYPE_EVENT: &str = "event";
const MSG_TYPE_PING: &str = "ping";
const MSG_TYPE_PONG: &str = "pong";

// Header key constants
const HEADER_TYPE: &str = "type";
const HEADER_SUM: &str = "sum";
const HEADER_SEQ: &str = "seq";
const HEADER_MESSAGE_ID: &str = "message_id";
const HEADER_BIZ_RT: &str = "biz_rt";

// Handshake error header keys (matching Go SDK consts)
const HEADER_HANDSHAKE_STATUS: &str = "Handshake-Status";
const HEADER_HANDSHAKE_MSG: &str = "Handshake-Msg";
const HEADER_HANDSHAKE_AUTH_ERR_CODE: &str = "Handshake-Autherrcode";

// Handshake error codes (matching Go SDK consts)
const AUTH_FAILED: i32 = 514;
const FORBIDDEN: i32 = 403;
const EXCEED_CONN_LIMIT: i32 = 1000040350;

// ── WS endpoint response ──

#[derive(serde::Deserialize)]
struct WsEndpointResp {
    code: i32,
    msg: Option<String>,
    data: Option<WsEndpointData>,
}

#[derive(serde::Deserialize)]
struct WsEndpointData {
    #[serde(rename = "URL")]
    url: String,
    #[serde(rename = "ClientConfig")]
    client_config: Option<ClientConfig>,
}

#[derive(serde::Deserialize, Debug, Clone, Default)]
#[allow(dead_code)]
struct ClientConfig {
    #[serde(rename = "ReconnectCount", default)]
    reconnect_count: i32,
    #[serde(rename = "ReconnectInterval", default)]
    reconnect_interval: u64,
    #[serde(rename = "ReconnectNonce", default)]
    reconnect_nonce: u64,
    #[serde(rename = "PingInterval", default)]
    ping_interval: u64,
}

// ── ACK response payload ──

#[derive(serde::Serialize)]
struct AckResponse {
    code: u16,
    #[serde(skip_serializing_if = "Option::is_none")]
    headers: Option<HashMap<String, String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    data: Option<serde_json::Value>,
}

// ── Fragment reassembly buffer ──

struct FragEntry {
    frames: Vec<Option<Vec<u8>>>,
    created: Instant,
}

impl FragEntry {
    fn new(sum: usize) -> Self {
        Self {
            frames: vec![None; sum],
            created: Instant::now(),
        }
    }

    fn insert(&mut self, index: usize, data: Vec<u8>) {
        if index < self.frames.len() {
            self.frames[index] = Some(data);
        }
    }

    fn complete(&self) -> bool {
        self.frames.iter().all(|f| f.is_some())
    }

    fn assemble(self) -> Vec<u8> {
        self.frames.into_iter().flatten().flatten().collect()
    }

    fn expired(&self) -> bool {
        self.created.elapsed() > Duration::from_secs(5)
    }
}

// ── Header helpers ──

fn get_header(headers: &[proto::Header], key: &str) -> Option<String> {
    headers
        .iter()
        .find(|h| h.key == key)
        .map(|h| h.value.clone())
}

fn get_header_int(headers: &[proto::Header], key: &str) -> i32 {
    get_header(headers, key)
        .and_then(|v| v.parse().ok())
        .unwrap_or(0)
}

fn make_header(key: &str, value: &str) -> proto::Header {
    proto::Header {
        key: key.to_string(),
        value: value.to_string(),
    }
}

// ── WebSocket client ──

type LifecycleCallback = Arc<dyn Fn() + Send + Sync>;
type ErrorCallback = Arc<dyn Fn(&LarkError) + Send + Sync>;

pub struct WsClient {
    config: Arc<Config>,
    dispatcher: Arc<EventDispatcher>,
    domain: String,
    headers: HashMap<String, String>,
    log_level: Option<tracing::Level>,
    ping_interval: Arc<AtomicU64>,
    reconnect_interval: Arc<AtomicU64>,
    reconnect_nonce: Arc<AtomicU64>,
    auto_reconnect: Arc<AtomicBool>,
    on_ready: Option<LifecycleCallback>,
    on_error: Option<ErrorCallback>,
    on_reconnecting: Option<LifecycleCallback>,
    on_reconnected: Option<LifecycleCallback>,
    on_disconnected: Option<LifecycleCallback>,
}

fn ws_log_enabled(log_level: Option<tracing::Level>, level: tracing::Level) -> bool {
    log_level.is_none_or(|max| level <= max)
}

impl WsClient {
    pub fn new(config: Config, dispatcher: EventDispatcher) -> Self {
        let domain = config.base_url.clone();
        let log_level = config.log_level;
        Self {
            config: Arc::new(config),
            dispatcher: Arc::new(dispatcher),
            domain,
            headers: HashMap::new(),
            log_level,
            ping_interval: Arc::new(AtomicU64::new(120)),
            reconnect_interval: Arc::new(AtomicU64::new(120)),
            reconnect_nonce: Arc::new(AtomicU64::new(30)),
            auto_reconnect: Arc::new(AtomicBool::new(true)),
            on_ready: None,
            on_error: None,
            on_reconnecting: None,
            on_reconnected: None,
            on_disconnected: None,
        }
    }

    /// Disable automatic reconnection. When set to `false`, `start()` will
    /// return after the first connection closes instead of reconnecting.
    pub fn auto_reconnect(self, enable: bool) -> Self {
        self.auto_reconnect.store(enable, Ordering::Relaxed);
        self
    }

    /// Set the OpenAPI domain used to request the WebSocket endpoint.
    ///
    /// Defaults to the [`Config`] base URL (`https://open.feishu.cn` unless the
    /// client builder overrides it). Set this to
    /// `https://open.larksuite.com` for Lark global tenants.
    pub fn domain(mut self, domain: impl Into<String>) -> Self {
        self.domain = domain.into();
        self
    }

    /// Set custom headers that will be added to the bootstrap HTTP request
    /// to the WebSocket endpoint.
    ///
    /// Mirrors the Go SDK's `WithHeaders` option.
    pub fn headers(mut self, headers: HashMap<String, String>) -> Self {
        self.headers = headers;
        self
    }

    /// Set the maximum tracing level emitted by this WebSocket client.
    ///
    /// This mirrors the Go SDK's `WithLogLevel` option while keeping Rust's
    /// [`tracing`] subscriber as the logging backend.
    pub fn log_level(mut self, level: tracing::Level) -> Self {
        self.log_level = Some(level);
        self
    }

    pub fn on_ready<F>(mut self, handler: F) -> Self
    where
        F: Fn() + Send + Sync + 'static,
    {
        self.on_ready = Some(Arc::new(handler));
        self
    }

    pub fn on_error<F>(mut self, handler: F) -> Self
    where
        F: Fn(&LarkError) + Send + Sync + 'static,
    {
        self.on_error = Some(Arc::new(handler));
        self
    }

    pub fn on_reconnecting<F>(mut self, handler: F) -> Self
    where
        F: Fn() + Send + Sync + 'static,
    {
        self.on_reconnecting = Some(Arc::new(handler));
        self
    }

    pub fn on_reconnected<F>(mut self, handler: F) -> Self
    where
        F: Fn() + Send + Sync + 'static,
    {
        self.on_reconnected = Some(Arc::new(handler));
        self
    }

    pub fn on_disconnected<F>(mut self, handler: F) -> Self
    where
        F: Fn() + Send + Sync + 'static,
    {
        self.on_disconnected = Some(Arc::new(handler));
        self
    }

    fn log_enabled(&self, level: tracing::Level) -> bool {
        ws_log_enabled(self.log_level, level)
    }

    fn apply_config(&self, conf: &ClientConfig) {
        if conf.ping_interval > 0 {
            self.ping_interval
                .store(conf.ping_interval, Ordering::Relaxed);
        }
        if conf.reconnect_interval > 0 {
            self.reconnect_interval
                .store(conf.reconnect_interval, Ordering::Relaxed);
        }
        if conf.reconnect_nonce > 0 {
            self.reconnect_nonce
                .store(conf.reconnect_nonce, Ordering::Relaxed);
        }
    }

    /// Run the WebSocket connection. If auto-reconnect is enabled (default),
    /// reconnects indefinitely on errors. Otherwise returns after one session.
    pub async fn start(self) -> Result<(), LarkError> {
        let mut consecutive_failures: u32 = 0;
        loop {
            let was_reconnecting = consecutive_failures > 0;
            match self.run_once(was_reconnecting).await {
                Ok(()) => {
                    if self.log_enabled(tracing::Level::INFO) {
                        tracing::info!("ws connection closed");
                    }
                    consecutive_failures = 0;
                }
                Err(e) => {
                    if self.log_enabled(tracing::Level::WARN) {
                        tracing::warn!("ws connection error: {e}");
                    }
                    if let Some(ref handler) = self.on_error {
                        handler(&e);
                    }
                    consecutive_failures += 1;
                }
            }
            if !self.auto_reconnect.load(Ordering::Relaxed) {
                return Ok(());
            }
            // Fast reconnect (5s) on first failure after a working connection.
            // Use server-configured interval on repeated failures to avoid
            // hammering the endpoint.
            let wait = if consecutive_failures <= 1 {
                Duration::from_secs(5)
            } else {
                let interval = self.reconnect_interval.load(Ordering::Relaxed);
                let nonce = self.reconnect_nonce.load(Ordering::Relaxed);
                let jitter = if nonce > 0 {
                    uuid::Uuid::new_v4().as_u64_pair().0 % nonce
                } else {
                    0
                };
                Duration::from_secs(interval + jitter)
            };
            if self.log_enabled(tracing::Level::INFO) {
                tracing::info!("reconnecting in {wait:?}");
            }
            if let Some(ref handler) = self.on_reconnecting {
                handler();
            }
            sleep(wait).await;
        }
    }

    async fn run_once(&self, was_reconnecting: bool) -> Result<(), LarkError> {
        let (url, service_id) = self.get_ws_url().await?;
        if self.log_enabled(tracing::Level::INFO) {
            tracing::info!("connecting to ws endpoint: {url}");
        }

        let (ws_stream, resp) = connect_async(&url)
            .await
            .map_err(|e| LarkError::Event(format!("ws connect failed: {e}")))?;

        // Parse handshake error headers on non-101 responses (Go SDK parseErr)
        if resp.status() != http::StatusCode::SWITCHING_PROTOCOLS {
            return Err(self.parse_handshake_error(&resp));
        }

        if self.log_enabled(tracing::Level::INFO) {
            tracing::info!("ws connected");
        }
        if was_reconnecting && let Some(ref handler) = self.on_reconnected {
            handler();
        }
        if let Some(ref handler) = self.on_ready {
            handler();
        }

        let mut pending_frags: HashMap<String, FragEntry> = HashMap::new();

        let (write, mut read) = ws_stream.split();
        let write = Arc::new(Mutex::new(write));

        // Start ping loop
        let ping_write = write.clone();
        let ping_interval = self.ping_interval.clone();
        let ping_service_id = service_id;
        let ping_log_level = self.log_level;
        let ping_handle = tokio::spawn(async move {
            loop {
                let secs = ping_interval.load(Ordering::Relaxed);
                sleep(Duration::from_secs(secs)).await;
                let frame = proto::Frame {
                    seq_id: 0,
                    log_id: 0,
                    service: ping_service_id,
                    method: METHOD_CONTROL,
                    headers: vec![make_header(HEADER_TYPE, MSG_TYPE_PING)],
                    payload_encoding: None,
                    payload_type: None,
                    payload: None,
                    log_id_new: None,
                };
                let encoded = frame.encode_to_vec();
                let mut w = ping_write.lock().await;
                if (*w).send(Message::Binary(encoded.into())).await.is_err() {
                    break;
                }
                if ws_log_enabled(ping_log_level, tracing::Level::DEBUG) {
                    tracing::debug!("ws ping sent");
                }
            }
        });

        while let Some(msg) = read.next().await {
            let msg = match msg {
                Ok(m) => m,
                Err(e) => {
                    if self.log_enabled(tracing::Level::WARN) {
                        tracing::warn!("ws recv error: {e}");
                    }
                    break;
                }
            };
            match msg {
                Message::Binary(data) => {
                    if let Err(e) = self
                        .handle_binary_message(&data, &mut pending_frags, &write)
                        .await
                        && self.log_enabled(tracing::Level::WARN)
                    {
                        tracing::warn!("ws frame handling error: {e}");
                    }
                }
                Message::Close(_) => {
                    if self.log_enabled(tracing::Level::INFO) {
                        tracing::info!("ws server closed connection");
                    }
                    break;
                }
                _ => {}
            }
        }

        ping_handle.abort();
        if let Some(ref handler) = self.on_disconnected {
            handler();
        }
        Ok(())
    }

    async fn handle_binary_message<S>(
        &self,
        data: &[u8],
        pending_frags: &mut HashMap<String, FragEntry>,
        write: &Arc<Mutex<S>>,
    ) -> Result<(), LarkError>
    where
        S: futures_util::Sink<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin,
    {
        let frame = proto::Frame::decode(data)
            .map_err(|e| LarkError::Event(format!("proto frame decode error: {e}")))?;

        match frame.method {
            METHOD_CONTROL => {
                let msg_type = get_header(&frame.headers, HEADER_TYPE).unwrap_or_default();
                match msg_type.as_str() {
                    MSG_TYPE_PING => {
                        if self.log_enabled(tracing::Level::DEBUG) {
                            tracing::debug!("ws ping received, seq={}", frame.seq_id);
                        }
                        let pong = proto::Frame {
                            method: METHOD_CONTROL,
                            service: frame.service,
                            seq_id: frame.seq_id,
                            log_id: frame.log_id,
                            headers: vec![make_header(HEADER_TYPE, MSG_TYPE_PONG)],
                            payload_encoding: None,
                            payload_type: None,
                            payload: None,
                            log_id_new: None,
                        };
                        let encoded = pong.encode_to_vec();
                        let mut w = write.lock().await;
                        let _ = (*w).send(Message::Binary(encoded.into())).await;
                    }
                    MSG_TYPE_PONG => {
                        if self.log_enabled(tracing::Level::DEBUG) {
                            tracing::debug!("ws pong received");
                        }
                        if let Some(payload) = &frame.payload
                            && !payload.is_empty()
                            && let Ok(conf) = serde_json::from_slice::<ClientConfig>(payload)
                        {
                            if self.log_enabled(tracing::Level::DEBUG) {
                                tracing::debug!("ws client config updated: {conf:?}");
                            }
                            self.apply_config(&conf);
                        }
                    }
                    _ => {
                        if self.log_enabled(tracing::Level::DEBUG) {
                            tracing::debug!("ws unknown control type: {msg_type}");
                        }
                    }
                }
            }
            METHOD_DATA => {
                let msg_type = get_header(&frame.headers, HEADER_TYPE).unwrap_or_default();

                // Card frames are not handled (same as Go SDK)
                if msg_type != MSG_TYPE_EVENT {
                    return Ok(());
                }

                let sum = get_header_int(&frame.headers, HEADER_SUM);
                let seq = get_header_int(&frame.headers, HEADER_SEQ);
                let msg_id = get_header(&frame.headers, HEADER_MESSAGE_ID).unwrap_or_default();

                let payload = if sum <= 1 {
                    frame.payload.clone().unwrap_or_default()
                } else {
                    // Clean expired entries
                    pending_frags.retain(|_, v| !v.expired());

                    let entry = pending_frags
                        .entry(msg_id.clone())
                        .or_insert_with(|| FragEntry::new(sum as usize));
                    entry.insert(seq as usize, frame.payload.clone().unwrap_or_default());
                    if entry.complete() {
                        match pending_frags.remove(&msg_id) {
                            Some(e) => e.assemble(),
                            None => return Ok(()),
                        }
                    } else {
                        return Ok(());
                    }
                };

                let start = Instant::now();
                let status = self.dispatch_event(payload).await;
                let biz_rt = start.elapsed().as_millis().to_string();

                // Send ACK frame
                let ack_code = if status { 200u16 } else { 500 };
                let ack_payload = serde_json::to_vec(&AckResponse {
                    code: ack_code,
                    headers: None,
                    data: None,
                })
                .unwrap_or_default();

                let mut ack_headers = frame.headers.clone();
                ack_headers.push(make_header(HEADER_BIZ_RT, &biz_rt));

                let ack_frame = proto::Frame {
                    seq_id: frame.seq_id,
                    log_id: frame.log_id,
                    service: frame.service,
                    method: frame.method,
                    headers: ack_headers,
                    payload_encoding: frame.payload_encoding.clone(),
                    payload_type: frame.payload_type.clone(),
                    payload: Some(ack_payload),
                    log_id_new: frame.log_id_new.clone(),
                };
                let encoded = ack_frame.encode_to_vec();
                let mut w = write.lock().await;
                let _ = (*w).send(Message::Binary(encoded.into())).await;
            }
            _ => {
                if self.log_enabled(tracing::Level::DEBUG) {
                    tracing::debug!("ws unknown method: {}", frame.method);
                }
            }
        }

        Ok(())
    }

    async fn dispatch_event(&self, payload: Vec<u8>) -> bool {
        use crate::event::EventReq;

        let req = EventReq {
            headers: HashMap::new(),
            body: payload,
            request_uri: String::new(),
        };

        let resp = self.dispatcher.handle(req).await;
        if resp.status_code != 200 {
            if self.log_enabled(tracing::Level::WARN) {
                tracing::warn!(
                    "event dispatch returned {}: {}",
                    resp.status_code,
                    String::from_utf8_lossy(&resp.body)
                );
            }
            return false;
        }

        true
    }

    async fn get_ws_url(&self) -> Result<(String, i32), LarkError> {
        let base = self.domain.trim_end_matches('/');
        let mut url = format!("{base}/callback/ws/endpoint");

        let mut body = serde_json::json!({
            "AppID": self.config.app_id,
            "AppSecret": self.config.app_secret,
        });
        let mut target_service_header = None;

        if let Some(ref provider) = self.config.client_assertion_provider {
            let aud = extract_ws_aud(&self.domain)?;
            let assertion = provider.retrieve_token(&aud).await.map_err(|e| {
                LarkError::ClientAssertion(format!("retrieve client assertion token failed: {e}"))
            })?;
            if assertion.value.is_empty() {
                return Err(LarkError::ClientAssertion(
                    "client assertion token is empty".to_string(),
                ));
            }
            body["AppSecret"] = serde_json::Value::String(String::new());
            body["ClientAssertion"] = serde_json::Value::String(assertion.value);
            if let Some(ref target) = assertion.target_info {
                let target_service = if target.target_service.contains("://") {
                    target.target_service.clone()
                } else {
                    format!("https://{}", target.target_service)
                };
                url = format!(
                    "{}{}/callback/ws/endpoint",
                    target_service.trim_end_matches('/'),
                    target.target_prefix
                );
                target_service_header = Some(aud);
            }
        } else if self.config.app_secret.is_empty() {
            return Err(LarkError::ClientAssertion(
                "appSecret and clientAssertionProvider cannot be nil".to_string(),
            ));
        }

        let mut req = self
            .config
            .http_client
            .post(&url)?
            .header_str("locale", "zh")
            .map_err(|e| LarkError::Event(format!("invalid locale header: {e}")))?;

        // Merge custom headers (Go SDK WithHeaders)
        for (k, v) in &self.headers {
            req = req
                .header_str(k, v)
                .map_err(|e| LarkError::Event(format!("invalid header {k}: {e}")))?;
        }
        if let Some(ref aud) = target_service_header {
            req = req
                .header_str(HEADER_X_TARGET_SERVICE, aud)
                .map_err(|e| LarkError::Event(format!("invalid target service header: {e}")))?;
        }

        let resp = req
            .json(&body)
            .map_err(|e| LarkError::Event(format!("failed to serialize endpoint body: {e}")))?
            .send()
            .await
            .map_err(|e| LarkError::Http(e.into_error()))?;

        let body: WsEndpointResp = resp.json().await?;

        if body.code != 0 {
            return Err(LarkError::Event(format!(
                "ws endpoint returned code {}: {}",
                body.code,
                body.msg.unwrap_or_default()
            )));
        }

        let data = body
            .data
            .ok_or_else(|| LarkError::Event("ws endpoint returned no data".to_string()))?;

        if let Some(ref conf) = data.client_config {
            self.apply_config(conf);
        }

        // Extract service_id from URL query params
        let service_id = data
            .url
            .split('?')
            .nth(1)
            .and_then(|qs| {
                qs.split('&').find_map(|pair| {
                    let (k, v) = pair.split_once('=')?;
                    if k == "service_id" {
                        v.parse::<i32>().ok()
                    } else {
                        None
                    }
                })
            })
            .unwrap_or(0);

        Ok((data.url, service_id))
    }

    fn parse_handshake_error(
        &self,
        resp: &tokio_tungstenite::tungstenite::handshake::client::Response,
    ) -> LarkError {
        let headers = resp.headers();

        let code_str = headers
            .get(HEADER_HANDSHAKE_STATUS)
            .and_then(|v| v.to_str().ok())
            .unwrap_or_default();
        let msg = headers
            .get(HEADER_HANDSHAKE_MSG)
            .and_then(|v| v.to_str().ok())
            .unwrap_or_default();

        let code: i32 = code_str.parse().unwrap_or(0);

        match code {
            AUTH_FAILED => {
                let auth_code_str = headers
                    .get(HEADER_HANDSHAKE_AUTH_ERR_CODE)
                    .and_then(|v| v.to_str().ok())
                    .unwrap_or_default();
                let auth_code: i32 = auth_code_str.parse().unwrap_or(0);
                if auth_code == EXCEED_CONN_LIMIT {
                    LarkError::Event(format!("ws client error: code={code}, msg={msg}"))
                } else {
                    LarkError::Event(format!("ws server error: code={code}, msg={msg}"))
                }
            }
            FORBIDDEN => LarkError::Event(format!("ws client error: code={code}, msg={msg}")),
            _ => LarkError::Event(format!("ws server error: code={code}, msg={msg}")),
        }
    }
}

fn extract_ws_aud(raw_url: &str) -> Result<String, LarkError> {
    let with_scheme;
    let url = if raw_url.contains("://") {
        raw_url
    } else {
        with_scheme = format!("https://{raw_url}");
        &with_scheme
    };
    let parsed = url::Url::parse(url)
        .map_err(|e| LarkError::ClientAssertion(format!("invalid ws domain: {e}")))?;
    parsed
        .host_str()
        .map(ToString::to_string)
        .ok_or_else(|| LarkError::ClientAssertion(format!("invalid ws domain: {raw_url}")))
}

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

    use std::sync::{Arc, Mutex as StdMutex};

    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

    // ── Builder/config tests ──

    #[tokio::test]
    async fn log_level_builder_overrides_config_default() {
        let mut config = Config::new("app_id", "app_secret");
        config.log_level = Some(tracing::Level::ERROR);
        let dispatcher = EventDispatcher::new("", "");

        let client = WsClient::new(config, dispatcher).log_level(tracing::Level::DEBUG);

        assert_eq!(client.log_level, Some(tracing::Level::DEBUG));
        assert!(client.log_enabled(tracing::Level::INFO));
        assert!(client.log_enabled(tracing::Level::DEBUG));
        assert!(!client.log_enabled(tracing::Level::TRACE));
    }

    #[tokio::test]
    async fn get_ws_url_uses_custom_domain() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let request = Arc::new(StdMutex::new(String::new()));
        let captured = Arc::clone(&request);

        let handle = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut buf = vec![0u8; 4096];
            let n = stream.read(&mut buf).await.unwrap();
            *captured.lock().unwrap() = String::from_utf8_lossy(&buf[..n]).into_owned();

            let body = r#"{"code":0,"msg":"ok","data":{"URL":"wss://example.test/ws?service_id=42","ClientConfig":{"PingInterval":15,"ReconnectInterval":16,"ReconnectNonce":7}}}"#;
            let response = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
                body.len()
            );
            stream.write_all(response.as_bytes()).await.unwrap();
            stream.shutdown().await.unwrap();
        });

        let config = Config::new("app_id", "app_secret");
        let dispatcher = EventDispatcher::new("", "");
        let client = WsClient::new(config, dispatcher).domain(format!("http://{addr}/"));

        let (url, service_id) = client.get_ws_url().await.unwrap();
        handle.await.unwrap();

        assert_eq!(url, "wss://example.test/ws?service_id=42");
        assert_eq!(service_id, 42);
        assert_eq!(client.ping_interval.load(Ordering::Relaxed), 15);
        assert_eq!(client.reconnect_interval.load(Ordering::Relaxed), 16);
        assert_eq!(client.reconnect_nonce.load(Ordering::Relaxed), 7);

        let request = request.lock().unwrap();
        assert!(request.starts_with("POST /callback/ws/endpoint HTTP/1.1"));
        assert!(request.contains(r#""AppID":"app_id""#));
        assert!(request.contains(r#""AppSecret":"app_secret""#));
    }

    // ── Protobuf frame encode/decode roundtrip ──

    #[test]
    fn frame_encode_decode_roundtrip_all_fields() {
        let frame = proto::Frame {
            seq_id: 1,
            log_id: 100,
            service: 42,
            method: METHOD_DATA,
            headers: vec![
                make_header(HEADER_TYPE, MSG_TYPE_EVENT),
                make_header(HEADER_SUM, "1"),
            ],
            payload_encoding: Some("json".into()),
            payload_type: Some("event".into()),
            payload: Some(b"hello world".to_vec()),
            log_id_new: Some("new-log-id".into()),
        };
        let encoded = frame.encode_to_vec();
        let decoded = proto::Frame::decode(encoded.as_slice()).unwrap();

        assert_eq!(decoded.seq_id, 1);
        assert_eq!(decoded.log_id, 100);
        assert_eq!(decoded.service, 42);
        assert_eq!(decoded.method, METHOD_DATA);
        assert_eq!(decoded.headers.len(), 2);
        assert_eq!(
            get_header(&decoded.headers, HEADER_TYPE).as_deref(),
            Some(MSG_TYPE_EVENT)
        );
        assert_eq!(
            get_header(&decoded.headers, HEADER_SUM).as_deref(),
            Some("1")
        );
        assert_eq!(decoded.payload_encoding.as_deref(), Some("json"));
        assert_eq!(decoded.payload_type.as_deref(), Some("event"));
        assert_eq!(decoded.payload.as_deref(), Some(b"hello world".as_slice()));
        assert_eq!(decoded.log_id_new.as_deref(), Some("new-log-id"));
    }

    #[test]
    fn frame_encode_decode_minimal_control_frame() {
        let frame = proto::Frame {
            seq_id: 0,
            log_id: 0,
            service: 1,
            method: METHOD_CONTROL,
            headers: vec![make_header(HEADER_TYPE, MSG_TYPE_PING)],
            payload_encoding: None,
            payload_type: None,
            payload: None,
            log_id_new: None,
        };
        let encoded = frame.encode_to_vec();
        assert!(!encoded.is_empty());

        let decoded = proto::Frame::decode(encoded.as_slice()).unwrap();
        assert_eq!(decoded.method, METHOD_CONTROL);
        assert_eq!(decoded.seq_id, 0);
        assert!(decoded.payload.is_none());
        assert_eq!(
            get_header(&decoded.headers, HEADER_TYPE).as_deref(),
            Some(MSG_TYPE_PING)
        );
    }

    // ── Fragment reassembly ──

    #[test]
    fn frag_entry_new_creates_correct_capacity() {
        let entry = FragEntry::new(3);
        assert_eq!(entry.frames.len(), 3);
        assert!(!entry.complete());
    }

    #[test]
    fn frag_entry_insert_and_complete_in_order() {
        let mut entry = FragEntry::new(2);
        entry.insert(0, b"part0".to_vec());
        assert!(!entry.complete());
        entry.insert(1, b"part1".to_vec());
        assert!(entry.complete());
    }

    #[test]
    fn frag_entry_assemble_concatenates_in_order() {
        let mut entry = FragEntry::new(3);
        entry.insert(2, b"three".to_vec());
        entry.insert(0, b"one-".to_vec());
        entry.insert(1, b"two-".to_vec());
        assert!(entry.complete());
        assert_eq!(entry.assemble(), b"one-two-three");
    }

    #[test]
    fn frag_entry_out_of_bounds_insert_is_noop() {
        let mut entry = FragEntry::new(1);
        entry.insert(5, b"oob".to_vec());
        assert!(!entry.complete());
        entry.insert(0, b"ok".to_vec());
        assert!(entry.complete());
    }

    #[test]
    fn frag_entry_not_expired_immediately() {
        let entry = FragEntry::new(1);
        assert!(!entry.expired());
    }

    // ── Header helpers ──

    #[test]
    fn get_header_finds_value() {
        let headers = vec![make_header("type", "event"), make_header("sum", "3")];
        assert_eq!(get_header(&headers, "type").as_deref(), Some("event"));
        assert_eq!(get_header(&headers, "missing"), None);
    }

    #[test]
    fn get_header_int_parses_and_defaults_to_zero() {
        let headers = vec![make_header("sum", "5")];
        assert_eq!(get_header_int(&headers, "sum"), 5);
        assert_eq!(get_header_int(&headers, "missing"), 0);
        assert_eq!(get_header_int(&[], "sum"), 0);
    }

    #[test]
    fn make_header_creates_correct_struct() {
        let h = make_header("key1", "val1");
        assert_eq!(h.key, "key1");
        assert_eq!(h.value, "val1");
    }
}
