sqlx-turso-driver 0.0.1

use crate::{
    Turso, TursoConnectOptions,
    error::{engine_error, unsupported},
};
use sqlx_core::{
    connection::{Connection, LogSettings},
    error::Error,
    transaction::Transaction,
};

sqlx_core::impl_acquire!(Turso, TursoConnection);

/// Owns one physical Turso connection. Intentionally not Clone.
pub struct TursoConnection {
    pub(crate) inner: turso::Connection,
    pub(crate) _database: turso::Database,
    pub(crate) log_settings: LogSettings,
    // Keep the actual statement until completion or checked cleanup. Dropping
    // a Rows/future alone would hide failures in the engine's best-effort Drop.
    pub(crate) pending_statement: Option<turso::Statement>,
    pub(crate) unusable: bool,
    pub(crate) transaction_state: crate::transaction::TransactionState,
    pub(crate) rejected_begin_drop: bool,
    pub(crate) migration_data_path: Option<std::path::PathBuf>,
    pub(crate) migration_lock: Option<crate::migrate::MigrationLock>,
    #[cfg(test)]
    pub(crate) transition_hook: Option<crate::transaction::tests::Hook>,
    #[cfg(test)]
    pub(crate) rollback_fault: bool,
    #[cfg(test)]
    pub(crate) migration_hook: Option<crate::migrate::tests::Hook>,
}
impl std::fmt::Debug for TursoConnection {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TursoConnection")
            .field("has_pending_statement", &self.pending_statement.is_some())
            .field("unusable", &self.unusable)
            .field("transaction_state", &self.transaction_state)
            .finish_non_exhaustive()
    }
}
impl TursoConnection {
    fn map_cacheflush_result(&mut self, result: turso::Result<()>) -> Result<(), Error> {
        result.map_err(|error| {
            // A failed pager flush has no verified recovery path. Preserve the
            // original error, but never let a later ping approve pool reuse.
            self.unusable = true;
            engine_error(error)
        })
    }

