drizzle 0.1.12

//! Async `SQLite` driver using [`turso`].
//!
//! # Quick start
//!
//! ```no_run
//! use drizzle::sqlite::turso::Drizzle;
//! use drizzle::sqlite::prelude::*;
//! use turso::Builder;
//!
//! #[SQLiteTable]
//! struct User {
//!     #[column(primary)]
//!     id: i32,
//!     name: String,
//! }
//!
//! #[derive(SQLiteSchema)]
//! struct AppSchema {
//!     user: User,
//! }
//!
//! #[tokio::main]
//! async fn main() -> drizzle::Result<()> {
//!     let db_builder = Builder::new_local(":memory:").build().await?;
//!     let conn = db_builder.connect()?;
//!     let (db, AppSchema { user }) = Drizzle::new(conn, AppSchema::new());
//!     db.create().await?;
//!
//!     // Insert
//!     db.insert(user).values([InsertUser::new("Alice")]).execute().await?;
//!
//!     // Select
//!     let users: Vec<SelectUser> = db.select(()).from(user).all().await?;
//!
//!     Ok(())
//! }
//! ```
//!
//! # Transactions
//!
//! Return `Ok(value)` to commit, `Err(...)` to rollback.
//!
//! ```no_run
//! # use drizzle::sqlite::turso::Drizzle;
//! # use drizzle::sqlite::prelude::*;
//! # use turso::Builder;
//! # #[SQLiteTable] struct User { #[column(primary)] id: i32, name: String }
//! # #[derive(SQLiteSchema)] struct S { user: User }
//! # #[tokio::main] async fn main() -> drizzle::Result<()> {
//! # let db_builder = Builder::new_local(":memory:").build().await?;
//! # let conn = db_builder.connect()?;
//! # let (mut db, S { user, .. }) = Drizzle::new(conn, S::new());
//! use drizzle::sqlite::connection::SQLiteTransactionType;
//!
//! let count = db.transaction(SQLiteTransactionType::Deferred, async |tx| {
//!     tx.insert(user).values([InsertUser::new("Alice")]).execute().await?;
//!     let users: Vec<SelectUser> = tx.select(()).from(user).all().await?;
//!     Ok(users.len())
//! }).await?;
//! # Ok(()) }
//! ```
//!
//! # Savepoints
//!
//! Savepoints nest inside transactions — a failed savepoint rolls back
//! without aborting the outer transaction.
//!
//! ```no_run
//! # use drizzle::sqlite::turso::Drizzle;
//! # use drizzle::sqlite::prelude::*;
//! # use drizzle::sqlite::connection::SQLiteTransactionType;
//! # use turso::Builder;
//! # #[SQLiteTable] struct User { #[column(primary)] id: i32, name: String }
//! # #[derive(SQLiteSchema)] struct S { user: User }
//! # #[tokio::main] async fn main() -> drizzle::Result<()> {
//! # let db_builder = Builder::new_local(":memory:").build().await?;
//! # let conn = db_builder.connect()?;
//! # let (mut db, S { user, .. }) = Drizzle::new(conn, S::new());
//! db.transaction(SQLiteTransactionType::Deferred, async |tx| {
//!     tx.insert(user).values([InsertUser::new("Alice")]).execute().await?;
//!
//!     // This savepoint fails — only its changes roll back
//!     let _: Result<(), _> = tx.savepoint(async |stx| {
//!         stx.insert(user).values([InsertUser::new("Bad")]).execute().await?;
//!         Err(drizzle::error::DrizzleError::Other("oops".into()))
//!     }).await;
//!
//!     let users: Vec<SelectUser> = tx.select(()).from(user).all().await?;
//!     assert_eq!(users.len(), 1); // only Alice
//!     Ok(())
//! }).await?;
//! # Ok(()) }
//! ```
//!
//! # Cloning for `tokio::spawn`
//!
//! `Drizzle` is cheaply cloneable — the underlying connection is shared.
//! Move a clone into spawned tasks for concurrent queries.
//!
//! ```no_run
//! # use drizzle::sqlite::turso::Drizzle;
//! # use drizzle::sqlite::prelude::*;
//! # use turso::Builder;
//! # #[SQLiteTable] struct User { #[column(primary)] id: i32, name: String }
//! # #[derive(SQLiteSchema)] struct S { user: User }
//! # #[tokio::main] async fn main() -> drizzle::Result<()> {
//! # let db_builder = Builder::new_local(":memory:").build().await?;
//! # let conn = db_builder.connect()?;
//! # let (db, S { user, .. }) = Drizzle::new(conn, S::new());
//! let db_clone = db.clone();
//! tokio::spawn(async move {
//!     db_clone
//!         .insert(user)
//!         .values([InsertUser::new("Bob")])
//!         .execute()
//!         .await
//!         .expect("insert from task");
//! }).await.unwrap();
//! # Ok(()) }
//! ```

mod prepared;

use drizzle_core::error::{DrizzleError, QueryContext, ResultExt};
use drizzle_core::prepared::prepare_render;
use drizzle_core::traits::ToSQL;
use turso::{Connection, IntoValue, Row};

#[cfg(feature = "sqlite")]
use drizzle_sqlite::{
    builder::{self, QueryBuilder},
    connection::SQLiteTransactionType,
    values::SQLiteValue,
};

crate::drizzle_prepare_impl!();
use crate::builder::sqlite::common;
use crate::builder::sqlite::rows::TursoRows as Rows;
use crate::transaction::sqlite::turso::Transaction;

