use std::sync::{Arc, LazyLock};
use std::time::{Duration, Instant};

use actix_web::{HttpMessage, HttpRequest, HttpResponse, web};
use actix_ws::Message as WsMessage;
use serde::Serialize;
use uuid::Uuid;

use queue::{ProjectRoomEvent, RoomMessageEvent, RoomMessageStreamChunkEvent};
use service::AppService;
use session::Session;

const MAX_TEXT_MESSAGE_LEN: usize = 64 * 1024;
const MAX_MESSAGES_PER_SECOND: u32 = 10;

const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);
const HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(60);
const MAX_IDLE_TIMEOUT: Duration = Duration::from_secs(300);
const RATE_LIMIT_WINDOW: Duration = Duration::from_secs(1);

/// Authenticate WebSocket request: try query parameter token first, then fall back to session.
async fn authenticate_ws_request(
    service: &AppService,
    req: &HttpRequest,
) -> Result<Uuid, actix_web::Error> {
    // Try query parameter token first (one-time use via Redis)
    if let Some(token) = req.uri().query().and_then(|q| {
        q.split('&')
            .find(|p| p.starts_with("token="))
            .and_then(|p| p.split('=').nth(1))
    }) {
        match service.ws_token.validate_token(token).await {
            Ok(uid) => {
                slog::debug!(service.logs, "WS: token auth successful for uid={}", uid);
                return Ok(uid);
            }
            Err(_) => {
                slog::warn!(service.logs, "WS: token auth failed");
                service
                    .room
                    .room_manager
                    .metrics
                    .ws_auth_failures
                    .increment(1);
                return Err(crate::error::ApiError(service::error::AppError::Unauthorized).into());
            }
        }
    }

    // Fall back to session-based auth
    let session = Session::get_session(&mut req.extensions_mut());
    match session.user() {
        Some(uid) => Ok(uid),
        None => {
            service
                .room
                .room_manager
                .metrics
                .ws_auth_failures
                .increment(1);
            Err(crate::error::ApiError(service::error::AppError::Unauthorized).into())
        }
    }
}

async fn check_ws_rate_limit(
    log: &slog::Logger,
    manager: &Arc<room::connection::RoomConnectionManager>,
    message_count: &mut u32,
    rate_window_start: &mut Instant,
) -> bool {
    if rate_window_start.elapsed() > RATE_LIMIT_WINDOW {
        *message_count = 0;
        *rate_window_start = Instant::now();
    }
    *message_count += 1;
    if *message_count > MAX_MESSAGES_PER_SECOND {
        slog::warn!(log, "WS rate limit exceeded");
        manager.metrics.ws_rate_limit_hits.increment(1);
        true
    } else {
        false
    }
}

#[derive(Clone, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum WsEventPayload {
    RoomMessage(RoomMessagePayload),
    ProjectEvent(ProjectEventPayload),
    AiStreamChunk(AiStreamChunkPayload),
}

#[derive(Clone, Serialize)]
pub struct AiStreamChunkPayload {
    pub message_id: Uuid,
    pub room_id: Uuid,
    pub content: String,
    pub done: bool,
    pub error: Option<String>,
}

impl From<RoomMessageStreamChunkEvent> for AiStreamChunkPayload {
    fn from(e: RoomMessageStreamChunkEvent) -> Self {
        Self {
            message_id: e.message_id,
            room_id: e.room_id,
            content: e.content,
            done: e.done,
            error: e.error,
        }
    }
}

impl From<Arc<RoomMessageStreamChunkEvent>> for AiStreamChunkPayload {
    fn from(e: Arc<RoomMessageStreamChunkEvent>) -> Self {
        AiStreamChunkPayload::from((&*e).clone())
    }
}

#[derive(Clone, Serialize)]
pub struct RoomMessagePayload {
    pub id: Uuid,
    pub room_id: Uuid,
    pub sender_type: String,
    pub sender_id: Option<Uuid>,
    pub thread_id: Option<Uuid>,
    pub content: String,
    pub content_type: String,
    pub send_at: chrono::DateTime<chrono::Utc>,
    pub seq: i64,
    pub display_name: Option<String>,
}

impl From<RoomMessageEvent> for RoomMessagePayload {
    fn from(e: RoomMessageEvent) -> Self {
        Self {
            id: e.id,
            room_id: e.room_id,
            sender_type: e.sender_type,
            sender_id: e.sender_id,
            thread_id: e.thread_id,
            content: e.content,
            content_type: e.content_type,
            send_at: e.send_at,
            seq: e.seq,
            display_name: e.display_name,
        }
    }
}

