sqlx-turso-driver 0.0.1

//! Forward-only, transactionally applied local SQLx migrations.
use crate::{
    Turso, TursoConnectOptions, TursoConnection,
    error::{engine_error, unsupported},
};
use fs4::fs_std::FileExt;
use futures_core::future::BoxFuture;
use sqlx_core::{
    connection::{ConnectOptions, Connection},
    error::Error,
    executor::Executor,
    migrate::{AppliedMigration, Migrate, MigrateDatabase, MigrateError, Migration},
    row::Row,
    sql_str::{AssertSqlSafe, SqlSafeStr},
};
use std::{
    fs::{File, OpenOptions},
    path::{Path, PathBuf},
    time::{Duration, Instant},
};

// Resolve the actual data path before SDK open so symlinks use the same WAL and
// migration lock path. Do not keep an extra descriptor on the engine's fcntl inode.
pub(crate) fn canonical_data_path(path: &Path) -> Result<PathBuf, Error> {
    match path.canonicalize() {
        Ok(canonical) => return Ok(canonical),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => return Err(error.into()),
    }
    match OpenOptions::new().write(true).create_new(true).open(path) {
        Ok(file) => drop(file),
        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
        Err(error) => return Err(error.into()),
    }
    Ok(path.canonicalize()?)
}

// Turso already locks the data inode. Use a persistent OS-locked sidecar keyed
// to its canonical path, never an existence flag and never deleted on unlock.
// Memory migrations use the uniquely borrowed physical connection.
pub(crate) struct MigrationLock {
    file: Option<File>,
}
impl MigrationLock {
    fn release(&mut self) -> Result<(), Error> {
        if let Some(file) = &self.file {
            FileExt::unlock(file)?;
        }
        self.file = None;
        Ok(())
    }
}
impl Drop for MigrationLock {
    fn drop(&mut self) {
        let _ = self.release();
    }
}

// Every migration method that can suspend owns this guard. On cancellation,
// release the advisory lock and quarantine the connection; do not infer SDK
// rollback/commit completion. D4 retains its statements/transition state.
struct Operation<'a> {
    conn: &'a mut TursoConnection,
    completed: bool,
}
impl<'a> Operation<'a> {
    fn new(conn: &'a mut TursoConnection) -> Self {
        Self {
            conn,
            completed: false,
        }
    }
    fn finish<T>(mut self, result: T) -> T {
        self.completed = true;
        result
    }
}
impl Drop for Operation<'_> {
    fn drop(&mut self) {
        if !self.completed {
            self.conn.unusable = true;
            self.conn.migration_lock.take();
        }
    }
}

// SQLx permits arbitrary table names; D5 accepts only a simple ASCII identifier.
// Reject schema qualification/quotes instead of interpolating untrusted SQL.
fn table(name: &str) -> Result<String, MigrateError> {
    let mut bytes = name.bytes();
    if !bytes
        .next()
        .is_some_and(|b| b.is_ascii_alphabetic() || b == b'_')
        || !bytes.all(|b| b.is_ascii_alphanumeric() || b == b'_')
    {
        return Err(
            unsupported("migration table names other than simple ASCII identifiers").into(),
        );
    }
    Ok(format!("\"{name}\""))
}

// Use exactly the pinned engine parser and consumed byte offsets. Prevalidate
// transaction-escaping commands, but prepare/execute statements sequentially:
// later statements may refer to schema created by earlier DDL.
fn statements(sql: &str) -> Result<Vec<&str>, Error> {
    use turso_parser::ast::{Cmd, Stmt};
    let mut rest = sql;
    let mut result = Vec::new();
    loop {
        let (command, end) = turso::core::dialect::sqlite::parse(rest)
            .map_err(|e| engine_error(turso::Error::Error(e.to_string())))?;
        let Some(command) = command else {
            break;
        };
        if end == 0 || end > rest.len() || !rest.is_char_boundary(end) {
            return Err(unsupported("invalid migration parser consumed offset"));
        }
        if let Cmd::Stmt(
            Stmt::Begin { .. }
            | Stmt::Commit { .. }
            | Stmt::Rollback { .. }
            | Stmt::Savepoint { .. }
            | Stmt::Release { .. }
            | Stmt::Attach { .. }
            | Stmt::Detach { .. }
            | Stmt::Pragma { .. }
            | Stmt::Vacuum { .. },
        ) = command
        {
            return Err(unsupported(
                "transaction control, attached schemas, PRAGMA or VACUUM in migrations",
            ));
        }
        result.push(&rest[..end]);
        rest = &rest[end..];
    }
    Ok(result)
}

