//! Service Router — handles PLP SERVICE_REQUEST/SERVICE_RESPONSE packets.
//!
//! This module replaces the need for a separate HTTP control-plane process.
//! Clients send `ServiceRequest` packets over their PLP connection and
//! receive `ServiceResponse` packets back, all on the same QUIC/WS link.

use bytes::Bytes;
use pulse_core::plp::{Header, Packet, PacketFlags, PacketType, PLP_VERSION};
use pulse_core::queue_storage::QueueStorageBackend;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, info, warn};

/// JSON payload for a ServiceRequest packet.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServiceRequest {
    /// Unique request id for correlating responses.
    pub id: String,
    /// Target service name (e.g. "sessions", "schemas", "queues", "metrics").
    pub service: String,
    /// Method to invoke on the service (e.g. "list", "create", "get").
    pub method: String,
    /// Arbitrary JSON parameters.
    #[serde(default)]
    pub params: serde_json::Value,
}

/// JSON payload for a ServiceResponse packet.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServiceResponse {
    /// The request id this response correlates to.
    pub id: String,
    /// HTTP-style status code (200, 400, 404, 500, etc.)
    pub status: u16,
    /// Response data.
    #[serde(default)]
    pub data: serde_json::Value,
    /// Optional error message.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

/// Shared state for control-plane services embedded in the relay.
pub struct ServiceState {
    /// Active sessions: session_id → session info.
    pub sessions: Arc<RwLock<HashMap<String, SessionInfo>>>,
    /// Webhook registry (shared with the HTTP listener).
    pub webhook_registry: Arc<crate::webhook::WebhookRegistry>,
    /// Embedded schema registry
    pub schemas: crate::schema_registry::SchemaStore,
    /// Queue store
    pub queues: crate::queues::QueueStore,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SessionInfo {
    pub session_id: String,
    pub relay_endpoint: String,
    pub status: String,
    pub created_at: u64,
}

impl ServiceState {
    pub fn new(queue_backend: Arc<dyn QueueStorageBackend>) -> Self {
        Self {
            sessions: Arc::new(RwLock::new(HashMap::new())),
            webhook_registry: Arc::new(crate::webhook::WebhookRegistry::new()),
            schemas: Arc::new(RwLock::new(crate::schema_registry::SchemaRegistry::new())),
            queues: Arc::new(crate::queues::QueueManager::new(queue_backend)),
        }
    }
}

/// Route a ServiceRequest to the appropriate handler and return a ServiceResponse.
pub async fn handle_service_request(
    req: ServiceRequest,
    state: &ServiceState,
) -> ServiceResponse {
    debug!(
        "[Service] request: service={} method={} id={}",
        req.service, req.method, req.id
    );

    match req.service.as_str() {
        "sessions" => handle_sessions(&req, state).await,
        "webhooks" => handle_webhooks(&req, state).await,
        "schemas" => crate::schema_registry::handle_schemas_router(&req, &state.schemas).await,
        "queues" => crate::queues::handle_queues_router(&req, &state.queues).await,
        "ping" => ServiceResponse {
            id: req.id,
            status: 200,
            data: serde_json::json!({ "pong": true }),
            error: None,
        },
        _ => ServiceResponse {
            id: req.id.clone(),
            status: 404,
            data: serde_json::Value::Null,
            error: Some(format!("unknown service: {}", req.service)),
        },
    }
}

async fn handle_sessions(req: &ServiceRequest, state: &ServiceState) -> ServiceResponse {
    match req.method.as_str() {
        "list" => {
            let sessions = state.sessions.read().await;
            let list: Vec<&SessionInfo> = sessions.values().collect();
            ServiceResponse {
                id: req.id.clone(),
                status: 200,
                data: serde_json::to_value(&list).unwrap_or_default(),
                error: None,
            }
        }
        "get" => {
            let session_id = req
                .params
                .get("session_id")
                .and_then(|v| v.as_str())
                .unwrap_or("");
            let sessions = state.sessions.read().await;
            match sessions.get(session_id) {
                Some(info) => ServiceResponse {
                    id: req.id.clone(),
                    status: 200,
                    data: serde_json::to_value(info).unwrap_or_default(),
                    error: None,
                },
                None => ServiceResponse {
                    id: req.id.clone(),
                    status: 404,
                    data: serde_json::Value::Null,
                    error: Some(format!("session not found: {}", session_id)),
                },
            }
        }
        _ => ServiceResponse {
            id: req.id.clone(),
            status: 400,
            data: serde_json::Value::Null,
            error: Some(format!("unknown method: sessions.{}", req.method)),
        },
    }
}

async fn handle_webhooks(req: &ServiceRequest, state: &ServiceState) -> ServiceResponse {
    match req.method.as_str() {
        "list" => {
            let conn_id = req
                .params
                .get("conn_id")
                .and_then(|v| v.as_str())
                .unwrap_or("");
            let urls = state.webhook_registry.list_for_conn(conn_id).await;
            ServiceResponse {
                id: req.id.clone(),
                status: 200,
                data: serde_json::to_value(&urls).unwrap_or_default(),
                error: None,
            }
        }
        _ => ServiceResponse {
            id: req.id.clone(),
            status: 400,
            data: serde_json::Value::Null,
            error: Some(format!("unknown method: webhooks.{}", req.method)),
        },
    }
}

/// Encode a ServiceResponse into a PLP packet.
pub fn encode_service_response(resp: &ServiceResponse, request_seq: u32) -> Packet {
    let payload = serde_json::to_vec(resp).unwrap_or_default();
    Packet {
        header: Header {
            version: PLP_VERSION,
            packet_type: PacketType::ServiceResponse,
            flags: PacketFlags::RELIABLE,
            stream_id: 0,
            sequence_number: request_seq,
            payload_length: payload.len() as u64,
        },
        payload: Bytes::from(payload),
    }
}

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