    pub(crate) fn clear_pending_statement(&mut self) -> Result<(), Error> {
        if self.unusable {
            return Err(unsupported("reuse after failed statement cleanup"));
        }
        if let Some(statement) = self.pending_statement.take()
            && let Err(error) = statement.reset()
        {
            self.unusable = true;
            return Err(engine_error(error));
        }
        Ok(())
    }
}
impl Connection for TursoConnection {
    type Database = Turso;
    type Options = TursoConnectOptions;
    async fn close(mut self) -> Result<(), Error> {
        self.ready().await?;
        if !self.inner.is_autocommit().map_err(engine_error)? {
            return Err(unsupported(
                "graceful close with an open raw SQL transaction",
            ));
        }
        self.inner.cacheflush().map_err(engine_error)?;
        drop(self);
        Ok(())
    }
    async fn close_hard(self) -> Result<(), Error> {
        drop(self);
        Ok(())
    }
    async fn ping(&mut self) -> Result<(), Error> {
        // SQLx Migrator skips unlock on checksum/dirty/source/schema errors.
        // A session lock still held at pool release means discard, never idle.
        if self.migration_lock.is_some() {
            return Err(unsupported("pool reuse while holding a migration lock"));
        }
        self.ready().await?;
        // Only owned, pending SQLx transactions are recovered. Untracked raw
        // BEGIN must still not escape into the next pool borrower as clean.
        if !self.inner.is_autocommit().map_err(engine_error)? {
            return Err(unsupported("pool reuse with an open raw SQL transaction"));
        }
        use sqlx_core::executor::Executor;
        (&mut *self).execute("SELECT 1").await?;
        Ok(())
    }
    async fn begin(&mut self) -> Result<Transaction<'_, Turso>, Error> {
        Transaction::begin(self, None).await
    }
    fn shrink_buffers(&mut self) {}
    async fn flush(&mut self) -> Result<(), Error> {
        self.ready().await?;
        let result = self.inner.cacheflush();
        self.map_cacheflush_result(result)
    }
    fn should_flush(&self) -> bool {
        self.pending_statement.is_some()
            || matches!(
                self.transaction_state,
                crate::transaction::TransactionState::RollbackPending
                    | crate::transaction::TransactionState::Unknown
            )
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{TursoDatabaseError, connect_pool};
    use sqlx::{ConnectOptions, Executor};
    use std::{error::Error as _, io::ErrorKind, time::Duration};
    use tokio::time::timeout;

    type TestResult = Result<(), Box<dyn std::error::Error>>;
    const BOUND: Duration = Duration::from_secs(2);

    // Synthetic cleanup Result only: no actual SDK/OS I/O fault is induced.
    // Catches forgotten poison state and replacement of the original mapped error.
    #[tokio::test]
    async fn synthetic_cacheflush_error_persistently_rejects_reuse() -> TestResult {
        let mut connection = TursoConnectOptions::memory().connect().await?;
        connection.map_cacheflush_result(Ok(()))?;
        connection.ping().await?;
        let error = connection
            .map_cacheflush_result(Err(turso::Error::IoError(
                ErrorKind::Other,
                "synthetic lifecycle cacheflush failure",
            )))
            .unwrap_err();
        let database_error = error
            .as_database_error()
            .expect("the first error must keep its Database mapping")
            .downcast_ref::<TursoDatabaseError>();
        assert!(matches!(
            database_error
                .source()
                .unwrap()
                .downcast_ref::<turso::Error>(),
            Some(turso::Error::IoError(
                ErrorKind::Other,
                "synthetic lifecycle cacheflush failure"
            ))
        ));
        assert!(connection.pending_statement.is_none());
        assert!(!connection.should_flush());
        assert!(matches!(connection.ping().await, Err(Error::Protocol(_))));
        assert!(matches!(
            connection.execute("SELECT 1").await,
            Err(Error::Protocol(_))
        ));
        assert!(matches!(
            sqlx::query_scalar::<Turso, i64>("SELECT 1")
                .fetch_optional(&mut connection)
                .await,
            Err(Error::Protocol(_))
        ));
        assert!(matches!(
            connection
                .prepare(sqlx::SqlSafeStr::into_sql_str("SELECT 1"))
                .await,
            Err(Error::Protocol(_))
        ));
        assert!(matches!(connection.flush().await, Err(Error::Protocol(_))));
        // A later success must never silently clear an established quarantine.
        connection.map_cacheflush_result(Ok(()))?;
        assert!(matches!(connection.ping().await, Err(Error::Protocol(_))));
        connection.close_hard().await?;
        Ok(())
    }

    // Synthetic failure at the cleanup-Result boundary; actual engine connections
    // and SQLx release/acquire are used to check file discard and explicit memory loss.
    #[tokio::test]
    async fn synthetic_cacheflush_error_quarantines_pool_connection() -> TestResult {
        for memory in [false, true] {
            let directory = tempfile::tempdir()?;
            let options = if memory {
                TursoConnectOptions::memory()
            } else {
                TursoConnectOptions::file(directory.path().join("flush-quarantine.db"))?
            };
            let pool = connect_pool(options, 1).await?;
            let mut connection = timeout(BOUND, pool.acquire()).await??;
            connection
                .execute("CREATE TABLE items(value INTEGER)")
                .await?;
            connection.execute("INSERT INTO items VALUES (7)").await?;
            connection
                .execute("CREATE TEMP TABLE local_state(value INTEGER)")
                .await?;
            assert!(matches!(
                connection.map_cacheflush_result(Err(turso::Error::IoError(
                    ErrorKind::Other,
                    "synthetic lifecycle cacheflush failure",
                ))),
                Err(Error::Database(_))
            ));
            drop(connection);
            let next = timeout(BOUND, pool.acquire()).await?;
            if memory {
                assert!(matches!(next, Err(Error::Configuration(_))), "{next:?}");
            } else {
                let mut next = next?;
                assert!(
                    next.execute("INSERT INTO local_state VALUES (1)")
                        .await
                        .is_err(),
                    "the failed-flush physical connection must not return to idle"
                );
                assert_eq!(
                    sqlx::query_scalar::<Turso, i64>("SELECT value FROM items")
                        .fetch_one(&mut *next)
                        .await?,
                    7
                );
                next.execute("INSERT INTO items VALUES (8)").await?;
                drop(next);
            }
            timeout(BOUND, pool.close()).await?;
        }
        Ok(())
    }
}