tuible 0.0.2-alpha.1

use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;
use std::sync::RwLock;
use std::sync::atomic::{AtomicBool, Ordering};

use futures_util::TryStreamExt as _;
use libsqlite3_sys::{SQLITE_OK, sqlite3_finalize, sqlite3_prepare_v3, sqlite3_stmt_readonly};
use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
use sqlx::{Column as _, Executor as _, Row as _, SqlSafeStr as _, Statement as _};

use super::error::DbError;
use super::model::{Column, QueryOutcome, Row, SchemaColumn, TablePage, Value};

const ROWID_ALIAS: &str = "__tuible_rowid__";

pub struct SqliteSource {
    pool: SqlitePool,
    read_only: bool,
    interruption: Arc<RwLock<Arc<AtomicBool>>>,
}

impl SqliteSource {
    pub async fn connect(path: &Path) -> Result<Self, DbError> {
        Self::connect_with_mode(path, false).await
    }

    pub async fn connect_read_only(path: &Path) -> Result<Self, DbError> {
        Self::connect_with_mode(path, true).await
    }

    async fn connect_with_mode(path: &Path, read_only: bool) -> Result<Self, DbError> {
        let options = SqliteConnectOptions::new()
            .filename(path)
            .read_only(read_only)
            .create_if_missing(false)
            .foreign_keys(true);
        let interruption = Arc::new(RwLock::new(Arc::new(AtomicBool::new(false))));
        let connection_interruption = Arc::clone(&interruption);
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .after_connect(move |connection, _metadata| {
                let connection_interruption = Arc::clone(&connection_interruption);
                Box::pin(async move {
                    connection
                        .lock_handle()
                        .await?
                        .set_progress_handler(1_000, move || {
                            connection_interruption
                                .read()
                                .is_ok_and(|interrupted| !interrupted.load(Ordering::Acquire))
                        });
                    Ok(())
                })
            })
            .connect_with(options)
            .await?;
        Ok(Self {
            pool,
            read_only,
            interruption,
        })
    }

    pub fn is_read_only(&self) -> bool {
        self.read_only
    }

    pub async fn prepare_operation(&self, interrupted: Arc<AtomicBool>) -> Result<(), DbError> {
        let connection = self.pool.acquire().await?;
        let mut active = self.interruption.write().map_err(|_| {
            DbError::Unsupported("SQLite cancellation state is unavailable".to_string())
        })?;
        *active = interrupted;
        drop(active);
        drop(connection);
        Ok(())
    }

    pub async fn list_tables(&self) -> Result<Vec<String>, DbError> {
        let names: Vec<(String,)> = sqlx::query_as(
            "SELECT name FROM sqlite_master \
             WHERE type = 'table' AND name NOT LIKE 'sqlite_%' \
             ORDER BY name",
        )
        .fetch_all(&self.pool)
        .await?;
        Ok(names.into_iter().map(|(name,)| name).collect())
    }

    pub async fn fetch_rows(
        &self,
        table: &str,
        limit: i64,
        offset: i64,
    ) -> Result<TablePage, DbError> {
        let limit = limit.max(1);
        let offset = offset.max(0);
        if !self.list_tables().await?.iter().any(|t| t == table) {
            return Err(DbError::UnknownTable(table.to_string()));
        }

        let quoted_table = quote_identifier(table);
        // Safe: the identifier is escaped above and pagination values are bound.
        // rowid variant fails on WITHOUT ROWID tables; fall back to plain SELECT.
        if let Some(rowid) = self.rowid_name(table).await? {
            let rowid = quote_identifier(rowid);
            let with_rowid = format!(
                "SELECT {rowid} AS \"{ROWID_ALIAS}\", * FROM {quoted_table} \
                 ORDER BY {rowid} LIMIT ? OFFSET ?"
            );
            if let Ok(mut rows) = sqlx::query(sqlx::AssertSqlSafe(with_rowid))
                .bind(limit.saturating_add(1))
                .bind(offset)
                .fetch_all(&self.pool)
                .await
            {
                let has_more = rows.len() > limit as usize;
                rows.truncate(limit as usize);
                if let Ok(page) = self
                    .build_page(table, &rows, true, offset as usize, has_more)
                    .await
                {
                    return Ok(page);
                }
            }
        }

        let plain = format!("SELECT * FROM {quoted_table} LIMIT ? OFFSET ?");
        let mut rows = sqlx::query(sqlx::AssertSqlSafe(plain))
            .bind(limit.saturating_add(1))
            .bind(offset)
            .fetch_all(&self.pool)
            .await?;
        let has_more = rows.len() > limit as usize;
        rows.truncate(limit as usize);
        self.build_page(table, &rows, false, offset as usize, has_more)
            .await
    }

