//! The portable SQLite engine — a thin, host-agnostic wrapper over
//! `turso_core` (Turso's SQLite core, the Rust rewrite of SQLite).
//!
//! This module has **no** `wasm_bindgen` and **no** WASI in it: it is plain
//! Rust that compiles and tests natively (`cargo test`) and also compiles to
//! `wasm32-unknown-unknown`. The per-host bindings (the browser Web Worker
//! surface, a future WASI/component export) wrap this.
//!
//! v0 is **in-memory**: a `MemoryIO`-backed `:memory:` database, so there are
//! no filesystem syscalls. Durable persistence (snapshot the database image to
//! the host's blob store — OPFS cache / R2 source-of-truth, per the harness
//! design) is the next slice; see the crate README.

use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use serde::{Deserialize, Serialize};
use turso_core::{Connection, Database, LimboError, MemoryIO, NonNan, Numeric, Value};

/// Each engine opens on a *unique* real path inside its own MemoryIO. A real
/// path (not `:memory:`) so the engine writes an actual SQLite file image we can
/// serialize/restore; *unique* because turso_core dedupes opens by path in a
/// global registry — a shared path would alias separate instances. The database
/// still lives entirely in RAM.
static NEXT_DB_ID: AtomicU64 = AtomicU64::new(1);

fn fresh_path() -> String {
    format!("qube-{}.db", NEXT_DB_ID.fetch_add(1, Ordering::Relaxed))
}

/// A single SQL value crossing the engine boundary.
///
/// This is the engine's own value type (not `turso_core::Value`) so the public
/// surface — and the host bindings that serialize it — stay independent of the
/// engine crate's internals. It maps 1:1 to SQLite's five storage classes.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", content = "value", rename_all = "lowercase")]
pub enum SqlValue {
    Null,
    Integer(i64),
    Real(f64),
    Text(String),
    Blob(Vec<u8>),
}

impl SqlValue {
    fn from_core(v: &Value) -> SqlValue {
        match v {
            Value::Null => SqlValue::Null,
            Value::Numeric(Numeric::Integer(i)) => SqlValue::Integer(*i),
            Value::Numeric(Numeric::Float(f)) => SqlValue::Real(f64::from(*f)),
            Value::Text(t) => SqlValue::Text(t.as_str().to_string()),
            Value::Blob(b) => SqlValue::Blob(b.clone()),
        }
    }

    fn into_core(self) -> Value {
        match self {
            SqlValue::Null => Value::Null,
            SqlValue::Integer(i) => Value::Numeric(Numeric::Integer(i)),
            // SQLite cannot store NaN — it coerces to NULL. Mirror that.
            SqlValue::Real(f) => match NonNan::new(f) {
                Some(nn) => Value::Numeric(Numeric::Float(nn)),
                None => Value::Null,
            },
            SqlValue::Text(s) => Value::from_text(s),
            SqlValue::Blob(b) => Value::from_blob(b),
        }
    }
}

/// The rows + column names returned by a `SELECT` (or any row-producing SQL).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct QueryResult {
    pub columns: Vec<String>,
    pub rows: Vec<Vec<SqlValue>>,
}

/// The side effects of a row-less statement (`INSERT`/`UPDATE`/`DELETE`).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunResult {
    /// Rows changed by the most recent statement.
    pub changes: i64,
    /// Rowid of the most recent successful `INSERT`.
    pub last_insert_rowid: i64,
}

/// Errors the engine can surface. Kept small and stringly-typed at the SQL
/// boundary — the host bindings turn these into JS exceptions / WIT errors.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "message", rename_all = "snake_case")]
pub enum EngineError {
    /// Open / connect failed.
    Open(String),
    /// Prepare/compile or execution failed (syntax error, constraint, …).
    Sql(String),
}

impl std::fmt::Display for EngineError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            EngineError::Open(m) => write!(f, "open error: {m}"),
            EngineError::Sql(m) => write!(f, "sql error: {m}"),
        }
    }
}

impl std::error::Error for EngineError {}

impl From<LimboError> for EngineError {
    fn from(e: LimboError) -> Self {
        EngineError::Sql(e.to_string())
    }
}

