//! WebSocket transport for attached and headless dashboard modes.

use crate::tui::TuiApp;
use crate::tui::app::ThroughputPoint;
use crate::tui::constants::chart;
use crate::tui::dashboard::DashboardState;
use crate::tui::log_capture::LogBuffer;
use crate::tui::{AttachedDashboard, RemoteDashboardStatus};
use anyhow::Context;
use futures::{Sink, SinkExt, Stream, StreamExt, future};
use serde::{Deserialize, Serialize};
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::watch;
use tokio_tungstenite::{
    accept_async_with_config, connect_async_with_config,
    tungstenite::{Message, protocol::WebSocketConfig},
};
use tracing::{info, warn};

pub(crate) const REMOTE_DASHBOARD_LOG_LINE_LIMIT: usize = 8;
pub(crate) const REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT: usize = 64;
const REMOTE_DASHBOARD_SNAPSHOT_HISTORY_LIMIT: usize = 1;
const ATTACHED_DASHBOARD_HISTORY_LIMIT: usize = chart::HISTORY_POINT_COUNT;
const REMOTE_DASHBOARD_TOP_USER_LIMIT: usize = 5;
const DASHBOARD_WEBSOCKET_MAX_MESSAGE_BYTES: usize = 1024 * 1024;

fn dashboard_websocket_config() -> WebSocketConfig {
    WebSocketConfig::default()
        .max_message_size(Some(DASHBOARD_WEBSOCKET_MAX_MESSAGE_BYTES))
        .max_frame_size(Some(DASHBOARD_WEBSOCKET_MAX_MESSAGE_BYTES))
}

#[allow(clippy::cast_precision_loss)]
fn dashboard_message(state: &DashboardState) -> anyhow::Result<Message> {
    let payload = postcard::to_allocvec(state).context("failed to serialize dashboard state")?;
    anyhow::ensure!(
        payload.len() <= DASHBOARD_WEBSOCKET_MAX_MESSAGE_BYTES,
        "dashboard state snapshot exceeds websocket message limit"
    );
    Ok(Message::Binary(payload.into()))
}