    async fn build_page(
        &self,
        table: &str,
        rows: &[sqlx::sqlite::SqliteRow],
        has_rowid: bool,
        offset: usize,
        has_more: bool,
    ) -> Result<TablePage, DbError> {
        let skip = usize::from(has_rowid);
        let columns = if let Some(first) = rows.first() {
            first
                .columns()
                .iter()
                .skip(skip)
                .map(|c| Column {
                    name: c.name().to_string(),
                })
                .collect()
        } else {
            self.get_columns(table).await?
        };

        let out_rows = rows
            .iter()
            .map(|row| Row {
                values: (skip..row.columns().len())
                    .map(|idx| decode_value(row, idx))
                    .collect(),
            })
            .collect();

        let rowids = has_rowid
            .then(|| {
                rows.iter()
                    .map(|row| row.try_get(0))
                    .collect::<Result<Vec<i64>, _>>()
            })
            .transpose()?;

        Ok(TablePage {
            columns,
            rows: out_rows,
            rowids,
            offset,
            has_more,
        })
    }

    pub async fn table_schema(&self, table: &str) -> Result<Vec<SchemaColumn>, DbError> {
        if !self.list_tables().await?.iter().any(|t| t == table) {
            return Err(DbError::UnknownTable(table.to_string()));
        }
        let rows: Vec<(String, String, i64, i64)> = sqlx::query_as(
            "SELECT name, type, \"notnull\", pk FROM pragma_table_info(?) ORDER BY cid",
        )
        .bind(table)
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .into_iter()
            .map(|(name, col_type, notnull, pk)| SchemaColumn {
                name,
                col_type,
                notnull: notnull != 0,
                pk: pk != 0,
            })
            .collect())
    }

    pub async fn schema_catalog(&self) -> Result<BTreeMap<String, Vec<SchemaColumn>>, DbError> {
        let rows: Vec<(String, String, String, i64, i64)> = sqlx::query_as(
            "SELECT m.name, p.name, p.type, p.\"notnull\", p.pk \
             FROM sqlite_master AS m \
             JOIN pragma_table_info(m.name) AS p \
             WHERE m.type = 'table' AND m.name NOT LIKE 'sqlite_%' \
             ORDER BY m.name, p.cid",
        )
        .fetch_all(&self.pool)
        .await?;
        let mut catalog = BTreeMap::new();
        for (table, name, col_type, notnull, pk) in rows {
            catalog
                .entry(table)
                .or_insert_with(Vec::new)
                .push(SchemaColumn {
                    name,
                    col_type,
                    notnull: notnull != 0,
                    pk: pk != 0,
                });
        }
        Ok(catalog)
    }

    pub async fn execute_sql(&self, sql: &str, max_rows: usize) -> Result<QueryOutcome, DbError> {
        if has_multiple_statements(sql) {
            return Err(DbError::MultipleStatements);
        }
        let statement = self
            .pool
            .prepare(sqlx::AssertSqlSafe(sql.to_string()).into_sql_str())
            .await?;
        let columns: Vec<Column> = statement
            .columns()
            .iter()
            .map(|column| Column {
                name: column.name().to_string(),
            })
            .collect();

        if !columns.is_empty() {
            let mut rows = Vec::with_capacity(max_rows.min(1_000) + 1);
            let mut stream = statement.query().fetch(&self.pool);
            while rows.len() <= max_rows {
                let Some(row) = stream.try_next().await? else {
                    break;
                };
                rows.push(row);
            }
            drop(stream);
            let truncated = rows.len() > max_rows;
            rows.truncate(max_rows);
            let read_only = self.statement_is_read_only(sql).await;
            Ok(QueryOutcome::Rows {
                columns,
                rows: rows.iter().map(row_to_row).collect(),
                truncated,
                next_token: None,
                read_only,
            })
        } else {
            let result = statement.query().execute(&self.pool).await?;
            Ok(QueryOutcome::Affected(result.rows_affected()))
        }
    }

    async fn statement_is_read_only(&self, sql: &str) -> bool {
        let Ok(length) = i32::try_from(sql.len()) else {
            return false;
        };
        let Ok(mut connection) = self.pool.acquire().await else {
            return false;
        };
        let Ok(mut handle) = connection.lock_handle().await else {
            return false;
        };
        let mut statement = std::ptr::null_mut();
        // SAFETY: the locked SQLx handle guarantees exclusive access to a live SQLite connection;
        // SQLite reads exactly `length` bytes from `sql`, and `statement` is finalized below.
        let status = unsafe {
            sqlite3_prepare_v3(
                handle.as_raw_handle().as_ptr(),
                sql.as_ptr().cast(),
                length,
                0,
                &mut statement,
                std::ptr::null_mut(),
            )
        };
        if status != SQLITE_OK || statement.is_null() {
            return false;
        }
        // SAFETY: `statement` was successfully prepared above and remains valid until finalized.
        let read_only = unsafe { sqlite3_stmt_readonly(statement) != 0 };
        // SAFETY: this is the sole owner of the prepared statement and does not use it afterward.
        unsafe {
            sqlite3_finalize(statement);
        }
        read_only
    }

    pub async fn execute_table_filter(
        &self,
        table: &str,
        sql: &str,
        max_rows: usize,
    ) -> Result<(TablePage, bool), DbError> {
        if has_multiple_statements(sql) {
            return Err(DbError::MultipleStatements);
        }
        let rowid = self.rowid_name(table).await?;
        let (statement, has_rowid) = if let Some(rowid) = rowid {
            let with_rowid = sql.replacen(
                "SELECT *",
                &format!("SELECT {} AS \"{ROWID_ALIAS}\", *", quote_identifier(rowid)),
                1,
            );
            match self
                .pool
                .prepare(sqlx::AssertSqlSafe(with_rowid).into_sql_str())
                .await
            {
                Ok(statement) => (statement, true),
                Err(_) => (
                    self.pool
                        .prepare(sqlx::AssertSqlSafe(sql.to_string()).into_sql_str())
                        .await?,
                    false,
                ),
            }
        } else {
            (
                self.pool
                    .prepare(sqlx::AssertSqlSafe(sql.to_string()).into_sql_str())
                    .await?,
                false,
            )
        };
        let value_start = usize::from(has_rowid);
        let columns = statement
            .columns()
            .iter()
            .skip(value_start)
            .map(|column| Column {
                name: column.name().to_string(),
            })
            .collect();
        let mut raw_rows = Vec::with_capacity(max_rows.min(1_000) + 1);
        let mut stream = statement.query().fetch(&self.pool);
        while raw_rows.len() <= max_rows {
            let Some(row) = stream.try_next().await? else {
                break;
            };
            raw_rows.push(row);
        }
        drop(stream);
        let truncated = raw_rows.len() > max_rows;
        raw_rows.truncate(max_rows);
        let rowids = has_rowid
            .then(|| {
                raw_rows
                    .iter()
                    .map(|row| row.try_get(0))
                    .collect::<Result<Vec<i64>, _>>()
            })
            .transpose()?;
        let rows = raw_rows
            .iter()
            .map(|row| Row {
                values: (value_start..row.len())
                    .map(|index| decode_value(row, index))
                    .collect(),
            })
            .collect();
        Ok((
            TablePage {
                columns,
                rows,
                rowids,
                offset: 0,
                // Filter cursors are intentionally bounded; paging must never fall back to table scans.
                has_more: false,
            },
            truncated,
        ))
    }

    pub async fn update_cell(
        &self,
        table: &str,
        rowid: i64,
        column: &str,
        value: &Value,
    ) -> Result<(), DbError> {
        let schema = self.table_schema(table).await?;
        if !schema.iter().any(|c| c.name == column) {
            return Err(DbError::UnknownColumn(column.to_string()));
        }
        let rowid_column = self
            .rowid_name(table)
            .await?
            .ok_or_else(|| DbError::NoRowId(table.to_string()))?;
        let query = format!(
            "UPDATE {} SET {} = ? WHERE {} = ?",
            quote_identifier(table),
            quote_identifier(column),
            quote_identifier(rowid_column)
        );
        let query = sqlx::query(sqlx::AssertSqlSafe(query));
        let query = match value {
            Value::Null => query.bind(Option::<String>::None),
            Value::Int(value) => query.bind(value),
            Value::Float(value) => query.bind(value),
            Value::Decimal(value) => query.bind(value),
            Value::Text(value) => query.bind(value),
            Value::Bool(value) => query.bind(value),
            Value::Bytes(value) => query.bind(value),
            Value::Json(value) => query.bind(value.to_string()),
        };
        query.bind(rowid).execute(&self.pool).await?;
        Ok(())
    }

    async fn get_columns(&self, table: &str) -> Result<Vec<Column>, DbError> {
        let rows: Vec<(String,)> =
            sqlx::query_as("SELECT name FROM pragma_table_info(?) ORDER BY cid")
                .bind(table)
                .fetch_all(&self.pool)
                .await?;
        Ok(rows.into_iter().map(|(name,)| Column { name }).collect())
    }

    async fn rowid_name(&self, table: &str) -> Result<Option<&'static str>, DbError> {
        let without_rowid: Option<i64> =
            sqlx::query_scalar("SELECT wr FROM pragma_table_list WHERE name = ?")
                .bind(table)
                .fetch_optional(&self.pool)
                .await?;
        if without_rowid == Some(1) {
            return Ok(None);
        }
        let columns = self.get_columns(table).await?;
        Ok(["rowid", "_rowid_", "oid"].into_iter().find(|candidate| {
            columns
                .iter()
                .all(|column| !column.name.eq_ignore_ascii_case(candidate))
        }))
    }
}

