//! SQLite pool initialization and projection-cursor persistence.
//!
//! Two durable stores back the server:
//! 1. Hydrant's fjall store (`HYDRANT_DATABASE_PATH`) for raw ATProto records.
//! 2. This SQLite projection database (`DATABASE_URL`) for app-specific derived
//!    views. The pool here owns the second.

use std::str::FromStr;

use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
use sqlx::{Sqlite, SqlitePool, Transaction};

use super::config::ServerConfig;

/// Open the SQLite pool, enable WAL mode, run migrations, and return the pool.
///
/// WAL mode is set on the connection options (not in a migration) because SQLite
/// silently ignores `journal_mode` changes inside the transaction sqlx uses to
/// run migrations.
pub async fn init_db(cfg: &ServerConfig) -> anyhow::Result<SqlitePool> {
    ensure_data_dir(&cfg.database_url)?;

    let options = SqliteConnectOptions::from_str(&cfg.database_url)?
        .create_if_missing(true)
        .journal_mode(SqliteJournalMode::Wal);

    let pool = SqlitePoolOptions::new()
        .max_connections(5)
        .connect_with(options)
        .await?;

    sqlx::migrate!("./migrations").run(&pool).await?;
    tracing::info!("sqlite migrations applied");
    Ok(pool)
}

fn ensure_data_dir(db_url: &str) -> anyhow::Result<()> {
    if let Some(path) = db_url.strip_prefix("sqlite:")
        && let Some(parent) = std::path::Path::new(path).parent()
    {
        std::fs::create_dir_all(parent)?;
    }
    Ok(())
}

struct CursorRow {
    value: String,
}

/// Load the last persisted projection cursor (the Hydrant event id), if any.
pub async fn load_cursor(db: &SqlitePool) -> anyhow::Result<Option<u64>> {
    let row: Option<CursorRow> = sqlx::query_as!(
        CursorRow,
        "SELECT value FROM projection_state WHERE key = ?",
        "cursor"
    )
    .fetch_optional(db)
    .await?;
    Ok(row.and_then(|r| r.value.parse().ok()))
}

/// Persist the projection cursor outside a projection transaction.
///
/// Used for events that are acknowledged but not projected (e.g. account status
/// events), so the durable cursor advances past them and a reconnect/resume does
/// not replay them indefinitely.
pub async fn advance_cursor(db: &SqlitePool, cursor: u64) -> anyhow::Result<()> {
    let mut tx = db.begin().await?;
    save_cursor(&mut tx, cursor).await?;
    tx.commit().await?;
    Ok(())
}

/// Persist the projection cursor inside an in-flight projection transaction.
pub async fn save_cursor(tx: &mut Transaction<'_, Sqlite>, cursor: u64) -> anyhow::Result<()> {
    let now = chrono::Utc::now()
        .timestamp_nanos_opt()
        .expect("current timestamp must fit in i64 nanoseconds");
    sqlx::query!(
        "INSERT INTO projection_state (key, value, updated_at) VALUES ('cursor', ?, ?) \
         ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
        cursor.to_string(),
        now
    )
    .execute(&mut **tx)
    .await?;
    Ok(())
}