//! Q-06: Queue Management API
//!
//! REST endpoints for managing durable message queues.
//!
//! Endpoints:
//! - POST   /api/v1/queues          → create a queue
//! - GET    /api/v1/queues          → list all queues with metrics
//! - GET    /api/v1/queues/{name}   → get queue details
//! - DELETE /api/v1/queues/{name}   → delete (purge + remove) a queue
//! - POST   /api/v1/queues/{name}/purge → purge all messages

use crate::service::{ServiceRequest, ServiceResponse};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

/// Shared queue store across the control plane.
pub type QueueStore = Arc<RwLock<HashMap<String, QueueEntry>>>;

/// A queue entry in the control plane's store.
/// Note: This is a lightweight management record; the actual DurableQueue
/// lives in the relay/core process. The control plane tracks metadata.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueEntry {
    pub name: String,
    pub mode: String,
    pub max_delivery_attempts: u32,
    pub default_ttl_secs: Option<u64>,
    pub created_at: i64,
    /// Metrics (updated periodically by relay heartbeats)
    pub depth: u64,
    pub dead_letter_depth: u64,
    pub total_enqueued: u64,
    pub subscriber_count: u32,
    /// Operational state
    pub paused: bool,
}

// ═══════════════════════════════════════════════════════════════════════════
// Request types
// ═══════════════════════════════════════════════════════════════════════════

#[derive(Debug, Deserialize)]
pub struct CreateQueueRequest {
    pub name: String,
    #[serde(default = "default_mode")]
    pub mode: String,
    #[serde(default = "default_max_attempts")]
    pub max_delivery_attempts: u32,
    pub default_ttl_secs: Option<u64>,
}

fn default_mode() -> String {
    "competing".to_string()
}
fn default_max_attempts() -> u32 {
    5
}

// ═══════════════════════════════════════════════════════════════════════════
// Handlers
// ═══════════════════════════════════════════════════════════════════════════

pub async fn handle_queues_router(req: &ServiceRequest, store: &QueueStore) -> ServiceResponse {
    match req.method.as_str() {
        "create" => {
            if let Ok(create_req) = serde_json::from_value::<CreateQueueRequest>(req.params.clone()) {
                create_queue(req.id.clone(), create_req, store).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("invalid params".into()),
                }
            }
        }
        "list" => list_queues(req.id.clone(), store).await,
        "get" => {
            if let Some(name) = req.params.get("name").and_then(|v| v.as_str()) {
                get_queue(req.id.clone(), name.to_string(), store).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("missing queue name".into()),
                }
            }
        }
        "delete" => {
            if let Some(name) = req.params.get("name").and_then(|v| v.as_str()) {
                delete_queue(req.id.clone(), name.to_string(), store).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("missing queue name".into()),
                }
            }
        }
        "purge" => {
            if let Some(name) = req.params.get("name").and_then(|v| v.as_str()) {
                purge_queue(req.id.clone(), name.to_string(), store).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("missing queue name".into()),
                }
            }
        }
        "pause" => {
            if let Some(name) = req.params.get("name").and_then(|v| v.as_str()) {
                pause_queue(req.id.clone(), name.to_string(), store).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("missing queue name".into()),
                }
            }
        }
        "resume" => {
            if let Some(name) = req.params.get("name").and_then(|v| v.as_str()) {
                resume_queue(req.id.clone(), name.to_string(), store).await
            } else {
                ServiceResponse {
                    id: req.id.clone(),
                    status: 400,
                    data: serde_json::Value::Null,
                    error: Some("missing queue name".into()),
                }
            }
        }
        _ => ServiceResponse {
            id: req.id.clone(),
            status: 400,
            data: serde_json::Value::Null,
            error: Some(format!("unknown queues method: {}", req.method)),
        },
    }
}

async fn create_queue(
    req_id: String,
    payload: CreateQueueRequest,
    store: &QueueStore,
) -> ServiceResponse {
    let mut st = store.write().await;

    if st.contains_key(&payload.name) {
        return ServiceResponse {
            id: req_id,
            status: 409,
            data: serde_json::Value::Null,
            error: Some("queue already exists".into()),
        };
    }

    let entry = QueueEntry {
        name: payload.name.clone(),
        mode: payload.mode,
        max_delivery_attempts: payload.max_delivery_attempts,
        default_ttl_secs: payload.default_ttl_secs,
        created_at: chrono::Utc::now().timestamp(),
        depth: 0,
        dead_letter_depth: 0,
        total_enqueued: 0,
        subscriber_count: 0,
        paused: false,
    };

    st.insert(payload.name.clone(), entry);

    tracing::info!("Queue '{}' created", payload.name);

    ServiceResponse {
        id: req_id,
        status: 201,
        data: serde_json::json!({
            "status": "created",
            "name": payload.name,
        }),
        error: None,
    }
}