pub type Drizzle<Schema = ()> = common::Drizzle<Connection, Schema>;
pub type DrizzleBuilder<'a, Schema, Builder, State> =
    common::DrizzleBuilder<'a, common::Drizzle<Connection, Schema>, Schema, Builder, State>;

async fn turso_execute_cached(
    conn: &Connection,
    sql: &str,
    params: Vec<turso::Value>,
) -> turso::Result<u64> {
    let mut stmt = conn.prepare_cached(sql).await?;
    stmt.execute(params).await
}

async fn turso_query_cached(
    conn: &Connection,
    sql: &str,
    params: Vec<turso::Value>,
) -> turso::Result<turso::Rows> {
    let mut stmt = conn.prepare_cached(sql).await?;
    stmt.query(params).await
}

impl<Schema> common::Drizzle<Connection, Schema> {
    pub async fn execute<'a, T>(
        &'a self,
        query: T,
    ) -> Result<u64, drizzle_core::error::DrizzleError>
    where
        T: ToSQL<'a, SQLiteValue<'a>>,
    {
        let query = query.to_sql();
        let (sql_str, params) = query.build();
        let driver_params: Vec<turso::Value> = params
            .iter()
            .copied()
            .map(|p| {
                p.into_value()
                    .map_err(drizzle_core::error::DrizzleError::from)
            })
            .collect::<Result<Vec<_>, _>>()
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        turso_execute_cached(&self.conn, &sql_str, driver_params)
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))
    }

    /// Runs the query and returns all matching rows (for SELECT queries)
    pub async fn all<'a, T, R, C>(&'a self, query: T) -> drizzle_core::error::Result<C>
    where
        R: for<'r> TryFrom<&'r Row>,
        for<'r> <R as TryFrom<&'r Row>>::Error: Into<DrizzleError>,
        T: ToSQL<'a, SQLiteValue<'a>>,
        C: Default + Extend<R>,
    {
        let sql = query.to_sql();
        let (sql_str, params) = sql.build();
        let driver_params: Vec<turso::Value> = params
            .iter()
            .copied()
            .map(|p| p.into_value().map_err(DrizzleError::from))
            .collect::<Result<Vec<_>, _>>()
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        let mut rows = turso_query_cached(&self.conn, &sql_str, driver_params)
            .await
            .map_err(DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        let mut out = C::default();
        while let Some(row) = rows
            .next()
            .await
            .map_err(DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?
        {
            out.extend(core::iter::once(R::try_from(&row).map_err(Into::into)?));
        }
        Ok(out)
    }

    /// Runs the query and returns a row cursor.
    pub async fn rows<'a, T, R>(&'a self, query: T) -> drizzle_core::error::Result<Rows<R>>
    where
        R: for<'r> TryFrom<&'r Row>,
        for<'r> <R as TryFrom<&'r Row>>::Error: Into<DrizzleError>,
        T: ToSQL<'a, SQLiteValue<'a>>,
    {
        let sql = query.to_sql();
        let (sql_str, params) = sql.build();
        let driver_params: Vec<turso::Value> = params
            .iter()
            .copied()
            .map(|p| p.into_value().map_err(DrizzleError::from))
            .collect::<Result<Vec<_>, _>>()
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        let rows = self
            .conn
            .query(&sql_str, driver_params)
            .await
            .map_err(DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        Ok(Rows::new(rows))
    }

    /// Runs the query and returns a single row (for SELECT queries)
    pub async fn get<'a, T, R>(&'a self, query: T) -> drizzle_core::error::Result<R>
    where
        R: for<'r> TryFrom<&'r Row>,
        for<'r> <R as TryFrom<&'r Row>>::Error: Into<DrizzleError>,
        T: ToSQL<'a, SQLiteValue<'a>>,
    {
        let sql = query.to_sql();
        let (sql_str, params) = sql.build();
        let driver_params: Vec<turso::Value> = params
            .iter()
            .copied()
            .map(|p| p.into_value().map_err(DrizzleError::from))
            .collect::<Result<Vec<_>, _>>()
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        let mut rows = turso_query_cached(&self.conn, &sql_str, driver_params)
            .await
            .map_err(DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        rows.next()
            .await
            .map_err(DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?
            .map_or_else(
                || Err(DrizzleError::NotFound),
                |row| R::try_from(&row).map_err(Into::into),
            )
    }

    /// Executes a transaction with the given callback.
    ///
    /// The transaction is committed when the callback returns `Ok` and
    /// rolled back on `Err`.
    ///
    /// ```no_run
    /// # use drizzle::sqlite::turso::Drizzle;
    /// # use drizzle::sqlite::prelude::*;
    /// # use drizzle::sqlite::connection::SQLiteTransactionType;
    /// # use turso::Builder;
    /// # #[SQLiteTable] struct User { #[column(primary)] id: i32, name: String }
    /// # #[derive(SQLiteSchema)] struct S { user: User }
    /// # #[tokio::main] async fn main() -> drizzle::Result<()> {
    /// # let db_builder = Builder::new_local(":memory:").build().await?;
    /// # let conn = db_builder.connect()?;
    /// # let (mut db, S { user, .. }) = Drizzle::new(conn, S::new());
    /// let count = db.transaction(SQLiteTransactionType::Deferred, async |tx| {
    ///     tx.insert(user).values([InsertUser::new("Alice")]).execute().await?;
    ///     let users: Vec<SelectUser> = tx.select(()).from(user).all().await?;
    ///     Ok(users.len())
    /// }).await?;
    /// # Ok(()) }
    /// ```
    pub async fn transaction<F, R>(
        &mut self,
        tx_type: SQLiteTransactionType,
        f: F,
    ) -> drizzle_core::error::Result<R>
    where
        Schema: Copy,
        F: AsyncFnOnce(&Transaction<Schema>) -> drizzle_core::error::Result<R>,
    {
        let tx = self.conn.transaction_with_behavior(tx_type.into()).await?;
        let transaction = Transaction::new(tx, tx_type, self.schema);

        match f(&transaction).await {
            Ok(result) => {
                transaction.commit().await?;
                Ok(result)
            }
            Err(e) => {
                let _ = transaction.rollback().await;
                Err(e)
            }
        }
    }
}

impl<Schema> Drizzle<Schema>
where
    Schema: drizzle_core::traits::SQLSchemaImpl + Default,
{
    /// Create schema objects from `SQLSchemaImpl`.
    pub async fn create(&self) -> drizzle_core::error::Result<()> {
        let schema = Schema::default();
        let statements = schema.create_statements()?;
        for sql in statements {
            self.conn.execute(&sql, ()).await?;
        }
        Ok(())
    }
}

impl<Schema> common::Drizzle<Connection, Schema> {
    /// Apply pending migrations from an embedded migration slice.
    ///
    /// Creates the migrations table if needed and runs pending migrations in a transaction.
    pub async fn migrate(
        &mut self,
        migrations: &[drizzle_migrations::Migration],
        tracking: drizzle_migrations::Tracking,
    ) -> drizzle_core::error::Result<drizzle_migrations::MigrateOutcome> {
        let set = drizzle_migrations::Migrations::with_tracking(
            migrations.to_vec(),
            drizzle_types::Dialect::SQLite,
            tracking,
        );

        ensure_sqlite_migration_table(&mut self.conn, &set).await?;
        let mut rows = self
            .conn
            .query(&set.applied_names_sql(), ())
            .await
            .map_err(DrizzleError::from)?;

        let mut applied_names: Vec<String> = Vec::new();
        while let Some(row) = rows.next().await.map_err(DrizzleError::from)? {
            if let Ok(name) = row.get::<String>(0) {
                applied_names.push(name);
            }
        }

        let pending: Vec<_> = set.pending(&applied_names).collect();

        if pending.is_empty() {
            return Ok(drizzle_migrations::MigrateOutcome::UpToDate);
        }

        let tx = self.conn.transaction().await.map_err(DrizzleError::from)?;

        let mut applied = Vec::with_capacity(pending.len());
        for migration in &pending {
            for stmt in migration.statements() {
                if !stmt.trim().is_empty() {
                    tx.execute(stmt, ()).await.map_err(DrizzleError::from)?;
                }
            }
            tx.execute(&set.record_migration_sql(migration), ())
                .await
                .map_err(DrizzleError::from)?;
            applied.push(migration.tag().to_string());
        }

        tx.commit().await.map_err(DrizzleError::from)?;

        Ok(drizzle_migrations::MigrateOutcome::Applied { tags: applied })
    }
}

async fn migration_table_has_name_column(
    conn: &turso::Connection,
    set: &drizzle_migrations::Migrations,
) -> drizzle_core::error::Result<bool> {
    let table_name = set.table_name().replace('\'', "''");
    let pragma_sql = format!("SELECT name FROM pragma_table_info('{table_name}')");
    let mut rows = conn
        .query(&pragma_sql, ())
        .await
        .map_err(DrizzleError::from)?;
    while let Some(row) = rows.next().await.map_err(DrizzleError::from)? {
        if let Ok(name) = row.get::<String>(0)
            && name == "name"
        {
            return Ok(true);
        }
    }
    Ok(false)
}

async fn load_legacy_applied_migrations(
    conn: &turso::Connection,
    set: &drizzle_migrations::Migrations,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::AppliedMigrationMetadata>> {
    let mut rows = conn
        .query(
            &format!(
                "SELECT id, hash, created_at FROM {} ORDER BY id ASC",
                set.table_ident_sql()
            ),
            (),
        )
        .await
        .map_err(DrizzleError::from)?;
    let mut applied = Vec::new();
    while let Some(row) = rows.next().await.map_err(DrizzleError::from)? {
        applied.push(drizzle_migrations::AppliedMigrationMetadata {
            id: row.get::<Option<i64>>(0).ok().flatten(),
            hash: row.get::<String>(1).map_err(DrizzleError::from)?,
            created_at: row.get::<i64>(2).map_err(DrizzleError::from)?,
        });
    }
    Ok(applied)
}

async fn backfill_migration_name_column(
    conn: &mut turso::Connection,
    set: &drizzle_migrations::Migrations,
    matched: Vec<drizzle_migrations::MatchedMigrationMetadata>,
) -> drizzle_core::error::Result<()> {
    let tx = conn.transaction().await.map_err(DrizzleError::from)?;
    tx.execute(
        &format!(
            "ALTER TABLE {} ADD COLUMN \"name\" text",
            set.table_ident_sql()
        ),
        (),
    )
    .await
    .map_err(DrizzleError::from)?;
    tx.execute(
        &format!(
            "ALTER TABLE {} ADD COLUMN \"applied_at\" TEXT",
            set.table_ident_sql()
        ),
        (),
    )
    .await
    .map_err(DrizzleError::from)?;

    for row in matched {
        let escaped_name = row.name.replace('\'', "''");
        let where_clause = if let Some(id) = row.id {
            format!("\"id\" = {id}")
        } else {
            format!(
                "\"created_at\" = {} AND \"hash\" = '{}'",
                row.created_at,
                row.hash.replace('\'', "''")
            )
        };
        tx.execute(
            &format!(
                "UPDATE {} SET \"name\" = '{}', \"applied_at\" = NULL WHERE {}",
                set.table_ident_sql(),
                escaped_name,
                where_clause
            ),
            (),
        )
        .await
        .map_err(DrizzleError::from)?;
    }

    tx.commit().await.map_err(DrizzleError::from)?;
    Ok(())
}

async fn ensure_sqlite_migration_table(
    conn: &mut turso::Connection,
    set: &drizzle_migrations::Migrations,
) -> drizzle_core::error::Result<()> {
    conn.execute(&set.create_table_sql(), ())
        .await
        .map_err(DrizzleError::from)?;

    if migration_table_has_name_column(conn, set).await? {
        return Ok(());
    }

    let applied = load_legacy_applied_migrations(conn, set).await?;
    let matched = drizzle_migrations::match_applied_migration_metadata(set.all(), &applied)
        .map_err(|e| DrizzleError::Other(e.to_string().into()))?;

    backfill_migration_name_column(conn, set, matched).await
}

async fn turso_introspect_query_tables(
    conn: &turso::Connection,
) -> drizzle_core::error::Result<Vec<(String, Option<String>)>> {
    use drizzle_migrations::sqlite::introspect::queries;
    let err = DrizzleError::from;

    let mut tables_rows = conn.query(queries::TABLES_QUERY, ()).await.map_err(err)?;
    let mut tables: Vec<(String, Option<String>)> = Vec::new();
    while let Some(row) = tables_rows.next().await.map_err(err)? {
        let name: String = row.get(0).unwrap_or_default();
        let sql: Option<String> = row.get(1).ok();
        tables.push((name, sql));
    }
    Ok(tables)
}

async fn turso_introspect_query_columns(
    conn: &turso::Connection,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::sqlite::introspect::RawColumnInfo>> {
    use drizzle_migrations::sqlite::introspect::{RawColumnInfo, queries};
    let err = DrizzleError::from;

    let mut columns_rows = conn.query(queries::COLUMNS_QUERY, ()).await.map_err(err)?;
    let mut raw_columns: Vec<RawColumnInfo> = Vec::new();
    while let Some(row) = columns_rows.next().await.map_err(err)? {
        raw_columns.push(RawColumnInfo {
            table: row.get(0).unwrap_or_default(),
            cid: row.get(1).unwrap_or(0),
            name: row.get(2).unwrap_or_default(),
            column_type: row.get(3).unwrap_or_default(),
            not_null: row.get::<i32>(4).unwrap_or(0) != 0,
            default_value: row.get(5).ok(),
            pk: row.get(6).unwrap_or(0),
            hidden: row.get(7).unwrap_or(0),
            sql: row.get(8).ok(),
        });
    }
    Ok(raw_columns)
}

async fn turso_introspect_query_indexes_and_fks(
    conn: &turso::Connection,
    tables: &[(String, Option<String>)],
) -> (
    Vec<drizzle_migrations::sqlite::introspect::RawIndexInfo>,
    Vec<drizzle_migrations::sqlite::introspect::RawIndexColumn>,
    Vec<drizzle_migrations::sqlite::introspect::RawForeignKey>,
) {
    use drizzle_migrations::sqlite::introspect::{
        RawForeignKey, RawIndexColumn, RawIndexInfo, queries,
    };

    let mut all_indexes: Vec<RawIndexInfo> = Vec::new();
    let mut all_index_columns: Vec<RawIndexColumn> = Vec::new();
    let mut all_fks: Vec<RawForeignKey> = Vec::new();

    for (table_name, _) in tables {
        if let Ok(mut idx_rows) = conn.query(&queries::indexes_query(table_name), ()).await {
            while let Ok(Some(row)) = idx_rows.next().await {
                let idx = RawIndexInfo {
                    table: table_name.clone(),
                    name: row.get(1).unwrap_or_default(),
                    unique: row.get::<i32>(2).unwrap_or(0) != 0,
                    origin: row.get(3).unwrap_or_default(),
                    partial: row.get::<i32>(4).unwrap_or(0) != 0,
                };

                if let Ok(mut col_rows) =
                    conn.query(&queries::index_info_query(&idx.name), ()).await
                {
                    while let Ok(Some(col_row)) = col_rows.next().await {
                        all_index_columns.push(RawIndexColumn {
                            index_name: idx.name.clone(),
                            seqno: col_row.get(0).unwrap_or(0),
                            cid: col_row.get(1).unwrap_or(0),
                            name: col_row.get(2).ok(),
                            desc: col_row.get::<i32>(3).unwrap_or(0) != 0,
                            coll: col_row.get(4).unwrap_or_default(),
                            key: col_row.get::<i32>(5).unwrap_or(0) != 0,
                        });
                    }
                }

                all_indexes.push(idx);
            }
        }

        if let Ok(mut fk_rows) = conn
            .query(&queries::foreign_keys_query(table_name), ())
            .await
        {
            while let Ok(Some(row)) = fk_rows.next().await {
                all_fks.push(RawForeignKey {
                    table: table_name.clone(),
                    id: row.get(0).unwrap_or(0),
                    seq: row.get(1).unwrap_or(0),
                    to_table: row.get(2).unwrap_or_default(),
                    from_column: row.get(3).unwrap_or_default(),
                    to_column: row.get(4).unwrap_or_default(),
                    on_update: row.get(5).unwrap_or_default(),
                    on_delete: row.get(6).unwrap_or_default(),
                    r#match: row.get(7).unwrap_or_default(),
                });
            }
        }
    }

    (all_indexes, all_index_columns, all_fks)
}

async fn turso_introspect_query_views(
    conn: &turso::Connection,
) -> Vec<drizzle_migrations::sqlite::introspect::RawViewInfo> {
    use drizzle_migrations::sqlite::introspect::{RawViewInfo, queries};

    let mut all_views: Vec<RawViewInfo> = Vec::new();
    if let Ok(mut views_rows) = conn.query(queries::VIEWS_QUERY, ()).await {
        while let Ok(Some(row)) = views_rows.next().await {
            let name: String = row.get(0).unwrap_or_default();
            let sql: String = row.get(1).unwrap_or_default();
            all_views.push(RawViewInfo { name, sql });
        }
    }
    all_views
}

fn turso_introspect_build_ddl(
    tables: &[(String, Option<String>)],
    raw_columns: &[drizzle_migrations::sqlite::introspect::RawColumnInfo],
    all_indexes: &[drizzle_migrations::sqlite::introspect::RawIndexInfo],
    all_index_columns: &[drizzle_migrations::sqlite::introspect::RawIndexColumn],
    all_fks: &[drizzle_migrations::sqlite::introspect::RawForeignKey],
    all_views: Vec<drizzle_migrations::sqlite::introspect::RawViewInfo>,
) -> drizzle_migrations::sqlite::SQLiteDDL {
    use drizzle_migrations::sqlite::{
        SQLiteDDL, Table as SqliteTable, View,
        introspect::{
            parse_generated_columns_from_table_sql, parse_view_sql, process_columns,
            process_foreign_keys, process_indexes, process_unique_constraints_from_indexes,
        },
    };
    use std::collections::{HashMap, HashSet};

    let table_sql_map: HashMap<String, String> = tables
        .iter()
        .filter_map(|(name, sql)| sql.as_ref().map(|s| (name.clone(), s.clone())))
        .collect();

    let mut generated_columns: HashMap<String, drizzle_migrations::sqlite::ddl::ParsedGenerated> =
        HashMap::new();
    for (table, sql) in &table_sql_map {
        generated_columns.extend(parse_generated_columns_from_table_sql(table, sql));
    }
    let pk_columns: HashSet<(String, String)> = raw_columns
        .iter()
        .filter(|c| c.pk > 0)
        .map(|c| (c.table.clone(), c.name.clone()))
        .collect();

    let (columns, primary_keys) = process_columns(raw_columns, &generated_columns, &pk_columns);
    let indexes = process_indexes(all_indexes, all_index_columns, &table_sql_map);
    let foreign_keys = process_foreign_keys(all_fks);
    let uniques = process_unique_constraints_from_indexes(all_indexes, all_index_columns);

    let mut ddl = SQLiteDDL::new();
    for (table_name, table_sql) in tables {
        let mut table = SqliteTable::new(table_name.clone());
        if let Some(sql) = table_sql {
            let sql_upper = sql.to_uppercase();
            table.strict = sql_upper.contains(" STRICT");
            table.without_rowid = sql_upper.contains("WITHOUT ROWID");
        }
        ddl.tables.push(table);
    }
    for col in columns {
        ddl.columns.push(col);
    }
    for idx in indexes {
        ddl.indexes.push(idx);
    }
    for fk in foreign_keys {
        ddl.fks.push(fk);
    }
    for pk in primary_keys {
        ddl.pks.push(pk);
    }
    for u in uniques {
        ddl.uniques.push(u);
    }
    for v in all_views {
        let mut view = View::new(v.name);
        if let Some(def) = parse_view_sql(&v.sql) {
            view.definition = Some(def.into());
        }
        ddl.views.push(view);
    }
    ddl
}

impl<Schema> common::Drizzle<Connection, Schema> {
    /// Introspect the live database and return a [`Snapshot`] of its current schema.
    pub async fn introspect(
        &self,
    ) -> drizzle_core::error::Result<drizzle_migrations::schema::Snapshot> {
        let tables = turso_introspect_query_tables(&self.conn).await?;
        let raw_columns = turso_introspect_query_columns(&self.conn).await?;
        let (all_indexes, all_index_columns, all_fks) =
            turso_introspect_query_indexes_and_fks(&self.conn, &tables).await;
        let all_views = turso_introspect_query_views(&self.conn).await;

        let ddl = turso_introspect_build_ddl(
            &tables,
            &raw_columns,
            &all_indexes,
            &all_index_columns,
            &all_fks,
            all_views,
        );

        let mut snapshot = drizzle_migrations::sqlite::SQLiteSnapshot::new();
        for entity in ddl.to_entities() {
            snapshot.add_entity(entity);
        }

        Ok(drizzle_migrations::schema::Snapshot::Sqlite(snapshot))
    }

    /// Introspect the live database, diff against the desired schema, and
    /// execute the SQL statements needed to bring the database in sync.
    ///
    /// This is a no-op if the database already matches.
    pub async fn push<S: drizzle_migrations::Schema>(
        &self,
        schema: &S,
    ) -> drizzle_core::error::Result<()> {
        let live = self.introspect().await?;
        let desired = schema.to_snapshot();
        let generated = drizzle_migrations::diff(&live, &desired)
            .map_err(|e| DrizzleError::Other(e.to_string().into()))?;
        for stmt in generated.statements {
            if !stmt.trim().is_empty() {
                self.conn
                    .execute(&stmt, ())
                    .await
                    .map_err(DrizzleError::from)?;
            }
        }
        Ok(())
    }
}

// =============================================================================
// Query API: find_many / find_first
// =============================================================================

#[cfg(feature = "query")]
use drizzle_core::query::DeserializeStore as _;
#[cfg(feature = "query")]
use drizzle_core::query::FromJsonObject as _;

// AllColumns: read base from individual row columns via TryFrom<Row>
#[cfg(feature = "query")]
impl<'a, Schema, T, Rels, Cl>
    common::DrizzleQueryBuilder<
        '_,
        'a,
        Connection,
        Schema,
        T,
        Rels,
        drizzle_core::query::AllColumns,
        Cl,
    >
{
    /// Executes the query and returns all matching rows with their relations.
    pub async fn find_many(
        self,
    ) -> drizzle_core::error::Result<
        Vec<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::Select,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::Select: for<'r> TryFrom<&'r ::turso::Row>,
        for<'r> <<T as drizzle_core::query::QueryTable>::Select as TryFrom<&'r ::turso::Row>>::Error:
            Into<drizzle_core::error::DrizzleError>,
        Rels: drizzle_core::query::BuildStore
            + drizzle_core::query::RenderRelations<'a, SQLiteValue<'a>>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        let num_base_cols = T::COLUMN_NAMES.len();

        let builder = self.builder;
        let mut rendered = Vec::new();
        builder.relations.render_into(&mut rendered);
        let query_sql = drizzle_core::query::build_query_sql(
            T::TABLE_NAME,
            T::COLUMN_NAMES,
            T::BLOB_COLUMNS,
            rendered,
            builder.where_sql,
            builder.order_by_sql,
            builder.limit,
            builder.offset,
            false,
        );
        let (sql, bind_params) = query_sql.build();

        let params: Vec<turso::Value> = bind_params
            .iter()
            .copied()
            .map(std::convert::Into::into)
            .collect();
        let mut raw_rows = turso_query_cached(&self.runner.conn, &sql, params)
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql, &bind_params))?;
        let mut results = Vec::new();

        while let Some(row) = raw_rows
            .next()
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql, &bind_params))?
        {
            let base = <T as drizzle_core::query::QueryTable>::Select::try_from(&row)
                .map_err(Into::into)?;

            let mut rel_col = num_base_cols;
            let mut next_rel = || {
                let json = row
                    .get::<Option<String>>(rel_col)
                    .map_err(drizzle_core::error::DrizzleError::from)?;
                rel_col += 1;
                Ok(json)
            };
            let store =
                <Rels as drizzle_core::query::BuildStore>::Store::from_json_columns(&mut next_rel)?;

            results.push(drizzle_core::query::QueryRow::new(base, store));
        }

        Ok(results)
    }
}