fn quote_identifier(identifier: &str) -> String {
    format!("\"{}\"", identifier.replace('"', "\"\""))
}

fn has_multiple_statements(sql: &str) -> bool {
    #[derive(Clone, Copy)]
    enum State {
        Normal,
        SingleQuote,
        DoubleQuote,
        Backtick,
        Bracket,
        LineComment,
        BlockComment,
    }

    let mut state = State::Normal;
    let mut terminated = false;
    let mut chars = sql.chars().peekable();
    while let Some(character) = chars.next() {
        match state {
            State::Normal => {
                if character == '-' && chars.peek() == Some(&'-') {
                    chars.next();
                    state = State::LineComment;
                } else if character == '/' && chars.peek() == Some(&'*') {
                    chars.next();
                    state = State::BlockComment;
                } else if terminated {
                    if !character.is_whitespace() && character != ';' {
                        return true;
                    }
                } else {
                    state = match character {
                        '\'' => State::SingleQuote,
                        '"' => State::DoubleQuote,
                        '`' => State::Backtick,
                        '[' => State::Bracket,
                        ';' => {
                            terminated = true;
                            State::Normal
                        }
                        _ => State::Normal,
                    };
                }
            }
            State::SingleQuote if character == '\'' => {
                if chars.peek() == Some(&'\'') {
                    chars.next();
                } else {
                    state = State::Normal;
                }
            }
            State::DoubleQuote if character == '"' => {
                if chars.peek() == Some(&'"') {
                    chars.next();
                } else {
                    state = State::Normal;
                }
            }
            State::Backtick if character == '`' => state = State::Normal,
            State::Bracket if character == ']' => state = State::Normal,
            State::LineComment if character == '\n' => state = State::Normal,
            State::BlockComment if character == '*' && chars.peek() == Some(&'/') => {
                chars.next();
                state = State::Normal;
            }
            _ => {}
        }
    }
    false
}

