prototype actor_store and db

Cargo.lock

···
  "constcat",
  "diesel",
  "diesel_migrations",
+ "dotenvy",
  "figment",
  "futures",
  "hex",
···
  "regex",
  "reqwest 0.12.15",
  "reqwest-middleware",
+ "rocket_sync_db_pools",
  "rsky-common",
  "rsky-lexicon",
  "rsky-pds",

Cargo.toml

···
 async-trait = "0.1.88"
 lazy_static = "1.5.0"
 secp256k1 = "0.28.2"
+dotenvy = "0.15.7"
+[dependencies.rocket_sync_db_pools]
+version = "=0.1.0"
+features = ["diesel_sqlite_pool"]

src/actor_store/actor_store.rs

···
 use std::{env, fmt};
 use tokio::sync::RwLock;
 
-use super::ActorDb;
+use crate::db::DbConn;
+
 use super::blob::BlobReader;
 use super::preference::PreferenceReader;
 use super::record::RecordReader;
···
 // Combination of RepoReader/Transactor, BlobReader/Transactor, SqlRepoReader/Transactor
 impl ActorStore {
     /// Concrete reader of an individual repo (hence BlobStoreSql which takes `did` param)
-    pub fn new(did: String, blobstore: BlobStoreSql, db: ActorDb) -> Self {
+    pub fn new(did: String, blobstore: BlobStoreSql, db: DbConn) -> Self {
+        let db = Arc::new(db);
         ActorStore {
             storage: Arc::new(RwLock::new(SqlRepoReader::new(
                 did.clone(),
···
     pub async fn destroy(&mut self) -> Result<()> {
         let did: String = self.did.clone();
         let storage_guard = self.storage.read().await;
-        let db: ActorDb = storage_guard.db.clone();
+        let db: Arc<DbConn> = storage_guard.db.clone();
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
         let blob_rows: Vec<String> = db
···
         }
         let did: String = self.did.clone();
         let storage_guard = self.storage.read().await;
-        let db: ActorDb = storage_guard.db.clone();
+        let db: Arc<DbConn> = storage_guard.db.clone();
         use rsky_pds::schema::pds::record::dsl as RecordSchema;
 
         let cid_strs: Vec<String> = cids.into_iter().map(|c| c.to_string()).collect();

src/actor_store/blob.rs

···
 //!
 //! Modified for SQLite backend
 
-use anyhow::{Result, bail};
+use std::sync::Arc;
+
+use anyhow::{Error, Result, bail};
 use cidv10::Cid;
 use diesel::dsl::{count_distinct, exists, not};
-use diesel::result::Error;
 use diesel::sql_types::{Integer, Nullable, Text};
 use diesel::*;
 use futures::stream::{self, StreamExt};
···
 use rsky_repo::types::{PreparedBlobRef, PreparedWrite};
 use sha2::Digest;
 
-use super::ActorDb;
 use super::sql_blob::BlobStoreSql;
+use crate::db::DbConn;
 
 pub struct BlobReader {
     pub blobstore: BlobStoreSql,
     pub did: String,
-    pub db: ActorDb,
+    pub db: Arc<DbConn>,
 }
 
 // Basically handles getting blob records from db
 impl BlobReader {
-    pub fn new(blobstore: BlobStoreSql, db: ActorDb) -> Self {
+    pub fn new(blobstore: BlobStoreSql, db: Arc<DbConn>) -> Self {
         // BlobReader {
         //     did: blobstore.bucket.clone(),
         //     blobstore,

src/actor_store/db.rs

···
-//! Database schema and connection management for the actor store.
-
-use crate::db::DatabaseConnection;
-use anyhow::{Context as _, Result};
-
-/// Type alias for the actor database.
-pub(crate) type ActorDb = DatabaseConnection;
-
-/// Gets a database connection for the actor store.
-///
-/// # Arguments
-///
-/// * `location` - The file path or URI for the SQLite database.
-/// * `disable_wal_auto_checkpoint` - Whether to disable the WAL auto-checkpoint.
-///
-/// # Returns
-///
-/// A `Result` containing the `ActorDb` instance or an error.
-pub async fn get_db(location: &str, disable_wal_auto_checkpoint: bool) -> Result<ActorDb> {
-    let pragmas = if disable_wal_auto_checkpoint {
-        Some(
-            &[
-                ("wal_autocheckpoint", "0"),
-                ("journal_mode", "WAL"),
-                ("synchronous", "NORMAL"),
-                ("foreign_keys", "ON"),
-            ][..],
-        )
-    } else {
-        Some(
-            &[
-                ("journal_mode", "WAL"),
-                ("synchronous", "NORMAL"),
-                ("foreign_keys", "ON"),
-            ][..],
-        )
-    };
-
-    let db = DatabaseConnection::new(location, pragmas)
-        .await
-        .context("Failed to initialize the actor database")?;
-
-    // Ensure WAL mode is properly set up
-    db.ensure_wal().await?;
-
-    // Run migrations
-    // TODO: make sure the migrations are populated?
-    db.run_migrations()
-        .context("Failed to run migrations on the actor database")?;
-
-    Ok(db)
-}

src/actor_store/mod.rs

···
 
 mod actor_store;
 mod blob;
-mod db;
 mod preference;
 mod record;
 mod sql_blob;
 mod sql_repo;
-
-pub(crate) use db::ActorDb;

src/actor_store/preference.rs

···
 //!
 //! Modified for SQLite backend
 
+use std::sync::Arc;
+
 use anyhow::{Result, bail};
 use diesel::*;
 use rsky_lexicon::app::bsky::actor::RefPreferences;
···
 use rsky_pds::auth_verifier::AuthScope;
 use rsky_pds::models::AccountPref;
 
-use super::ActorDb;
+use crate::db::DbConn;
 
 pub struct PreferenceReader {
     pub did: String,
-    pub db: ActorDb,
+    pub db: Arc<DbConn>,
 }
 
 impl PreferenceReader {
-    pub fn new(did: String, db: ActorDb) -> Self {
+    pub fn new(did: String, db: Arc<DbConn>) -> Self {
         PreferenceReader { did, db }
     }
 

src/actor_store/record.rs

···
 use rsky_syntax::aturi::AtUri;
 use std::env;
 use std::str::FromStr;
+use std::sync::Arc;
 
-use crate::actor_store::db::ActorDb;
+use crate::db::DbConn;
 
 /// Combined handler for record operations with both read and write capabilities.
 pub(crate) struct RecordReader {
     /// Database connection.
-    pub db: ActorDb,
+    pub db: Arc<DbConn>,
     /// DID of the actor.
     pub did: String,
 }
 
 impl RecordReader {
     /// Create a new record handler.
-    pub(crate) fn new(did: String, db: ActorDb) -> Self {
+    pub(crate) fn new(did: String, db: Arc<DbConn>) -> Self {
         Self { did, db }
     }
 
···
         let record_backlinks = get_backlinks(uri, record)?;
         let conflicts: Vec<Vec<Record>> = stream::iter(record_backlinks)
             .then(|backlink| async move {
-                Ok::<Vec<Record>, Error>(
+                Ok::<Vec<Record>, anyhow::Error>(
                     self.get_record_backlinks(
                         uri.get_collection(),
                         backlink.path,

src/actor_store/sql_blob.rs

···
-use std::{path::PathBuf, str::FromStr as _};
+use std::{path::PathBuf, str::FromStr as _, sync::Arc};
 
 use anyhow::Result;
 use cidv10::Cid;
 use rsky_common::get_random_str;
 
-use crate::db::DatabaseConnection;
+use crate::db::DbConn;
 
 /// Type for stream of blob data
 pub type BlobStream = Box<dyn std::io::Read + Send>;
···
 /// Placeholder implementation for blob store
 #[derive(Clone)]
 pub(crate) struct BlobStoreSql {
-    client: DatabaseConnection,
+    client: Arc<DbConn>,
     path: PathBuf,
 }
 

src/actor_store/sql_repo.rs

···
-//! Based on https://github.com/blacksky-algorithms/rsky/blob/main/rsky-pds/src/actor_store/repo/sql_repo.rs
+//! Based on https://github.com/blacksky-algorithms/rsky/blob/main/rsky-pds/src/actor_store/record/mod.rs
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
···
 use std::sync::Arc;
 use tokio::sync::RwLock;
 
-use super::ActorDb;
+use crate::db::DbConn;
 
 #[derive(Clone, Debug)]
 pub struct SqlRepoReader {
     pub cache: Arc<RwLock<BlockMap>>,
-    pub db: ActorDb,
+    pub db: Arc<DbConn>,
     pub root: Option<Cid>,
     pub rev: Option<String>,
     pub now: String,
···
         cid: &'a Cid,
     ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>>> + Send + Sync + 'a>> {
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
         let cid = cid.clone();
 
         Box::pin(async move {
···
         cids: Vec<Cid>,
     ) -> Pin<Box<dyn Future<Output = Result<BlocksAndMissing>> + Send + Sync + 'a>> {
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
 
         Box::pin(async move {
             use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
···
         rev: String,
     ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'a>> {
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
         let bytes_cloned = bytes.clone();
         Box::pin(async move {
             use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
···
         rev: String,
     ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'a>> {
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
 
         Box::pin(async move {
             use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
···
         is_create: Option<bool>,
     ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'a>> {
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
         let now: String = self.now.clone();
 
         Box::pin(async move {
···
 
 // Basically handles getting ipld blocks from db
 impl SqlRepoReader {
-    pub fn new(did: String, now: Option<String>, db: ActorDb) -> Self {
+    pub fn new(did: String, now: Option<String>, db: Arc<DbConn>) -> Self {
         let now = now.unwrap_or_else(rsky_common::now);
         SqlRepoReader {
             cache: Arc::new(RwLock::new(BlockMap::new())),
···
         cursor: &Option<CidAndRev>,
     ) -> Result<Vec<RepoBlock>> {
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
         let since = since.clone();
         let cursor = cursor.clone();
         use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
···
 
     pub async fn count_blocks(&self) -> Result<i64> {
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
         use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
 
         let res = db
···
     /// Proactively cache all blocks from a particular commit (to prevent multiple roundtrips)
     pub async fn cache_rev(&mut self, rev: String) -> Result<()> {
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
         use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
 
-        let res: Vec<(String, Vec<u8>)> = db
+        let result: Vec<(String, Vec<u8>)> = db
             .run(move |conn| {
                 RepoBlockSchema::repo_block
                     .filter(RepoBlockSchema::did.eq(did))
···
                     .get_results::<(String, Vec<u8>)>(conn)
             })
             .await?;
-        for row in res {
+        for row in result {
             let mut cache_guard = self.cache.write().await;
             cache_guard.set(Cid::from_str(&row.0)?, row.1)
         }
···
             return Ok(());
         }
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
         use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
 
         let cid_strings: Vec<String> = cids.into_iter().map(|c| c.to_string()).collect();
···
 
     pub async fn get_root_detailed(&self) -> Result<CidAndRev> {
         let did: String = self.did.clone();
-        let db: ActorDb = self.db.clone();
+        let db: Arc<DbConn> = self.db.clone();
         use rsky_pds::schema::pds::repo_root::dsl as RepoRootSchema;
 
         let res = db

src/db/mod.rs

···
-use anyhow::{Context, Result};
-use diesel::connection::SimpleConnection;
-use diesel::r2d2::{ConnectionManager, Pool, PooledConnection};
-use diesel::sqlite::Sqlite;
-use diesel::*;
-use diesel_migrations::{EmbeddedMigrations, MigrationHarness, embed_migrations};
+use anyhow::Result;
+use diesel::prelude::*;
+use dotenvy::dotenv;
+use rocket_sync_db_pools::database;
 use std::env;
-use std::path::Path;
-use std::time::Duration;
+use std::fmt::{Debug, Formatter};
 
-pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("migrations");
-pub type SqlitePool = Pool<ConnectionManager<SqliteConnection>>;
-pub type SqlitePooledConnection = PooledConnection<ConnectionManager<SqliteConnection>>;
+#[database("sqlite_db")]
+pub struct DbConn(PgConnection);
 
-/// Database type for all queries
-pub type DbType = Sqlite;
+impl Debug for DbConn {
+    fn fmt(&self, _f: &mut Formatter<'_>) -> std::fmt::Result {
+        todo!()
+    }
+}
 
 #[tracing::instrument(skip_all)]
-pub async fn establish_connection_for_sequencer() -> Result<DatabaseConnection> {
+pub fn establish_connection_for_sequencer() -> Result<SqliteConnection> {
+    dotenv().ok();
+    tracing::debug!("Establishing database connection for Sequencer");
     let database_url = env::var("BLUEPDS_DB").unwrap_or("sqlite://data/sqlite.db".into());
-    let pragmas = Some(
-        &[
-            ("journal_mode", "WAL"),
-            ("synchronous", "NORMAL"),
-            ("foreign_keys", "ON"),
-        ][..],
-    );
-    let db = DatabaseConnection::new(&database_url, pragmas)
-        .await
-        .context("Failed to initialize the actor database")?;
-    db.ensure_wal().await?;
-    db.run_migrations()
-        .context("Failed to run migrations on the actor database")?;
+    let db = SqliteConnection::establish(&database_url).map_err(|error| {
+        let context = format!("Error connecting to {database_url:?}");
+        anyhow::Error::new(error).context(context)
+    })?;
     Ok(db)
 }
-
-/// Database connection wrapper
-#[derive(Clone, Debug)]
-pub struct DatabaseConnection {
-    pub pool: SqlitePool,
-}
-
-impl DatabaseConnection {
-    /// Create a new database connection with optional pragmas
-    pub async fn new(path: &str, pragmas: Option<&[(&str, &str)]>) -> Result<Self> {
-        // Create the database directory if it doesn't exist
-        if let Some(parent) = Path::new(path).parent() {
-            if !parent.exists() {
-                tokio::fs::create_dir_all(parent)
-                    .await
-                    .context(format!("Failed to create directory: {:?}", parent))?;
-            }
-        }
-
-        // Sanitize the path for connection string
-        let database_url = format!("sqlite:{}", path);
-
-        // Create a connection manager
-        let manager = ConnectionManager::<SqliteConnection>::new(database_url);
-
-        // Create the connection pool with SQLite-specific configurations
-        let pool = Pool::builder()
-            .max_size(10)
-            .connection_timeout(Duration::from_secs(30))
-            .test_on_check_out(true)
-            .build(manager)
-            .context("Failed to create connection pool")?;
-
-        // Initialize the database with pragmas
-        if let Some(pragmas) = pragmas {
-            let conn = &mut pool.get().context("Failed to get connection from pool")?;
-
-            // Apply all pragmas
-            for (pragma, value) in pragmas {
-                let sql = format!("PRAGMA {} = {}", pragma, value);
-                conn.batch_execute(&sql)
-                    .context(format!("Failed to set pragma {}", pragma))?;
-            }
-        }
-
-        let db = DatabaseConnection { pool };
-        Ok(db)
-    }
-
-    /// Run migrations on the database
-    pub fn run_migrations(&self) -> Result<()> {
-        let mut conn = self
-            .pool
-            .get()
-            .context("Failed to get connection for migrations")?;
-
-        conn.run_pending_migrations(MIGRATIONS)
-            .map_err(|e| anyhow::anyhow!("Failed to run migrations: {}", e))?;
-
-        Ok(())
-    }
-
-    /// Ensure WAL mode is enabled
-    pub async fn ensure_wal(&self) -> Result<()> {
-        let conn = &mut self.pool.get().context("Failed to get connection")?;
-        conn.batch_execute("PRAGMA journal_mode = WAL;")?;
-        conn.batch_execute("PRAGMA synchronous = NORMAL;")?;
-        conn.batch_execute("PRAGMA foreign_keys = ON;")?;
-        Ok(())
-    }
-
-    /// Execute a database operation with retries for busy errors
-    pub async fn run<F, T>(&self, operation: F) -> Result<T>
-    where
-        F: FnOnce(&mut SqliteConnection) -> QueryResult<T> + Send,
-        T: Send + 'static,
-    {
-        let mut retries = 0;
-        let max_retries = 5;
-        let mut last_error = None;
-
-        while retries < max_retries {
-            let mut conn = self.pool.get().context("Failed to get connection")?;
-            match operation(&mut conn) {
-                Ok(result) => return Ok(result),
-                // TODO: Busy error handling
-                // Err(diesel::result::Error::DatabaseError(
-                //     diesel::result::DatabaseErrorKind::DatabaseIsLocked,
-                //     _,
-                // )) => {
-                //     retries += 1;
-                //     let backoff_ms = 10 * (1 << retries); // Exponential backoff
-                //     last_error = Some(diesel::result::Error::DatabaseError(
-                //         diesel::result::DatabaseErrorKind::DatabaseIsLocked,
-                //         Box::new("Database is locked".to_string()),
-                //     ));
-                //     tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
-                // }
-                Err(e) => return Err(e.into()),
-            }
-        }
-
-        Err(anyhow::anyhow!(
-            "Max retries exceeded: {}",
-            last_error.unwrap_or_else(|| result::Error::RollbackTransaction)
-        ))
-    }
-
-    /// Check if currently in a transaction
-    pub fn assert_transaction(&self) -> Result<()> {
-        // SQLite doesn't have a straightforward way to check transaction state
-        // We'll implement a simplified version that just returns Ok for now
-        Ok(())
-    }
-
-    /// Run a transaction with retry logic for busy database errors
-    pub async fn transaction<T, F>(&self, f: F) -> Result<T>
-    where
-        F: FnOnce(&mut SqliteConnection) -> Result<T> + Send,
-        T: Send + 'static,
-    {
-        self.run(|conn| {
-            conn.transaction(|tx| f(tx).map_err(|e| result::Error::RollbackTransaction))
-        })
-        .await
-    }
-
-    /// Run a transaction with no retry logic
-    pub async fn transaction_no_retry<T, F>(&self, f: F) -> Result<T>
-    where
-        F: FnOnce(&mut SqliteConnection) -> std::result::Result<T, result::Error> + Send,
-        T: Send + 'static,
-    {
-        let mut conn = self
-            .pool
-            .get()
-            .context("Failed to get connection for transaction")?;
-
-        conn.transaction(|tx| f(tx))
-            .map_err(|e| anyhow::anyhow!("Transaction error: {:?}", e))
-    }
-}
-
-/// Create a connection pool for SQLite
-pub async fn create_sqlite_pool(database_url: &str) -> Result<SqlitePool> {
-    let manager = ConnectionManager::<SqliteConnection>::new(database_url);
-    let pool = Pool::builder()
-        .max_size(10)
-        .connection_timeout(Duration::from_secs(30))
-        .test_on_check_out(true)
-        .build(manager)
-        .context("Failed to create connection pool")?;
-
-    // Apply recommended SQLite settings
-    let conn = &mut pool.get()?;
-    conn.batch_execute(
-        "PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL; PRAGMA foreign_keys = ON;",
-    )?;
-
-    Ok(pool)
-}