// AllColumns find_first: requires no LIMIT set yet (internally adds LIMIT 1)
#[cfg(feature = "query")]
impl<'a, Schema, T, Rels, W, Ord>
    common::DrizzleQueryBuilder<
        '_,
        'a,
        Connection,
        Schema,
        T,
        Rels,
        drizzle_core::query::AllColumns,
        drizzle_core::query::Clauses<W, Ord, drizzle_core::query::NoLimit>,
    >
{
    /// Executes the query and returns the first matching row, or `None`.
    pub async fn find_first(
        self,
    ) -> drizzle_core::error::Result<
        Option<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::Select,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::Select: for<'r> TryFrom<&'r ::turso::Row>,
        for<'r> <<T as drizzle_core::query::QueryTable>::Select as TryFrom<&'r ::turso::Row>>::Error:
            Into<drizzle_core::error::DrizzleError>,
        Rels: drizzle_core::query::BuildStore
            + drizzle_core::query::RenderRelations<'a, SQLiteValue<'a>>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        Ok(self.limit(1).find_many().await?.into_iter().next())
    }
}

// PartialColumns: read base from a single JSON "__base" column via FromJsonObject
#[cfg(feature = "query")]
impl<'a, Schema, T, Rels, Cl>
    common::DrizzleQueryBuilder<
        '_,
        'a,
        Connection,
        Schema,
        T,
        Rels,
        drizzle_core::query::PartialColumns,
        Cl,
    >
{
    /// Executes the query and returns all matching rows with their relations.
    ///
    /// Base columns are deserialized from a JSON `"__base"` column.
    pub async fn find_many(
        self,
    ) -> drizzle_core::error::Result<
        Vec<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildStore
            + drizzle_core::query::RenderRelations<'a, SQLiteValue<'a>>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        let builder = self.builder;
        let column_names = &builder.cols.columns;
        let mut rendered = Vec::new();
        builder.relations.render_into(&mut rendered);
        let col_refs: Vec<&str> = column_names.clone();
        let query_sql = drizzle_core::query::build_query_sql(
            T::TABLE_NAME,
            &col_refs,
            T::BLOB_COLUMNS,
            rendered,
            builder.where_sql,
            builder.order_by_sql,
            builder.limit,
            builder.offset,
            true,
        );
        let (sql, bind_params) = query_sql.build();

        let params: Vec<turso::Value> = bind_params
            .iter()
            .copied()
            .map(std::convert::Into::into)
            .collect();
        let mut raw_rows = turso_query_cached(&self.runner.conn, &sql, params)
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql, &bind_params))?;
        let mut results = Vec::new();

        while let Some(row) = raw_rows
            .next()
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql, &bind_params))?
        {
            // Column 0 is the JSON "__base" object
            let base_json: String = row
                .get::<String>(0)
                .map_err(drizzle_core::error::DrizzleError::from)?;
            let base = <T as drizzle_core::query::QueryTable>::PartialSelect::from_json_str(
                &base_json, "base",
            )?;

            let mut rel_col = 1usize;
            let mut next_rel = || {
                let json = row.get::<String>(rel_col).ok();
                rel_col += 1;
                Ok(json)
            };
            let store =
                <Rels as drizzle_core::query::BuildStore>::Store::from_json_columns(&mut next_rel)?;

            results.push(drizzle_core::query::QueryRow::new(base, store));
        }

        Ok(results)
    }
}

