//! Dead Letter Queue (DLQ) operations for job queue.
//!
//! The DLQ stores jobs that have failed permanently (exhausted all retry attempts).
//! This allows manual inspection and recovery of failed jobs.
//!
//! # Atomicity
//!
//! Multi-step Redis operations use Lua scripts executed via `redis::Script`,
//! which provides EVALSHA + NOSCRIPT fallback automatically:
//!
//! - **`remove`**: `LREM` + `DEL` + `HDEL` (status hash) in a single script — no orphaned list
//!   entries and the status hash stays in sync.
//! - **`clear`**: `LRANGE` + N×(`DEL` + `HDEL`) + `DEL list` in a single script — consistent
//!   snapshot; jobs added concurrently are either fully included or fully excluded, and the status
//!   hash is kept in sync throughout.

use redis::aio::ConnectionManager;
use uuid::Uuid;

use super::Job;
use crate::error::Result;

// ── Lua scripts ───────────────────────────────────────────────────────────────

/// Atomically remove one job from the DLQ list, delete its data key, and remove
/// its entry from the status hash.
///
/// - `KEYS[1]` — `queue:dlq` (the DLQ list)
/// - `KEYS[2]` — `job:{uuid}` (the job data key)
/// - `KEYS[3]` — `queue:status` (the status hash)
/// - `ARGV[1]` — `dlq:{uuid}` (the list member value to remove)
/// - `ARGV[2]` — job UUID string (status hash field)
const DLQ_REMOVE_SCRIPT: &str = r"
redis.call('LREM', KEYS[1], 0, ARGV[1])
redis.call('DEL', KEYS[2])
redis.call('HDEL', KEYS[3], ARGV[2])
return 1
";

/// Atomically clear the entire DLQ: delete every job data key, remove every entry
/// from the status hash, then remove the list itself.
///
/// - `KEYS[1]` — `queue:dlq` (the DLQ list)
/// - `KEYS[2]` — `queue:status` (the status hash)
///
/// Returns the number of jobs that were removed.
const DLQ_CLEAR_SCRIPT: &str = r"
local members = redis.call('LRANGE', KEYS[1], 0, -1)
for _, member in ipairs(members) do
    local job_id = string.sub(member, 5)
    redis.call('DEL', 'job:' .. job_id)
    redis.call('HDEL', KEYS[2], job_id)
end
redis.call('DEL', KEYS[1])
return #members
";

/// Dead Letter Queue operations
#[derive(Clone)]
pub struct DeadLetterQueueManager {
    conn: ConnectionManager,
}

impl DeadLetterQueueManager {
    /// Create a new DLQ manager
    #[must_use]
    pub const fn new(conn: ConnectionManager) -> Self {
        Self { conn }
    }

