use sqlx::{FromRow, SqlitePool, sqlite::SqliteConnectOptions, sqlite::SqlitePoolOptions};
use std::sync::OnceLock;
use tokio::sync::OnceCell;

/// Direction of a fetch cursor for Bluesky firehose pagination.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FetchDirection {
    Forward,
    Reverse,
}

impl FetchDirection {
    fn as_str(self) -> &'static str {
        match self {
            FetchDirection::Forward => "forward",
            FetchDirection::Reverse => "reverse",
        }
    }
}

/// A cursor position for fetching chat messages from the firehose.
#[derive(Debug, Clone)]
pub struct FetchCursor {
    pub cursor: Option<String>,
    pub boundary_ms: i64,
}

/// A chat message stored in the local SQLite cache.
#[derive(Debug, Clone, FromRow)]
pub struct CachedChatMessage {
    pub streamer_did: String,
    pub author_did: String,
    pub rkey: String,
    pub created_at_ms: i64,
    pub text: String,
    pub reply_parent_uri: Option<String>,
    pub reply_root_uri: Option<String>,
    pub facets_json: Option<String>,
}

/// SQLite-backed cache for chat messages and fetch cursors.
pub struct ChatCache {
    pool: SqlitePool,
    schema_init: OnceCell<()>,
}

impl ChatCache {
    /// Open (or create) a SQLite database at `path`.
    ///
    /// The connection is established lazily; schema creation is deferred until
    /// the first operation via [`ensure_schema`].
    pub fn open(path: &str) -> Result<Self, sqlx::Error> {
        let options = SqliteConnectOptions::new()
            .filename(path)
            .create_if_missing(true)
            .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);

        let pool = SqlitePoolOptions::new().connect_lazy_with(options);