// PartialColumns find_first: requires no LIMIT set yet
#[cfg(feature = "query")]
impl<'a, Schema, T, Rels, W, Ord>
    common::DrizzleQueryBuilder<
        '_,
        'a,
        Connection,
        Schema,
        T,
        Rels,
        drizzle_core::query::PartialColumns,
        drizzle_core::query::Clauses<W, Ord, drizzle_core::query::NoLimit>,
    >
{
    /// Executes the query and returns the first matching row, or `None`.
    pub async fn find_first(
        self,
    ) -> drizzle_core::error::Result<
        Option<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildStore
            + drizzle_core::query::RenderRelations<'a, SQLiteValue<'a>>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        Ok(self.limit(1).find_many().await?.into_iter().next())
    }
}

#[cfg(feature = "query")]
impl<'a, T, Rels>
    common::DrizzlePreparedQuery<'a, Connection, T, Rels, drizzle_core::query::AllColumns>
{
    /// Executes the prepared relational query and returns all matching rows.
    pub async fn find_many<const N: usize>(
        &self,
        conn: &impl prepared::TursoExecutor,
        params: [drizzle_core::param::ParamBind<'a, SQLiteValue<'a>>; N],
    ) -> drizzle_core::error::Result<
        Vec<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::Select,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::Select: for<'r> TryFrom<&'r ::turso::Row>,
        for<'r> <<T as drizzle_core::query::QueryTable>::Select as TryFrom<&'r ::turso::Row>>::Error:
            Into<drizzle_core::error::DrizzleError>,
        Rels: drizzle_core::query::BuildStore,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        debug_assert_eq!(
            N,
            self.inner.external_param_count(),
            "parameter count mismatch: expected {} params but got {}",
            self.inner.external_param_count(),
            N
        );

        let num_base_cols = T::COLUMN_NAMES.len();
        let (sql_str, params) = self.inner.bind(params)?;
        let mut driver_params = Vec::with_capacity(self.inner.params.len());
        driver_params.extend(params.map(Into::into));
        let mut raw_rows = conn.fetch(sql_str, driver_params).await?;
        let mut results = Vec::new();

        while let Some(row) = raw_rows.next().await? {
            let base = <T as drizzle_core::query::QueryTable>::Select::try_from(&row)
                .map_err(Into::into)?;

            let mut rel_col = num_base_cols;
            let mut next_rel = || {
                let json = row
                    .get::<Option<String>>(rel_col)
                    .map_err(drizzle_core::error::DrizzleError::from)?;
                rel_col += 1;
                Ok(json)
            };
            let store =
                <Rels as drizzle_core::query::BuildStore>::Store::from_json_columns(&mut next_rel)?;

            results.push(drizzle_core::query::QueryRow::new(base, store));
        }

        Ok(results)
    }

    /// Executes the prepared relational query and returns the first row, if any.
    ///
    /// To apply `LIMIT 1` in SQL, call `.limit(1)` before `.prepare()`.
    pub async fn find_first<const N: usize>(
        &self,
        conn: &impl prepared::TursoExecutor,
        params: [drizzle_core::param::ParamBind<'a, SQLiteValue<'a>>; N],
    ) -> drizzle_core::error::Result<
        Option<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::Select,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::Select: for<'r> TryFrom<&'r ::turso::Row>,
        for<'r> <<T as drizzle_core::query::QueryTable>::Select as TryFrom<&'r ::turso::Row>>::Error:
            Into<drizzle_core::error::DrizzleError>,
        Rels: drizzle_core::query::BuildStore,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        Ok(self.find_many(conn, params).await?.into_iter().next())
    }
}

