//! SQLite-backed KV store for WASM plugins.
//!
//! Each plugin gets isolated key-value storage scoped by `(plugin_id, namespace)`.
//! Values can have optional TTLs, and expired entries are lazily pruned.

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

/// SQLite-backed KV store for WASM plugins.
pub struct PluginKvStore {
    db: Mutex<Connection>,
}

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

impl PluginKvStore {
    /// Open a KV store 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 KV database: {e}"))?;
        let store = Self {
            db: Mutex::new(conn),
        };
        store.init_schema()?;
        Ok(store)
    }

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

    fn init_schema(&self) -> Result<(), String> {
        let db = self.db.lock();
        db.execute_batch(
            "CREATE TABLE IF NOT EXISTS plugin_kv (
                plugin_id TEXT NOT NULL,
                namespace TEXT NOT NULL,
                key TEXT NOT NULL,
                value TEXT NOT NULL,
                expires_at INTEGER,
                created_at INTEGER NOT NULL,
                updated_at INTEGER NOT NULL,
                PRIMARY KEY (plugin_id, namespace, key)
            );
            CREATE INDEX IF NOT EXISTS idx_kv_expires
                ON plugin_kv(expires_at) WHERE expires_at IS NOT NULL;",
        )
        .map_err(|e| format!("init KV schema: {e}"))?;
        Ok(())
    }

    /// Get a value. Returns `None` if the key doesn't exist or is expired.
    pub fn get(
        &self,
        plugin_id: &str,
        namespace: &str,
        key: &str,
    ) -> Result<Option<String>, String> {
        let now = now_epoch_secs();
        let db = self.db.lock();
        let mut stmt = db
            .prepare(
                "SELECT value FROM plugin_kv
                 WHERE plugin_id = ?1 AND namespace = ?2 AND key = ?3
                   AND (expires_at IS NULL OR expires_at > ?4)",
            )
            .map_err(|e| format!("prepare KV get: {e}"))?;

        let result = stmt.query_row(rusqlite::params![plugin_id, namespace, key, now], |row| {
            row.get::<_, String>(0)
        });

        match result {
            Ok(val) => Ok(Some(val)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(format!("KV get: {e}")),
        }
    }

    /// Set a value with optional TTL in seconds.
    pub fn set(
        &self,
        plugin_id: &str,
        namespace: &str,
        key: &str,
        value: &str,
        ttl_secs: Option<u64>,
    ) -> Result<(), String> {
        let now = now_epoch_secs();
        let expires_at = ttl_secs.map(|ttl| now + ttl as i64);
        let db = self.db.lock();
        db.execute(
            "INSERT INTO plugin_kv (plugin_id, namespace, key, value, expires_at, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6)
             ON CONFLICT (plugin_id, namespace, key)
             DO UPDATE SET value = excluded.value,
                           expires_at = excluded.expires_at,
                           updated_at = excluded.updated_at",
            rusqlite::params![plugin_id, namespace, key, value, expires_at, now],
        )
        .map_err(|e| format!("KV set: {e}"))?;
        Ok(())
    }

    /// Delete a key. Returns `true` if the key existed.
    pub fn delete(&self, plugin_id: &str, namespace: &str, key: &str) -> Result<bool, String> {
        let db = self.db.lock();
        let changes = db
            .execute(
                "DELETE FROM plugin_kv
                 WHERE plugin_id = ?1 AND namespace = ?2 AND key = ?3",
                rusqlite::params![plugin_id, namespace, key],
            )
            .map_err(|e| format!("KV delete: {e}"))?;
        Ok(changes > 0)
    }

    /// List keys matching an optional prefix. Returns up to `limit` keys.
    pub fn list(
        &self,
        plugin_id: &str,
        namespace: &str,
        prefix: Option<&str>,
        limit: usize,
    ) -> Result<Vec<String>, String> {
        let now = now_epoch_secs();
        let db = self.db.lock();

        let (sql, like_pattern);
        if let Some(pfx) = prefix {
            // Escape SQL LIKE special chars in the prefix, then append %
            let escaped = pfx
                .replace('\\', "\\\\")
                .replace('%', "\\%")
                .replace('_', "\\_");
            like_pattern = format!("{escaped}%");
            sql = "SELECT key FROM plugin_kv
                   WHERE plugin_id = ?1 AND namespace = ?2
                     AND key LIKE ?3 ESCAPE '\\'
                     AND (expires_at IS NULL OR expires_at > ?4)
                   ORDER BY key
                   LIMIT ?5";
        } else {
            like_pattern = String::new();
            sql = "SELECT key FROM plugin_kv
                   WHERE plugin_id = ?1 AND namespace = ?2
                     AND (expires_at IS NULL OR expires_at > ?3)
                   ORDER BY key
                   LIMIT ?4";
        }

        if prefix.is_some() {
            let mut stmt = db
                .prepare(sql)
                .map_err(|e| format!("prepare KV list: {e}"))?;
            let rows = stmt
                .query_map(
                    rusqlite::params![plugin_id, namespace, &like_pattern, now, limit as i64],
                    |row| row.get::<_, String>(0),
                )
                .map_err(|e| format!("KV list: {e}"))?;
            rows.collect::<Result<Vec<_>, _>>()
                .map_err(|e| format!("KV list collect: {e}"))
        } else {
            let mut stmt = db
                .prepare(sql)
                .map_err(|e| format!("prepare KV list: {e}"))?;
            let rows = stmt
                .query_map(
                    rusqlite::params![plugin_id, namespace, now, limit as i64],
                    |row| row.get::<_, String>(0),
                )
                .map_err(|e| format!("KV list: {e}"))?;
            rows.collect::<Result<Vec<_>, _>>()
                .map_err(|e| format!("KV list collect: {e}"))
        }
    }

    /// Remove expired entries. Returns the number of rows deleted.
    pub fn cleanup_expired(&self) -> Result<usize, String> {
        let now = now_epoch_secs();
        let db = self.db.lock();
        let changes = db
            .execute(
                "DELETE FROM plugin_kv WHERE expires_at IS NOT NULL AND expires_at <= ?1",
                rusqlite::params![now],
            )
            .map_err(|e| format!("KV cleanup: {e}"))?;
        Ok(changes)
    }
}

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

    fn store() -> PluginKvStore {
        PluginKvStore::open_memory().unwrap()
    }

    #[test]
    fn get_set_delete_basic() {
        let kv = store();
        // Get non-existent key
        assert_eq!(kv.get("p1", "ns", "k1").unwrap(), None);

        // Set and get
        kv.set("p1", "ns", "k1", "hello", None).unwrap();
        assert_eq!(kv.get("p1", "ns", "k1").unwrap(), Some("hello".into()));

        // Overwrite
        kv.set("p1", "ns", "k1", "world", None).unwrap();
        assert_eq!(kv.get("p1", "ns", "k1").unwrap(), Some("world".into()));

        // Delete existing
        assert!(kv.delete("p1", "ns", "k1").unwrap());
        assert_eq!(kv.get("p1", "ns", "k1").unwrap(), None);

        // Delete non-existent
        assert!(!kv.delete("p1", "ns", "k1").unwrap());
    }

    #[test]
    fn ttl_expiry() {
        let kv = store();

        // Set with TTL of 0 seconds — should already be expired
        // We set it, then immediately try to get it. With TTL=0, expires_at == now,
        // and the query requires expires_at > now, so it should return None.
        kv.set("p1", "ns", "ttl-key", "ephemeral", Some(0)).unwrap();
        // expires_at == now, condition is `expires_at > now` → None
        assert_eq!(kv.get("p1", "ns", "ttl-key").unwrap(), None);

        // Set with a long TTL — should still be visible
        kv.set("p1", "ns", "ttl-key2", "durable", Some(3600))
            .unwrap();
        assert_eq!(
            kv.get("p1", "ns", "ttl-key2").unwrap(),
            Some("durable".into())
        );

        // No-TTL key should always be visible
        kv.set("p1", "ns", "no-ttl", "forever", None).unwrap();
        assert_eq!(
            kv.get("p1", "ns", "no-ttl").unwrap(),
            Some("forever".into())
        );
    }

    #[test]
    fn list_with_prefix_filter() {
        let kv = store();
        kv.set("p1", "ns", "user:1", "alice", None).unwrap();
        kv.set("p1", "ns", "user:2", "bob", None).unwrap();
        kv.set("p1", "ns", "session:1", "s1", None).unwrap();
        kv.set("p1", "ns", "session:2", "s2", None).unwrap();

        let users = kv.list("p1", "ns", Some("user:"), 100).unwrap();
        assert_eq!(users, vec!["user:1", "user:2"]);

        let sessions = kv.list("p1", "ns", Some("session:"), 100).unwrap();
        assert_eq!(sessions, vec!["session:1", "session:2"]);

        // No prefix — all keys
        let all = kv.list("p1", "ns", None, 100).unwrap();
        assert_eq!(all.len(), 4);
    }

    #[test]
    fn list_with_limit() {
        let kv = store();
        for i in 0..10 {
            kv.set(
                "p1",
                "ns",
                &format!("key-{i:02}"),
                &format!("val-{i}"),
                None,
            )
            .unwrap();
        }

        let limited = kv.list("p1", "ns", None, 3).unwrap();
        assert_eq!(limited.len(), 3);
        // Should be sorted alphabetically (key-00, key-01, key-02)
        assert_eq!(limited, vec!["key-00", "key-01", "key-02"]);
    }

    #[test]
    fn cleanup_expired_removes_entries() {
        let kv = store();
        // Insert an entry that is already expired (TTL=0)
        kv.set("p1", "ns", "expired", "gone", Some(0)).unwrap();
        // Insert an entry that is NOT expired
        kv.set("p1", "ns", "alive", "here", Some(3600)).unwrap();
        // Insert an entry with no TTL
        kv.set("p1", "ns", "forever", "yes", None).unwrap();

        let cleaned = kv.cleanup_expired().unwrap();
        assert_eq!(cleaned, 1);

        // "alive" and "forever" should still be there
        assert_eq!(kv.get("p1", "ns", "alive").unwrap(), Some("here".into()));
        assert_eq!(kv.get("p1", "ns", "forever").unwrap(), Some("yes".into()));
    }

    #[test]
    fn namespace_isolation() {
        let kv = store();
        // Plugin A writes to ns-a
        kv.set("plugin-a", "ns-a", "key1", "a-value", None).unwrap();
        // Plugin B writes to ns-b with the same key
        kv.set("plugin-b", "ns-b", "key1", "b-value", None).unwrap();

        // Each plugin only sees its own data
        assert_eq!(
            kv.get("plugin-a", "ns-a", "key1").unwrap(),
            Some("a-value".into())
        );
        assert_eq!(
            kv.get("plugin-b", "ns-b", "key1").unwrap(),
            Some("b-value".into())
        );

        // Cross-access returns None
        assert_eq!(kv.get("plugin-a", "ns-b", "key1").unwrap(), None);
        assert_eq!(kv.get("plugin-b", "ns-a", "key1").unwrap(), None);

        // Same plugin, different namespace
        kv.set("plugin-a", "ns-x", "key1", "x-value", None).unwrap();
        assert_eq!(
            kv.get("plugin-a", "ns-a", "key1").unwrap(),
            Some("a-value".into())
        );
        assert_eq!(
            kv.get("plugin-a", "ns-x", "key1").unwrap(),
            Some("x-value".into())
        );

        // List is scoped
        let keys_a = kv.list("plugin-a", "ns-a", None, 100).unwrap();
        assert_eq!(keys_a, vec!["key1"]);
        let keys_b = kv.list("plugin-b", "ns-a", None, 100).unwrap();
        assert!(keys_b.is_empty());
    }

    #[test]
    fn list_excludes_expired() {
        let kv = store();
        kv.set("p1", "ns", "alive", "yes", Some(3600)).unwrap();
        kv.set("p1", "ns", "dead", "no", Some(0)).unwrap();
        kv.set("p1", "ns", "forever", "always", None).unwrap();

        let keys = kv.list("p1", "ns", None, 100).unwrap();
        assert_eq!(keys, vec!["alive", "forever"]);
    }

    #[test]
    fn set_updates_existing_ttl() {
        let kv = store();
        // Set with TTL
        kv.set("p1", "ns", "k", "v1", Some(3600)).unwrap();
        assert_eq!(kv.get("p1", "ns", "k").unwrap(), Some("v1".into()));

        // Update: remove TTL
        kv.set("p1", "ns", "k", "v2", None).unwrap();
        assert_eq!(kv.get("p1", "ns", "k").unwrap(), Some("v2".into()));

        // Update: add TTL=0 (expire immediately)
        kv.set("p1", "ns", "k", "v3", Some(0)).unwrap();
        assert_eq!(kv.get("p1", "ns", "k").unwrap(), None);
    }

    #[test]
    fn list_with_special_prefix_chars() {
        let kv = store();
        kv.set("p1", "ns", "100%_done", "yes", None).unwrap();
        kv.set("p1", "ns", "100%_todo", "no", None).unwrap();
        kv.set("p1", "ns", "100x_other", "maybe", None).unwrap();

        // The "%" in the prefix should be escaped so it doesn't act as a wildcard
        let keys = kv.list("p1", "ns", Some("100%"), 100).unwrap();
        assert_eq!(keys, vec!["100%_done", "100%_todo"]);
    }
}