    /// DLQ key in Redis
    pub(crate) const fn dlq_key() -> &'static str {
        "queue:dlq"
    }

    /// Status hash key — shared with `RedisJobQueue`
    const fn status_key() -> &'static str {
        "queue:status"
    }

    /// Job storage key
    pub(crate) fn job_key(job_id: Uuid) -> String {
        format!("job:{job_id}")
    }

    /// DLQ member key
    pub(crate) fn dlq_member(job_id: Uuid) -> String {
        format!("dlq:{job_id}")
    }

    /// Get all jobs in the DLQ
    ///
    /// # Returns
    ///
    /// Vector of jobs currently in the DLQ
    ///
    /// # Errors
    ///
    /// Returns error if Redis operation fails
    pub async fn list_all(&self) -> Result<Vec<Job>> {
        let members: Vec<String> = redis::cmd("LRANGE")
            .arg(Self::dlq_key())
            .arg(0)
            .arg(-1)
            .query_async(&mut self.conn.clone())
            .await?;

        let mut jobs = Vec::new();

        for member in members {
            // Extract job ID from member (format: "dlq:{job_id}")
            if let Some(job_id_str) = member.strip_prefix("dlq:") {
                if let Ok(job_id) = Uuid::parse_str(job_id_str) {
                    if let Ok(Some(job_json)) = self.get_job(job_id).await {
                        if let Ok(job) = serde_json::from_str::<Job>(&job_json) {
                            jobs.push(job);
                        }
                    }
                }
            }
        }

        Ok(jobs)
    }

    /// Get a specific job from the DLQ by ID
    ///
    /// # Arguments
    ///
    /// * `job_id` - ID of the job to retrieve
    ///
    /// # Errors
    ///
    /// Returns error if Redis operation fails
    pub async fn get(&self, job_id: Uuid) -> Result<Option<Job>> {
        let job_json = self.get_job(job_id).await?;

        match job_json {
            Some(json) => {
                let job = serde_json::from_str::<Job>(&json)
                    .map_err(|e| crate::error::ObserverError::SerializationError(e.to_string()))?;
                Ok(Some(job))
            },
            None => Ok(None),
        }
    }

    /// Get count of jobs in the DLQ
    ///
    /// # Errors
    ///
    /// Returns error if Redis operation fails
    pub async fn count(&self) -> Result<usize> {
        let count: usize = redis::cmd("LLEN")
            .arg(Self::dlq_key())
            .query_async(&mut self.conn.clone())
            .await?;

        Ok(count)
    }

    /// Get the size of the DLQ (same as count, for compatibility)
    ///
    /// # Errors
    ///
    /// Returns error if Redis operation fails
    pub async fn size(&self) -> Result<usize> {
        self.count().await
    }

    /// Remove a job from the DLQ atomically.
    ///
    /// Uses a Lua script (`DLQ_REMOVE_SCRIPT`) to `LREM` the list entry and
    /// `DEL` the job data key in a single round-trip, eliminating the window
    /// where the list entry could exist without its backing data.
    ///
    /// # Arguments
    ///
    /// * `job_id` - ID of the job to remove
    ///
    /// # Errors
    ///
    /// Returns error if Redis operation fails.
    pub async fn remove(&self, job_id: Uuid) -> Result<()> {
        redis::Script::new(DLQ_REMOVE_SCRIPT)
            .key(Self::dlq_key()) // KEYS[1]
            .key(Self::job_key(job_id)) // KEYS[2]
            .key(Self::status_key()) // KEYS[3]
            .arg(Self::dlq_member(job_id)) // ARGV[1]
            .arg(job_id.to_string()) // ARGV[2]
            .invoke_async::<i64>(&mut self.conn.clone())
            .await?;
        Ok(())
    }

    /// Clear all jobs from the DLQ atomically.
    ///
    /// Uses a Lua script (`DLQ_CLEAR_SCRIPT`) to read the list and delete all
    /// backing job keys in one atomic operation, preventing partial clears when
    /// jobs are concurrently enqueued to the DLQ.
    ///
    /// # Errors
    ///
    /// Returns error if Redis operation fails.
    pub async fn clear(&self) -> Result<()> {
        redis::Script::new(DLQ_CLEAR_SCRIPT)
            .key(Self::dlq_key()) // KEYS[1]
            .key(Self::status_key()) // KEYS[2]
            .invoke_async::<i64>(&mut self.conn.clone())
            .await?;
        Ok(())
    }

    /// Get DLQ statistics
    ///
    /// # Returns
    ///
    /// Statistics about the current DLQ state
    ///
    /// # Errors
    ///
    /// Returns error if Redis operation fails
    pub async fn stats(&self) -> Result<DlqStats> {
        let count = self.count().await?;
        let all_jobs = self.list_all().await?;

        let mut by_action_type: std::collections::HashMap<String, usize> =
            std::collections::HashMap::new();

        for job in &all_jobs {
            *by_action_type.entry(job.action_type().to_string()).or_insert(0) += 1;
        }

        Ok(DlqStats {
            total_jobs: count,
            by_action_type,
        })
    }

    /// Internal helper to get job JSON
    async fn get_job(&self, job_id: Uuid) -> Result<Option<String>> {
        let json: Option<String> = redis::cmd("GET")
            .arg(Self::job_key(job_id))
            .query_async(&mut self.conn.clone())
            .await?;

        Ok(json)
    }
}

/// Statistics about the Dead Letter Queue
#[derive(Debug, Clone)]
pub struct DlqStats {
    /// Total number of jobs in the DLQ
    pub total_jobs:     usize,
    /// Count of jobs by action type
    pub by_action_type: std::collections::HashMap<String, usize>,
}

impl std::fmt::Display for DlqStats {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "DLQ Stats: {} total jobs", self.total_jobs)?;

        if !self.by_action_type.is_empty() {
            write!(f, " (")?;
            let parts: Vec<String> = self
                .by_action_type
                .iter()
                .map(|(action_type, count)| format!("{action_type}: {count}"))
                .collect();
            write!(f, "{})", parts.join(", "))?;
        }

        Ok(())
    }
}
