//! Durable queue for WASM plugins, backed by SQLite.
//!
//! Provides a per-plugin, per-queue-name message queue with delayed delivery,
//! automatic retries, and dead-letter handling. Items that exceed `max_attempts`
//! are moved to `Dead` status.

use parking_lot::Mutex;
use rusqlite::Connection;
use std::path::Path;
use std::time::{SystemTime, UNIX_EPOCH};

/// Durable queue for WASM plugins, backed by SQLite.
pub struct PluginQueue {
    db: Mutex<Connection>,
}

/// A single item in the queue.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct QueueItem {
    pub id: String,
    pub queue_name: String,
    pub plugin_id: String,
    pub payload: String,
    pub status: QueueItemStatus,
    pub created_at: i64,
    pub available_at: i64,
    pub attempts: u32,
    pub max_attempts: u32,
}

/// Lifecycle status of a queue item.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum QueueItemStatus {
    Pending,
    Processing,
    Completed,
    Failed,
    /// Exceeded max_attempts.
    Dead,
}

impl QueueItemStatus {
    /// String representation for SQLite storage. Will be used by host function wiring.
    #[allow(dead_code)]
    fn as_str(&self) -> &'static str {
        match self {
            Self::Pending => "Pending",
            Self::Processing => "Processing",
            Self::Completed => "Completed",
            Self::Failed => "Failed",
            Self::Dead => "Dead",
        }
    }

    fn from_str(s: &str) -> Self {
        match s {
            "Pending" => Self::Pending,
            "Processing" => Self::Processing,
            "Completed" => Self::Completed,
            "Failed" => Self::Failed,
            "Dead" => Self::Dead,
            _ => Self::Failed,
        }
    }
}

fn now_epoch_ms() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as i64
}

impl PluginQueue {
    /// Open a queue backed by a file at `db_path`.
    pub fn new(db_path: &Path) -> Result<Self, String> {
        let conn = Connection::open(db_path).map_err(|e| format!("open queue database: {e}"))?;
        let queue = Self {
            db: Mutex::new(conn),
        };
        queue.init_schema()?;
        Ok(queue)
    }

    /// Open an in-memory queue (useful for tests).
    pub fn open_memory() -> Result<Self, String> {
        let conn = Connection::open_in_memory()
            .map_err(|e| format!("open in-memory queue database: {e}"))?;
        let queue = Self {
            db: Mutex::new(conn),
        };
        queue.init_schema()?;
        Ok(queue)
    }

    fn init_schema(&self) -> Result<(), String> {
        let db = self.db.lock();
        db.execute_batch(
            "CREATE TABLE IF NOT EXISTS plugin_queue (
                id TEXT PRIMARY KEY,
                queue_name TEXT NOT NULL,
                plugin_id TEXT NOT NULL,
                payload TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT 'Pending',
                created_at INTEGER NOT NULL,
                available_at INTEGER NOT NULL,
                attempts INTEGER NOT NULL DEFAULT 0,
                max_attempts INTEGER NOT NULL DEFAULT 3,
                updated_at INTEGER NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_queue_pop
                ON plugin_queue(plugin_id, queue_name, status, available_at);",
        )
        .map_err(|e| format!("init queue schema: {e}"))?;
        Ok(())
    }

    /// Push an item to the queue. Returns the item ID.
    ///
    /// `delay_ms` optionally delays the item's availability by the given
    /// number of milliseconds from now.
    pub fn push(
        &self,
        plugin_id: &str,
        queue_name: &str,
        payload: &str,
        delay_ms: Option<u64>,
    ) -> Result<String, String> {
        let now = now_epoch_ms();
        let available_at = now + delay_ms.unwrap_or(0) as i64;
        let id = uuid::Uuid::now_v7().to_string();

        let db = self.db.lock();
        db.execute(
            "INSERT INTO plugin_queue
                (id, queue_name, plugin_id, payload, status, created_at, available_at, attempts, max_attempts, updated_at)
             VALUES (?1, ?2, ?3, ?4, 'Pending', ?5, ?6, 0, 3, ?5)",
            rusqlite::params![id, queue_name, plugin_id, payload, now, available_at],
        )
        .map_err(|e| format!("queue push: {e}"))?;

        Ok(id)
    }