#[cfg(feature = "query")]
impl<'a, T, Rels>
    common::DrizzlePreparedQuery<'a, Connection, T, Rels, drizzle_core::query::PartialColumns>
{
    /// Executes the prepared relational query and returns all matching rows.
    pub async fn find_many<const N: usize>(
        &self,
        conn: &impl prepared::TursoExecutor,
        params: [drizzle_core::param::ParamBind<'a, SQLiteValue<'a>>; N],
    ) -> drizzle_core::error::Result<
        Vec<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildStore,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        debug_assert_eq!(
            N,
            self.inner.external_param_count(),
            "parameter count mismatch: expected {} params but got {}",
            self.inner.external_param_count(),
            N
        );

        let (sql_str, params) = self.inner.bind(params)?;
        let mut driver_params = Vec::with_capacity(self.inner.params.len());
        driver_params.extend(params.map(Into::into));
        let mut raw_rows = conn.fetch(sql_str, driver_params).await?;
        let mut results = Vec::new();

        while let Some(row) = raw_rows.next().await? {
            let base_json: String = row
                .get::<String>(0)
                .map_err(drizzle_core::error::DrizzleError::from)?;
            let base = <T as drizzle_core::query::QueryTable>::PartialSelect::from_json_str(
                &base_json, "base",
            )?;

            let mut rel_col = 1usize;
            let mut next_rel = || {
                let json = row
                    .get::<Option<String>>(rel_col)
                    .map_err(drizzle_core::error::DrizzleError::from)?;
                rel_col += 1;
                Ok(json)
            };
            let store =
                <Rels as drizzle_core::query::BuildStore>::Store::from_json_columns(&mut next_rel)?;

            results.push(drizzle_core::query::QueryRow::new(base, store));
        }

        Ok(results)
    }

    /// Executes the prepared relational query and returns the first row, if any.
    ///
    /// To apply `LIMIT 1` in SQL, call `.limit(1)` before `.prepare()`.
    pub async fn find_first<const N: usize>(
        &self,
        conn: &impl prepared::TursoExecutor,
        params: [drizzle_core::param::ParamBind<'a, SQLiteValue<'a>>; N],
    ) -> drizzle_core::error::Result<
        Option<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildStore,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        Ok(self.find_many(conn, params).await?.into_iter().next())
    }
}