fn row_to_row(row: &sqlx::sqlite::SqliteRow) -> Row {
    let values = (0..row.columns().len())
        .map(|idx| decode_value(row, idx))
        .collect();
    Row { values }
}

// Tries decode targets in order and takes the first that succeeds — permissive
// on purpose, since SQLite types are per-value, not per-column.
fn decode_value(row: &sqlx::sqlite::SqliteRow, idx: usize) -> Value {
    if let Ok(v) = row.try_get::<Option<i64>, _>(idx) {
        return v.map(Value::Int).unwrap_or(Value::Null);
    }
    if let Ok(v) = row.try_get::<Option<f64>, _>(idx) {
        return v.map(Value::Float).unwrap_or(Value::Null);
    }
    if let Ok(v) = row.try_get::<Option<String>, _>(idx) {
        return v.map(Value::Text).unwrap_or(Value::Null);
    }
    if let Ok(v) = row.try_get::<Option<Vec<u8>>, _>(idx) {
        return v.map(Value::Bytes).unwrap_or(Value::Null);
    }
    Value::Null
}

/// Creates the demo SQLite database at `path` (if it doesn't already exist)
/// and seeds it with sample data, so `tuible demo` always has something to show.
pub async fn ensure_demo_db(path: &Path) -> Result<(), DbError> {
    if path.exists() {
        // Reseed demo files created before the events table existed.
        let source = SqliteSource::connect(path).await?;
        let current = source.list_tables().await?;
        source.pool.close().await;
        if current.contains(&"events".to_string()) {
            return Ok(());
        }
        std::fs::remove_file(path)?;
    }
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let url = format!("sqlite://{}?mode=rwc", path.display());
    let pool = SqlitePool::connect(&url).await?;
    sqlx::raw_sql(include_str!("../../fixtures/demo_seed.sql"))
        .execute(&pool)
        .await?;
    pool.close().await;
    Ok(())
}

