//! Q-06: Queue Management API
//!
//! Service-RPC endpoints for managing durable message queues.
//!
//! Methods (service: "queues"):
//! - create     → create a queue (backed by the configured storage backend)
//! - list       → list all queues with live metrics
//! - get        → get queue details
//! - delete     → delete (purge + remove) a queue
//! - purge      → purge all messages
//! - pause      → pause consumer delivery
//! - resume     → resume consumer delivery
//! - enqueue    → publish a message to a queue
//! - dequeue    → consume the next pending message
//! - ack        → acknowledge a message by sequence number

use crate::service::{ServiceRequest, ServiceResponse};
use pulse_core::queue::{DurableQueue, QueueConfig, SubscriptionMode};
use pulse_core::queue_storage::QueueStorageBackend;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

// ═══════════════════════════════════════════════════════════════════════════
// Types
// ═══════════════════════════════════════════════════════════════════════════

/// Manages all live queues — each backed by a real `DurableQueue` with the
/// configured storage backend.
pub struct QueueManager {
    /// backend shared across all queues
    backend: Arc<dyn QueueStorageBackend>,
    /// live DurableQueue instances keyed by name
    queues: RwLock<HashMap<String, Arc<DurableQueue>>>,
    /// metadata for management API responses
    meta: RwLock<HashMap<String, QueueMeta>>,
}

/// Lightweight metadata for management API responses.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueMeta {
    pub name: String,
    pub mode: String,
    pub max_delivery_attempts: u32,
    pub default_ttl_secs: Option<u64>,
    pub created_at: i64,
    pub paused: bool,
}

pub type QueueStore = Arc<QueueManager>;

impl QueueManager {
    pub fn new(backend: Arc<dyn QueueStorageBackend>) -> Self {
        Self {
            backend,
            queues: RwLock::new(HashMap::new()),
            meta: RwLock::new(HashMap::new()),
        }
    }
}

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

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

#[derive(Debug, Deserialize)]
struct EnqueueRequest {
    queue: String,
    payload: String,
    idempotency_key: Option<String>,
    ttl_secs: Option<u64>,
}

#[derive(Debug, Deserialize)]
struct DequeueRequest {
    queue: String,
    #[serde(default = "default_subscriber")]
    subscriber_id: String,
}

#[derive(Debug, Deserialize)]
struct AckRequest {
    queue: String,
    sequence: u64,
    #[serde(default = "default_subscriber")]
    subscriber_id: String,
}

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

fn parse_mode(s: &str) -> SubscriptionMode {
    match s {
        "fan_out" | "fanout" => SubscriptionMode::FanOut,
        "exclusive" => SubscriptionMode::Exclusive,
        _ => SubscriptionMode::Competing,
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Router
// ═══════════════════════════════════════════════════════════════════════════

pub async fn handle_queues_router(req: &ServiceRequest, store: &QueueStore) -> ServiceResponse {
    match req.method.as_str() {
        "create" => {
            if let Ok(r) = serde_json::from_value::<CreateQueueRequest>(req.params.clone()) {
                create_queue(req.id.clone(), r, store).await
            } else {
                bad_request(&req.id, "invalid params for create")
            }
        }
        "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, store).await
            } else {
                bad_request(&req.id, "missing queue name")
            }
        }
        "delete" => {
            if let Some(name) = req.params.get("name").and_then(|v| v.as_str()) {
                delete_queue(req.id.clone(), name, store).await
            } else {
                bad_request(&req.id, "missing queue name")
            }
        }
        "purge" => {
            if let Some(name) = req.params.get("name").and_then(|v| v.as_str()) {
                purge_queue(req.id.clone(), name, store).await
            } else {
                bad_request(&req.id, "missing queue name")
            }
        }
        "pause" => {
            if let Some(name) = req.params.get("name").and_then(|v| v.as_str()) {
                pause_queue(req.id.clone(), name, store).await
            } else {
                bad_request(&req.id, "missing queue name")
            }
        }
        "resume" => {
            if let Some(name) = req.params.get("name").and_then(|v| v.as_str()) {
                resume_queue(req.id.clone(), name, store).await
            } else {
                bad_request(&req.id, "missing queue name")
            }
        }
        // ── Message operations ──────────────────────────────────────────────
        "enqueue" => {
            if let Ok(r) = serde_json::from_value::<EnqueueRequest>(req.params.clone()) {
                enqueue_message(req.id.clone(), r, store).await
            } else {
                bad_request(&req.id, "invalid params for enqueue")
            }
        }
        "dequeue" => {
            if let Ok(r) = serde_json::from_value::<DequeueRequest>(req.params.clone()) {
                dequeue_message(req.id.clone(), r, store).await
            } else {
                bad_request(&req.id, "invalid params for dequeue")
            }
        }
        "ack" => {
            if let Ok(r) = serde_json::from_value::<AckRequest>(req.params.clone()) {
                ack_message(req.id.clone(), r, store).await
            } else {
                bad_request(&req.id, "invalid params for ack")
            }
        }
        _ => ServiceResponse {
            id: req.id.clone(),
            status: 400,
            data: serde_json::Value::Null,
            error: Some(format!("unknown queues method: {}", req.method)),
        },
    }
}