async fn apply_version(
    conn: &mut TursoConnection,
    name: &str,
    migration: &Migration,
) -> Result<Duration, Error> {
    if conn.migration_lock.is_none() {
        return Err(unsupported(
            "migration apply without exclusive migration lock",
        ));
    }
    if migration.no_tx || migration.migration_type.is_down_migration() {
        return Err(unsupported("nontransactional or down migration apply"));
    }
    let batch = statements(migration.sql.as_str())?;
    let started = Instant::now();
    let mut tx = conn.begin().await?;
    let result: Result<(), Error> = async {
        for statement in batch {
            // Slices of the already trusted Migration SQL, not user value interpolation.
            (&mut *tx).execute(AssertSqlSafe(statement).into_sql_str()).await?;
        }
        #[cfg(test)]
        if let Some(hook) = tx.migration_hook.take() {
            let _ = hook.reached.send(());
            let _ = hook.resume.await;
        }
        let nanos = i64::try_from(started.elapsed().as_nanos()).unwrap_or(i64::MAX);
        sqlx_core::query::query(AssertSqlSafe(format!(
            "INSERT INTO {name} (version, description, success, checksum, execution_time) VALUES (?, ?, ?, ?, ?)"
        )))
            .bind(migration.version).bind(migration.description.as_ref()).bind(true)
            .bind(migration.checksum.as_ref()).bind(nanos).execute(&mut *tx).await?;
        Ok(())
    }.await;
    if let Err(error) = result {
        // Rollback failure preserves the original migration error and D4 Unknown
        // quarantine. No version is acknowledged as successfully applied.
        let _ = tx.rollback().await;
        return Err(error);
    }
    tx.commit().await?;
    Ok(started.elapsed())
}

