use std::time::{Duration, Instant};

use anyhow::Result;
use base64ct::{Base64, Encoding};
use ed25519_dalek::{Signer, SigningKey};
use futures_util::{SinkExt, StreamExt};
use serde_json::json;
use tokio::sync::{mpsc, watch};
use tokio_tungstenite::{connect_async, tungstenite::Message};

use crate::tui::event::{AppEvent, LatencySample, WsMessage};

/// Parameters for public WS subscriptions — symbol + kline interval.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubscriptionParams {
    pub symbol: String,
    pub kline_interval: String,
}

/// Connect to public WebSocket streams with auto-reconnection.
/// Monitors `params_rx` for symbol/interval changes and re-subscribes.
pub async fn connect_public(
    ws_url: String,
    event_tx: mpsc::UnboundedSender<AppEvent>,
    mut params_rx: watch::Receiver<SubscriptionParams>,
) {
    let mut retry_count: u32 = 0;

    loop {
        let current_params = params_rx.borrow_and_update().clone();
        match try_connect_public(&ws_url, &current_params, &event_tx, &mut params_rx).await {
            Ok(None) => {
                // Clean shutdown (receiver dropped)
                break;
            }
            Ok(Some(_)) => {
                retry_count = 0;
            }
            Err(_e) => {
                let _ = event_tx.send(AppEvent::WsUpdate(WsMessage::Disconnected));
                retry_count += 1;
                let _ = event_tx.send(AppEvent::WsUpdate(WsMessage::Reconnecting(retry_count)));

                // Exponential backoff: 1s, 2s, 4s, 8s, 16s, 30s max
                let delay = Duration::from_secs((1u64 << retry_count.min(4)).min(30));
                tokio::time::sleep(delay).await;
            }
        }
    }
}

async fn try_connect_public(
    ws_url: &str,
    params: &SubscriptionParams,
    event_tx: &mpsc::UnboundedSender<AppEvent>,
    params_rx: &mut watch::Receiver<SubscriptionParams>,
) -> Result<Option<String>> {
    let url = ws_url.trim_end_matches('/').to_string();
    let (ws_stream, _) = tokio::time::timeout(Duration::from_secs(10), connect_async(&url))
        .await
        .map_err(|_| anyhow::anyhow!("WebSocket connection timed out"))??;

    let (mut write, mut read) = ws_stream.split();

    let mut current = params.clone();

    // Subscribe to public channels
    subscribe_channels(&mut write, &current).await?;

    let _ = event_tx.send(AppEvent::WsUpdate(WsMessage::Connected));

    // Periodic ping for RTT measurement (every 5 seconds)
    let mut ping_interval = tokio::time::interval(Duration::from_secs(5));
    ping_interval.tick().await; // consume the immediate first tick
    let mut last_ping_sent: Option<Instant> = None;

    loop {
        tokio::select! {
            msg = read.next() => {
                match msg {
                    Some(Ok(Message::Text(text))) => {
                        if let Ok(value) = serde_json::from_str::<serde_json::Value>(&text) {
                            let stream = value.get("stream").and_then(|s| s.as_str()).unwrap_or("");
                            let data = value.get("data").cloned().unwrap_or(value.clone());

                            // Extract latency from streams where T = engine timestamp (microseconds).
                            // Skip kline — its T field is the kline end time in seconds.
                            if !stream.starts_with("kline.") {
                                if let Some(sample) = extract_latency(&data) {
                                    let _ = event_tx.send(AppEvent::WsUpdate(
                                        WsMessage::LatencySample(sample),
                                    ));
                                }
                            }

                            let ws_msg = if stream.starts_with("ticker.") {
                                Some(WsMessage::Ticker(data))
                            } else if stream.starts_with("depth.") {
                                Some(WsMessage::Depth(data))
                            } else if stream.starts_with("trade.") {
                                Some(WsMessage::Trade(data))
                            } else if stream.starts_with("kline.") {
                                Some(WsMessage::Kline(data))
                            } else if stream.starts_with("markPrice.") {
                                Some(WsMessage::MarkPrice(data))
                            } else {
                                None
                            };

                            if let Some(msg) = ws_msg {
                                if event_tx.send(AppEvent::WsUpdate(msg)).is_err() {
                                    return Ok(None); // Receiver dropped
                                }
                            }
                        }
                    }
                    Some(Ok(Message::Close(_))) | None => {
                        return Err(anyhow::anyhow!("WebSocket closed"));
                    }
                    Some(Ok(Message::Ping(data))) => {
                        let _ = write.send(Message::Pong(data)).await;
                    }
                    Some(Ok(Message::Pong(_))) => {
                        if let Some(sent) = last_ping_sent.take() {
                            let rtt = sent.elapsed();
                            let _ = event_tx.send(AppEvent::WsUpdate(WsMessage::PingRtt(rtt)));
                        }
                    }
                    Some(Err(e)) => {
                        return Err(anyhow::anyhow!("WebSocket error: {e}"));
                    }
                    _ => {}
                }
            }
            _ = ping_interval.tick() => {
                last_ping_sent = Some(Instant::now());
                let _ = write.send(Message::Ping(vec![])).await;
            }
            Ok(()) = params_rx.changed() => {
                let new_params = params_rx.borrow_and_update().clone();
                // Unsub old, sub new — stay on the same connection
                unsubscribe_channels(&mut write, &current).await;
                current = new_params;
                subscribe_channels(&mut write, &current).await?;
            }
        }
    }
}