fn bad_request(id: &str, msg: &str) -> ServiceResponse {
    ServiceResponse {
        id: id.to_string(),
        status: 400,
        data: serde_json::Value::Null,
        error: Some(msg.to_string()),
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Queue CRUD handlers
// ═══════════════════════════════════════════════════════════════════════════

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

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

    let config = QueueConfig {
        max_delivery_attempts: payload.max_delivery_attempts,
        default_ttl_secs: payload.default_ttl_secs,
        mode: parse_mode(&payload.mode),
        ..QueueConfig::default()
    };

    let dq = DurableQueue::with_backend(
        &payload.name,
        config,
        Arc::clone(&store.backend),
    );

    // Try to recover any previously persisted messages
    match dq.recover() {
        Ok(n) if n > 0 => {
            tracing::info!("Queue '{}' created — recovered {} pending messages", payload.name, n);
        }
        _ => {
            tracing::info!("Queue '{}' created", payload.name);
        }
    }

    let meta = QueueMeta {
        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(),
        paused: false,
    };

    queues.insert(payload.name.clone(), Arc::new(dq));
    meta_map.insert(payload.name.clone(), meta);

    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 queues = store.queues.read().await;
    let meta_map = store.meta.read().await;

    let mut entries = Vec::new();
    for (name, dq) in queues.iter() {
        let meta = meta_map.get(name);
        let depth = dq.depth();
        let dlq_depth = dq.dead_letter_depth();

        entries.push(serde_json::json!({
            "name": name,
            "mode": meta.map(|m| m.mode.as_str()).unwrap_or("competing"),
            "depth": depth,
            "dead_letter_depth": dlq_depth,
            "paused": meta.map(|m| m.paused).unwrap_or(false),
            "created_at": meta.map(|m| m.created_at).unwrap_or(0),
        }));
    }

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

async fn get_queue(req_id: String, name: &str, store: &QueueStore) -> ServiceResponse {
    // Auto-create-on-first-reference: if the queue doesn't exist yet, create it
    // with default settings and attempt recovery from the backend.
    let dq = get_or_create_queue(name, store).await;
    let meta_map = store.meta.read().await;
    let meta = meta_map.get(name);

    ServiceResponse {
        id: req_id,
        status: 200,
        data: serde_json::json!({
            "name": name,
            "mode": meta.map(|m| m.mode.as_str()).unwrap_or("competing"),
            "depth": dq.depth(),
            "dead_letter_depth": dq.dead_letter_depth(),
            "paused": meta.map(|m| m.paused).unwrap_or(false),
            "created_at": meta.map(|m| m.created_at).unwrap_or(0),
        }),
        error: None,
    }
}

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

    match queues.remove(name) {
        Some(_) => {
            meta_map.remove(name);
            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: &str, store: &QueueStore) -> ServiceResponse {
    let queues = store.queues.read().await;

    match queues.get(name) {
        Some(dq) => {
            let purged = dq.purge();
            tracing::info!("Queue '{}' purged ({} messages)", name, purged);
            ServiceResponse {
                id: req_id,
                status: 200,
                data: serde_json::json!({
                    "status": "purged",
                    "name": name,
                    "messages_purged": purged,
                }),
                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: &str, store: &QueueStore) -> ServiceResponse {
    let mut meta_map = store.meta.write().await;

    match meta_map.get_mut(name) {
        Some(m) => {
            m.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: &str, store: &QueueStore) -> ServiceResponse {
    let mut meta_map = store.meta.write().await;

    match meta_map.get_mut(name) {
        Some(m) => {
            m.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()),
        },
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Message operations — enqueue / dequeue / ack
// ═══════════════════════════════════════════════════════════════════════════

/// Auto-create a queue if it doesn't exist. Returns the DurableQueue Arc.
/// This is the central "get-or-create" helper used by enqueue, dequeue, ack, and info.
async fn get_or_create_queue(name: &str, store: &QueueStore) -> Arc<DurableQueue> {
    // Fast path: read lock
    {
        let queues = store.queues.read().await;
        if let Some(dq) = queues.get(name) {
            return Arc::clone(dq);
        }
    }
    // Slow path: write lock + create
    let mut queues = store.queues.write().await;
    let mut meta_map = store.meta.write().await;
    // Double-check after acquiring write lock
    if let Some(dq) = queues.get(name) {
        return Arc::clone(dq);
    }
    let dq = Arc::new(DurableQueue::with_backend(
        name,
        QueueConfig::default(),
        Arc::clone(&store.backend),
    ));
    // Try recovery for any pre-existing data in the backend
    match dq.recover() {
        Ok(n) if n > 0 => {
            tracing::info!("Queue '{}' auto-created — recovered {} pending messages", name, n);
        }
        _ => {
            tracing::info!("Queue '{}' auto-created on first reference", name);
        }
    }
    queues.insert(name.to_string(), Arc::clone(&dq));
    meta_map.insert(name.to_string(), QueueMeta {
        name: name.to_string(),
        mode: "competing".to_string(),
        max_delivery_attempts: 5,
        default_ttl_secs: None,
        created_at: chrono::Utc::now().timestamp(),
        paused: false,
    });
    dq
}

async fn enqueue_message(
    req_id: String,
    r: EnqueueRequest,
    store: &QueueStore,
) -> ServiceResponse {
    let dq = get_or_create_queue(&r.queue, store).await;

    let payload = r.payload.into_bytes();
    match dq.enqueue(payload, r.idempotency_key.clone(), r.ttl_secs) {
        Some(seq) => {
            tracing::debug!("Enqueued message seq={} to queue '{}'", seq, r.queue);
            ServiceResponse {
                id: req_id,
                status: 200,
                data: serde_json::json!({
                    "status": "enqueued",
                    "queue": r.queue,
                    "sequence": seq,
                }),
                error: None,
            }
        }
        None => {
            // Distinguish dedup from error: dedup only when an idempotency key was supplied
            if r.idempotency_key.is_some() {
                ServiceResponse {
                    id: req_id,
                    status: 200,
                    data: serde_json::json!({
                        "status": "deduplicated",
                        "queue": r.queue,
                    }),
                    error: None,
                }
            } else {
                tracing::error!("Enqueue failed for queue '{}' — backend error (no dedup key)", r.queue);
                ServiceResponse {
                    id: req_id,
                    status: 500,
                    data: serde_json::Value::Null,
                    error: Some(format!("enqueue failed for queue '{}' — backend error", r.queue)),
                }
            }
        }
    }
}

async fn dequeue_message(
    req_id: String,
    r: DequeueRequest,
    store: &QueueStore,
) -> ServiceResponse {
    // Auto-create queue on first reference (no more 404 for dequeue)
    let dq = get_or_create_queue(&r.queue, store).await;

    // Ensure subscriber is registered
    dq.subscribe(&r.subscriber_id);

    match dq.dequeue(&r.subscriber_id) {
        Some(msg) => ServiceResponse {
            id: req_id,
            status: 200,
            data: serde_json::json!({
                "status": "ok",
                "queue": r.queue,
                "message": {
                    "sequence": msg.sequence,
                    "payload": String::from_utf8_lossy(&msg.payload),
                    "delivery_count": msg.delivery_count,
                    "enqueued_at": msg.enqueued_at,
                },
            }),
            error: None,
        },
        None => ServiceResponse {
            id: req_id,
            status: 200,
            data: serde_json::json!({
                "status": "empty",
                "queue": r.queue,
            }),
            error: None,
        },
    }
}

async fn ack_message(
    req_id: String,
    r: AckRequest,
    store: &QueueStore,
) -> ServiceResponse {
    // Auto-create queue on first reference (no more 404 for ack)
    let dq = get_or_create_queue(&r.queue, store).await;

    if dq.ack(r.sequence, &r.subscriber_id) {
        ServiceResponse {
            id: req_id,
            status: 200,
            data: serde_json::json!({
                "status": "acked",
                "queue": r.queue,
                "sequence": r.sequence,
            }),
            error: None,
        }
    } else {
        ServiceResponse {
            id: req_id,
            status: 404,
            data: serde_json::Value::Null,
            error: Some(format!("message {} not found in queue '{}'", r.sequence, r.queue)),
        }
    }
}

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

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

    fn new_store() -> QueueStore {
        Arc::new(QueueManager::new(Arc::new(InMemoryBackend::new())))
    }

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

        let req = ServiceRequest {
            id: "r1".into(),
            service: "queues".into(),
            method: "create".into(),
            params: serde_json::json!({ "name": "events" }),
        };
        let resp = handle_queues_router(&req, &store).await;
        assert_eq!(resp.status, 201);

        let req = ServiceRequest {
            id: "r2".into(),
            service: "queues".into(),
            method: "list".into(),
            params: serde_json::json!({}),
        };
        let resp = handle_queues_router(&req, &store).await;
        assert_eq!(resp.status, 200);
        assert_eq!(resp.data["count"], 1);
    }

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

        // Enqueue (auto-creates queue)
        let req = ServiceRequest {
            id: "r1".into(),
            service: "queues".into(),
            method: "enqueue".into(),
            params: serde_json::json!({
                "queue": "tasks",
                "payload": "hello world"
            }),
        };
        let resp = handle_queues_router(&req, &store).await;
        assert_eq!(resp.status, 200);
        assert_eq!(resp.data["status"], "enqueued");
        let seq = resp.data["sequence"].as_u64().unwrap();
        assert_eq!(seq, 1);

        // Dequeue
        let req = ServiceRequest {
            id: "r2".into(),
            service: "queues".into(),
            method: "dequeue".into(),
            params: serde_json::json!({
                "queue": "tasks",
                "subscriber_id": "worker-1"
            }),
        };
        let resp = handle_queues_router(&req, &store).await;
        assert_eq!(resp.status, 200);
        assert_eq!(resp.data["status"], "ok");
        assert_eq!(resp.data["message"]["payload"], "hello world");
        assert_eq!(resp.data["message"]["sequence"], 1);

        // Ack
        let req = ServiceRequest {
            id: "r3".into(),
            service: "queues".into(),
            method: "ack".into(),
            params: serde_json::json!({
                "queue": "tasks",
                "sequence": 1,
                "subscriber_id": "worker-1"
            }),
        };
        let resp = handle_queues_router(&req, &store).await;
        assert_eq!(resp.status, 200);
        assert_eq!(resp.data["status"], "acked");

        // Dequeue again should be empty
        let req = ServiceRequest {
            id: "r4".into(),
            service: "queues".into(),
            method: "dequeue".into(),
            params: serde_json::json!({
                "queue": "tasks",
                "subscriber_id": "worker-1"
            }),
        };
        let resp = handle_queues_router(&req, &store).await;
        assert_eq!(resp.data["status"], "empty");
    }

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

        // Create
        let req = ServiceRequest {
            id: "r1".into(),
            service: "queues".into(),
            method: "create".into(),
            params: serde_json::json!({ "name": "temp" }),
        };
        handle_queues_router(&req, &store).await;

        // Enqueue a message so we can verify it's gone after delete
        let req = ServiceRequest {
            id: "r1b".into(),
            service: "queues".into(),
            method: "enqueue".into(),
            params: serde_json::json!({ "queue": "temp", "payload": "will be deleted" }),
        };
        let resp = handle_queues_router(&req, &store).await;
        assert_eq!(resp.data["status"], "enqueued");

        // Delete
        let req = ServiceRequest {
            id: "r2".into(),
            service: "queues".into(),
            method: "delete".into(),
            params: serde_json::json!({ "name": "temp" }),
        };
        let resp = handle_queues_router(&req, &store).await;
        assert_eq!(resp.status, 200);

        // List should not contain the deleted queue
        let req = ServiceRequest {
            id: "r3".into(),
            service: "queues".into(),
            method: "list".into(),
            params: serde_json::json!({}),
        };
        let resp = handle_queues_router(&req, &store).await;
        assert_eq!(resp.data["count"], 0);
    }

    #[tokio::test]
    async fn persistence_survives_across_queue_instances() {
        // Simulate persistence: use the same backend, create queue, enqueue,
        // then "restart" by creating a new queue with the same backend.
        let backend: Arc<dyn QueueStorageBackend> = Arc::new(InMemoryBackend::new());

        // First "session"
        {
            let store = Arc::new(QueueManager::new(Arc::clone(&backend)));
            let req = ServiceRequest {
                id: "r1".into(),
                service: "queues".into(),
                method: "enqueue".into(),
                params: serde_json::json!({
                    "queue": "durable",
                    "payload": "persistent msg"
                }),
            };
            let resp = handle_queues_router(&req, &store).await;
            assert_eq!(resp.data["status"], "enqueued");
        }

        // Second "session" (simulates restart)
        {
            let store = Arc::new(QueueManager::new(Arc::clone(&backend)));
            // Create queue — should recover the message
            let req = ServiceRequest {
                id: "r2".into(),
                service: "queues".into(),
                method: "create".into(),
                params: serde_json::json!({ "name": "durable" }),
            };
            let resp = handle_queues_router(&req, &store).await;
            assert_eq!(resp.status, 201);

            // Dequeue should find the recovered message
            let req = ServiceRequest {
                id: "r3".into(),
                service: "queues".into(),
                method: "dequeue".into(),
                params: serde_json::json!({
                    "queue": "durable",
                    "subscriber_id": "w1"
                }),
            };
            let resp = handle_queues_router(&req, &store).await;
            assert_eq!(resp.data["status"], "ok");
            assert_eq!(resp.data["message"]["payload"], "persistent msg");
        }
    }
}