async fn list_queues(req_id: String, store: &QueueStore) -> ServiceResponse {
    let st = store.read().await;
    let queues: Vec<&QueueEntry> = st.values().collect();

    ServiceResponse {
        id: req_id,
        status: 200,
        data: serde_json::json!({
            "queues": queues,
            "count": queues.len(),
        }),
        error: None,
    }
}

async fn get_queue(req_id: String, name: String, store: &QueueStore) -> ServiceResponse {
    let st = store.read().await;

    match st.get(&name) {
        Some(q) => ServiceResponse {
            id: req_id,
            status: 200,
            data: serde_json::to_value(q).unwrap(),
            error: None,
        },
        None => ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some("queue not found".into()),
        },
    }
}

async fn delete_queue(req_id: String, name: String, store: &QueueStore) -> ServiceResponse {
    let mut st = store.write().await;

    match st.remove(&name) {
        Some(_) => {
            tracing::info!("Queue '{}' deleted", name);
            ServiceResponse {
                id: req_id,
                status: 200,
                data: serde_json::json!({ "status": "deleted", "name": name }),
                error: None,
            }
        }
        None => ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some("queue not found".into()),
        },
    }
}

async fn purge_queue(req_id: String, name: String, store: &QueueStore) -> ServiceResponse {
    let mut st = store.write().await;

    match st.get_mut(&name) {
        Some(q) => {
            let old_depth = q.depth;
            q.depth = 0;
            tracing::info!("Queue '{}' purged ({} messages)", name, old_depth);
            ServiceResponse {
                id: req_id,
                status: 200,
                data: serde_json::json!({
                    "status": "purged",
                    "name": name,
                    "messages_purged": old_depth,
                }),
                error: None,
            }
        }
        None => ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some("queue not found".into()),
        },
    }
}

async fn pause_queue(req_id: String, name: String, store: &QueueStore) -> ServiceResponse {
    let mut st = store.write().await;

    match st.get_mut(&name) {
        Some(q) => {
            q.paused = true;
            ServiceResponse {
                id: req_id,
                status: 200,
                data: serde_json::json!({ "status": "paused", "name": name }),
                error: None,
            }
        }
        None => ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some("queue not found".into()),
        },
    }
}

async fn resume_queue(req_id: String, name: String, store: &QueueStore) -> ServiceResponse {
    let mut st = store.write().await;

    match st.get_mut(&name) {
        Some(q) => {
            q.paused = false;
            ServiceResponse {
                id: req_id,
                status: 200,
                data: serde_json::json!({ "status": "resumed", "name": name }),
                error: None,
            }
        }
        None => ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some("queue not found".into()),
        },
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Tests
// ═══════════════════════════════════════════════════════════════════════════

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

    fn new_store() -> QueueStore {
        Arc::new(RwLock::new(HashMap::new()))
    }

    #[tokio::test]
    async fn create_and_list_queues() {
        let store = new_store();

        store.write().await.insert(
            "events".to_string(),
            QueueEntry {
                name: "events".to_string(),
                mode: "competing".to_string(),
                max_delivery_attempts: 5,
                default_ttl_secs: None,
                created_at: 0,
                depth: 42,
                dead_letter_depth: 3,
                total_enqueued: 100,
                subscriber_count: 2,
                paused: false,
            },
        );

        let queues = store.read().await;
        assert_eq!(queues.len(), 1);
        assert_eq!(queues.get("events").unwrap().depth, 42);
    }

    #[tokio::test]
    async fn delete_removes_queue() {
        let store = new_store();

        store.write().await.insert(
            "temp".to_string(),
            QueueEntry {
                name: "temp".to_string(),
                mode: "fan_out".to_string(),
                max_delivery_attempts: 3,
                default_ttl_secs: Some(60),
                created_at: 0,
                depth: 10,
                dead_letter_depth: 0,
                total_enqueued: 10,
                subscriber_count: 1,
                paused: false,
            },
        );

        store.write().await.remove("temp");
        assert!(store.read().await.get("temp").is_none());
    }

    #[tokio::test]
    async fn pause_and_resume_queue() {
        let store = new_store();

        store.write().await.insert(
            "jobs".to_string(),
            QueueEntry {
                name: "jobs".to_string(),
                mode: "exclusive".to_string(),
                max_delivery_attempts: 5,
                default_ttl_secs: None,
                created_at: 0,
                depth: 0,
                dead_letter_depth: 0,
                total_enqueued: 0,
                subscriber_count: 0,
                paused: false,
            },
        );

        {
            let mut queues = store.write().await;
            queues.get_mut("jobs").unwrap().paused = true;
        }
        assert!(store.read().await.get("jobs").unwrap().paused);

        {
            let mut queues = store.write().await;
            queues.get_mut("jobs").unwrap().paused = false;
        }
        assert!(!store.read().await.get("jobs").unwrap().paused);
    }
}
