add db module

src/db/cast.rs

···
+//! Type-safe casting utilities.
+
+use serde::{Serialize, de::DeserializeOwned};
+use std::fmt;
+
+/// Represents an ISO 8601 date string (e.g., "2023-01-01T12:00:00Z").
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DateISO(String);
+
+impl DateISO {
+    /// Converts a `chrono::DateTime<Utc>` to a `DateISO`.
+    pub fn from_date(date: chrono::DateTime<chrono::Utc>) -> Self {
+        Self(date.to_rfc3339())
+    }
+
+    /// Converts a `DateISO` back to a `chrono::DateTime<Utc>`.
+    pub fn to_date(&self) -> Result<chrono::DateTime<chrono::Utc>, chrono::ParseError> {
+        self.0.parse::<chrono::DateTime<chrono::Utc>>()
+    }
+}
+
+impl fmt::Display for DateISO {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+/// Represents a JSON-encoded string.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct JsonEncoded<T: Serialize>(String, std::marker::PhantomData<T>);
+
+impl<T: Serialize> JsonEncoded<T> {
+    /// Encodes a value into a JSON string.
+    pub fn to_json(value: &T) -> Result<Self, serde_json::Error> {
+        let json = serde_json::to_string(value)?;
+        Ok(Self(json, std::marker::PhantomData))
+    }
+
+    /// Decodes a JSON string back into a value.
+    pub fn from_json(json_str: &str) -> Result<T, serde_json::Error>
+    where
+        T: DeserializeOwned,
+    {
+        serde_json::from_str(json_str)
+    }
+
+    /// Returns the underlying JSON string.
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl<T: Serialize> fmt::Display for JsonEncoded<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use serde::{Deserialize, Serialize};
+
+    #[test]
+    fn test_date_iso() {
+        let now = Utc::now();
+        let date_iso = DateISO::from_date(now);
+        let parsed_date = date_iso.to_date().unwrap();
+        assert_eq!(now.to_rfc3339(), parsed_date.to_rfc3339());
+    }
+
+    #[derive(Serialize, Deserialize, PartialEq, Debug)]
+    struct TestStruct {
+        name: String,
+        value: i32,
+    }
+
+    #[test]
+    fn test_json_encoded() {
+        let test_value = TestStruct {
+            name: "example".to_string(),
+            value: 42,
+        };
+
+        // Encode to JSON
+        let encoded = JsonEncoded::to_json(&test_value).unwrap();
+        assert_eq!(encoded.as_str(), r#"{"name":"example","value":42}"#);
+
+        // Decode from JSON
+        let decoded: TestStruct = JsonEncoded::from_json(encoded.as_str()).unwrap();
+        assert_eq!(decoded, test_value);
+    }
+}

src/db/db.rs

···
+//! Database connection and transaction management.
+
+use sqlx::{
+    Sqlite, Transaction,
+    sqlite::{SqliteConnectOptions, SqlitePool, SqliteQueryResult, SqliteTransactionManager},
+};
+use std::collections::VecDeque;
+use std::str::FromStr;
+use std::sync::{Arc, Mutex};
+use tokio::sync::Mutex as AsyncMutex;
+
+use crate::db::util::retry_sqlite;
+
+/// Default pragmas for SQLite.
+const DEFAULT_PRAGMAS: &[(&str, &str)] = &[
+    // Add default pragmas here if needed, e.g., ("foreign_keys", "ON")
+];
+
+/// Database struct for managing SQLite connections and transactions.
+pub struct Database {
+    /// SQLite connection pool.
+    db: Arc<SqlitePool>,
+    /// Flag indicating if the database is destroyed.
+    destroyed: Arc<Mutex<bool>>,
+    /// Queue of commit hooks.
+    commit_hooks: Arc<AsyncMutex<VecDeque<Box<dyn FnOnce() + Send>>>>,
+}
+
+impl Database {
+    /// Creates a new database instance with the given location and optional pragmas.
+    pub async fn new(location: &str, pragmas: Option<&[(&str, &str)]>) -> sqlx::Result<Self> {
+        let mut options = SqliteConnectOptions::from_str(location)?.create_if_missing(true);
+
+        // Apply default and user-provided pragmas.
+        for &(key, value) in DEFAULT_PRAGMAS.iter().chain(pragmas.unwrap_or(&[])) {
+            options = options.pragma(key.to_string(), value.to_string());
+        }
+
+        let pool = SqlitePool::connect_with(options).await?;
+        Ok(Self {
+            db: Arc::new(pool),
+            destroyed: Arc::new(Mutex::new(false)),
+            commit_hooks: Arc::new(AsyncMutex::new(VecDeque::new())),
+        })
+    }
+
+    /// Ensures the database is using Write-Ahead Logging (WAL) mode.
+    pub async fn ensure_wal(&self) -> sqlx::Result<()> {
+        let mut conn = self.db.acquire().await?;
+        sqlx::query("PRAGMA journal_mode = WAL")
+            .execute(&mut *conn)
+            .await?;
+        Ok(())
+    }
+
+    /// Executes a transaction without retry logic.
+    pub async fn transaction_no_retry<F, T>(&self, func: F) -> sqlx::Result<T>
+    where
+        F: FnOnce(&mut Transaction<'_, Sqlite>) -> sqlx::Result<T>,
+    {
+        let mut tx = self.db.begin().await?;
+        let result = func(&mut tx)?;
+        tx.commit().await?;
+        self.run_commit_hooks().await;
+        Ok(result)
+    }
+
+    /// Executes a transaction with retry logic.
+    pub async fn transaction<F, T>(&self, func: F) -> sqlx::Result<T>
+    where
+        F: FnOnce(&mut Transaction<'_, Sqlite>) -> sqlx::Result<T> + Copy,
+    {
+        retry_sqlite(|| self.transaction_no_retry(func)).await
+    }
+
+    /// Executes a query with retry logic.
+    pub async fn execute_with_retry<F, T>(&self, query: F) -> sqlx::Result<T>
+    where
+        F: Fn() -> std::pin::Pin<Box<dyn futures::Future<Output = sqlx::Result<T>> + Send>> + Copy,
+    {
+        retry_sqlite(|| query()).await
+    }
+
+    /// Adds a commit hook to be executed after a successful transaction.
+    pub async fn on_commit<F>(&self, hook: F)
+    where
+        F: FnOnce() + Send + 'static,
+    {
+        let mut hooks = self.commit_hooks.lock().await;
+        hooks.push_back(Box::new(hook));
+    }
+
+    /// Closes the database connection pool.
+    pub async fn close(&self) -> sqlx::Result<()> {
+        let mut destroyed = self.destroyed.lock().unwrap();
+        if *destroyed {
+            return Ok(());
+        }
+        *destroyed = true;
+        drop(self.db.clone()); // Drop the pool to close connections.
+        Ok(())
+    }
+
+    /// Runs all commit hooks in the queue.
+    async fn run_commit_hooks(&self) {
+        let mut hooks = self.commit_hooks.lock().await;
+        while let Some(hook) = hooks.pop_front() {
+            hook();
+        }
+    }
+}

src/db/migrator.rs

···
+//! Database migration management.
+
+use sqlx::{SqlitePool, migrate::Migrator};
+use std::path::Path;
+use thiserror::Error;
+
+/// Error type for migration-related issues.
+#[derive(Debug, Error)]
+pub enum MigrationError {
+    #[error("Migration failed: {0}")]
+    MigrationFailed(String),
+    #[error("Unknown failure occurred while migrating")]
+    UnknownFailure,
+}
+
+/// Migrator struct for managing database migrations.
+pub struct DatabaseMigrator {
+    /// SQLx migrator instance.
+    migrator: Migrator,
+    /// SQLite connection pool.
+    db: SqlitePool,
+}
+
+impl DatabaseMigrator {
+    /// Creates a new `DatabaseMigrator` instance.
+    ///
+    /// # Arguments
+    /// - `migrations_path`: Path to the directory containing migration files.
+    /// - `db`: SQLite connection pool.
+    pub async fn new(migrations_path: &Path, db: SqlitePool) -> Self {
+        let migrator = Migrator::new(migrations_path)
+            .await
+            .expect("Failed to initialize migrator");
+        Self { migrator, db }
+    }
+
+    /// Migrates the database to a specific migration or throws an error.
+    ///
+    /// # Arguments
+    /// - `migration`: The target migration name.
+    ///
+    /// # Unimplemented
+    /// This currently runs all migrations instead of a specific one.
+    pub async fn migrate_to_or_throw(&self, _migration: &str) -> Result<(), MigrationError> {
+        // TODO: Implement migration to a specific version
+        // For now, we will just run all migrations
+        let result = self.migrator.run(&self.db).await;
+
+        match result {
+            Ok(_) => Ok(()),
+            Err(err) => Err(MigrationError::MigrationFailed(err.to_string())),
+        }
+    }
+
+    /// Migrates the database to the latest migration or throws an error.
+    pub async fn migrate_to_latest_or_throw(&self) -> Result<(), MigrationError> {
+        let result = self.migrator.run(&self.db).await;
+
+        match result {
+            Ok(_) => Ok(()),
+            Err(err) => Err(MigrationError::MigrationFailed(err.to_string())),
+        }
+    }
+}

src/db/mod.rs

···
+mod cast;
+mod db;
+mod migrator;
+mod pagination;
+mod util;

src/db/pagination.rs

···
+use std::fmt::Debug;
+
+/// Represents a cursor with primary and secondary parts.
+#[derive(Debug, Clone)]
+pub struct Cursor {
+    pub primary: String,
+    pub secondary: String,
+}
+
+/// Represents a labeled result with primary and secondary parts.
+#[derive(Debug, Clone)]
+pub struct LabeledResult {
+    pub primary: String,
+    pub secondary: String,
+}
+
+/// Trait defining the interface for a keyset-paginated cursor.
+pub trait GenericKeyset<R, LR: Debug> {
+    fn label_result(&self, result: R) -> LR;
+    fn labeled_result_to_cursor(&self, labeled: LR) -> Cursor;
+    fn cursor_to_labeled_result(&self, cursor: Cursor) -> LR;
+
+    fn pack_from_result(&self, results: Vec<R>) -> Option<String> {
+        todo!()
+        // results
+        //     .last()
+        //     .map(|result| self.pack(Some(self.label_result(result.clone()))))
+    }
+
+    fn pack(&self, labeled: Option<LR>) -> Option<String> {
+        labeled.map(|l| self.pack_cursor(self.labeled_result_to_cursor(l)))
+    }
+
+    fn unpack(&self, cursor_str: Option<String>) -> Option<LR> {
+        cursor_str
+            .and_then(|cursor| self.unpack_cursor(cursor))
+            .map(|c| self.cursor_to_labeled_result(c))
+    }
+
+    fn pack_cursor(&self, cursor: Cursor) -> String {
+        format!("{}::{}", cursor.primary, cursor.secondary)
+    }
+
+    fn unpack_cursor(&self, cursor_str: String) -> Option<Cursor> {
+        let parts: Vec<&str> = cursor_str.split("::").collect();
+        if parts.len() == 2 {
+            Some(Cursor {
+                primary: parts[0].to_string(),
+                secondary: parts[1].to_string(),
+            })
+        } else {
+            None
+        }
+    }
+}
+
+/// A concrete implementation of `GenericKeyset` for time and CID-based pagination.
+pub struct TimeCidKeyset;
+
+impl TimeCidKeyset {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl GenericKeyset<CreatedAtCidResult, LabeledResult> for TimeCidKeyset {
+    fn label_result(&self, result: CreatedAtCidResult) -> LabeledResult {
+        LabeledResult {
+            primary: result.created_at,
+            secondary: result.cid,
+        }
+    }
+
+    fn labeled_result_to_cursor(&self, labeled: LabeledResult) -> Cursor {
+        Cursor {
+            primary: labeled.primary,
+            secondary: labeled.secondary,
+        }
+    }
+
+    fn cursor_to_labeled_result(&self, cursor: Cursor) -> LabeledResult {
+        LabeledResult {
+            primary: cursor.primary,
+            secondary: cursor.secondary,
+        }
+    }
+}
+
+/// Represents a database result with created_at and cid fields.
+#[derive(Debug, Clone)]
+pub struct CreatedAtCidResult {
+    pub created_at: String,
+    pub cid: String,
+}
+
+/// Pagination options for queries.
+pub struct PaginationOptions<'a> {
+    pub limit: Option<usize>,
+    pub cursor: Option<String>,
+    pub direction: Option<&'a str>,
+    pub try_index: Option<bool>,
+}
+
+/// Applies pagination to a query.
+pub fn paginate<K>(query: &mut String, opts: PaginationOptions, keyset: &K) -> String
+where
+    K: GenericKeyset<CreatedAtCidResult, LabeledResult>,
+{
+    let PaginationOptions {
+        limit,
+        cursor,
+        direction,
+        try_index,
+    } = opts;
+
+    let direction = direction.unwrap_or("desc");
+    let labeled = cursor.and_then(|c| keyset.unpack(Some(c)));
+    let keyset_sql = labeled.map(|l| get_sql(&l, direction, try_index.unwrap_or(false)));
+
+    if let Some(sql) = keyset_sql {
+        query.push_str(&format!(" WHERE {}", sql));
+    }
+
+    if let Some(l) = limit {
+        query.push_str(&format!(" LIMIT {}", l));
+    }
+
+    query.push_str(&format!(
+        " ORDER BY primary {} secondary {}",
+        direction, direction
+    ));
+
+    query.clone()
+}
+
+/// Generates SQL conditions for pagination.
+fn get_sql(labeled: &LabeledResult, direction: &str, try_index: bool) -> String {
+    if try_index {
+        if direction == "asc" {
+            format!(
+                "(primary, secondary) > ('{}', '{}')",
+                labeled.primary, labeled.secondary
+            )
+        } else {
+            format!(
+                "(primary, secondary) < ('{}', '{}')",
+                labeled.primary, labeled.secondary
+            )
+        }
+    } else {
+        if direction == "asc" {
+            format!(
+                "(primary > '{}' OR (primary = '{}' AND secondary > '{}'))",
+                labeled.primary, labeled.primary, labeled.secondary
+            )
+        } else {
+            format!(
+                "(primary < '{}' OR (primary = '{}' AND secondary < '{}'))",
+                labeled.primary, labeled.primary, labeled.secondary
+            )
+        }
+    }
+}

src/db/util.rs

···
+//! This module contains utility functions and types for working with SQLite databases using SQLx.
+
+use sqlx::Error;
+use std::collections::HashSet;
+
+/// Returns a SQL clause to check if a record is not soft-deleted.
+pub fn not_soft_deleted_clause(alias: &str) -> String {
+    format!(r#"{}."takedownRef" IS NULL"#, alias)
+}
+
+/// Checks if a record is soft-deleted.
+pub fn is_soft_deleted(takedown_ref: Option<&str>) -> bool {
+    takedown_ref.is_some()
+}
+
+/// SQL clause to count all rows.
+pub const COUNT_ALL: &str = "COUNT(*)";
+
+/// SQL clause to count distinct rows based on a reference.
+pub fn count_distinct(ref_col: &str) -> String {
+    format!("COUNT(DISTINCT {})", ref_col)
+}
+
+/// Generates a SQL clause for the `excluded` column in an `ON CONFLICT` clause.
+pub fn excluded(col: &str) -> String {
+    format!("excluded.{}", col)
+}
+
+/// Generates a SQL clause for a large `WHERE IN` clause using a hash lookup.
+/// # DEPRECATED
+/// Use SQLx parameterized queries instead.
+#[deprecated = "Use SQLx parameterized queries instead"]
+pub fn values_list(vals: &[&str]) -> String {
+    let values = vals
+        .iter()
+        .map(|val| format!("('{}')", val))
+        .collect::<Vec<_>>()
+        .join(", ");
+    format!("(VALUES {})", values)
+}
+
+/// Retries an asynchronous SQLite operation with exponential backoff.
+pub async fn retry_sqlite<F, Fut, T>(operation: F) -> Result<T, sqlx::Error>
+where
+    F: Fn() -> Fut,
+    Fut: std::future::Future<Output = Result<T, sqlx::Error>>,
+{
+    let max_retries = 60;
+    let mut attempt = 0;
+
+    while attempt < max_retries {
+        match operation().await {
+            Ok(result) => return Ok(result),
+            Err(err) if is_retryable_sqlite_error(&err) => {
+                if let Some(wait_ms) = get_wait_ms_sqlite(attempt, 5000) {
+                    tokio::time::sleep(std::time::Duration::from_millis(wait_ms)).await;
+                    attempt += 1;
+                } else {
+                    return Err(err);
+                }
+            }
+            Err(err) => return Err(err),
+        }
+    }
+
+    Err(sqlx::Error::Protocol("Max retries exceeded".into()))
+}
+
+/// Checks if an error is retryable for SQLite.
+fn is_retryable_sqlite_error(err: &Error) -> bool {
+    matches!(
+        err,
+        Error::Database(db_err) if {
+            let code = db_err.code().unwrap_or_default().to_string();
+            RETRY_ERRORS.contains(code.as_str())
+        }
+    )
+}
+
+/// Calculates the wait time for retries based on SQLite's backoff strategy.
+fn get_wait_ms_sqlite(attempt: usize, timeout: u64) -> Option<u64> {
+    const DELAYS: [u64; 12] = [1, 2, 5, 10, 15, 20, 25, 25, 25, 50, 50, 100];
+    const TOTALS: [u64; 12] = [0, 1, 3, 8, 18, 33, 53, 78, 103, 128, 178, 228];
+
+    if attempt >= DELAYS.len() {
+        let delay = DELAYS.last().unwrap();
+        let prior = TOTALS.last().unwrap() + delay * (attempt as u64 - (DELAYS.len() as u64 - 1));
+        if prior + delay > timeout {
+            return None;
+        }
+        Some(*delay)
+    } else {
+        let delay = DELAYS[attempt];
+        let prior = TOTALS[attempt];
+        if prior + delay > timeout {
+            None
+        } else {
+            Some(delay)
+        }
+    }
+}
+
+/// Checks if an error is a unique constraint violation.
+pub fn is_err_unique_violation(err: &Error) -> bool {
+    matches!(
+        err,
+        Error::Database(db_err) if {
+            let code = db_err.code().unwrap_or_default();
+            code == "23505" || code == "SQLITE_CONSTRAINT_UNIQUE"
+        }
+    )
+}
+
+lazy_static::lazy_static! {
+    /// Set of retryable SQLite error codes.
+    static ref RETRY_ERRORS: HashSet<&'static str> = {
+        let mut set = HashSet::new();
+        set.insert("SQLITE_BUSY");
+        set.insert("SQLITE_BUSY_SNAPSHOT");
+        set.insert("SQLITE_BUSY_RECOVERY");
+        set.insert("SQLITE_BUSY_TIMEOUT");
+        set
+    };
+}