use futures_util::StreamExt;
use serde_json::Value;
use tokio::sync::mpsc;
use tokio_tungstenite::connect_async;

/// WebSocket events received from the server.
#[derive(Debug, Clone)]
pub enum WsEvent {
    Connected,
    StreamStart {
        turn_id: String,
        model: Option<String>,
    },
    StreamEnd {
        turn_id: String,
    },
    ModelSelection {
        model: String,
    },
    AgentWorking {
        activity: String,
    },
    AgentIdle,
    Raw(Value),
    Disconnected(String),
}

/// Connect to the Roboticus WebSocket and forward parsed events.
pub async fn connect(
    base_url: &str,
    ticket: Option<&str>,
    tx: mpsc::UnboundedSender<WsEvent>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let ws_url = base_url
        .replacen("http://", "ws://", 1)
        .replacen("https://", "wss://", 1);
    let url = match ticket {
        Some(t) => format!("{ws_url}/ws?ticket={}", urlencoding::encode(t)),
        None => format!("{ws_url}/ws"),
    };

    let (ws_stream, _) = connect_async(&url).await?;
    let _ = tx.send(WsEvent::Connected);

    let (_write, mut read) = ws_stream.split();
    while let Some(msg) = read.next().await {
        match msg {
            Ok(tokio_tungstenite::tungstenite::Message::Text(text)) => {
                match serde_json::from_str::<Value>(&text) {
                    Ok(val) => {
                        let event = parse_event(&val);
                        if tx.send(event).is_err() {
                            break;
                        }
                    }
                    Err(e) => {
                        tracing::warn!(
                            error = %e,
                            raw_message = %text,
                            "failed to parse WebSocket message as JSON"
                        );
                    }
                }
            }
            Ok(tokio_tungstenite::tungstenite::Message::Close(_)) => {
                let _ = tx.send(WsEvent::Disconnected("server closed".into()));
                break;
            }
            Err(e) => {
                let _ = tx.send(WsEvent::Disconnected(e.to_string()));
                break;
            }
            _ => {}
        }
    }
    Ok(())
}

fn parse_event(val: &Value) -> WsEvent {
    let event_type = val.get("type").and_then(|v| v.as_str()).unwrap_or_default();
    match event_type {
        "stream_start" => WsEvent::StreamStart {
            turn_id: val
                .get("turn_id")
                .and_then(|v| v.as_str())
                .unwrap_or_default()
                .to_string(),
            model: val.get("model").and_then(|v| v.as_str()).map(String::from),
        },
        "stream_end" => WsEvent::StreamEnd {
            turn_id: val
                .get("turn_id")
                .and_then(|v| v.as_str())
                .unwrap_or_default()
                .to_string(),
        },
        "model_selection" | "model_shift" => WsEvent::ModelSelection {
            model: val
                .get("model")
                .and_then(|v| v.as_str())
                .unwrap_or_default()
                .to_string(),
        },
        "agent_working" => {
            let activity = val
                .get("activity")
                .and_then(|v| v.as_str())
                .unwrap_or("working")
                .to_string();
            // Pipeline progress events (classifying, memory, roster, planning) are
            // for operator logs only — they should not drive the chat typing indicator.
            let workstation = val
                .get("workstation")
                .and_then(|v| v.as_str())
                .unwrap_or("");
            if workstation == "pipeline" {
                // Route to Raw so it appears in logs but not the chat spinner
                WsEvent::Raw(val.clone())
            } else {
                WsEvent::AgentWorking { activity }
            }
        }
        "agent_idle" => WsEvent::AgentIdle,
        _ => WsEvent::Raw(val.clone()),
    }
}