fn dashboard_state_from_message(
    message: Result<Message, tokio_tungstenite::tungstenite::Error>,
) -> Option<DashboardState> {
    match message {
        Ok(Message::Binary(bytes)) if bytes.len() <= DASHBOARD_WEBSOCKET_MAX_MESSAGE_BYTES => {
            postcard::from_bytes::<DashboardState>(bytes.as_ref()).ok()
        }
        _ => None,
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
enum DashboardClientRequest {
    SetLogTail { lines: usize },
}

fn dashboard_client_request_message(request: DashboardClientRequest) -> anyhow::Result<Message> {
    Ok(Message::Text(serde_json::to_string(&request)?.into()))
}

fn dashboard_client_request_from_message(
    message: Result<Message, tokio_tungstenite::tungstenite::Error>,
) -> Option<DashboardClientRequest> {
    match message {
        Ok(Message::Text(text)) if text.len() <= DASHBOARD_WEBSOCKET_MAX_MESSAGE_BYTES => {
            serde_json::from_str::<DashboardClientRequest>(text.as_str()).ok()
        }
        _ => None,
    }
}

fn sanitize_log_tail_limit(lines: usize) -> usize {
    lines.clamp(
        REMOTE_DASHBOARD_LOG_LINE_LIMIT,
        REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT,
    )
}

fn dashboard_source_stream(
    source: impl Stream<Item = Result<Message, tokio_tungstenite::tungstenite::Error>>,
) -> impl Stream<Item = DashboardState> {
    source.filter_map(|message| async move { dashboard_state_from_message(message) })
}

fn dashboard_peer_label(peer_addr: Option<SocketAddr>) -> String {
    peer_addr
        .map(|addr| format!(" from {addr}"))
        .unwrap_or_default()
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReaderSessionOutcome {
    RetryAfterDelay,
}

async fn send_dashboard_state_stream<S>(
    mut sink: S,
    mut state_rx: watch::Receiver<Arc<DashboardState>>,
    mut log_tail_rx: watch::Receiver<usize>,
    log_buffer: LogBuffer,
) -> anyhow::Result<()>
where
    S: Sink<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin,
{
    let mut state = state_rx.borrow().clone();
    let mut log_tail = *log_tail_rx.borrow();

    loop {
        sink.send(dashboard_message_for_client(
            state.as_ref(),
            &log_buffer,
            log_tail,
        )?)
        .await?;

        tokio::select! {
            changed = state_rx.changed() => {
                if changed.is_err() {
                    break;
                }
                state = state_rx.borrow().clone();
            }
            changed = log_tail_rx.changed() => {
                if changed.is_err() {
                    break;
                }
                log_tail = *log_tail_rx.borrow();
            }
        }
    }

    Ok(())
}

async fn receive_dashboard_client_requests<S>(
    source: S,
    log_tail_tx: watch::Sender<usize>,
) -> anyhow::Result<()>
where
    S: Stream<Item = Result<Message, tokio_tungstenite::tungstenite::Error>> + Unpin,
{
    source
        .filter_map(|message| future::ready(dashboard_client_request_from_message(message)))
        .for_each(|request| {
            let log_tail_tx = log_tail_tx.clone();
            async move {
                let DashboardClientRequest::SetLogTail { lines } = request;
                let _ = log_tail_tx.send(sanitize_log_tail_limit(lines));
            }
        })
        .await;
    Ok(())
}

async fn handle_dashboard_client(
    stream: TcpStream,
    state_rx: watch::Receiver<Arc<DashboardState>>,
    log_buffer: LogBuffer,
) -> anyhow::Result<()> {
    let peer_addr = stream.peer_addr().ok();
    info!(
        "Dashboard websocket client connected{}",
        dashboard_peer_label(peer_addr)
    );

    let ws_stream = accept_async_with_config(stream, Some(dashboard_websocket_config())).await?;
    let (sink, source) = ws_stream.split();
    let (log_tail_tx, log_tail_rx) = watch::channel(REMOTE_DASHBOARD_LOG_LINE_LIMIT);

    futures::try_join!(
        send_dashboard_state_stream(sink, state_rx, log_tail_rx, log_buffer),
        receive_dashboard_client_requests(source, log_tail_tx)
    )?;

    info!(
        "Dashboard websocket client disconnected{}",
        dashboard_peer_label(peer_addr)
    );
    Ok(())
}

fn spawn_dashboard_client_task(
    stream: TcpStream,
    state_rx: watch::Receiver<Arc<DashboardState>>,
    log_buffer: LogBuffer,
) {
    tokio::spawn(async move {
        if let Err(err) = handle_dashboard_client(stream, state_rx, log_buffer).await {
            warn!("Dashboard websocket client error: {err}");
        }
    });
}

fn dashboard_message_for_client(
    state: &DashboardState,
    log_buffer: &LogBuffer,
    log_tail: usize,
) -> anyhow::Result<Message> {
    if log_tail == REMOTE_DASHBOARD_LOG_LINE_LIMIT {
        return dashboard_message(state);
    }

    let mut tailored_state = state.clone();
    tailored_state.log_lines = log_buffer.recent_lines(log_tail);
    dashboard_message(&tailored_state)
}

fn publish_dashboard_snapshot(app: &mut TuiApp, state_tx: &watch::Sender<Arc<DashboardState>>) {
    app.update();
    let state = Arc::new(remote_dashboard_state(app));
    let _ = state_tx.send(state);
}

fn handle_dashboard_accept_result(
    accept_result: std::io::Result<(TcpStream, SocketAddr)>,
    state_tx: &watch::Sender<Arc<DashboardState>>,
    log_buffer: &LogBuffer,
) {
    match accept_result {
        Ok((stream, _)) => {
            let rx = state_tx.subscribe();
            spawn_dashboard_client_task(stream, rx, log_buffer.clone());
        }
        Err(err) => {
            warn!(error = %err, "Dashboard websocket accept failed; continuing");
        }
    }
}

/// Bind the dashboard websocket listener.
///
/// # Errors
/// Returns an error when the address is not loopback or the socket cannot be bound.
pub async fn bind_dashboard_listener(listen_addr: SocketAddr) -> anyhow::Result<TcpListener> {
    anyhow::ensure!(
        listen_addr.ip().is_loopback(),
        "Dashboard websocket listener must bind to a loopback address; got {listen_addr}"
    );
    TcpListener::bind(listen_addr).await.with_context(|| {
        format!(
            "Failed to bind dashboard websocket listener at {listen_addr}; ensure that socket is free and distinct from the proxy listener"
        )
    })
}

/// Run the headless dashboard publisher on an already-bound listener.
///
/// # Errors
/// Returns an error if websocket publishing fails after startup.
pub async fn run_dashboard_publisher_on_listener(
    mut app: TuiApp,
    listener: TcpListener,
    mut shutdown_rx: tokio::sync::mpsc::Receiver<()>,
) -> anyhow::Result<()> {
    let (state_tx, _state_rx) = watch::channel(Arc::new(remote_dashboard_state(&app)));
    let log_buffer = app.log_buffer().as_ref().clone();
    let mut update_interval = tokio::time::interval(Duration::from_millis(250));
    update_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

    loop {
        tokio::select! {
            _ = shutdown_rx.recv() => {
                break;
            }
            _ = update_interval.tick() => {
                publish_dashboard_snapshot(&mut app, &state_tx);
            }
            accept_result = listener.accept() => {
                handle_dashboard_accept_result(accept_result, &state_tx, &log_buffer);
            }
        }
    }

    Ok(())
}

/// Run the headless dashboard websocket publisher.
///
/// # Errors
/// Returns an error if the listener cannot be bound or the publisher exits with
/// an unrecoverable websocket error.
pub async fn run_dashboard_publisher(
    app: TuiApp,
    listen_addr: SocketAddr,
    shutdown_rx: tokio::sync::mpsc::Receiver<()>,
) -> anyhow::Result<()> {
    let listener = bind_dashboard_listener(listen_addr).await?;
    run_dashboard_publisher_on_listener(app, listener, shutdown_rx).await
}

/// Spawn a background task that keeps the attached TUI synced to a websocket dashboard.
pub(crate) fn spawn_dashboard_reader(
    connect_addr: SocketAddr,
    state_tx: watch::Sender<AttachedDashboard>,
    log_tail_rx: watch::Receiver<usize>,
) -> tokio::task::JoinHandle<()> {
    spawn_dashboard_reader_with_retry_delay(
        connect_addr,
        state_tx,
        log_tail_rx,
        Duration::from_secs(1),
    )
}

fn spawn_dashboard_reader_with_retry_delay(
    connect_addr: SocketAddr,
    state_tx: watch::Sender<AttachedDashboard>,
    log_tail_rx: watch::Receiver<usize>,
    retry_delay: Duration,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let ws_url = format!("ws://{connect_addr}");

        info!("Connecting attached dashboard client to {ws_url}");

        loop {
            let session_result =
                dashboard_reader_session(&ws_url, state_tx.clone(), log_tail_rx.clone()).await;
            if !dashboard_reader_should_retry(&session_result) {
                if let Err(err) = session_result {
                    warn!("Attached dashboard websocket reader stopped for {ws_url}: {err}");
                }
                break;
            }

            let last_error = retry_reason(&session_result);
            let last_snapshot = state_tx.borrow().latest_state.clone();
            let _ = state_tx.send(AttachedDashboard::reconnecting(
                connect_addr,
                retry_delay,
                last_snapshot,
                last_error.clone(),
            ));
            warn!(
                "Attached dashboard websocket disconnected or unavailable; retrying in {retry_delay:?}: {last_error}"
            );
            tokio::time::sleep(retry_delay).await;
            info!("Reconnecting attached dashboard client to {ws_url}");
        }
    })
}

fn dashboard_reader_should_retry(outcome: &anyhow::Result<ReaderSessionOutcome>) -> bool {
    matches!(outcome, Ok(ReaderSessionOutcome::RetryAfterDelay))
}

async fn dashboard_reader_session(
    ws_url: &str,
    state_tx: watch::Sender<AttachedDashboard>,
    mut log_tail_rx: watch::Receiver<usize>,
) -> anyhow::Result<ReaderSessionOutcome> {
    let connect_addr = dashboard_target_from_ws_url(ws_url)?;
    let (ws_stream, _) =
        match connect_async_with_config(ws_url, Some(dashboard_websocket_config()), false).await {
            Ok(connection) => connection,
            Err(err) => {
                warn!("Failed to connect attached dashboard client to {ws_url}: {err}");
                return Ok(ReaderSessionOutcome::RetryAfterDelay);
            }
        };

    info!("Attached dashboard websocket connected to {ws_url}");
    let last_snapshot = state_tx.borrow().latest_state.clone();
    let _ = state_tx.send(AttachedDashboard::connected(connect_addr, last_snapshot));
    let (sink, source) = ws_stream.split();
    let send_requests = send_dashboard_client_requests(sink, &mut log_tail_rx);
    let forward_states = forward_dashboard_states(source, state_tx);
    tokio::pin!(send_requests);
    tokio::pin!(forward_states);

    match future::select(send_requests, forward_states).await {
        future::Either::Left((result, _)) | future::Either::Right((result, _)) => result?,
    }

    warn!("Attached dashboard websocket stream ended for {ws_url}");
    Ok(ReaderSessionOutcome::RetryAfterDelay)
}

async fn send_dashboard_client_requests<S>(
    mut sink: S,
    log_tail_rx: &mut watch::Receiver<usize>,
) -> anyhow::Result<()>
where
    S: Sink<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin,
{
    let mut current_tail = *log_tail_rx.borrow();
    if current_tail != REMOTE_DASHBOARD_LOG_LINE_LIMIT {
        sink.send(dashboard_client_request_message(
            DashboardClientRequest::SetLogTail {
                lines: current_tail,
            },
        )?)
        .await?;
    }

    loop {
        if log_tail_rx.changed().await.is_err() {
            break;
        }
        current_tail = *log_tail_rx.borrow();
        sink.send(dashboard_client_request_message(
            DashboardClientRequest::SetLogTail {
                lines: current_tail,
            },
        )?)
        .await?;
    }

    Ok(())
}

fn trim_remote_dashboard_state(state: &mut DashboardState) {
    state.top_users.truncate(REMOTE_DASHBOARD_TOP_USER_LIMIT);
}

fn remote_dashboard_state(app: &TuiApp) -> DashboardState {
    let mut state = app.snapshot_state_with_limits(
        Some(REMOTE_DASHBOARD_LOG_LINE_LIMIT),
        Some(REMOTE_DASHBOARD_SNAPSHOT_HISTORY_LIMIT),
    );
    trim_remote_dashboard_state(&mut state);
    state
}

fn append_history_samples(
    previous: &[ThroughputPoint],
    incoming: &[ThroughputPoint],
    limit: usize,
) -> Vec<ThroughputPoint> {
    let incoming_start = incoming.len().saturating_sub(limit);
    let incoming = &incoming[incoming_start..];
    let previous_limit = limit.saturating_sub(incoming.len());
    let previous_start = previous.len().saturating_sub(previous_limit);
    let previous = &previous[previous_start..];

    let mut accumulated = Vec::with_capacity(previous.len() + incoming.len());
    accumulated.extend_from_slice(previous);
    accumulated.extend_from_slice(incoming);
    accumulated
}

fn accumulate_dashboard_state(
    previous: Option<&DashboardState>,
    mut incoming: DashboardState,
) -> DashboardState {
    incoming.client_history = append_history_samples(
        previous
            .map(|state| state.client_history.as_slice())
            .unwrap_or_default(),
        &incoming.client_history,
        ATTACHED_DASHBOARD_HISTORY_LIMIT,
    );

    for (idx, backend) in incoming.backend_views.iter_mut().enumerate() {
        backend.history = append_history_samples(
            previous
                .and_then(|state| state.backend_views.get(idx))
                .map(|backend| backend.history.as_slice())
                .unwrap_or_default(),
            &backend.history,
            ATTACHED_DASHBOARD_HISTORY_LIMIT,
        );
    }

    incoming
}

async fn forward_dashboard_states<S>(
    source: S,
    state_tx: watch::Sender<AttachedDashboard>,
) -> anyhow::Result<()>
where
    S: Stream<Item = Result<Message, tokio_tungstenite::tungstenite::Error>> + Unpin,
{
    let mut accumulated = state_tx.borrow().latest_state.clone();
    let source = dashboard_source_stream(source);
    futures::pin_mut!(source);

    while let Some(snapshot) = source.next().await {
        let connect_addr = match &state_tx.borrow().status {
            RemoteDashboardStatus::Connecting { target }
            | RemoteDashboardStatus::Connected { target }
            | RemoteDashboardStatus::Reconnecting { target, .. } => target.to_owned(),
        };
        let state = Arc::new(accumulate_dashboard_state(accumulated.as_deref(), snapshot));
        state_tx
            .send(AttachedDashboard::connected(
                connect_addr,
                Some(state.clone()),
            ))
            .map_err(|_| anyhow::anyhow!("attached dashboard state receiver dropped"))?;
        accumulated = Some(state);
    }

    Ok(())
}

fn retry_reason(outcome: &anyhow::Result<ReaderSessionOutcome>) -> String {
    outcome
        .as_ref()
        .map_or_else(std::string::ToString::to_string, |_| {
            "connection dropped".to_string()
        })
}

fn dashboard_target_from_ws_url(ws_url: &str) -> anyhow::Result<SocketAddr> {
    ws_url
        .strip_prefix("ws://")
        .context("dashboard websocket URL must start with ws://")?
        .parse()
        .with_context(|| {
            format!("dashboard websocket URL contains an invalid socket address: {ws_url}")
        })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::Server;
    use crate::metrics::MetricsCollector;
    use crate::router::BackendSelector;
    use crate::tui::app::{ThroughputPoint, ViewMode};
    use crate::tui::dashboard::{BackendDisplay, BackendView, BufferPoolStats, DashboardMetrics};
    use crate::tui::{LogBuffer, TuiAppBuilder};
    use crate::types::tui::{HistorySize, Throughput, Timestamp};
    use crate::types::{HostName, MaxConnections, Port, ServerName};
    use futures::{FutureExt, stream};
    use std::sync::Arc;
    use tokio::sync::mpsc;
    use tokio_tungstenite::accept_async;

    fn create_test_servers(count: usize) -> Arc<[Server]> {
        (0..count)
            .map(|i| {
                Server::builder(
                    format!("backend{i}.example.com"),
                    Port::try_new(119).unwrap(),
                )
                .name(format!("Backend {i}"))
                .build()
                .unwrap()
            })
            .collect::<Vec<_>>()
            .into()
    }

    fn build_test_app() -> TuiApp {
        let metrics = MetricsCollector::new(1);
        let router = Arc::new(BackendSelector::new());
        let servers = create_test_servers(1);

        TuiApp::new(metrics, router, servers)
    }

    fn empty_dashboard_state() -> DashboardState {
        DashboardState {
            metrics: DashboardMetrics::default(),
            backend_views: Vec::new(),
            top_users: Vec::new(),
            client_history: Vec::new(),
            system_stats: crate::tui::SystemStats::default(),
            view_mode: ViewMode::Normal,
            show_details: false,
            log_lines: Vec::new(),
            buffer_pool: None,
        }
    }

    fn dashboard_state_with_sample(idx: usize) -> DashboardState {
        let sample = idx as f64;
        DashboardState {
            backend_views: vec![BackendView {
                server: BackendDisplay {
                    host: HostName::try_new("backend.example.com".to_string()).unwrap(),
                    port: Port::try_new(119).unwrap(),
                    name: ServerName::try_new("Backend".to_string()).unwrap(),
                    max_connections: MaxConnections::try_new(10).unwrap(),
                },
                stats: crate::metrics::BackendStats::default(),
                active_connections: 0,
                health_status: crate::metrics::BackendHealthStatus::Healthy,
                pending_count: 0,
                load_ratio: None,
                stateful_count: 0,
                traffic_share: None,
                history: vec![ThroughputPoint::new_backend(
                    Timestamp::now(),
                    Throughput::new(sample),
                    Throughput::new(sample + 1.0),
                    crate::types::tui::CommandsPerSecond::new(sample + 2.0),
                )],
            }],
            client_history: vec![ThroughputPoint::new_client(
                Timestamp::now(),
                Throughput::new(sample),
                Throughput::new(sample + 1.0),
            )],
            log_lines: vec![format!("line {idx}")],
            ..empty_dashboard_state()
        }
    }

    fn encode_dashboard_state(state: &DashboardState) -> Vec<u8> {
        postcard::to_allocvec(state).expect("serialize dashboard state")
    }

    fn decode_dashboard_state(bytes: &[u8]) -> DashboardState {
        postcard::from_bytes::<DashboardState>(bytes).expect("deserialize dashboard state")
    }

    #[test]
    fn dashboard_state_from_message_parses_binary_frames() {
        let state = DashboardState {
            log_lines: vec!["hello".to_string()],
            ..empty_dashboard_state()
        };
        let message = Message::Binary(encode_dashboard_state(&state).into());

        let parsed = dashboard_state_from_message(Ok(message)).expect("state should parse");
        assert_eq!(parsed.log_lines, vec!["hello".to_string()]);
    }

    #[test]
    fn dashboard_state_from_message_rejects_text_frames() {
        let state = DashboardState {
            log_lines: vec!["hello".to_string()],
            ..empty_dashboard_state()
        };
        let message = Message::Text(serde_json::to_string(&state).unwrap().into());

        assert!(dashboard_state_from_message(Ok(message)).is_none());
    }

    #[test]
    fn dashboard_state_from_message_ignores_non_state_and_invalid_payloads() {
        assert!(dashboard_state_from_message(Ok(Message::Binary(vec![1, 2, 3].into()))).is_none());
        assert!(dashboard_state_from_message(Ok(Message::Ping(vec![1, 2].into()))).is_none());
        assert!(
            dashboard_state_from_message(Err(
                tokio_tungstenite::tungstenite::Error::ConnectionClosed
            ))
            .is_none()
        );
        assert!(dashboard_state_from_message(Ok(Message::Text("{not-json}".into()))).is_none());
    }

    #[test]
    fn dashboard_state_from_message_rejects_oversized_binary_frames() {
        let payload = vec![b' '; DASHBOARD_WEBSOCKET_MAX_MESSAGE_BYTES + 1];

        assert!(dashboard_state_from_message(Ok(Message::Binary(payload.into()))).is_none());
    }

    #[test]
    fn dashboard_client_request_round_trips() {
        let message = dashboard_client_request_message(DashboardClientRequest::SetLogTail {
            lines: REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT,
        })
        .expect("serialize request");

        let parsed =
            dashboard_client_request_from_message(Ok(message)).expect("request should parse");

        assert_eq!(
            parsed,
            DashboardClientRequest::SetLogTail {
                lines: REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT
            }
        );
    }

    #[test]
    fn dashboard_client_request_from_message_rejects_oversized_text_frames() {
        let payload = " ".repeat(DASHBOARD_WEBSOCKET_MAX_MESSAGE_BYTES + 1);

        assert!(dashboard_client_request_from_message(Ok(Message::Text(payload.into()))).is_none());
    }

    #[test]
    fn sanitize_log_tail_limit_clamps_to_supported_range() {
        assert_eq!(sanitize_log_tail_limit(1), REMOTE_DASHBOARD_LOG_LINE_LIMIT);
        assert_eq!(
            sanitize_log_tail_limit(REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT + 10),
            REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT
        );
    }

    #[test]
    fn dashboard_peer_label_formats_optional_peer_addresses() {
        assert_eq!(dashboard_peer_label(None), "");
        assert_eq!(
            dashboard_peer_label(Some("127.0.0.1:1234".parse().unwrap())),
            " from 127.0.0.1:1234"
        );
    }

    #[test]
    fn dashboard_reader_should_retry_only_for_retry_outcomes() {
        assert!(dashboard_reader_should_retry(&Ok(
            ReaderSessionOutcome::RetryAfterDelay
        )));
        assert!(!dashboard_reader_should_retry(&Err(anyhow::anyhow!(
            "boom"
        ))));
    }

    #[test]
    fn retry_reason_prefers_source_error_text() {
        assert_eq!(
            retry_reason(&Ok(ReaderSessionOutcome::RetryAfterDelay)),
            "connection dropped"
        );
        assert_eq!(
            retry_reason(&Err(anyhow::anyhow!("connection refused"))),
            "connection refused"
        );
    }

    #[test]
    fn dashboard_target_from_ws_url_parses_socket_addresses() {
        let addr = dashboard_target_from_ws_url("ws://127.0.0.1:8120").expect("target addr");
        assert_eq!(addr, "127.0.0.1:8120".parse().unwrap());
        assert!(dashboard_target_from_ws_url("http://127.0.0.1:8120").is_err());
    }

    #[tokio::test]
    async fn publish_dashboard_snapshot_updates_watch_channel() {
        let mut app = build_test_app();
        let (state_tx, mut state_rx) = watch::channel(Arc::new(app.snapshot_state()));

        publish_dashboard_snapshot(&mut app, &state_tx);

        state_rx.changed().await.expect("snapshot should update");
        let snapshot = state_rx.borrow().clone();
        assert_eq!(snapshot.backend_views.len(), 1);
        assert_eq!(snapshot.view_mode, ViewMode::Normal);
    }

    #[tokio::test]
    async fn dashboard_publisher_bind_error_mentions_dashboard_listener() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind test port");
        let addr = listener.local_addr().expect("local addr");
        let (_shutdown_tx, shutdown_rx) = mpsc::channel::<()>(1);

        let err = run_dashboard_publisher(build_test_app(), addr, shutdown_rx)
            .await
            .expect_err("bind should fail");

        let err = format!("{err:#}");
        assert!(err.contains("Failed to bind dashboard websocket listener"));
        assert!(err.contains("ensure that socket is free"));
    }

    #[test]
    fn dashboard_accept_errors_are_nonfatal() {
        let app = build_test_app();
        let (state_tx, _state_rx) = watch::channel(Arc::new(remote_dashboard_state(&app)));

        handle_dashboard_accept_result(
            Err(std::io::Error::other("synthetic accept error")),
            &state_tx,
            app.log_buffer().as_ref(),
        );
    }

    #[tokio::test]
    async fn publish_dashboard_snapshot_sends_only_latest_history_sample() {
        let metrics = MetricsCollector::new(1);
        let router = Arc::new(BackendSelector::new());
        let servers = create_test_servers(1);
        let log_buffer = LogBuffer::new();
        for i in 0..100 {
            log_buffer.push(format!("line {i}"));
        }

        let mut app = TuiAppBuilder::new(metrics, router, servers)
            .with_log_buffer(log_buffer)
            .with_history_size(HistorySize::new(90))
            .build();

        for _ in 0..80 {
            app.update();
        }

        let (state_tx, mut state_rx) = watch::channel(Arc::new(app.snapshot_state()));
        publish_dashboard_snapshot(&mut app, &state_tx);
        state_rx.changed().await.expect("snapshot should update");

        let snapshot = state_rx.borrow().clone();
        assert_eq!(snapshot.log_lines.len(), REMOTE_DASHBOARD_LOG_LINE_LIMIT);
        assert_eq!(
            snapshot.log_lines.first().map(String::as_str),
            Some("line 92")
        );
        assert_eq!(
            snapshot.client_history.len(),
            REMOTE_DASHBOARD_SNAPSHOT_HISTORY_LIMIT,
        );
        assert_eq!(
            snapshot.backend_views[0].history.len(),
            REMOTE_DASHBOARD_SNAPSHOT_HISTORY_LIMIT,
        );
    }

    #[tokio::test]
    async fn reader_accumulates_remote_history_on_attached_client() {
        let connect_addr: SocketAddr = "127.0.0.1:8120".parse().unwrap();
        let (state_tx, mut state_rx) = watch::channel(AttachedDashboard::connecting(connect_addr));

        let source = stream::iter((0..80).map(|idx| {
            Ok(Message::Binary(
                encode_dashboard_state(&dashboard_state_with_sample(idx)).into(),
            ))
        }));

        forward_dashboard_states(source, state_tx)
            .await
            .expect("forward dashboard states");

        let observed =
            wait_for_log_lines(&mut state_rx, &["line 79"], Duration::from_secs(3)).await;

        assert_eq!(
            observed.client_history.len(),
            ATTACHED_DASHBOARD_HISTORY_LIMIT
        );
        assert_eq!(
            observed.backend_views[0].history.len(),
            ATTACHED_DASHBOARD_HISTORY_LIMIT,
        );
        assert_eq!(observed.client_history[0].sent_per_sec().get(), 20.0);
        assert_eq!(
            observed.backend_views[0].history[0].sent_per_sec().get(),
            20.0,
        );
        assert_eq!(
            observed.client_history.last().unwrap().sent_per_sec().get(),
            79.0,
        );
        assert_eq!(
            observed.backend_views[0]
                .history
                .last()
                .unwrap()
                .sent_per_sec()
                .get(),
            79.0,
        );
    }

    #[test]
    fn trim_remote_dashboard_state_limits_top_users() {
        let mut state = DashboardState {
            metrics: crate::tui::dashboard::DashboardMetrics::default(),
            backend_views: Vec::new(),
            top_users: (0..10)
                .map(|idx| crate::tui::dashboard::DashboardUserStats {
                    username: format!("user-{idx}"),
                    active_connections: 0,
                    total_connections: crate::types::TotalConnections::new(0),
                    bytes_sent: crate::types::BytesSent::ZERO,
                    bytes_received: crate::types::BytesReceived::ZERO,
                    bytes_sent_per_sec: crate::types::BytesPerSecondRate::new(0),
                    bytes_received_per_sec: crate::types::BytesPerSecondRate::new(0),
                    total_commands: crate::metrics::CommandCount::new(0),
                    errors: crate::metrics::ErrorCount::new(0),
                })
                .collect(),
            client_history: Vec::new(),
            system_stats: crate::tui::SystemStats::default(),
            view_mode: ViewMode::Normal,
            show_details: false,
            log_lines: Vec::new(),
            buffer_pool: None,
        };

        trim_remote_dashboard_state(&mut state);

        assert_eq!(state.top_users.len(), REMOTE_DASHBOARD_TOP_USER_LIMIT);
        assert_eq!(
            state.top_users.first().map(|user| user.username.as_str()),
            Some("user-0")
        );
    }

    #[tokio::test]
    async fn dashboard_publisher_rejects_non_loopback_listener() {
        let err = bind_dashboard_listener("0.0.0.0:8120".parse().unwrap())
            .await
            .expect_err("non-loopback bind should fail");

        assert!(format!("{err:#}").contains("must bind to a loopback address"));
    }

    #[tokio::test]
    async fn reader_session_reports_retry_for_disconnected_listener() {
        let addr = "127.0.0.1:1".parse().unwrap();

        let (state_tx, _state_rx) = watch::channel(AttachedDashboard::connecting(addr));
        let (_log_tail_tx, log_tail_rx) = watch::channel(REMOTE_DASHBOARD_LOG_LINE_LIMIT);
        let outcome = dashboard_reader_session(&format!("ws://{addr}"), state_tx, log_tail_rx)
            .await
            .expect("reader session");

        assert_eq!(outcome, ReaderSessionOutcome::RetryAfterDelay);
    }

    async fn wait_for_log_lines(
        state_rx: &mut watch::Receiver<AttachedDashboard>,
        expected: &[&str],
        timeout: Duration,
    ) -> Arc<DashboardState> {
        tokio::time::timeout(timeout, async {
            loop {
                if let Some(state) = state_rx.borrow().latest_state.clone()
                    && state
                        .log_lines
                        .iter()
                        .map(String::as_str)
                        .eq(expected.iter().copied())
                {
                    return state;
                }
                state_rx.changed().await.expect("reader still alive");
            }
        })
        .await
        .expect("observe expected dashboard state")
    }

    #[tokio::test]
    async fn publisher_serves_initial_snapshot() {
        let app = build_test_app();
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind test port");
        let addr = listener.local_addr().expect("local addr");

        let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
        let publisher = tokio::spawn(run_dashboard_publisher_on_listener(
            app,
            listener,
            shutdown_rx,
        ));

        let (ws_stream, _) = tokio::time::timeout(Duration::from_secs(3), async {
            loop {
                match tokio_tungstenite::connect_async(format!("ws://{addr}")).await {
                    Ok(result) => break result,
                    Err(_) => tokio::time::sleep(Duration::from_millis(50)).await,
                }
            }
        })
        .await
        .expect("publisher should accept connections");
        let (mut sink, mut source) = ws_stream.split();

        let message = tokio::time::timeout(Duration::from_secs(3), source.next())
            .await
            .expect("receive snapshot")
            .expect("ws message")
            .expect("valid ws frame");

        let Message::Binary(bytes) = message else {
            panic!("expected binary frame");
        };
        let state = decode_dashboard_state(bytes.as_ref());

        assert_eq!(state.backend_views.len(), 1);
        assert_eq!(state.view_mode, ViewMode::Normal);
        assert!(state.buffer_pool.is_none());

        sink.close().await.expect("close websocket");
        let _ = shutdown_tx.send(()).await;
        tokio::time::timeout(Duration::from_secs(3), publisher)
            .await
            .expect("publisher task should exit")
            .expect("publisher join")
            .expect("publisher should exit cleanly");
    }

    #[tokio::test]
    async fn reader_applies_latest_state() {
        let connect_addr: SocketAddr = "127.0.0.1:8120".parse().unwrap();
        let (state_tx, mut state_rx) = watch::channel(AttachedDashboard::connecting(connect_addr));

        let first = DashboardState {
            backend_views: Vec::new(),
            top_users: Vec::new(),
            client_history: vec![ThroughputPoint::new_client(
                Timestamp::now(),
                Throughput::new(1.0),
                Throughput::new(2.0),
            )],
            metrics: DashboardMetrics::default(),
            system_stats: crate::tui::SystemStats::default(),
            view_mode: ViewMode::Normal,
            show_details: false,
            log_lines: vec!["first".to_string()],
            buffer_pool: Some(BufferPoolStats {
                available: 1,
                in_use: 0,
                total: 1,
            }),
        };
        let mut second = first.clone();
        second.log_lines = vec!["second".to_string()];
        second.show_details = true;

        let source = stream::iter(vec![
            Ok(Message::Binary(encode_dashboard_state(&first).into())),
            Ok(Message::Binary(encode_dashboard_state(&second).into())),
        ]);

        forward_dashboard_states(source, state_tx)
            .await
            .expect("forward dashboard states");

        let observed = wait_for_log_lines(&mut state_rx, &["second"], Duration::from_secs(3)).await;

        assert_eq!(observed.log_lines, vec!["second".to_string()]);
        assert!(observed.show_details);
        assert_eq!(
            state_rx.borrow().status,
            RemoteDashboardStatus::Connected {
                target: connect_addr
            }
        );
    }

    #[tokio::test]
    async fn reader_ignores_malformed_frames() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind test port");
        let addr = listener.local_addr().expect("local addr");

        let (state_tx, mut state_rx) = watch::channel(AttachedDashboard::connecting(addr));
        let (_log_tail_tx, log_tail_rx) = watch::channel(REMOTE_DASHBOARD_LOG_LINE_LIMIT);
        let reader = spawn_dashboard_reader_with_retry_delay(
            addr,
            state_tx,
            log_tail_rx,
            Duration::from_millis(25),
        );

        let (stream, _) = tokio::time::timeout(Duration::from_secs(3), listener.accept())
            .await
            .expect("reader should connect")
            .expect("accept reader connection");
        let mut ws = accept_async(stream).await.expect("accept websocket");

        let good = DashboardState {
            log_lines: vec!["good".to_string()],
            ..empty_dashboard_state()
        };

        ws.send(Message::Binary(encode_dashboard_state(&good).into()))
            .await
            .expect("send good state");

        let observed = wait_for_log_lines(&mut state_rx, &["good"], Duration::from_secs(3)).await;
        assert_eq!(observed.log_lines, vec!["good".to_string()]);

        ws.send(Message::Binary(vec![1, 2, 3].into()))
            .await
            .expect("send binary frame");
        ws.send(Message::Text("{not-json}".into()))
            .await
            .expect("send invalid json");

        tokio::time::sleep(Duration::from_millis(100)).await;
        let latest = state_rx
            .borrow()
            .latest_state
            .clone()
            .expect("state should remain present");
        assert_eq!(latest.log_lines, vec!["good".to_string()]);

        drop(ws);
        reader.abort();
        let _ = reader.await;
    }

    #[tokio::test]
    async fn reader_recovers_after_disconnect() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind test port");
        let addr = listener.local_addr().expect("local addr");

        let (state_tx, mut state_rx) = watch::channel(AttachedDashboard::connecting(addr));
        let (log_tail_tx, log_tail_rx) = watch::channel(REMOTE_DASHBOARD_LOG_LINE_LIMIT);
        let reader = spawn_dashboard_reader_with_retry_delay(
            addr,
            state_tx,
            log_tail_rx,
            Duration::from_millis(25),
        );

        let first_conn = tokio::time::timeout(Duration::from_secs(3), listener.accept())
            .await
            .expect("reader should connect initially")
            .expect("accept initial connection");
        let mut first_ws = accept_async(first_conn.0)
            .await
            .expect("accept initial websocket");

        let first = DashboardState {
            log_lines: vec!["first".to_string()],
            ..empty_dashboard_state()
        };
        first_ws
            .send(Message::Binary(encode_dashboard_state(&first).into()))
            .await
            .expect("send first state");

        let first_observed =
            wait_for_log_lines(&mut state_rx, &["first"], Duration::from_secs(3)).await;
        assert_eq!(first_observed.log_lines, vec!["first".to_string()]);
        assert_eq!(
            state_rx.borrow().status,
            RemoteDashboardStatus::Connected { target: addr }
        );

        drop(first_ws);

        tokio::time::timeout(Duration::from_secs(3), async {
            loop {
                if matches!(
                    &state_rx.borrow().status,
                    RemoteDashboardStatus::Reconnecting { target, .. } if *target == addr
                ) {
                    break;
                }
                state_rx.changed().await.expect("reader still alive");
            }
        })
        .await
        .expect("reader should enter reconnecting state");

        let second_conn = tokio::time::timeout(Duration::from_secs(5), listener.accept())
            .await
            .expect("reader should reconnect")
            .expect("accept reconnect connection");
        let mut second_ws = accept_async(second_conn.0)
            .await
            .expect("accept reconnect websocket");
        assert!(
            second_ws.next().now_or_never().is_none(),
            "default log tail should not send an initial control frame"
        );
        let _ = log_tail_tx.send(REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT);
        let request = tokio::time::timeout(Duration::from_secs(3), second_ws.next())
            .await
            .expect("receive log-tail request")
            .expect("request frame")
            .expect("valid ws frame");
        assert_eq!(
            dashboard_client_request_from_message(Ok(request)),
            Some(DashboardClientRequest::SetLogTail {
                lines: REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT
            })
        );

        let second = DashboardState {
            view_mode: ViewMode::LogFullscreen,
            show_details: true,
            log_lines: vec!["second".to_string()],
            buffer_pool: Some(BufferPoolStats {
                available: 2,
                in_use: 1,
                total: 3,
            }),
            ..empty_dashboard_state()
        };
        second_ws
            .send(Message::Binary(encode_dashboard_state(&second).into()))
            .await
            .expect("send second state");

        let second_observed =
            wait_for_log_lines(&mut state_rx, &["second"], Duration::from_secs(5)).await;

        assert_eq!(second_observed.log_lines, vec!["second".to_string()]);
        assert_eq!(second_observed.view_mode, ViewMode::LogFullscreen);
        assert!(second_observed.show_details);
        assert_eq!(
            second_observed.buffer_pool.as_ref().map(|pool| pool.total),
            Some(3)
        );
        assert_eq!(
            state_rx.borrow().status,
            RemoteDashboardStatus::Connected { target: addr }
        );

        drop(second_ws);
        reader.abort();
        let _ = reader.await;
    }

    #[test]
    fn dashboard_message_for_client_expands_log_tail_on_request() {
        let log_buffer = LogBuffer::new();
        for i in 0..100 {
            log_buffer.push(format!("line {i}"));
        }

        let state = DashboardState {
            log_lines: log_buffer.recent_lines(REMOTE_DASHBOARD_LOG_LINE_LIMIT),
            ..empty_dashboard_state()
        };

        let message = dashboard_message_for_client(
            &state,
            &log_buffer,
            REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT,
        )
        .expect("serialize tailored state");
        let Message::Binary(bytes) = message else {
            panic!("expected binary frame");
        };
        let tailored = decode_dashboard_state(bytes.as_ref());

        assert_eq!(
            tailored.log_lines.len(),
            REMOTE_DASHBOARD_FULLSCREEN_LOG_LINE_LIMIT
        );
        assert_eq!(
            tailored.log_lines.first().map(String::as_str),
            Some("line 36")
        );
        assert_eq!(
            tailored.log_lines.last().map(String::as_str),
            Some("line 99")
        );
    }
}