impl Migrate for TursoConnection {
    fn create_schema_if_not_exists<'e>(
        &'e mut self,
        schema: &'e str,
    ) -> BoxFuture<'e, Result<(), MigrateError>> {
        Box::pin(async move { Err(MigrateError::CreateSchemasNotSupported(schema.to_owned())) })
    }
    fn ensure_migrations_table<'e>(
        &'e mut self,
        name: &'e str,
    ) -> BoxFuture<'e, Result<(), MigrateError>> {
        Box::pin(async move {
            let name = table(name)?;
            let op = Operation::new(self);
            let result = op.conn.execute(AssertSqlSafe(format!(
                "CREATE TABLE IF NOT EXISTS {name} (version BIGINT PRIMARY KEY, description TEXT NOT NULL, installed_on TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, success BOOLEAN NOT NULL, checksum BLOB NOT NULL, execution_time BIGINT NOT NULL)"
            )).into_sql_str()).await.map(|_| ()).map_err(MigrateError::from);
            op.finish(result)
        })
    }
    fn dirty_version<'e>(
        &'e mut self,
        name: &'e str,
    ) -> BoxFuture<'e, Result<Option<i64>, MigrateError>> {
        Box::pin(async move {
            let name = table(name)?;
            let op = Operation::new(self);
            let result = sqlx_core::query_scalar::query_scalar(AssertSqlSafe(format!(
                "SELECT version FROM {name} WHERE success = 0 ORDER BY version LIMIT 1"
            )))
            .fetch_optional(&mut *op.conn)
            .await
            .map_err(MigrateError::from);
            op.finish(result)
        })
    }
    fn list_applied_migrations<'e>(
        &'e mut self,
        name: &'e str,
    ) -> BoxFuture<'e, Result<Vec<AppliedMigration>, MigrateError>> {
        Box::pin(async move {
            let name = table(name)?;
            let op = Operation::new(self);
            let result = async {
                let rows = sqlx_core::query::query(AssertSqlSafe(format!(
                    "SELECT version, checksum FROM {name} WHERE success = 1 ORDER BY version"
                )))
                .fetch_all(&mut *op.conn)
                .await?;
                rows.into_iter()
                    .map(|row| {
                        Ok(AppliedMigration {
                            version: row.try_get(0)?,
                            checksum: row.try_get::<Vec<u8>, _>(1)?.into(),
                        })
                    })
                    .collect::<Result<Vec<_>, Error>>()
            }
            .await
            .map_err(MigrateError::from);
            op.finish(result)
        })
    }
    fn lock(&mut self) -> BoxFuture<'_, Result<(), MigrateError>> {
        Box::pin(async move {
            if self.migration_lock.is_some() {
                return Err(unsupported("migration lock already held by this connection").into());
            }
            let op = Operation::new(self);
            let result = async {
                op.conn.ready().await?;
                let file = op
                    .conn
                    .migration_data_path
                    .as_ref()
                    .map(|path| {
                        let mut lock_path = path.as_os_str().to_owned();
                        lock_path.push(".sqlx-migrate.lock");
                        OpenOptions::new()
                            .read(true)
                            .write(true)
                            .create(true)
                            .truncate(false)
                            .open(PathBuf::from(lock_path))
                    })
                    .transpose()?;
                if let Some(file) = &file
                    && !FileExt::try_lock_exclusive(file)?
                {
                    return Err(Error::Io(std::io::Error::new(
                        std::io::ErrorKind::WouldBlock,
                        "sqlx-turso: exclusive migration lock is held",
                    )));
                }
                op.conn.migration_lock = Some(MigrationLock { file });
                Ok(())
            }
            .await
            .map_err(MigrateError::from);
            op.finish(result)
        })
    }
    fn unlock(&mut self) -> BoxFuture<'_, Result<(), MigrateError>> {
        Box::pin(async move {
            let Some(lock) = &mut self.migration_lock else {
                return Err(unsupported("migration unlock without owned lock").into());
            };
            if let Err(error) = lock.release() {
                self.unusable = true;
                return Err(error.into());
            }
            self.migration_lock.take();
            Ok(())
        })
    }
    fn apply<'e>(
        &'e mut self,
        name: &'e str,
        migration: &'e Migration,
    ) -> BoxFuture<'e, Result<Duration, MigrateError>> {
        Box::pin(async move {
            let name = table(name)?;
            let op = Operation::new(self);
            let result = apply_version(op.conn, &name, migration)
                .await
                .map_err(|error| MigrateError::ExecuteMigration(error, migration.version));
            op.finish(result)
        })
    }
    fn revert<'e>(
        &'e mut self,
        _: &'e str,
        _: &'e Migration,
    ) -> BoxFuture<'e, Result<Duration, MigrateError>> {
        Box::pin(async {
            Err(unsupported("migration revert; restore a verified backup instead").into())
        })
    }
}