impl<S, Schema, State, Table, Mk, Rw, Grouped>
    DrizzleBuilder<'_, S, QueryBuilder<'_, Schema, State, Table, Mk, Rw, Grouped>, State>
where
    State: builder::ExecutableState,
{
    /// Runs the query and returns the number of affected rows
    pub async fn execute(self) -> drizzle_core::error::Result<u64> {
        let (sql_str, params) = self.builder.sql.build();
        let driver_params: Vec<turso::Value> = params
            .iter()
            .copied()
            .map(std::convert::Into::into)
            .collect();
        turso_execute_cached(&self.runner.conn, &sql_str, driver_params)
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))
    }

    /// Runs the query and returns all matching rows using the builder's row type.
    pub async fn all<R, Proof, AggProof>(self) -> drizzle_core::error::Result<Vec<R>>
    where
        for<'r> Mk: drizzle_core::row::DecodeSelectedRef<&'r ::turso::Row, R>
            + drizzle_core::row::MarkerScopeValidFor<Proof>
            + drizzle_core::row::StrictDecodeMarker
            + drizzle_core::row::MarkerColumnCountValid<::turso::Row, Rw, R>,
        Mk: drizzle_core::row::MarkerAggValidFor<Grouped, AggProof>,
    {
        let (sql_str, params) = self.builder.sql.build();
        let driver_params: Vec<turso::Value> = params
            .iter()
            .copied()
            .map(std::convert::Into::into)
            .collect();
        let mut rows = turso_query_cached(&self.runner.conn, &sql_str, driver_params)
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?;
        let mut decoded = Vec::new();
        while let Some(row) = rows
            .next()
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?
        {
            decoded.push(<Mk as drizzle_core::row::DecodeSelectedRef<
                &::turso::Row,
                R,
            >>::decode(&row)?);
        }
        Ok(decoded)
    }

    /// Runs the query and returns a row cursor using the builder's row type.
    pub async fn rows(self) -> drizzle_core::error::Result<Rows<Rw>>
    where
        Rw: for<'r> TryFrom<&'r turso::Row>,
        for<'r> <Rw as TryFrom<&'r turso::Row>>::Error: Into<drizzle_core::error::DrizzleError>,
    {
        let (sql_str, params) = self.builder.sql.build();
        let driver_params: Vec<turso::Value> = params
            .iter()
            .copied()
            .map(std::convert::Into::into)
            .collect();

        let rows = self
            .runner
            .conn
            .query(&sql_str, driver_params)
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?;
        Ok(Rows::with_sql(rows, sql_str))
    }

    /// Runs the query and returns a single row using the builder's row type.
    pub async fn get<R, Proof, AggProof>(self) -> drizzle_core::error::Result<R>
    where
        for<'r> Mk: drizzle_core::row::DecodeSelectedRef<&'r ::turso::Row, R>
            + drizzle_core::row::MarkerScopeValidFor<Proof>
            + drizzle_core::row::StrictDecodeMarker
            + drizzle_core::row::MarkerColumnCountValid<::turso::Row, Rw, R>,
        Mk: drizzle_core::row::MarkerAggValidFor<Grouped, AggProof>,
    {
        let (sql_str, params) = self.builder.sql.build();
        let driver_params: Vec<turso::Value> = params
            .iter()
            .copied()
            .map(std::convert::Into::into)
            .collect();
        let mut rows = turso_query_cached(&self.runner.conn, &sql_str, driver_params)
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?;
        rows.next()
            .await
            .map_err(drizzle_core::error::DrizzleError::from)
            .with_query(|| QueryContext::new(&sql_str, &params))?
            .map_or_else(
                || Err(drizzle_core::error::DrizzleError::NotFound),
                |row| <Mk as drizzle_core::row::DecodeSelectedRef<&::turso::Row, R>>::decode(&row),
            )
    }
}