/// An open, single-connection SQLite database.
///
/// Single connection is deliberate for v0: the harness runs the engine inside
/// one OPFS Web Worker, so all access is serialized through that worker anyway.
pub struct Engine {
    // The MemoryIO backing the database file — retained so we can serialize the
    // file image (export) and so it lives as long as the connection.
    io: Arc<MemoryIO>,
    // The unique path this database was opened on (its key in `io`).
    path: String,
    // Held to keep the database alive for as long as the connection.
    _db: Arc<Database>,
    conn: Arc<Connection>,
}

impl Engine {
    /// Open a fresh, empty database (held in RAM via MemoryIO).
    pub fn open_in_memory() -> Result<Engine, EngineError> {
        Self::open_with(Arc::new(MemoryIO::new()), fresh_path())
    }

    /// Open a database from a serialized SQLite image (e.g. an uploaded
    /// `.sqlite` file). The bytes are loaded into RAM, then opened.
    pub fn open_from_bytes(bytes: &[u8]) -> Result<Engine, EngineError> {
        let io = Arc::new(MemoryIO::new());
        let path = fresh_path();
        io.set_file_bytes(&path, bytes);
        Self::open_with(io, path)
    }

    fn open_with(io: Arc<MemoryIO>, path: String) -> Result<Engine, EngineError> {
        let db = Database::open_file(io.clone(), &path)
            .map_err(|e| EngineError::Open(e.to_string()))?;
        let conn = db.connect().map_err(|e| EngineError::Open(e.to_string()))?;
        Ok(Engine {
            io,
            path,
            _db: db,
            conn,
        })
    }

    /// Serialize the whole database to a SQLite file image (the `.sqlite` bytes).
    pub fn export(&self) -> Result<Vec<u8>, EngineError> {
        // Fold any WAL frames into the main db file so the image is complete.
        let _ = self.conn.execute("PRAGMA wal_checkpoint(TRUNCATE);");
        let _ = self.conn.cacheflush();
        Ok(self.io.file_bytes(&self.path).unwrap_or_default())
    }

    /// Execute one or more statements with no parameters and no result rows
    /// (DDL, batch setup). Statements are `;`-separated.
    pub fn execute_batch(&self, sql: &str) -> Result<(), EngineError> {
        self.conn.execute(sql)?;
        Ok(())
    }

    /// Run a single row-producing statement and collect all rows.
    pub fn query(&self, sql: &str, params: &[SqlValue]) -> Result<QueryResult, EngineError> {
        let mut stmt = self.conn.prepare(sql)?;
        bind_params(&mut stmt, params);

        let columns: Vec<String> = (0..stmt.num_columns())
            .map(|i| stmt.get_column_name(i).into_owned())
            .collect();

        let raw = stmt.run_collect_rows()?;
        let rows = raw
            .iter()
            .map(|r| r.iter().map(SqlValue::from_core).collect())
            .collect();

        Ok(QueryResult { columns, rows })
    }

    /// Run a single row-less statement (`INSERT`/`UPDATE`/`DELETE`) and report
    /// its side effects. Any rows it happens to produce are discarded.
    pub fn run(&self, sql: &str, params: &[SqlValue]) -> Result<RunResult, EngineError> {
        let mut stmt = self.conn.prepare(sql)?;
        bind_params(&mut stmt, params);
        // Drive to completion; we don't keep the rows.
        let _ = stmt.run_collect_rows()?;
        Ok(RunResult {
            changes: self.conn.changes(),
            last_insert_rowid: self.conn.last_insert_rowid(),
        })
    }
}

/// Bind `params` to a prepared statement positionally (SQLite parameters are
/// 1-indexed).
fn bind_params(stmt: &mut turso_core::Statement, params: &[SqlValue]) {
    for (i, p) in params.iter().enumerate() {
        // i + 1 is always >= 1, so the NonZero construction never fails.
        let idx = NonZeroUsize::new(i + 1).expect("parameter index is 1-based and non-zero");
        stmt.bind_at(idx, p.clone().into_core());
    }
}