#[cfg(test)]
#[allow(clippy::result_large_err)]
mod tests {
    use super::*;
    use futures_util::SinkExt;
    use serde_json::json;
    use std::sync::{Arc, Mutex};
    use tokio::net::TcpListener;
    use tokio_tungstenite::accept_hdr_async;
    use tokio_tungstenite::tungstenite::Message;
    use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};

    // ── URL construction ────────────────────────────────────────────

    #[test]
    fn url_http_to_ws_no_ticket() {
        let base = "http://localhost:3000";
        let ws_url = base
            .replacen("http://", "ws://", 1)
            .replacen("https://", "wss://", 1);
        let url = format!("{ws_url}/ws");
        assert_eq!(url, "ws://localhost:3000/ws");
    }

    #[test]
    fn url_https_to_wss_no_ticket() {
        let base = "https://example.com";
        let ws_url = base
            .replacen("http://", "ws://", 1)
            .replacen("https://", "wss://", 1);
        let url = format!("{ws_url}/ws");
        assert_eq!(url, "wss://example.com/ws");
    }

    #[test]
    fn url_with_ticket_encoding_special_chars() {
        let base = "http://localhost:3000";
        let ws_url = base
            .replacen("http://", "ws://", 1)
            .replacen("https://", "wss://", 1);
        let ticket = "tok&en=val ue";
        let url = format!("{ws_url}/ws?ticket={}", urlencoding::encode(ticket));
        assert_eq!(url, "ws://localhost:3000/ws?ticket=tok%26en%3Dval%20ue");
    }

    #[test]
    fn url_with_ticket_encoding_unicode() {
        let base = "https://api.example.com";
        let ws_url = base
            .replacen("http://", "ws://", 1)
            .replacen("https://", "wss://", 1);
        let ticket = "日本語";
        let url = format!("{ws_url}/ws?ticket={}", urlencoding::encode(ticket));
        assert!(url.starts_with("wss://api.example.com/ws?ticket="));
        assert!(!url.contains("日本語")); // should be percent-encoded
    }

    #[test]
    fn url_with_plain_ticket() {
        let base = "http://localhost:8080";
        let ws_url = base
            .replacen("http://", "ws://", 1)
            .replacen("https://", "wss://", 1);
        let ticket = "abc123";
        let url = format!("{ws_url}/ws?ticket={}", urlencoding::encode(ticket));
        assert_eq!(url, "ws://localhost:8080/ws?ticket=abc123");
    }

    // ── parse_event ─────────────────────────────────────────────────

    #[test]
    fn parse_event_stream_start() {
        let val = json!({"type": "stream_start", "turn_id": "t-42"});
        let event = parse_event(&val);
        match event {
            WsEvent::StreamStart { turn_id, .. } => assert_eq!(turn_id, "t-42"),
            other => panic!("expected StreamStart, got {other:?}"),
        }
    }

    #[test]
    fn parse_event_stream_start_missing_turn_id() {
        let val = json!({"type": "stream_start"});
        let event = parse_event(&val);
        match event {
            WsEvent::StreamStart { turn_id, .. } => assert_eq!(turn_id, ""),
            other => panic!("expected StreamStart, got {other:?}"),
        }
    }

    #[test]
    fn parse_event_stream_end() {
        let val = json!({"type": "stream_end", "turn_id": "t-99"});
        let event = parse_event(&val);
        match event {
            WsEvent::StreamEnd { turn_id } => assert_eq!(turn_id, "t-99"),
            other => panic!("expected StreamEnd, got {other:?}"),
        }
    }

    #[test]
    fn parse_event_model_selection() {
        let val = json!({"type": "model_selection", "model": "claude-3"});
        let event = parse_event(&val);
        match event {
            WsEvent::ModelSelection { model } => assert_eq!(model, "claude-3"),
            other => panic!("expected ModelSelection, got {other:?}"),
        }
    }

    #[test]
    fn parse_event_model_shift_alias() {
        let val = json!({"type": "model_shift", "model": "gpt-4o"});
        let event = parse_event(&val);
        match event {
            WsEvent::ModelSelection { model } => assert_eq!(model, "gpt-4o"),
            other => panic!("expected ModelSelection, got {other:?}"),
        }
    }

    #[test]
    fn parse_event_unknown_type_becomes_raw() {
        let val = json!({"type": "heartbeat", "ts": 12345});
        let event = parse_event(&val);
        match event {
            WsEvent::Raw(v) => assert_eq!(v["type"], "heartbeat"),
            other => panic!("expected Raw, got {other:?}"),
        }
    }

    #[test]
    fn parse_event_no_type_field_becomes_raw() {
        let val = json!({"data": "something"});
        let event = parse_event(&val);
        match event {
            WsEvent::Raw(v) => assert_eq!(v["data"], "something"),
            other => panic!("expected Raw, got {other:?}"),
        }
    }

    #[test]
    fn parse_event_agent_activity_variants() {
        match parse_event(&json!({"type": "agent_working", "activity": "Planning"})) {
            WsEvent::AgentWorking { activity } => assert_eq!(activity, "Planning"),
            other => panic!("expected AgentWorking, got {other:?}"),
        }
        assert!(matches!(
            parse_event(&json!({"type": "agent_idle"})),
            WsEvent::AgentIdle
        ));
    }

    #[tokio::test]
    async fn connect_emits_connected_parsed_events_and_disconnected() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let request_path = Arc::new(Mutex::new(String::new()));
        let request_path_clone = Arc::clone(&request_path);

        tokio::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            let ws_stream = accept_hdr_async(stream, move |req: &Request, resp: Response| {
                *request_path_clone.lock().unwrap() = req.uri().to_string();
                Ok(resp)
            })
            .await
            .unwrap();
            let (mut write, _) = ws_stream.split();
            write
                .send(Message::Text(
                    json!({"type":"stream_start","turn_id":"turn-1","model":"gpt-test"})
                        .to_string(),
                ))
                .await
                .unwrap();
            write
                .send(Message::Text(
                    json!({"type":"agent_working","activity":"Planning"}).to_string(),
                ))
                .await
                .unwrap();
            write.close().await.unwrap();
        });

        let (tx, mut rx) = mpsc::unbounded_channel();
        connect(&format!("http://{}", addr), Some("ticket with space"), tx)
            .await
            .unwrap();

        assert!(matches!(rx.recv().await, Some(WsEvent::Connected)));
        match rx.recv().await {
            Some(WsEvent::StreamStart { turn_id, model }) => {
                assert_eq!(turn_id, "turn-1");
                assert_eq!(model.as_deref(), Some("gpt-test"));
            }
            other => panic!("expected StreamStart, got {other:?}"),
        }
        match rx.recv().await {
            Some(WsEvent::AgentWorking { activity }) => assert_eq!(activity, "Planning"),
            other => panic!("expected AgentWorking, got {other:?}"),
        }
        match rx.recv().await {
            Some(WsEvent::Disconnected(reason)) => assert!(reason.contains("server closed")),
            other => panic!("expected Disconnected, got {other:?}"),
        }
        assert_eq!(
            request_path.lock().unwrap().as_str(),
            "/ws?ticket=ticket%20with%20space"
        );
    }

    #[tokio::test]
    async fn connect_ignores_invalid_json_messages() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        tokio::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            let ws_stream = accept_hdr_async(stream, |_req: &Request, resp: Response| Ok(resp))
                .await
                .unwrap();
            let (mut write, _) = ws_stream.split();
            write
                .send(Message::Text("not-json".to_string()))
                .await
                .unwrap();
            write.close().await.unwrap();
        });

        let (tx, mut rx) = mpsc::unbounded_channel();
        connect(&format!("http://{}", addr), None, tx)
            .await
            .unwrap();

        assert!(matches!(rx.recv().await, Some(WsEvent::Connected)));
        match rx.recv().await {
            Some(WsEvent::Disconnected(reason)) => assert!(reason.contains("server closed")),
            other => panic!("expected Disconnected, got {other:?}"),
        }
        assert!(
            rx.try_recv().is_err(),
            "invalid JSON should not emit an event"
        );
    }
}