        Ok(Self {
            pool,
            schema_init: OnceCell::new(),
        })
    }

    /// Ensure the database schema has been created. Called before every
    /// operation so that the first caller initialises it exactly once.
    async fn ensure_schema(&self) -> Result<(), sqlx::Error> {
        self.schema_init
            .get_or_try_init(|| async {
                sqlx::query(
                    r#"
                    CREATE TABLE IF NOT EXISTS messages (
                        streamer_did      TEXT NOT NULL,
                        author_did        TEXT NOT NULL,
                        rkey              TEXT NOT NULL,
                        created_at_ms     INTEGER NOT NULL,
                        text              TEXT NOT NULL,
                        reply_parent_uri  TEXT,
                        reply_root_uri    TEXT,
                        facets_json       TEXT,
                        PRIMARY KEY (author_did, rkey)
                    )
                    "#,
                )
                .execute(&self.pool)
                .await?;

                sqlx::query(
                    r#"
                    CREATE INDEX IF NOT EXISTS messages_streamer_time
                        ON messages (streamer_did, created_at_ms)
                    "#,
                )
                .execute(&self.pool)
                .await?;

                sqlx::query(
                    r#"
                    CREATE TABLE IF NOT EXISTS fetch_cursors (
                        streamer_did  TEXT NOT NULL,
                        direction     TEXT NOT NULL CHECK(direction IN ('forward', 'reverse')),
                        cursor        TEXT,
                        boundary_ms   INTEGER NOT NULL,
                        PRIMARY KEY (streamer_did, direction)
                    )
                    "#,
                )
                .execute(&self.pool)
                .await?;

                sqlx::query(
                    r#"
                    CREATE TABLE IF NOT EXISTS video_catalog (
                        id          INTEGER PRIMARY KEY DEFAULT 1,
                        data_json   TEXT NOT NULL,
                        updated_at  INTEGER NOT NULL
                    )
                    "#,
                )
                .execute(&self.pool)
                .await?;

                Ok(())
            })
            .await
            .copied()
    }

    /// Bulk-insert messages, silently ignoring duplicates (same author_did + rkey).
    pub async fn store_messages(&self, messages: &[CachedChatMessage]) -> Result<(), sqlx::Error> {
        self.ensure_schema().await?;

        let mut tx = self.pool.begin().await?;
        for msg in messages {
            sqlx::query(
                r#"
                INSERT OR IGNORE INTO messages
                    (streamer_did, author_did, rkey, created_at_ms, text,
                     reply_parent_uri, reply_root_uri, facets_json)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                "#,
            )
            .bind(&msg.streamer_did)
            .bind(&msg.author_did)
            .bind(&msg.rkey)
            .bind(msg.created_at_ms)
            .bind(&msg.text)
            .bind(&msg.reply_parent_uri)
            .bind(&msg.reply_root_uri)
            .bind(&msg.facets_json)
            .execute(&mut *tx)
            .await?;
        }
        tx.commit().await?;
        Ok(())
    }

    /// Retrieve all messages for `streamer_did` whose timestamp falls within
    /// `[start_ms, end_ms]`, ordered chronologically.
    pub async fn get_messages(
        &self,
        streamer_did: &str,
        start_ms: i64,
        end_ms: i64,
    ) -> Result<Vec<CachedChatMessage>, sqlx::Error> {
        self.ensure_schema().await?;

        sqlx::query_as::<_, CachedChatMessage>(
            r#"
            SELECT streamer_did, author_did, rkey, created_at_ms, text,
                   reply_parent_uri, reply_root_uri, facets_json
            FROM messages
            WHERE streamer_did = ?
              AND created_at_ms BETWEEN ? AND ?
            ORDER BY created_at_ms ASC
            "#,
        )
        .bind(streamer_did)
        .bind(start_ms)
        .bind(end_ms)
        .fetch_all(&self.pool)
        .await
    }

    /// Retrieve the stored fetch cursor for `(streamer_did, direction)`.
    pub async fn get_fetch_cursor(
        &self,
        streamer_did: &str,
        direction: FetchDirection,
    ) -> Result<Option<FetchCursor>, sqlx::Error> {
        self.ensure_schema().await?;

        let row: Option<(Option<String>, i64)> = sqlx::query_as(
            r#"
            SELECT cursor, boundary_ms
            FROM fetch_cursors
            WHERE streamer_did = ? AND direction = ?
            "#,
        )
        .bind(streamer_did)
        .bind(direction.as_str())
        .fetch_optional(&self.pool)
        .await?;

        Ok(row.map(|(cursor, boundary_ms)| FetchCursor {
            cursor,
            boundary_ms,
        }))
    }

    /// Upsert the fetch cursor for `(streamer_did, direction)`.
    pub async fn set_fetch_cursor(
        &self,
        streamer_did: &str,
        direction: FetchDirection,
        cursor: Option<&str>,
        boundary_ms: i64,
    ) -> Result<(), sqlx::Error> {
        self.ensure_schema().await?;

        sqlx::query(
            r#"
            INSERT INTO fetch_cursors (streamer_did, direction, cursor, boundary_ms)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(streamer_did, direction)
            DO UPDATE SET cursor = excluded.cursor, boundary_ms = excluded.boundary_ms
            "#,
        )
        .bind(streamer_did)
        .bind(direction.as_str())
        .bind(cursor)
        .bind(boundary_ms)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    /// Store the full video catalog as a JSON blob, replacing any previous entry.
    pub async fn store_video_catalog<T: serde::Serialize>(
        &self,
        videos: &[T],
    ) -> Result<(), sqlx::Error> {
        self.ensure_schema().await?;

        let json =
            serde_json::to_string(videos).map_err(|e| sqlx::Error::Protocol(e.to_string()))?;
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs() as i64;

        sqlx::query(
            r#"
            INSERT INTO video_catalog (id, data_json, updated_at)
            VALUES (1, ?, ?)
            ON CONFLICT(id)
            DO UPDATE SET data_json = excluded.data_json, updated_at = excluded.updated_at
            "#,
        )
        .bind(&json)
        .bind(now)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    /// Load the cached video catalog, if any. Returns `(videos, updated_at_unix_secs)`.
    pub async fn load_video_catalog<T: serde::de::DeserializeOwned>(
        &self,
    ) -> Result<Option<(Vec<T>, i64)>, sqlx::Error> {
        self.ensure_schema().await?;

        let row: Option<(String, i64)> = sqlx::query_as(
            r#"
            SELECT data_json, updated_at
            FROM video_catalog
            WHERE id = 1
            "#,
        )
        .fetch_optional(&self.pool)
        .await?;

        match row {
            Some((json, updated_at)) => {
                let videos: Vec<T> = serde_json::from_str(&json)
                    .map_err(|e| sqlx::Error::Protocol(e.to_string()))?;
                Ok(Some((videos, updated_at)))
            }
            None => Ok(None),
        }
    }
}

// ---------------------------------------------------------------------------
// Global accessor
// ---------------------------------------------------------------------------

static CACHE: OnceLock<ChatCache> = OnceLock::new();

/// Return a reference to the process-wide [`ChatCache`].
///
/// Panics on first call if the database cannot be opened (e.g. bad path or
/// permissions).
pub(crate) fn shared_cache() -> &'static ChatCache {
    CACHE.get_or_init(|| {
        let dir = std::env::var("VODPLACE_CACHE_DIR").unwrap_or_else(|_| "./data".to_string());
        std::fs::create_dir_all(&dir).expect("failed to create cache directory");
        let path = format!("{dir}/cache.db");
        ChatCache::open(&path).expect("failed to open chat message cache database")
    })
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

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

    /// Create an in-memory SQLite chat cache for tests.
    ///
    /// Uses `max_connections(1)` with an eager connect so all operations share
    /// the same connection and see the same in-memory database. A lazy pool would
    /// open separate connections, each with an empty DB, causing schema-not-found
    /// failures after `ensure_schema` runs on one connection.
    async fn test_cache() -> ChatCache {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("failed to open in-memory test database");
        let cache = ChatCache {
            pool,
            schema_init: OnceCell::new(),
        };
        cache.ensure_schema().await.expect("schema init failed");
        cache
    }

    fn test_message(rkey: &str, created_at_ms: i64) -> CachedChatMessage {
        CachedChatMessage {
            streamer_did: "did:plc:streamer1".to_string(),
            author_did: "did:plc:author1".to_string(),
            rkey: rkey.to_string(),
            created_at_ms,
            text: format!("message at {created_at_ms}"),
            reply_parent_uri: None,
            reply_root_uri: None,
            facets_json: None,
        }
    }

    #[tokio::test]
    async fn store_and_retrieve_messages_same_window() {
        let cache = test_cache().await;

        let messages = vec![
            test_message("rkey1", 1000),
            test_message("rkey2", 2000),
            test_message("rkey3", 3000),
        ];

        cache.store_messages(&messages).await.expect("store failed");

        let result = cache
            .get_messages("did:plc:streamer1", 1000, 3000)
            .await
            .expect("get failed");

        assert_eq!(result.len(), 3);
        assert_eq!(result[0].rkey, "rkey1");
        assert_eq!(result[1].rkey, "rkey2");
        assert_eq!(result[2].rkey, "rkey3");
    }

    #[tokio::test]
    async fn query_subset_time_window() {
        let cache = test_cache().await;

        let messages = vec![
            test_message("rkey1", 1000),
            test_message("rkey2", 2000),
            test_message("rkey3", 3000),
            test_message("rkey4", 4000),
        ];

        cache.store_messages(&messages).await.expect("store failed");

        // Ask for only the middle two
        let result = cache
            .get_messages("did:plc:streamer1", 2000, 3000)
            .await
            .expect("get failed");

        assert_eq!(result.len(), 2);
        assert_eq!(result[0].rkey, "rkey2");
        assert_eq!(result[1].rkey, "rkey3");
    }

    #[tokio::test]
    async fn set_and_get_forward_cursor() {
        let cache = test_cache().await;

        cache
            .set_fetch_cursor(
                "did:plc:streamer1",
                FetchDirection::Forward,
                Some("cursor-abc"),
                5000,
            )
            .await
            .expect("set failed");

        let got = cache
            .get_fetch_cursor("did:plc:streamer1", FetchDirection::Forward)
            .await
            .expect("get failed")
            .expect("cursor not found");

        assert_eq!(got.cursor.as_deref(), Some("cursor-abc"));
        assert_eq!(got.boundary_ms, 5000);
    }

    #[tokio::test]
    async fn set_and_get_reverse_cursor() {
        let cache = test_cache().await;

        cache
            .set_fetch_cursor(
                "did:plc:streamer1",
                FetchDirection::Reverse,
                Some("cursor-xyz"),
                1000,
            )
            .await
            .expect("set failed");

        let got = cache
            .get_fetch_cursor("did:plc:streamer1", FetchDirection::Reverse)
            .await
            .expect("get failed")
            .expect("cursor not found");

        assert_eq!(got.cursor.as_deref(), Some("cursor-xyz"));
        assert_eq!(got.boundary_ms, 1000);
    }

    #[tokio::test]
    async fn forward_and_reverse_cursors_are_independent() {
        let cache = test_cache().await;

        cache
            .set_fetch_cursor(
                "did:plc:streamer1",
                FetchDirection::Forward,
                Some("fwd-cursor"),
                9000,
            )
            .await
            .expect("set forward failed");

        cache
            .set_fetch_cursor(
                "did:plc:streamer1",
                FetchDirection::Reverse,
                Some("rev-cursor"),
                1000,
            )
            .await
            .expect("set reverse failed");

        let fwd = cache
            .get_fetch_cursor("did:plc:streamer1", FetchDirection::Forward)
            .await
            .expect("get forward failed")
            .expect("forward cursor not found");

        let rev = cache
            .get_fetch_cursor("did:plc:streamer1", FetchDirection::Reverse)
            .await
            .expect("get reverse failed")
            .expect("reverse cursor not found");

        assert_eq!(fwd.cursor.as_deref(), Some("fwd-cursor"));
        assert_eq!(fwd.boundary_ms, 9000);
        assert_eq!(rev.cursor.as_deref(), Some("rev-cursor"));
        assert_eq!(rev.boundary_ms, 1000);
    }

    #[tokio::test]
    async fn duplicate_messages_insert_or_ignore() {
        let cache = test_cache().await;

        let messages = vec![test_message("rkey1", 1000)];
        cache
            .store_messages(&messages)
            .await
            .expect("first store failed");

        // Store the same message again — must not error and must not duplicate
        cache
            .store_messages(&messages)
            .await
            .expect("second store failed");

        let result = cache
            .get_messages("did:plc:streamer1", 1000, 1000)
            .await
            .expect("get failed");

        assert_eq!(result.len(), 1, "duplicate message was not ignored");
    }
}