//! Redis storage backend for durable message queues.
//!
//! Persists queue messages to Redis using a combination of Redis data structures:
//! - **Sorted Set** (`pulse:q:{name}:msgs`) — messages keyed by sequence, scored by sequence
//! - **Hash** (`pulse:q:{name}:meta:{seq}`) — per-message metadata (state, delivery_count, etc.)
//! - **Set** (`pulse:q:{name}:dedup`) — idempotency key dedup
//! - **Sorted Set** (`pulse:q:{name}:dlq`) — dead-letter messages
//!
//! This approach gives us O(log N) enqueue/dequeue with range queries for
//! pending messages, atomic state transitions, and TTL-based eviction.
//!
//! Enable with: `cargo build --features queue-redis`
//!
//! # When to use
//!
//! - When multiple relay instances need to share queue state
//! - When you already have Redis in your infrastructure
//! - For ephemeral/cache-like queues where Redis's optional persistence (AOF/RDB) is acceptable

use pulse_core::queue::{MessageState, QueueMessage};
use pulse_core::queue_storage::{QueueStorageBackend, QueueStorageError};
use redis::{Client, Commands, Connection};
use std::sync::Mutex;

// ═══════════════════════════════════════════════════════════════════════════
// Key helpers
// ═══════════════════════════════════════════════════════════════════════════

fn msgs_key(queue: &str) -> String {
    format!("pulse:q:{queue}:msgs")
}

fn meta_key(queue: &str, seq: u64) -> String {
    format!("pulse:q:{queue}:meta:{seq}")
}

fn dedup_key(queue: &str) -> String {
    format!("pulse:q:{queue}:dedup")
}

fn dlq_key(queue: &str) -> String {
    format!("pulse:q:{queue}:dlq")
}

fn dlq_meta_key(queue: &str, seq: u64) -> String {
    format!("pulse:q:{queue}:dlq:meta:{seq}")
}

// ═══════════════════════════════════════════════════════════════════════════
// State mapping
// ═══════════════════════════════════════════════════════════════════════════

fn state_to_str(state: MessageState) -> &'static str {
    match state {
        MessageState::Pending => "pending",
        MessageState::InFlight => "inflight",
        MessageState::Acked => "acked",
        MessageState::DeadLettered => "dead",
    }
}