    /// Pop the next available item. Marks it as `Processing`.
    ///
    /// Returns the oldest `Pending` item whose `available_at <= now`.
    pub fn pop(&self, plugin_id: &str, queue_name: &str) -> Result<Option<QueueItem>, String> {
        let now = now_epoch_ms();
        let db = self.db.lock();

        // Find the oldest available pending item
        let row = db
            .prepare(
                "SELECT id, queue_name, plugin_id, payload, status,
                        created_at, available_at, attempts, max_attempts
                 FROM plugin_queue
                 WHERE plugin_id = ?1 AND queue_name = ?2
                   AND status = 'Pending' AND available_at <= ?3
                 ORDER BY available_at ASC, created_at ASC
                 LIMIT 1",
            )
            .map_err(|e| format!("prepare queue pop: {e}"))?
            .query_row(rusqlite::params![plugin_id, queue_name, now], |row| {
                Ok(QueueItem {
                    id: row.get(0)?,
                    queue_name: row.get(1)?,
                    plugin_id: row.get(2)?,
                    payload: row.get(3)?,
                    status: QueueItemStatus::from_str(&row.get::<_, String>(4)?),
                    created_at: row.get(5)?,
                    available_at: row.get(6)?,
                    attempts: row.get(7)?,
                    max_attempts: row.get(8)?,
                })
            });

        match row {
            Ok(mut item) => {
                // Mark as Processing and increment attempts
                item.attempts += 1;
                item.status = QueueItemStatus::Processing;
                db.execute(
                    "UPDATE plugin_queue
                     SET status = 'Processing', attempts = ?1, updated_at = ?2
                     WHERE id = ?3",
                    rusqlite::params![item.attempts, now, item.id],
                )
                .map_err(|e| format!("queue pop update: {e}"))?;
                Ok(Some(item))
            }
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(format!("queue pop: {e}")),
        }
    }

    /// Acknowledge successful processing. Marks the item as `Completed`.
    pub fn ack(&self, plugin_id: &str, queue_name: &str, item_id: &str) -> Result<(), String> {
        let now = now_epoch_ms();
        let db = self.db.lock();
        let changes = db
            .execute(
                "UPDATE plugin_queue
                 SET status = 'Completed', updated_at = ?1
                 WHERE id = ?2 AND plugin_id = ?3 AND queue_name = ?4
                   AND status = 'Processing'",
                rusqlite::params![now, item_id, plugin_id, queue_name],
            )
            .map_err(|e| format!("queue ack: {e}"))?;

        if changes == 0 {
            return Err(format!(
                "queue ack: item {item_id} not found or not in Processing state"
            ));
        }
        Ok(())
    }

    /// Mark an item as failed. If under `max_attempts`, re-queue as `Pending`.
    /// Otherwise, move to `Dead` status.
    pub fn nack(&self, plugin_id: &str, queue_name: &str, item_id: &str) -> Result<(), String> {
        let now = now_epoch_ms();
        let db = self.db.lock();

        // Get current attempts and max_attempts
        let (attempts, max_attempts): (u32, u32) = db
            .prepare(
                "SELECT attempts, max_attempts FROM plugin_queue
                 WHERE id = ?1 AND plugin_id = ?2 AND queue_name = ?3
                   AND status = 'Processing'",
            )
            .map_err(|e| format!("prepare queue nack: {e}"))?
            .query_row(rusqlite::params![item_id, plugin_id, queue_name], |row| {
                Ok((row.get(0)?, row.get(1)?))
            })
            .map_err(|e| format!("queue nack lookup: {e}"))?;

        let new_status = if attempts >= max_attempts {
            "Dead"
        } else {
            "Pending"
        };

        db.execute(
            "UPDATE plugin_queue
             SET status = ?1, updated_at = ?2
             WHERE id = ?3",
            rusqlite::params![new_status, now, item_id],
        )
        .map_err(|e| format!("queue nack update: {e}"))?;

        Ok(())
    }

    /// Get queue depth (number of `Pending` items).
    pub fn depth(&self, plugin_id: &str, queue_name: &str) -> Result<usize, String> {
        let db = self.db.lock();
        let count: i64 = db
            .prepare(
                "SELECT COUNT(*) FROM plugin_queue
                 WHERE plugin_id = ?1 AND queue_name = ?2 AND status = 'Pending'",
            )
            .map_err(|e| format!("prepare queue depth: {e}"))?
            .query_row(rusqlite::params![plugin_id, queue_name], |row| row.get(0))
            .map_err(|e| format!("queue depth: {e}"))?;
        Ok(count as usize)
    }