impl From<Arc<RoomMessageEvent>> for RoomMessagePayload {
    fn from(e: Arc<RoomMessageEvent>) -> Self {
        RoomMessagePayload::from((&*e).clone())
    }
}

impl From<&RoomMessageEvent> for RoomMessagePayload {
    fn from(e: &RoomMessageEvent) -> Self {
        Self {
            id: e.id,
            room_id: e.room_id,
            sender_type: e.sender_type.clone(),
            sender_id: e.sender_id,
            thread_id: e.thread_id,
            content: e.content.clone(),
            content_type: e.content_type.clone(),
            send_at: e.send_at,
            seq: e.seq,
            display_name: e.display_name.clone(),
        }
    }
}

#[derive(Clone, Serialize)]
pub struct ProjectEventPayload {
    pub event_type: String,
    pub project_id: Uuid,
    pub room_id: Option<Uuid>,
    pub category_id: Option<Uuid>,
    pub message_id: Option<Uuid>,
    pub seq: Option<i64>,
    pub timestamp: chrono::DateTime<chrono::Utc>,
}

impl From<ProjectRoomEvent> for ProjectEventPayload {
    fn from(e: ProjectRoomEvent) -> Self {
        Self {
            event_type: e.event_type,
            project_id: e.project_id,
            room_id: e.room_id,
            category_id: e.category_id,
            message_id: e.message_id,
            seq: e.seq,
            timestamp: e.timestamp,
        }
    }
}

impl From<Arc<ProjectRoomEvent>> for ProjectEventPayload {
    fn from(e: Arc<ProjectRoomEvent>) -> Self {
        ProjectEventPayload::from((&*e).clone())
    }
}

impl From<&ProjectRoomEvent> for ProjectEventPayload {
    fn from(e: &ProjectRoomEvent) -> Self {
        Self {
            event_type: e.event_type.clone(),
            project_id: e.project_id,
            room_id: e.room_id,
            category_id: e.category_id,
            message_id: e.message_id,
            seq: e.seq,
            timestamp: e.timestamp,
        }
    }
}