fn str_to_state(s: &str) -> MessageState {
    match s {
        "pending" => MessageState::Pending,
        "inflight" => MessageState::InFlight,
        "acked" => MessageState::Acked,
        "dead" => MessageState::DeadLettered,
        _ => MessageState::Pending,
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Redis Backend
// ═══════════════════════════════════════════════════════════════════════════

/// Redis-backed storage for durable message queues.
///
/// Uses synchronous Redis commands behind a `Mutex<Connection>`.
/// Stores the `Client` so the connection can be re-established if it drops.
/// For production deployments with high concurrency, consider using
/// a connection pool (e.g., `r2d2` or `bb8`).
pub struct RedisQueueBackend {
    client: Client,
    conn: Mutex<Connection>,
}

impl RedisQueueBackend {
    /// Connect to Redis and create a new backend.
    pub fn connect(url: &str) -> Result<Self, QueueStorageError> {
        let client = Client::open(url)
            .map_err(|e| QueueStorageError::Io(format!("Redis client error: {e}")))?;
        let conn = client
            .get_connection()
            .map_err(|e| QueueStorageError::Io(format!("Redis connection error: {e}")))?;
        Ok(Self {
            client,
            conn: Mutex::new(conn),
        })
    }

    /// Create from an existing Redis connection.
    pub fn from_connection(conn: Connection) -> Self {
        // When created from an existing connection, we don't have a client for reconnection.
        // Create a dummy client pointing to localhost (won't be used for reconnect).
        let client = Client::open("redis://localhost:6379").unwrap();
        Self {
            client,
            conn: Mutex::new(conn),
        }
    }

    /// Try to reconnect if the existing connection is broken.
    fn ensure_connected(&self) -> Result<std::sync::MutexGuard<'_, Connection>, QueueStorageError> {
        let mut conn = self.conn.lock().unwrap();
        // Test the connection with a PING
        let ok: Result<String, _> = redis::cmd("PING").query(&mut *conn);
        if ok.is_err() {
            eprintln!("[RedisQueue] Connection lost, attempting reconnect...");
            match self.client.get_connection() {
                Ok(new_conn) => {
                    *conn = new_conn;
                    eprintln!("[RedisQueue] Reconnected successfully");
                }
                Err(e) => {
                    return Err(QueueStorageError::Io(format!(
                        "Redis reconnect failed: {e}"
                    )));
                }
            }
        }
        Ok(conn)
    }
}

impl QueueStorageBackend for RedisQueueBackend {
    fn append(&self, queue: &str, msg: &QueueMessage) -> Result<(), QueueStorageError> {
        let mut conn = self.ensure_connected()?;

        // Store payload in the sorted set (score = sequence for ordering)
        let payload_b64 = base64_encode(&msg.payload);
        let _: () = conn
            .zadd(msgs_key(queue), &payload_b64, msg.sequence as f64)
            .map_err(|e| QueueStorageError::Io(format!("zadd failed: {e}")))?;

        // Store metadata in a hash
        let mk = meta_key(queue, msg.sequence);
        let _: () = redis::pipe()
            .hset(&mk, "state", state_to_str(msg.state))
            .hset(&mk, "dc", msg.delivery_count)
            .hset(&mk, "ts", msg.enqueued_at)
            .hset(&mk, "ttl", msg.ttl_secs.unwrap_or(0))
            .hset(
                &mk,
                "key",
                msg.idempotency_key.as_deref().unwrap_or(""),
            )
            .query(&mut *conn)
            .map_err(|e| QueueStorageError::Io(format!("hset meta failed: {e}")))?;

        Ok(())
    }

    fn load_pending(&self, queue: &str) -> Result<Vec<QueueMessage>, QueueStorageError> {
        let mut conn = self.ensure_connected()?;

        // Get all members from the sorted set (payload + score=sequence)
        let members: Vec<(String, f64)> = conn
            .zrangebyscore_withscores(msgs_key(queue), "-inf", "+inf")
            .map_err(|e| QueueStorageError::Io(format!("zrangebyscore failed: {e}")))?;

        let mut result = Vec::new();
        for (payload_b64, score) in members {
            let seq = score as u64;
            let mk = meta_key(queue, seq);

            // Read metadata
            let meta: Vec<String> = redis::cmd("HMGET")
                .arg(&mk)
                .arg("state")
                .arg("dc")
                .arg("ts")
                .arg("ttl")
                .arg("key")
                .query(&mut *conn)
                .unwrap_or_default();

            if meta.len() < 5 {
                continue;
            }

            let state = str_to_state(&meta[0]);
            if state != MessageState::Pending && state != MessageState::InFlight {
                continue;
            }

            let delivery_count: u32 = meta[1].parse().unwrap_or(0);
            let enqueued_at: u64 = meta[2].parse().unwrap_or(0);
            let ttl_raw: u64 = meta[3].parse().unwrap_or(0);
            let ttl_secs = if ttl_raw == 0 { None } else { Some(ttl_raw) };
            let idempotency_key = if meta[4].is_empty() {
                None
            } else {
                Some(meta[4].clone())
            };

            let payload = base64_decode(&payload_b64);

            result.push(QueueMessage {
                sequence: seq,
                payload,
                idempotency_key,
                ttl_secs,
                enqueued_at,
                delivery_count,
                state,
            });
        }

        Ok(result)
    }

    fn update_state(
        &self,
        queue: &str,
        sequence: u64,
        new_state: MessageState,
        delivery_count: u32,
    ) -> Result<(), QueueStorageError> {
        let mut conn = self.ensure_connected()?;
        let mk = meta_key(queue, sequence);

        let exists: bool = conn
            .exists(&mk)
            .map_err(|e| QueueStorageError::Io(format!("exists check failed: {e}")))?;

        if !exists {
            return Err(QueueStorageError::NotFound);
        }

        let _: () = redis::pipe()
            .hset(&mk, "state", state_to_str(new_state))
            .hset(&mk, "dc", delivery_count)
            .query(&mut *conn)
            .map_err(|e| QueueStorageError::Io(format!("update_state failed: {e}")))?;

        // If acked or dead-lettered, remove from the main sorted set
        if new_state == MessageState::Acked || new_state == MessageState::DeadLettered {
            let _: () = conn
                .zrembyscore(msgs_key(queue), sequence as f64, sequence as f64)
                .map_err(|e| QueueStorageError::Io(format!("zrem failed: {e}")))?;
            // Clean up metadata after a short delay (keep for compact)
        }

        Ok(())
    }

    fn dead_letter(&self, queue: &str, msg: &QueueMessage) -> Result<(), QueueStorageError> {
        let mut conn = self.ensure_connected()?;

        let payload_b64 = base64_encode(&msg.payload);
        let _: () = conn
            .zadd(dlq_key(queue), &payload_b64, msg.sequence as f64)
            .map_err(|e| QueueStorageError::Io(format!("dlq zadd failed: {e}")))?;

        let mk = dlq_meta_key(queue, msg.sequence);
        let _: () = redis::pipe()
            .hset(&mk, "dc", msg.delivery_count)
            .hset(&mk, "ts", msg.enqueued_at)
            .query(&mut *conn)
            .map_err(|e| QueueStorageError::Io(format!("dlq meta failed: {e}")))?;

        Ok(())
    }

    fn compact(&self, queue: &str, before_seq: u64) -> Result<u64, QueueStorageError> {
        let mut conn = self.ensure_connected()?;

        // Remove metadata hashes for sequences below threshold
        // that are in terminal states
        let mut removed: u64 = 0;
        for seq in 1..before_seq {
            let mk = meta_key(queue, seq);
            let state_str: Option<String> = conn
                .hget(&mk, "state")
                .unwrap_or(None);

            if let Some(s) = state_str {
                let state = str_to_state(&s);
                if state == MessageState::Acked || state == MessageState::DeadLettered {
                    let _: () = conn.del(&mk).unwrap_or(());
                    removed += 1;
                }
            }
        }

        Ok(removed)
    }

    fn has_idempotency_key(&self, queue: &str, key: &str) -> Result<bool, QueueStorageError> {
        let mut conn = self.ensure_connected()?;
        let exists: bool = conn
            .sismember(dedup_key(queue), key)
            .map_err(|e| QueueStorageError::Io(format!("sismember failed: {e}")))?;
        Ok(exists)
    }

    fn record_idempotency_key(&self, queue: &str, key: &str) -> Result<(), QueueStorageError> {
        let mut conn = self.ensure_connected()?;
        let _: () = conn
            .sadd(dedup_key(queue), key)
            .map_err(|e| QueueStorageError::Io(format!("sadd failed: {e}")))?;
        Ok(())
    }

    fn evict_idempotency_keys(&self, queue: &str, keep: usize) -> Result<(), QueueStorageError> {
        let mut conn = self.ensure_connected()?;
        let dk = dedup_key(queue);

        let size: usize = conn
            .scard(&dk)
            .map_err(|e| QueueStorageError::Io(format!("scard failed: {e}")))?;

        if size <= keep {
            return Ok(());
        }

        // Redis SETs are unordered, so we pop random members to trim
        let to_remove = size - keep;
        for _ in 0..to_remove {
            let _: Option<String> = conn.spop(&dk).unwrap_or(None);
        }

        Ok(())
    }

    fn purge(&self, queue: &str) -> Result<usize, QueueStorageError> {
        let mut conn = self.ensure_connected()?;
        let mk = msgs_key(queue);

        let count: usize = conn
            .zcard(&mk)
            .map_err(|e| QueueStorageError::Io(format!("zcard failed: {e}")))?;

        // Delete the sorted set and all metadata
        let _: () = conn.del(&mk).unwrap_or(());
        let _: () = conn.del(dedup_key(queue)).unwrap_or(());

        // Clean up metadata keys (best effort — may miss some)
        // In production, use SCAN to find and delete all meta keys
        Ok(count)
    }

    fn load_dead_letters(&self, queue: &str) -> Result<Vec<QueueMessage>, QueueStorageError> {
        let mut conn = self.ensure_connected()?;

        let members: Vec<(String, f64)> = conn
            .zrangebyscore_withscores(dlq_key(queue), "-inf", "+inf")
            .map_err(|e| QueueStorageError::Io(format!("dlq zrange failed: {e}")))?;

        let mut result = Vec::new();
        for (payload_b64, score) in members {
            let seq = score as u64;
            let mk = dlq_meta_key(queue, seq);

            let dc: u32 = conn.hget(&mk, "dc").unwrap_or(0);
            let ts: u64 = conn.hget(&mk, "ts").unwrap_or(0);

            result.push(QueueMessage {
                sequence: seq,
                payload: base64_decode(&payload_b64),
                idempotency_key: None,
                ttl_secs: None,
                enqueued_at: ts,
                delivery_count: dc,
                state: MessageState::DeadLettered,
            });
        }

        Ok(result)
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Base64 helpers (payloads are binary, Redis stores strings)
// ═══════════════════════════════════════════════════════════════════════════

fn base64_encode(data: &[u8]) -> String {
    use base64::Engine;
    base64::engine::general_purpose::STANDARD.encode(data)
}

fn base64_decode(s: &str) -> Vec<u8> {
    use base64::Engine;
    base64::engine::general_purpose::STANDARD
        .decode(s)
        .unwrap_or_default()
}

// ═══════════════════════════════════════════════════════════════════════════
// Tests (require a running Redis instance, so marked #[ignore])
// ═══════════════════════════════════════════════════════════════════════════

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

    /// These tests require a Redis instance. Run them with:
    /// ```shell
    /// REDIS_URL=redis://localhost:6379 \
    ///   cargo test -p sansavision-pulse --features queue-redis -- redis_queue --ignored
    /// ```
    fn get_test_url() -> Option<String> {
        std::env::var("REDIS_URL").ok()
    }

    #[test]
    #[ignore = "requires Redis — set REDIS_URL to run"]
    fn redis_queue_lifecycle() {
        let url = get_test_url().expect("REDIS_URL not set");
        let backend = RedisQueueBackend::connect(&url).unwrap();

        let q = "redis-test-lifecycle";

        // Clean up from previous runs
        let _ = backend.purge(q);

        // Append
        let msg = QueueMessage {
            sequence: 1,
            payload: b"hello-redis".to_vec(),
            idempotency_key: Some("redis-key-1".to_string()),
            ttl_secs: Some(3600),
            enqueued_at: 1000,
            delivery_count: 0,
            state: MessageState::Pending,
        };
        backend.append(q, &msg).unwrap();

        // Load pending
        let pending = backend.load_pending(q).unwrap();
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].payload, b"hello-redis");

        // Idempotency
        backend.record_idempotency_key(q, "redis-key-1").unwrap();
        assert!(backend.has_idempotency_key(q, "redis-key-1").unwrap());
        assert!(!backend.has_idempotency_key(q, "redis-key-2").unwrap());

        // Update state
        backend
            .update_state(q, 1, MessageState::Acked, 1)
            .unwrap();
        let pending = backend.load_pending(q).unwrap();
        assert!(pending.is_empty());

        // Purge
        let _ = backend.purge(q);
    }
}