    use pulse_core::queue_storage::InMemoryBackend;

    fn test_state() -> ServiceState {
        ServiceState::new(Arc::new(InMemoryBackend::new()))
    }

    #[tokio::test]
    async fn ping_service_returns_pong() {
        let state = test_state();
        let req = ServiceRequest {
            id: "r1".to_string(),
            service: "ping".to_string(),
            method: "".to_string(),
            params: serde_json::Value::Null,
        };
        let resp = handle_service_request(req, &state).await;
        assert_eq!(resp.status, 200);
        assert_eq!(resp.data["pong"], true);
    }

    #[tokio::test]
    async fn unknown_service_returns_404() {
        let state = test_state();
        let req = ServiceRequest {
            id: "r2".to_string(),
            service: "nonexistent".to_string(),
            method: "foo".to_string(),
            params: serde_json::Value::Null,
        };
        let resp = handle_service_request(req, &state).await;
        assert_eq!(resp.status, 404);
        assert!(resp.error.unwrap().contains("unknown service"));
    }

    #[tokio::test]
    async fn sessions_list_returns_empty_initially() {
        let state = test_state();
        let req = ServiceRequest {
            id: "r3".to_string(),
            service: "sessions".to_string(),
            method: "list".to_string(),
            params: serde_json::Value::Null,
        };
        let resp = handle_service_request(req, &state).await;
        assert_eq!(resp.status, 200);
        assert_eq!(resp.data, serde_json::json!([]));
    }

    #[tokio::test]
    async fn sessions_get_returns_404_for_missing() {
        let state = test_state();
        let req = ServiceRequest {
            id: "r4".to_string(),
            service: "sessions".to_string(),
            method: "get".to_string(),
            params: serde_json::json!({ "session_id": "nope" }),
        };
        let resp = handle_service_request(req, &state).await;
        assert_eq!(resp.status, 404);
    }

    #[tokio::test]
    async fn service_response_encodes_to_extended_packet() {
        let resp = ServiceResponse {
            id: "r5".to_string(),
            status: 200,
            data: serde_json::json!({ "ok": true }),
            error: None,
        };
        let pkt = encode_service_response(&resp, 42);
        assert_eq!(pkt.header.packet_type, PacketType::ServiceResponse);
        assert_eq!(pkt.header.sequence_number, 42);
        // Verify roundtrip encode/decode
        let mut buf = Vec::new();
        pkt.encode(&mut buf);
        let decoded = Packet::decode(&mut buf.as_slice()).unwrap();
        assert_eq!(decoded.header.packet_type, PacketType::ServiceResponse);
        let decoded_resp: ServiceResponse =
            serde_json::from_slice(&decoded.payload).unwrap();
        assert_eq!(decoded_resp.id, "r5");
        assert_eq!(decoded_resp.status, 200);
    }
}