#[cfg(test)]
mod tests {
    #![allow(clippy::expect_used, clippy::unwrap_used)]

    use super::*;
    use tempfile::tempdir;

    #[tokio::test]
    async fn creates_and_seeds_demo_db() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");

        ensure_demo_db(&path).await.unwrap();

        assert!(path.exists());
        let pool = SqlitePool::connect(&format!("sqlite://{}", path.display()))
            .await
            .unwrap();
        let count: (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(count.0, 3);
    }

    #[tokio::test]
    async fn is_idempotent_when_db_already_exists() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");

        ensure_demo_db(&path).await.unwrap();
        ensure_demo_db(&path).await.unwrap();

        assert!(path.exists());
    }

    #[tokio::test]
    async fn reseeds_outdated_demo_db_missing_events_table() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();

        let source = SqliteSource::connect(&path).await.unwrap();
        source.execute_sql("DROP TABLE events", 100).await.unwrap();
        drop(source);

        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();
        assert!(
            source
                .list_tables()
                .await
                .unwrap()
                .contains(&"events".to_string())
        );
    }

    #[tokio::test]
    async fn events_table_has_loose_types_and_json() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let page = source.fetch_rows("events", 10, 0).await.unwrap();

        let score_idx = page.columns.iter().position(|c| c.name == "score").unwrap();
        let scores: Vec<_> = page
            .rows
            .iter()
            .map(|r| r.values[score_idx].clone())
            .collect();
        assert!(scores.iter().any(|v| matches!(v, Value::Float(_))));
        assert!(scores.iter().any(|v| matches!(v, Value::Text(_))));
        assert!(scores.iter().any(|v| matches!(v, Value::Null)));

        let payload_idx = page
            .columns
            .iter()
            .position(|c| c.name == "payload")
            .unwrap();
        assert!(matches!(&page.rows[0].values[payload_idx], Value::Text(s) if s.starts_with('{')));
    }

    #[tokio::test]
    async fn lists_seeded_tables() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();

        let source = SqliteSource::connect(&path).await.unwrap();
        let mut tables = source.list_tables().await.unwrap();
        tables.sort();

        assert_eq!(
            tables,
            vec![
                "authors".to_string(),
                "books".to_string(),
                "events".to_string()
            ]
        );
    }

    #[tokio::test]
    async fn fetch_rows_returns_seeded_authors() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let page = source.fetch_rows("authors", 10, 0).await.unwrap();

        assert!(page.columns.iter().any(|c| c.name == "name"));
        assert_eq!(page.rows.len(), 3);
    }

    #[tokio::test]
    async fn fetch_rows_rejects_unknown_table() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let result = source.fetch_rows("secrets", 10, 0).await;

        assert!(matches!(result, Err(DbError::UnknownTable(_))));
    }

    #[tokio::test]
    async fn fetch_rows_includes_rowids_without_leaking_column() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let page = source.fetch_rows("authors", 10, 0).await.unwrap();

        assert_eq!(page.columns.len(), 3);
        assert_eq!(page.rows[0].values.len(), 3);
        assert_eq!(page.rowids, Some(vec![1, 2, 3]));
    }

    #[tokio::test]
    async fn table_schema_describes_columns() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let schema = source.table_schema("books").await.unwrap();

        let title = schema.iter().find(|c| c.name == "title").unwrap();
        assert_eq!(title.col_type, "TEXT");
        assert!(title.notnull);
        assert!(!title.pk);
        assert!(schema.iter().find(|c| c.name == "id").unwrap().pk);
    }

    #[tokio::test]
    async fn schema_catalog_loads_all_tables_in_one_query() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let catalog = source.schema_catalog().await.unwrap();

        assert_eq!(catalog.len(), 3);
        assert!(
            catalog["authors"]
                .iter()
                .any(|column| column.name == "name")
        );
        assert!(catalog["books"].iter().any(|column| column.name == "title"));
    }

    #[tokio::test]
    async fn execute_sql_returns_rows_for_select() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let outcome = source
            .execute_sql("SELECT name FROM authors WHERE country = 'Poland'", 100)
            .await
            .unwrap();

        match outcome {
            QueryOutcome::Rows {
                columns,
                rows,
                truncated,
                read_only,
                ..
            } => {
                assert_eq!(columns[0].name, "name");
                assert_eq!(rows.len(), 1);
                assert_eq!(rows[0].values[0], Value::Text("Stanislaw Lem".into()));
                assert!(!truncated);
                assert!(read_only);
            }
            QueryOutcome::Affected(_) | QueryOutcome::Executed => panic!("expected rows"),
        }
    }

    #[tokio::test]
    async fn statement_read_only_classification_handles_ctes_pragmas_and_returning_writes() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        assert!(
            source
                .statement_is_read_only("WITH one(n) AS (VALUES (1)) SELECT n FROM one")
                .await
        );
        assert!(
            source
                .statement_is_read_only("PRAGMA table_info(authors)")
                .await
        );
        assert!(
            !source
                .statement_is_read_only("UPDATE authors SET name = name RETURNING id")
                .await
        );
        assert!(
            !source
                .statement_is_read_only("PRAGMA user_version = 7")
                .await
        );
        assert!(
            !source
                .statement_is_read_only("PRAGMA journal_mode = WAL")
                .await
        );
    }

    #[tokio::test]
    async fn filtered_table_queries_preserve_rowids_for_editing() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let (page, truncated) = source
            .execute_table_filter(
                "authors",
                "SELECT * FROM \"authors\" WHERE \"country\" = 'Poland'",
                100,
            )
            .await
            .unwrap();

        assert!(!truncated);
        assert_eq!(page.rows.len(), 1);
        assert_eq!(page.rowids.as_ref().map(Vec::len), Some(1));
        assert_eq!(page.columns[0].name, "id");
    }

    #[tokio::test]
    async fn execute_sql_reports_affected_for_update() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let outcome = source
            .execute_sql("UPDATE books SET year = 1970 WHERE author_id = 1", 100)
            .await
            .unwrap();

        assert!(matches!(outcome, QueryOutcome::Affected(2)));
    }

    #[tokio::test]
    async fn update_cell_persists_new_value() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        source
            .update_cell("authors", 2, "country", &Value::Text("PL".into()))
            .await
            .unwrap();

        let page = source.fetch_rows("authors", 10, 0).await.unwrap();
        assert!(
            page.rows
                .iter()
                .any(|r| r.values.contains(&Value::Text("PL".into())))
        );
    }

    #[tokio::test]
    async fn update_cell_rejects_unknown_column() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let result = source
            .update_cell("authors", 1, "salary", &Value::Int(1))
            .await;

        assert!(matches!(result, Err(DbError::UnknownColumn(_))));
    }

    #[tokio::test]
    async fn execute_sql_preserves_columns_for_empty_results() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let outcome = source
            .execute_sql("SELECT id, name FROM authors WHERE 0", 100)
            .await
            .unwrap();

        match outcome {
            QueryOutcome::Rows {
                columns,
                rows,
                truncated,
                ..
            } => {
                assert_eq!(
                    columns
                        .iter()
                        .map(|column| column.name.as_str())
                        .collect::<Vec<_>>(),
                    ["id", "name"]
                );
                assert!(rows.is_empty());
                assert!(!truncated);
            }
            QueryOutcome::Affected(_) | QueryOutcome::Executed => panic!("expected rows"),
        }
    }

    #[tokio::test]
    async fn execute_sql_bounds_returned_rows() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let outcome = source
            .execute_sql("SELECT id FROM books ORDER BY id", 2)
            .await
            .unwrap();

        assert!(matches!(
            outcome,
            QueryOutcome::Rows { rows, truncated: true, .. } if rows.len() == 2
        ));
    }

    #[tokio::test]
    async fn read_only_connection_rejects_writes() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("demo.sqlite");
        ensure_demo_db(&path).await.unwrap();
        let source = SqliteSource::connect_read_only(&path).await.unwrap();

        let result = source.execute_sql("DELETE FROM books", 100).await;

        assert!(result.is_err());
    }

    #[tokio::test]
    async fn quoted_identifiers_can_be_fetched_and_updated() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("quoted.sqlite");
        std::fs::File::create(&path).unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();
        source
            .execute_sql("CREATE TABLE \"odd\"\"table\" (\"say\"\"hi\" TEXT)", 100)
            .await
            .unwrap();
        source
            .execute_sql("INSERT INTO \"odd\"\"table\" VALUES ('hello')", 100)
            .await
            .unwrap();

        let page = source.fetch_rows("odd\"table", 10, 0).await.unwrap();
        source
            .update_cell(
                "odd\"table",
                page.rowids.as_ref().unwrap()[0],
                "say\"hi",
                &Value::Text("updated".into()),
            )
            .await
            .unwrap();

        let page = source.fetch_rows("odd\"table", 10, 0).await.unwrap();
        assert_eq!(page.rows[0].values[0], Value::Text("updated".into()));
    }

    #[tokio::test]
    async fn shadowed_rowid_uses_an_available_alias() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("shadowed.sqlite");
        std::fs::File::create(&path).unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();
        source
            .execute_sql("CREATE TABLE shadowed (rowid TEXT, value INTEGER)", 100)
            .await
            .unwrap();
        source
            .execute_sql("INSERT INTO shadowed VALUES ('public', 1)", 100)
            .await
            .unwrap();

        let page = source.fetch_rows("shadowed", 10, 0).await.unwrap();
        let hidden_rowid = page.rowids.as_ref().unwrap()[0];
        source
            .update_cell("shadowed", hidden_rowid, "value", &Value::Int(2))
            .await
            .unwrap();

        let page = source.fetch_rows("shadowed", 10, 0).await.unwrap();
        assert_eq!(page.rows[0].values[1], Value::Int(2));
    }

    #[tokio::test]
    async fn without_rowid_tables_are_browsable_but_not_editable() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("without-rowid.sqlite");
        std::fs::File::create(&path).unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();
        source
            .execute_sql(
                "CREATE TABLE keyed (id TEXT PRIMARY KEY, value TEXT) WITHOUT ROWID",
                100,
            )
            .await
            .unwrap();
        source
            .execute_sql("INSERT INTO keyed VALUES ('a', 'hello')", 100)
            .await
            .unwrap();

        let page = source.fetch_rows("keyed", 10, 0).await.unwrap();

        assert_eq!(page.rows.len(), 1);
        assert!(page.rowids.is_none());

        let (filtered, _) = source
            .execute_table_filter("keyed", "SELECT * FROM \"keyed\" WHERE \"id\" = 'a'", 10)
            .await
            .unwrap();
        assert_eq!(filtered.rows.len(), 1);
        assert!(filtered.rowids.is_none());
    }

    #[tokio::test]
    async fn table_pages_report_offset_and_more_rows() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("pages.sqlite");
        std::fs::File::create(&path).unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();
        source
            .execute_sql("CREATE TABLE numbers (value INTEGER)", 100)
            .await
            .unwrap();
        source
            .execute_sql("INSERT INTO numbers VALUES (1), (2), (3)", 100)
            .await
            .unwrap();

        let first = source.fetch_rows("numbers", 2, 0).await.unwrap();
        let second = source.fetch_rows("numbers", 2, 2).await.unwrap();

        assert_eq!(first.offset, 0);
        assert!(first.has_more);
        assert_eq!(second.offset, 2);
        assert!(!second.has_more);
        assert_eq!(second.rows[0].values[0], Value::Int(3));
    }

    #[test]
    fn multiple_statement_detection_ignores_literals_and_comments() {
        assert!(!has_multiple_statements("SELECT ';' AS value; -- trailing"));
        assert!(!has_multiple_statements(
            "SELECT \"semi;colon\"; /* trailing */"
        ));
        assert!(has_multiple_statements("SELECT 1; /* next */ SELECT 2"));
        assert!(has_multiple_statements("SELECT 1; DELETE FROM jobs"));
    }

    #[tokio::test]
    async fn execute_sql_rejects_multiple_statements() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("single-statement.sqlite");
        std::fs::File::create(&path).unwrap();
        let source = SqliteSource::connect(&path).await.unwrap();

        let result = source.execute_sql("SELECT 1; SELECT 2", 10).await;

        assert!(matches!(result, Err(DbError::MultipleStatements)));
    }

    #[tokio::test]
    async fn progress_handler_interrupts_a_running_query() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("interrupt.sqlite");
        std::fs::File::create(&path).unwrap();
        let source = Arc::new(SqliteSource::connect(&path).await.unwrap());
        source.pool.acquire().await.unwrap().close().await.unwrap();
        let interrupted = Arc::new(AtomicBool::new(false));
        source
            .prepare_operation(Arc::clone(&interrupted))
            .await
            .unwrap();
        let query_source = Arc::clone(&source);
        let query = tokio::spawn(async move {
            query_source
                .execute_sql(
                    "WITH RECURSIVE numbers(value) AS (VALUES(1) UNION ALL SELECT value + 1 FROM numbers WHERE value < 100000000) SELECT sum(value) FROM numbers",
                    1,
                )
                .await
        });
        tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        interrupted.store(true, Ordering::Release);

        let result = tokio::time::timeout(std::time::Duration::from_secs(1), query)
            .await
            .unwrap()
            .unwrap();

        assert!(result.is_err());
    }

    #[tokio::test]
    async fn canceled_generation_cannot_interrupt_its_replacement() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("generations.sqlite");
        std::fs::File::create(&path).unwrap();
        let source = Arc::new(SqliteSource::connect(&path).await.unwrap());
        let first_interrupted = Arc::new(AtomicBool::new(false));
        source
            .prepare_operation(Arc::clone(&first_interrupted))
            .await
            .unwrap();
        let first_source = Arc::clone(&source);
        let first = tokio::spawn(async move {
            first_source
                .execute_sql(
                    "WITH RECURSIVE numbers(value) AS (VALUES(1) UNION ALL SELECT value + 1 FROM numbers WHERE value < 100000000) SELECT sum(value) FROM numbers",
                    1,
                )
                .await
        });
        tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        first_interrupted.store(true, Ordering::Release);

        let second_interrupted = Arc::new(AtomicBool::new(false));
        source
            .prepare_operation(Arc::clone(&second_interrupted))
            .await
            .unwrap();
        let second = source.execute_sql("SELECT 42", 1).await.unwrap();
        let first = first.await.unwrap();

        assert!(first.is_err());
        assert!(matches!(second, QueryOutcome::Rows { rows, .. } if rows.len() == 1));
        assert!(first_interrupted.load(Ordering::Acquire));
        assert!(!second_interrupted.load(Ordering::Acquire));
    }
}