/// Extract latency metrics from WS message data containing E (event_time) and T (timestamp) fields.
/// Both are in microseconds since epoch.
fn extract_latency(data: &serde_json::Value) -> Option<LatencySample> {
    let event_time_us = data.get("E").and_then(|v| v.as_i64())?;
    let engine_time_us = data.get("T").and_then(|v| v.as_i64())?;
    let now_us = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .ok()?
        .as_micros() as i64;
    // Use abs for network — small negative values from clock skew still indicate low latency
    let network_us = (now_us - event_time_us).abs();
    let engine_us = (event_time_us - engine_time_us).max(0);

    // Sanity check: reject if either metric exceeds 60s (likely wrong timestamp semantics)
    const MAX_REASONABLE_US: i64 = 60_000_000;
    if network_us > MAX_REASONABLE_US || engine_us > MAX_REASONABLE_US {
        return None;
    }

    Some(LatencySample {
        network_us,
        engine_us,
        timestamp: Instant::now(),
    })
}

fn channel_params(params: &SubscriptionParams) -> Vec<String> {
    let symbol = &params.symbol;
    let interval = &params.kline_interval;
    vec![
        format!("ticker.{symbol}"),
        format!("depth.{symbol}"),
        format!("trade.{symbol}"),
        format!("kline.{interval}.{symbol}"),
        format!("markPrice.{symbol}"),
    ]
}

async fn subscribe_channels<S>(write: &mut S, params: &SubscriptionParams) -> Result<()>
where
    S: SinkExt<Message> + Unpin,
    S::Error: std::fmt::Display,
{
    let msg = json!({ "method": "SUBSCRIBE", "params": channel_params(params) });
    write
        .send(Message::Text(msg.to_string()))
        .await
        .map_err(|e| anyhow::anyhow!("subscribe send failed: {e}"))
}

async fn unsubscribe_channels<S>(write: &mut S, params: &SubscriptionParams)
where
    S: SinkExt<Message> + Unpin,
{
    let msg = json!({ "method": "UNSUBSCRIBE", "params": channel_params(params) });
    let _ = write.send(Message::Text(msg.to_string())).await;
}

// ---------- Private WebSocket ----------