impl MigrateDatabase for Turso {
    async fn create_database(url: &str) -> Result<(), Error> {
        let options = local_options(url)?;
        options.connect().await?.close().await
    }
    async fn database_exists(url: &str) -> Result<bool, Error> {
        let options = local_options(url)?;
        match std::fs::metadata(options.path.expect("local_options requires a file")) {
            Ok(metadata) if metadata.is_file() => Ok(true),
            Ok(_) => Err(unsupported("non-file database path")),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(error) => Err(error.into()),
        }
    }
    async fn drop_database(_: &str) -> Result<(), Error> {
        // An advisory migration lock does not prove all ordinary SDK handles are
        // closed. Deleting a data/WAL file with live handles is outside D5.
        Err(unsupported(
            "database deletion; close all owners and manage verified backups explicitly",
        ))
    }
}
fn local_options(url: &str) -> Result<TursoConnectOptions, Error> {
    let options: TursoConnectOptions = url.parse()?;
    if options.path.is_none() {
        return Err(unsupported("file lifecycle for in-memory databases"));
    }
    Ok(options)
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::{TursoConnectOptions, connect_pool};
    use sqlx::{ConnectOptions, Connection};
    use sqlx_core::sql_str::SqlSafeStr;
    use std::time::Duration;
    use tokio::sync::oneshot;
    pub(crate) struct Hook {
        pub(crate) reached: oneshot::Sender<()>,
        pub(crate) resume: oneshot::Receiver<()>,
    }
    type TestResult = Result<(), Box<dyn std::error::Error>>;

    // Test-only await barriers around real SDK work, never simulated SQL/OS failures.
    async fn cancellation_case(phase: u8, pooled: bool) -> TestResult {
        tokio::time::timeout(Duration::from_secs(5), async {
            let directory = tempfile::tempdir()?;
            let options = TursoConnectOptions::file(directory.path().join("cancel-migration.db"))?;
            let pool = connect_pool(options.clone(), 1).await?;
            let mut checkout = pool.acquire().await?;
            checkout.ensure_migrations_table("_sqlx_migrations").await?;
            if !pooled { checkout.lock().await?; }
            let (reached, mut at_barrier) = oneshot::channel();
            let (_resume, resume) = oneshot::channel();
            if phase == 1 {
                checkout.migration_hook = Some(Hook { reached, resume });
            } else {
                use crate::transaction::tests::{Hook, Phase};
                checkout.transition_hook = Some(Hook { phase: match phase { 0 => Phase::AfterBegin, 2 => Phase::BeforeCommit, _ => Phase::AfterCommit }, reached, resume });
            }
            let migration = Migration::new(1, "cancellation".into(), sqlx_core::migrate::MigrationType::Simple,
                "CREATE TABLE cancelled_schema(id INTEGER); INSERT INTO cancelled_schema VALUES (1);".into_sql_str(), false);
            if pooled {
                drop(checkout);
                let migrator = sqlx_core::migrate::Migrator::with_migrations(vec![migration]);
                let mut run = Box::pin(migrator.run(&pool));
                tokio::select! {
                    result = &mut run => panic!("migration missed cancellation barrier: {result:?}"),
                    reached = &mut at_barrier => reached.expect("barrier lost"),
                }
                drop(run);
                let next = pool.acquire().await?;
                drop(next);
            } else {
                let mut apply = checkout.apply("_sqlx_migrations", &migration);
                tokio::select! {
                    result = &mut apply => panic!("apply missed cancellation barrier: {result:?}"),
                    reached = &mut at_barrier => reached.expect("barrier lost"),
                }
                drop(apply);
                assert!(checkout.ping().await.is_err(), "cancelled migration authorized reuse");
                // Must release the advisory lock on cancellation even before owner discard.
                let mut competitor = options.connect().await?;
                competitor.lock().await?;
                competitor.unlock().await?;
                competitor.close_hard().await?;
                assert!(checkout.close().await.is_err(), "cancelled owner must not close as clean");
            }
            pool.close().await;
            let mut reopened = options.connect().await?;
            // After actual COMMIT, cancellation loses acknowledgement, not atomicity:
            // schema AND version must persist together. Earlier phases persist neither.
            let committed = usize::from(phase == 3);
            assert_eq!(sqlx::query_scalar::<Turso, i64>("SELECT count(*) FROM sqlite_schema WHERE name='cancelled_schema'").fetch_one(&mut reopened).await?, committed as i64);
            assert_eq!(reopened.list_applied_migrations("_sqlx_migrations").await?.len(), committed);
            reopened.close().await?;
            TestResult::Ok(())
        }).await.expect("migration cancellation deadline")
    }
    #[tokio::test]
    async fn cancelled_apply_releases_lock_and_quarantines() -> TestResult {
        for phase in 0..4 {
            cancellation_case(phase, false).await?;
        }
        Ok(())
    }
    #[tokio::test]
    async fn cancelled_pool_migrator_discards_lock_owner() -> TestResult {
        for phase in 0..4 {
            cancellation_case(phase, true).await?;
        }
        Ok(())
    }
}