    /// Clean up `Completed` and `Dead` items older than `retention_secs`.
    /// Returns the number of rows deleted.
    pub fn cleanup(&self, retention_secs: u64) -> Result<usize, String> {
        let cutoff = now_epoch_ms() - (retention_secs as i64 * 1000);
        let db = self.db.lock();
        let changes = db
            .execute(
                "DELETE FROM plugin_queue
                 WHERE status IN ('Completed', 'Dead')
                   AND updated_at < ?1",
                rusqlite::params![cutoff],
            )
            .map_err(|e| format!("queue cleanup: {e}"))?;
        Ok(changes)
    }
}

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

    fn queue() -> PluginQueue {
        PluginQueue::open_memory().unwrap()
    }

    #[test]
    fn push_pop_ack_lifecycle() {
        let q = queue();
        let id = q.push("p1", "tasks", r#"{"job":"build"}"#, None).unwrap();
        assert!(!id.is_empty());

        // Depth should be 1
        assert_eq!(q.depth("p1", "tasks").unwrap(), 1);

        // Pop returns the item
        let item = q.pop("p1", "tasks").unwrap().expect("should have an item");
        assert_eq!(item.id, id);
        assert_eq!(item.payload, r#"{"job":"build"}"#);
        assert_eq!(item.status, QueueItemStatus::Processing);
        assert_eq!(item.attempts, 1);

        // Depth should be 0 (item is Processing, not Pending)
        assert_eq!(q.depth("p1", "tasks").unwrap(), 0);

        // Pop again returns None
        assert!(q.pop("p1", "tasks").unwrap().is_none());

        // Ack completes the item
        q.ack("p1", "tasks", &id).unwrap();
    }

    #[test]
    fn delayed_delivery() {
        let q = queue();
        // Push with a large delay
        let _id = q.push("p1", "delayed", "payload", Some(60_000)).unwrap();

        // Depth is 1 (pending items)
        assert_eq!(q.depth("p1", "delayed").unwrap(), 1);

        // Pop returns None because available_at is in the future
        assert!(q.pop("p1", "delayed").unwrap().is_none());
    }

    #[test]
    fn nack_and_retry() {
        let q = queue();
        let id = q.push("p1", "retries", "job-data", None).unwrap();

        // Pop and nack — should go back to Pending (attempt 1 of 3)
        let item = q.pop("p1", "retries").unwrap().unwrap();
        assert_eq!(item.attempts, 1);
        q.nack("p1", "retries", &id).unwrap();

        // Pop again — attempt 2
        let item = q.pop("p1", "retries").unwrap().unwrap();
        assert_eq!(item.attempts, 2);
        q.nack("p1", "retries", &id).unwrap();

        // Pop again — attempt 3
        let item = q.pop("p1", "retries").unwrap().unwrap();
        assert_eq!(item.attempts, 3);

        // Nack at max_attempts → Dead
        q.nack("p1", "retries", &id).unwrap();

        // Should not appear in pop anymore
        assert!(q.pop("p1", "retries").unwrap().is_none());
        assert_eq!(q.depth("p1", "retries").unwrap(), 0);
    }

    #[test]
    fn dead_letter_on_max_attempts() {
        let q = queue();
        let id = q.push("p1", "dl", "fail-job", None).unwrap();

        // Exhaust all 3 attempts
        for i in 1..=3 {
            let item = q.pop("p1", "dl").unwrap().unwrap();
            assert_eq!(item.attempts, i);
            q.nack("p1", "dl", &id).unwrap();
        }

        // Item should now be Dead — verify by checking depth is 0
        assert_eq!(q.depth("p1", "dl").unwrap(), 0);

        // Pop returns None
        assert!(q.pop("p1", "dl").unwrap().is_none());
    }

    #[test]
    fn depth_returns_correct_count() {
        let q = queue();
        assert_eq!(q.depth("p1", "q").unwrap(), 0);

        q.push("p1", "q", "a", None).unwrap();
        q.push("p1", "q", "b", None).unwrap();
        q.push("p1", "q", "c", None).unwrap();
        assert_eq!(q.depth("p1", "q").unwrap(), 3);

        // Pop one — depth drops
        q.pop("p1", "q").unwrap();
        assert_eq!(q.depth("p1", "q").unwrap(), 2);

        // Different queue should have 0 depth
        assert_eq!(q.depth("p1", "other").unwrap(), 0);

        // Different plugin should have 0 depth
        assert_eq!(q.depth("p2", "q").unwrap(), 0);
    }

    #[test]
    fn cleanup_removes_old_completed() {
        let q = queue();
        let id1 = q.push("p1", "q", "job1", None).unwrap();
        let id2 = q.push("p1", "q", "job2", None).unwrap();
        let _id3 = q.push("p1", "q", "job3", None).unwrap();

        // Complete first two
        q.pop("p1", "q").unwrap();
        q.ack("p1", "q", &id1).unwrap();
        q.pop("p1", "q").unwrap();
        q.ack("p1", "q", &id2).unwrap();

        // Cleanup with 0 retention — should remove completed items
        // The items were just completed so updated_at ~= now. With 0 retention,
        // cutoff = now, and updated_at < cutoff is false. Use a large retention
        // to verify the mechanism, then use a tiny future offset.
        let cleaned = q.cleanup(0).unwrap();
        // Items were just updated, so updated_at >= cutoff. None cleaned.
        assert_eq!(cleaned, 0);

        // Manually set updated_at to the past to simulate old items
        {
            let db = q.db.lock();
            db.execute(
                "UPDATE plugin_queue SET updated_at = 0 WHERE status = 'Completed'",
                [],
            )
            .unwrap();
        }

        let cleaned = q.cleanup(0).unwrap();
        assert_eq!(cleaned, 2);

        // Third item (still pending) should survive
        assert_eq!(q.depth("p1", "q").unwrap(), 1);
    }

    #[test]
    fn fifo_ordering() {
        let q = queue();
        let id1 = q.push("p1", "fifo", "first", None).unwrap();
        let id2 = q.push("p1", "fifo", "second", None).unwrap();
        let id3 = q.push("p1", "fifo", "third", None).unwrap();

        let item1 = q.pop("p1", "fifo").unwrap().unwrap();
        assert_eq!(item1.id, id1);
        assert_eq!(item1.payload, "first");
        q.ack("p1", "fifo", &id1).unwrap();

        let item2 = q.pop("p1", "fifo").unwrap().unwrap();
        assert_eq!(item2.id, id2);
        assert_eq!(item2.payload, "second");
        q.ack("p1", "fifo", &id2).unwrap();

        let item3 = q.pop("p1", "fifo").unwrap().unwrap();
        assert_eq!(item3.id, id3);
        assert_eq!(item3.payload, "third");
        q.ack("p1", "fifo", &id3).unwrap();

        assert!(q.pop("p1", "fifo").unwrap().is_none());
    }

    #[test]
    fn ack_wrong_state_fails() {
        let q = queue();
        let id = q.push("p1", "q", "data", None).unwrap();

        // Ack without pop (item is Pending, not Processing)
        let result = q.ack("p1", "q", &id);
        assert!(result.is_err());
    }

    #[test]
    fn plugin_isolation() {
        let q = queue();
        let _id_a = q.push("plugin-a", "shared", "a-payload", None).unwrap();
        let _id_b = q.push("plugin-b", "shared", "b-payload", None).unwrap();

        // Plugin A can only pop its own items
        let item_a = q.pop("plugin-a", "shared").unwrap().unwrap();
        assert_eq!(item_a.payload, "a-payload");

        // Plugin A's pop doesn't return plugin B's items
        assert!(q.pop("plugin-a", "shared").unwrap().is_none());

        // Plugin B can pop its own
        let item_b = q.pop("plugin-b", "shared").unwrap().unwrap();
        assert_eq!(item_b.payload, "b-payload");
    }

    #[test]
    fn multiple_queues_isolated() {
        let q = queue();
        q.push("p1", "queue-a", "a-data", None).unwrap();
        q.push("p1", "queue-b", "b-data", None).unwrap();

        assert_eq!(q.depth("p1", "queue-a").unwrap(), 1);
        assert_eq!(q.depth("p1", "queue-b").unwrap(), 1);

        let item = q.pop("p1", "queue-a").unwrap().unwrap();
        assert_eq!(item.payload, "a-data");

        // queue-b still has its item
        assert_eq!(q.depth("p1", "queue-b").unwrap(), 1);
    }
}