#[derive(Clone, Serialize)]
pub struct WsOutEvent {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub room_id: Option<Uuid>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub project_id: Option<Uuid>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event: Option<WsEventPayload>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

pub(crate) fn validate_origin(req: &HttpRequest) -> bool {
    static ALLOWED_ORIGINS: LazyLock<Vec<String>> = LazyLock::new(|| {
        std::env::var("WS_ALLOWED_ORIGINS")
            .map(|v| v.split(',').map(|s| s.trim().to_string()).collect())
            .unwrap_or_else(|_| {
                vec![
                    "http://localhost".to_string(),
                    "https://localhost".to_string(),
                    "http://127.0.0.1".to_string(),
                    "https://127.0.0.1".to_string(),
                    "ws://localhost".to_string(),
                    "wss://localhost".to_string(),
                    "ws://127.0.0.1".to_string(),
                    "wss://127.0.0.1".to_string(),
                ]
            })
    });

    let Some(origin) = req.headers().get("origin") else {
        return true;
    };
    let Ok(origin_str) = origin.to_str() else {
        return false;
    };

    // Exact match (with port)
    if ALLOWED_ORIGINS.iter().any(|allowed| origin_str == *allowed) {
        return true;
    }

    // Strip port: http://localhost:5173 -> http://localhost, http://[::1]:5173 -> http://[::1]
    let origin_without_port = if let Some((scheme_host, port)) = origin_str.rsplit_once(':') {
        if port.chars().all(|c| c.is_ascii_digit()) {
            scheme_host.to_string()
        } else {
            origin_str.to_string()
        }
    } else {
        origin_str.to_string()
    };

    if ALLOWED_ORIGINS
        .iter()
        .any(|allowed| origin_without_port == *allowed)
    {
        return true;
    }

    // Also check if the full origin starts with any allowed prefix
    ALLOWED_ORIGINS
        .iter()
        .any(|allowed| origin_str.starts_with(allowed))
}

pub async fn ws_room(
    room_id: web::Path<Uuid>,
    service: web::Data<AppService>,
    req: HttpRequest,
    stream: web::Payload,
) -> Result<HttpResponse, actix_web::Error> {
    let room_id = room_id.into_inner();

    // Authenticate: try query parameter token first, then session
    let user_id = authenticate_ws_request(&service, &req).await?;

    let origin_val = req
        .headers()
        .get("origin")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("(none)");
    slog::debug!(
        service.logs,
        "WS room connection attempt user_id={} room_id={} origin={}",
        user_id,
        room_id,
        origin_val
    );

    if !validate_origin(&req) {
        slog::warn!(
            service.logs,
            "WS room: origin rejected user_id={} room_id={} origin={}",
            user_id,
            room_id,
            origin_val
        );
        service
            .room
            .room_manager
            .metrics
            .ws_auth_failures
            .increment(1);
        return Err(crate::error::ApiError(service::error::AppError::BadRequest(
            "Invalid origin".into(),
        ))
        .into());
    }

    if let Err(e) = service.room.check_room_access(room_id, user_id).await {
        slog::warn!(
            service.logs,
            "WS room: access denied for user_id={} room_id={} error={}",
            user_id,
            room_id,
            e
        );
        return Err(crate::error::ApiError::from(e).into());
    }

    let manager = service.room.room_manager.clone();
    manager.metrics.ws_connections_active.increment(1.0);
    manager.metrics.ws_connections_total.increment(1);
    manager.metrics.incr_room_connections(room_id).await;

    let (response, mut session, mut msg_stream) = actix_ws::handle(&req, stream)?;

    actix::spawn(async move {
        let mut receiver = match manager.subscribe(room_id, user_id).await {
            Ok(r) => r,
            Err(e) => {
                slog::error!(service.logs, "Failed to subscribe to room: {}", e);
                return;
            }
        };
        let mut stream_rx = manager.subscribe_room_stream(room_id).await;
        let mut shutdown_rx = manager.subscribe_shutdown();

        let mut last_heartbeat = Instant::now();
        let mut last_activity = Instant::now();
        let mut heartbeat_interval = tokio::time::interval(HEARTBEAT_INTERVAL);
        heartbeat_interval.tick().await;

        let mut message_count: u32 = 0;
        let mut rate_window_start = Instant::now();

        loop {
            tokio::select! {
                _ = heartbeat_interval.tick() => {
                    if last_heartbeat.elapsed() > HEARTBEAT_TIMEOUT {
                        slog::warn!(service.logs, "WS room {} heartbeat timeout for user {}", room_id, user_id);
                        manager.metrics.ws_heartbeat_timeout_total.increment(1);
                        let _ = session.close(Some(actix_ws::CloseCode::Policy.into())).await;
                        break;
                    }

                    if last_activity.elapsed() > MAX_IDLE_TIMEOUT {
                        slog::info!(service.logs, "WS room {} idle timeout for user {}", room_id, user_id);
                        manager.metrics.ws_idle_timeout_total.increment(1);
                        let _ = session.close(Some(actix_ws::CloseCode::Normal.into())).await;
                        break;
                    }

                    if session.ping(b"").await.is_err() {
                        break;
                    }
                    manager.metrics.ws_heartbeat_sent_total.increment(1);
                }
                _ = shutdown_rx.recv() => {
                    slog::info!(service.logs, "WS room {} shutdown", room_id);
                    let _ = session.close(Some(actix_ws::CloseCode::Normal.into())).await;
                    break;
                }
                msg = msg_stream.recv() => {
                    match msg {
                        Some(Ok(WsMessage::Ping(bytes))) => {
                            if session.pong(&bytes).await.is_err() {
                                break;
                            }
                            last_heartbeat = Instant::now();
                        }
                        Some(Ok(WsMessage::Pong(_))) => {
                            last_heartbeat = Instant::now();
                        }
                        #[allow(unused_assignments)]
                        Some(Ok(WsMessage::Text(text))) => {
                            if last_activity.elapsed() > MAX_IDLE_TIMEOUT {
                                slog::info!(service.logs, "WS room {} idle timeout for user {}", room_id, user_id);
                                manager.metrics.ws_idle_timeout_total.increment(1);
                                let _ = session.close(Some(actix_ws::CloseCode::Normal.into())).await;
                                break;
                            }
                            last_activity = Instant::now();
                            if check_ws_rate_limit(&service.logs, &manager, &mut message_count, &mut rate_window_start).await {
                                let _ = session.text(serde_json::json!({
                                    "type": "error",
                                    "error": "rate_limit_exceeded",
                                    "max_per_second": MAX_MESSAGES_PER_SECOND
                                }).to_string()).await;
                                break;
                            }

                            if text.len() > MAX_TEXT_MESSAGE_LEN {
                                slog::warn!(service.logs, "WS room {} message too long from user {}: {} bytes", room_id, user_id, text.len());
                                let _ = session.text(serde_json::json!({
                                    "type": "error",
                                    "error": "message_too_long",
                                    "max_bytes": MAX_TEXT_MESSAGE_LEN
                                }).to_string()).await;
                                break;
                            }

                            slog::warn!(service.logs, "WS room {} unexpected text message from user {} ({} bytes) — WS is push-only, use REST to send messages", room_id, user_id, text.len());
                            let _ = session.text(serde_json::json!({
                                "type": "error",
                                "error": "ws_push_only",
                                "message": "WebSocket is for receiving messages only. Use the REST API to send messages."
                            }).to_string()).await;
                            break;
                        }
                        Some(Ok(WsMessage::Binary(_))) => {
                            if check_ws_rate_limit(&service.logs, &manager, &mut message_count, &mut rate_window_start).await {
                                break;
                            }
                            slog::warn!(service.logs, "WS room {} unexpected binary from user {}", room_id, user_id);
                            break;
                        }
                        Some(Ok(WsMessage::Close(reason))) => {
                            let _ = session.close(reason).await;
                            break;
                        }
                        Some(Ok(_)) => {}
                        Some(Err(e)) => {
                            slog::warn!(service.logs, "WS room error: {}", e);
                            break;
                        }
                        None => break,
                    }
                }
                event = receiver.recv() => {
                    match event {
                        Ok(event) => {
                            let payload = WsOutEvent {
                                room_id: Some(room_id),
                                project_id: None,
                                event: Some(WsEventPayload::RoomMessage(event.into())),
                                error: None,
                            };
                            match serde_json::to_string(&payload) {
                                Ok(json) => {
                                    if session.text(json).await.is_err() {
                                        break;
                                    }
                                }
                                Err(e) => {
                                    slog::error!(service.logs, "WS serialize error: {}", e);
                                    break;
                                }
                            }
                        }
                        Err(_) => break,
                    }
                }
                chunk_event = stream_rx.recv() => {
                    match chunk_event {
                        Ok(chunk) => {
                            let payload = WsOutEvent {
                                room_id: Some(room_id),
                                project_id: None,
                                event: Some(WsEventPayload::AiStreamChunk(chunk.into())),
                                error: None,
                            };
                            match serde_json::to_string(&payload) {
                                Ok(json) => {
                                    if session.text(json).await.is_err() {
                                        break;
                                    }
                                }
                                Err(e) => {
                                    slog::error!(service.logs, "WS streaming serialize error: {}", e);
                                }
                            }
                        }
                        Err(_) => {}
                    }
                }
            }
        }

        manager.unsubscribe(room_id, user_id).await;
        manager.metrics.ws_connections_active.decrement(1.0);
        manager.metrics.ws_disconnections_total.increment(1);
        manager.metrics.dec_room_connections(room_id).await;
    });

    Ok(response)
}

pub async fn ws_project(
    project_id: web::Path<Uuid>,
    service: web::Data<AppService>,
    req: HttpRequest,
    stream: web::Payload,
) -> Result<HttpResponse, actix_web::Error> {
    let project_id = project_id.into_inner();

    // Authenticate: try query parameter token first, then session
    let user_id = authenticate_ws_request(&service, &req).await?;

    if !validate_origin(&req) {
        service
            .room
            .room_manager
            .metrics
            .ws_auth_failures
            .increment(1);
        return Err(crate::error::ApiError(service::error::AppError::BadRequest(
            "Invalid origin".into(),
        ))
        .into());
    }

    if let Err(e) = service.room.check_project_member(project_id, user_id).await {
        service
            .room
            .room_manager
            .metrics
            .ws_auth_failures
            .increment(1);
        return Err(crate::error::ApiError::from(e).into());
    }

    if let Err(e) = service
        .room
        .room_manager
        .check_project_connection_rate(project_id, user_id)
        .await
    {
        service
            .room
            .room_manager
            .metrics
            .ws_rate_limit_hits
            .increment(1);
        return Err(crate::error::ApiError::from(e).into());
    }

    let manager = service.room.room_manager.clone();
    manager.metrics.ws_connections_active.increment(1.0);
    manager.metrics.ws_connections_total.increment(1);

    let (response, mut session, mut msg_stream) = actix_ws::handle(&req, stream)?;

    actix::spawn(async move {
        let mut receiver = match manager.subscribe_project(project_id, user_id).await {
            Ok(r) => r,
            Err(e) => {
                slog::error!(service.logs, "Failed to subscribe to project: {}", e);
                return;
            }
        };
        let mut shutdown_rx = manager.subscribe_shutdown();

        let mut last_heartbeat = Instant::now();
        let mut last_activity = Instant::now();
        let mut heartbeat_interval = tokio::time::interval(HEARTBEAT_INTERVAL);
        heartbeat_interval.tick().await;

        let mut message_count: u32 = 0;
        let mut rate_window_start = Instant::now();

        loop {
            tokio::select! {
                _ = heartbeat_interval.tick() => {
                    if last_heartbeat.elapsed() > HEARTBEAT_TIMEOUT {
                        slog::warn!(service.logs, "WS project {} heartbeat timeout for user {}", project_id, user_id);
                        manager.metrics.ws_heartbeat_timeout_total.increment(1);
                        let _ = session.close(Some(actix_ws::CloseCode::Policy.into())).await;
                        break;
                    }

                    if last_activity.elapsed() > MAX_IDLE_TIMEOUT {
                        slog::info!(service.logs, "WS project {} idle timeout for user {}", project_id, user_id);
                        manager.metrics.ws_idle_timeout_total.increment(1);
                        let _ = session.close(Some(actix_ws::CloseCode::Normal.into())).await;
                        break;
                    }

                    if session.ping(b"").await.is_err() {
                        break;
                    }
                    manager.metrics.ws_heartbeat_sent_total.increment(1);
                }
                _ = shutdown_rx.recv() => {
                    slog::info!(service.logs, "WS project {} shutdown", project_id);
                    let _ = session.close(Some(actix_ws::CloseCode::Normal.into())).await;
                    break;
                }
                msg = msg_stream.recv() => {
                    match msg {
                        Some(Ok(WsMessage::Ping(bytes))) => {
                            if session.pong(&bytes).await.is_err() {
                                break;
                            }
                            last_heartbeat = Instant::now();
                        }
                        Some(Ok(WsMessage::Pong(_))) => {
                            last_heartbeat = Instant::now();
                        }
                        #[allow(unused_assignments)]
                        Some(Ok(WsMessage::Text(text))) => {
                            if last_activity.elapsed() > MAX_IDLE_TIMEOUT {
                                slog::info!(service.logs, "WS project {} idle timeout for user {}", project_id, user_id);
                                manager.metrics.ws_idle_timeout_total.increment(1);
                                let _ = session.close(Some(actix_ws::CloseCode::Normal.into())).await;
                                break;
                            }
                            last_activity = Instant::now();
                            slog::warn!(service.logs, "WS project {} unexpected text from user {} ({} bytes) — WS is push-only", project_id, user_id, text.len());
                            let _ = session.text(serde_json::json!({
                                "type": "error",
                                "error": "ws_push_only",
                                "message": "WebSocket is for receiving events only."
                            }).to_string()).await;
                            break;
                        }
                        Some(Ok(WsMessage::Binary(_))) => {
                            if check_ws_rate_limit(&service.logs, &manager, &mut message_count, &mut rate_window_start).await {
                                slog::warn!(service.logs, "WS project {} rate limit exceeded for user {}", project_id, user_id);
                                let _ = session.text(serde_json::json!({
                                    "type": "error",
                                    "error": "rate_limit_exceeded",
                                    "max_per_second": MAX_MESSAGES_PER_SECOND
                                }).to_string()).await;
                                break;
                            }
                            slog::warn!(service.logs, "WS project {} unexpected binary from user {}", project_id, user_id);
                            break;
                        }
                        Some(Ok(WsMessage::Close(reason))) => {
                            let _ = session.close(reason).await;
                            break;
                        }
                        Some(Ok(_)) => {}
                        Some(Err(e)) => {
                            slog::warn!(service.logs, "WS project error: {}", e);
                            break;
                        }
                        None => break,
                    }
                }
                event = receiver.recv() => {
                    match event {
                        Ok(event) => {
                            let payload = WsOutEvent {
                                room_id: event.room_id,
                                project_id: Some(project_id),
                                event: Some(WsEventPayload::ProjectEvent(event.into())),
                                error: None,
                            };
                            match serde_json::to_string(&payload) {
                                Ok(json) => {
                                    if session.text(json).await.is_err() {
                                        break;
                                    }
                                }
                                Err(e) => {
                                    slog::error!(service.logs, "WS serialize error: {}", e);
                                    break;
                                }
                            }
                        }
                        Err(_) => break,
                    }
                }
            }
        }

        manager.unsubscribe_project(project_id, user_id).await;
        manager.metrics.ws_connections_active.decrement(1.0);
        manager.metrics.ws_disconnections_total.increment(1);
    });

    Ok(response)
}