/// Connect to private WebSocket streams with auto-reconnection.
/// Authenticates with ED25519 signature and subscribes to account streams.
pub async fn connect_private(
    ws_url: String,
    api_secret: String,
    event_tx: mpsc::UnboundedSender<AppEvent>,
) {
    let mut retry_count: u32 = 0;

    loop {
        match try_connect_private(&ws_url, &api_secret, &event_tx).await {
            Ok(()) => {
                // Clean shutdown
                break;
            }
            Err(_e) => {
                let _ = event_tx.send(AppEvent::WsUpdate(WsMessage::PrivateDisconnected));
                retry_count += 1;
                let _ = event_tx.send(AppEvent::WsUpdate(WsMessage::PrivateReconnecting(
                    retry_count,
                )));

                // Exponential backoff: 1s, 2s, 4s, 8s, 16s, 30s max
                let delay = Duration::from_secs((1u64 << retry_count.min(4)).min(30));
                tokio::time::sleep(delay).await;
            }
        }
    }
}

async fn try_connect_private(
    ws_url: &str,
    api_secret: &str,
    event_tx: &mpsc::UnboundedSender<AppEvent>,
) -> Result<()> {
    // Decode the base64 API secret into an ED25519 signing key
    let secret_bytes =
        Base64::decode_vec(api_secret).map_err(|_| anyhow::anyhow!("Invalid base64 API secret"))?;
    if secret_bytes.len() != 64 {
        anyhow::bail!(
            "Invalid API secret length: expected 64 bytes, got {}",
            secret_bytes.len()
        );
    }
    let signing_key = SigningKey::from_keypair_bytes(
        secret_bytes
            .as_slice()
            .try_into()
            .map_err(|_| anyhow::anyhow!("Invalid keypair bytes"))?,
    )
    .map_err(|e| anyhow::anyhow!("Invalid signing key: {e}"))?;
    let verifying_key = signing_key.verifying_key();
    let verifying_key_b64 = Base64::encode_string(verifying_key.as_bytes());

    let url = ws_url.trim_end_matches('/').to_string();
    let (ws_stream, _) = tokio::time::timeout(Duration::from_secs(10), connect_async(&url))
        .await
        .map_err(|_| anyhow::anyhow!("Private WebSocket connection timed out"))??;

    let (mut write, mut read) = ws_stream.split();

    // Build auth signature: instruction=subscribe&timestamp={ms}&window=5000
    let timestamp = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as u64;
    let window = 5000u64;
    let sign_payload = format!("instruction=subscribe&timestamp={timestamp}&window={window}");
    let signature = signing_key.sign(sign_payload.as_bytes());
    let signature_b64 = Base64::encode_string(&signature.to_bytes());

    // Subscribe to private streams with auth
    let subscribe_msg = json!({
        "method": "SUBSCRIBE",
        "params": ["account.orderUpdate", "account.balanceUpdate"],
        "signature": [
            verifying_key_b64,
            signature_b64,
            timestamp.to_string(),
            window.to_string(),
        ]
    });
    write
        .send(Message::Text(subscribe_msg.to_string()))
        .await
        .map_err(|e| anyhow::anyhow!("Private subscribe send failed: {e}"))?;

    let _ = event_tx.send(AppEvent::WsUpdate(WsMessage::PrivateConnected));

    loop {
        match read.next().await {
            Some(Ok(Message::Text(text))) => {
                if let Ok(value) = serde_json::from_str::<serde_json::Value>(&text) {
                    let stream = value.get("stream").and_then(|s| s.as_str()).unwrap_or("");
                    let data = value.get("data").cloned().unwrap_or(value.clone());

                    let ws_msg = if stream == "account.orderUpdate" {
                        Some(WsMessage::OrderUpdate(data))
                    } else if stream == "account.balanceUpdate" {
                        Some(WsMessage::BalanceUpdate(data))
                    } else {
                        None
                    };

                    if let Some(msg) = ws_msg {
                        if event_tx.send(AppEvent::WsUpdate(msg)).is_err() {
                            return Ok(()); // Receiver dropped
                        }
                    }
                }
            }
            Some(Ok(Message::Close(_))) | None => {
                return Err(anyhow::anyhow!("Private WebSocket closed"));
            }
            Some(Ok(Message::Ping(data))) => {
                let _ = write.send(Message::Pong(data)).await;
            }
            Some(Err(e)) => {
                return Err(anyhow::anyhow!("Private WebSocket error: {e}"));
            }
            _ => {}
        }
    }
}
