slave-framework 0.1.0

//! Connections to the application's local Turso database.

use std::path::Path;
use std::time::Duration;

use turso::{Builder, Connection};

use crate::{Error, Result, env};

/// Database file used when `DATABASE_PATH` is not set.
pub const DEFAULT_PATH: &str = "database/database.db";

/// How long a connection waits for a lock held by another connection before
/// failing with "database is locked". Without it, concurrent writers fail at once.
pub const BUSY_TIMEOUT: Duration = Duration::from_secs(5);

/// Returns the database path from `DATABASE_PATH`, or [`DEFAULT_PATH`].
pub fn path_from_env() -> String {
    env::var_or("DATABASE_PATH", DEFAULT_PATH)
}

/// A shareable handle to a local Turso database. Cheap to clone.
///
/// Open one per process and call [`connect`](Database::connect) per unit of
/// work (for example, per request).
#[derive(Clone, Debug)]
pub struct Database {
    inner: turso::Database,
}

impl Database {
    /// Opens (creating if necessary) the database at `path`. `:memory:` opens
    /// a private in-memory database.
    pub async fn open(path: &str) -> Result<Self> {
        if path != ":memory:" {
            if let Some(dir) = Path::new(path)
                .parent()
                .filter(|d| !d.as_os_str().is_empty())
            {
                std::fs::create_dir_all(dir).map_err(|source| Error::DatabaseDirectory {
                    path: dir.to_path_buf(),
                    source,
                })?;
            }
        }
        let inner = Builder::new_local(path).build().await?;
        Ok(Self { inner })
    }

    /// Opens a new connection with a busy timeout and foreign-key enforcement enabled.
    pub async fn connect(&self) -> Result<Connection, turso::Error> {
        let conn = self.inner.connect()?;
        conn.busy_timeout(BUSY_TIMEOUT)?;
        conn.execute("PRAGMA foreign_keys = ON", ()).await?;
        Ok(conn)
    }
}

/// Opens the database at `path` and returns a single configured connection.
pub async fn connect(path: &str) -> Result<Connection> {
    Ok(Database::open(path).await?.connect().await?)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn concurrent_writers_wait_instead_of_failing() {
        let dir = std::env::temp_dir().join(format!("slave-db-{}", std::process::id()));
        let path = dir.join("concurrent.db");
        let db = Database::open(path.to_str().unwrap()).await.unwrap();
        let conn = db.connect().await.unwrap();
        conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, n INTEGER)", ())
            .await
            .unwrap();

        let tasks: Vec<_> = (0..20)
            .map(|n| {
                let db = db.clone();
                tokio::spawn(async move {
                    let conn = db.connect().await?;
                    conn.execute("INSERT INTO t (n) VALUES (?1)", (n,)).await
                })
            })
            .collect();
        for task in tasks {
            task.await.unwrap().unwrap();
        }

        let mut rows = conn.query("SELECT count(*) FROM t", ()).await.unwrap();
        let count: i64 = rows.next().await.unwrap().unwrap().get(0).unwrap();
        assert_eq!(count, 20);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn foreign_keys_are_enforced() {
        let conn = connect(":memory:").await.unwrap();
        conn.execute("CREATE TABLE a (id INTEGER PRIMARY KEY)", ())
            .await
            .unwrap();
        conn.execute(
            "CREATE TABLE b (a_id INTEGER NOT NULL REFERENCES a(id))",
            (),
        )
        .await
        .unwrap();
        let err = conn.execute("INSERT INTO b (a_id) VALUES (1)", ()).await;
        assert!(err.is_err(), "insert with a dangling reference must fail");
    }
}