implement actor_store

src/actor_store/blob.rs

···
     /// DID of the actor
     pub did: String,
     /// Database connection
-    pub db: deadpool_diesel::Connection<SqliteConnection>,
+    pub db: deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 }
 
 impl BlobReader {
     /// Create a new blob reader
-    pub fn new(blobstore: BlobStoreSql, db: deadpool_diesel::Connection<SqliteConnection>) -> Self {
+    pub fn new(
+        blobstore: BlobStoreSql,
+        db: deadpool_diesel::Pool<
+            deadpool_diesel::Manager<SqliteConnection>,
+            deadpool_diesel::sqlite::Object,
+        >,
+    ) -> Self {
         BlobReader {
             did: blobstore.did.clone(),
             blobstore,
···
         let did = self.did.clone();
         let found = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 BlobSchema::blob
                     .filter(BlobSchema::did.eq(did))
···
         let did = self.did.clone();
         let res = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 let results = RecordBlobSchema::record_blob
                     .filter(RecordBlobSchema::blobCid.eq(cid.to_string()))
···
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
         let did = self.did.clone();
-        self.db.interact(move |conn| {
+        self.db.get().await?.interact(move |conn| {
             let BlobMetadata {
                 temp_key,
                 size,
···
         let uris_clone = uris.clone();
         let deleted_repo_blobs: Vec<models::RecordBlob> = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 RecordBlobSchema::record_blob
                     .filter(RecordBlobSchema::recordUri.eq_any(&uris_clone))
···
         // Now perform the delete
         let uris_clone = uris.clone();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 delete(RecordBlobSchema::record_blob)
                     .filter(RecordBlobSchema::recordUri.eq_any(uris_clone))
···
         let did_clone = self.did.clone();
         let duplicated_cids: Vec<String> = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 RecordBlobSchema::record_blob
                     .filter(RecordBlobSchema::blobCid.eq_any(cids_clone))
···
         let cids = cids_to_delete.clone();
         let did_clone = self.did.clone();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 delete(BlobSchema::blob)
                     .filter(BlobSchema::cid.eq_any(cids))
···
 
         let found = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 BlobSchema::blob
                     .filter(
···
                     .await?;
             }
             self.db
+                .get()
+                .await?
                 .interact(move |conn| {
                     update(BlobSchema::blob)
                         .filter(BlobSchema::tempKey.eq(found.temp_key))
···
         let did = self.did.clone();
 
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 insert_into(RecordBlobSchema::record_blob)
                     .values((
···
 
         let did = self.did.clone();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 let res = BlobSchema::blob
                     .filter(BlobSchema::did.eq(&did))
···
 
         let did = self.did.clone();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 let res: i64 = RecordBlobSchema::record_blob
                     .filter(RecordBlobSchema::did.eq(&did))
···
 
         let did = self.did.clone();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 let ListMissingBlobsOpts { cursor, limit } = opts;
 
···
                 builder = builder.filter(RecordBlobSchema::blobCid.gt(cursor));
             }
             self.db
+                .get()
+                .await?
                 .interact(move |conn| builder.load(conn))
                 .await
                 .expect("Failed to list blobs")?
···
                 builder = builder.filter(RecordBlobSchema::blobCid.gt(cursor));
             }
             self.db
+                .get()
+                .await?
                 .interact(move |conn| builder.load(conn))
                 .await
                 .expect("Failed to list blobs")?
···
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 let res = BlobSchema::blob
                     .filter(BlobSchema::cid.eq(cid.to_string()))
···
         let did_clone = self.did.clone();
 
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 update(BlobSchema::blob)
                     .filter(BlobSchema::cid.eq(blob_cid))

src/actor_store/mod.rs

···
     pub fn new(
         did: String,
         blobstore: BlobStoreSql,
-        db: deadpool_diesel::Connection<SqliteConnection>,
+        db: deadpool_diesel::Pool<
+            deadpool_diesel::Manager<SqliteConnection>,
+            deadpool_diesel::sqlite::Object,
+        >,
+        conn: deadpool_diesel::sqlite::Object,
     ) -> Self {
         ActorStore {
-            storage: Arc::new(RwLock::new(SqlRepoReader::new(
-                did.clone(),
-                None,
-                db.clone(),
-            ))),
+            storage: Arc::new(RwLock::new(SqlRepoReader::new(did.clone(), None, conn))),
             record: RecordReader::new(did.clone(), db.clone()),
             pref: PreferenceReader::new(did.clone(), db.clone()),
             did,
-            blob: BlobReader::new(blobstore, db.clone()), // Unlike TS impl, just use blob reader vs generator
+            blob: BlobReader::new(blobstore, db.clone()),
         }
     }
 
···
     pub async fn destroy(&mut self) -> Result<()> {
         let did: String = self.did.clone();
         let storage_guard = self.storage.read().await;
-        let db: deadpool_diesel::Connection<SqliteConnection> = storage_guard.db.clone();
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
-        let blob_rows: Vec<String> = db
-            .run(move |conn| {
+        let blob_rows: Vec<String> = storage_guard
+            .db
+            .interact(move |conn| {
                 BlobSchema::blob
                     .filter(BlobSchema::did.eq(did))
                     .select(BlobSchema::cid)
                     .get_results(conn)
             })
-            .await?;
+            .await
+            .expect("Failed to get blob rows")?;
         let cids = blob_rows
             .into_iter()
             .map(|row| Ok(Cid::from_str(&row)?))
···
         }
         let did: String = self.did.clone();
         let storage_guard = self.storage.read().await;
-        let db: deadpool_diesel::Connection<SqliteConnection> = 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();
         let touched_uri_strs: Vec<String> = touched_uris.iter().map(|t| t.to_string()).collect();
-        let res: Vec<String> = db
-            .run(move |conn| {
+        let res: Vec<String> = storage_guard
+            .db
+            .interact(move |conn| {
                 RecordSchema::record
                     .filter(RecordSchema::did.eq(did))
                     .filter(RecordSchema::cid.eq_any(cid_strs))
···
                     .select(RecordSchema::cid)
                     .get_results(conn)
             })
-            .await?;
+            .await
+            .expect("Failed to get duplicate record cids")?;
         res.into_iter()
             .map(|row| Cid::from_str(&row).map_err(|error| anyhow::Error::new(error)))
             .collect::<Result<Vec<Cid>>>()

src/actor_store/preference.rs

···
 
 pub struct PreferenceReader {
     pub did: String,
-    pub db: deadpool_diesel::Connection<SqliteConnection>,
+    pub db: deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 }
 
 impl PreferenceReader {
-    pub fn new(did: String, db: deadpool_diesel::Connection<SqliteConnection>) -> Self {
+    pub fn new(
+        did: String,
+        db: deadpool_diesel::Pool<
+            deadpool_diesel::Manager<SqliteConnection>,
+            deadpool_diesel::sqlite::Object,
+        >,
+    ) -> Self {
         PreferenceReader { did, db }
     }
 
···
 
         let did = self.did.clone();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 let prefs_res = AccountPrefSchema::account_pref
                     .filter(AccountPrefSchema::did.eq(&did))
···
         scope: AuthScope,
     ) -> Result<()> {
         let did = self.did.clone();
-        self.db
+        self.db.get().await?
             .interact(move |conn| {
                 match values
                     .iter()

src/actor_store/record.rs

···
 /// Combined handler for record operations with both read and write capabilities.
 pub(crate) struct RecordReader {
     /// Database connection.
-    pub db: deadpool_diesel::Connection<SqliteConnection>,
+    pub db: deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
     /// DID of the actor.
     pub did: String,
 }
 
 impl RecordReader {
     /// Create a new record handler.
-    pub(crate) fn new(did: String, db: deadpool_diesel::Connection<SqliteConnection>) -> Self {
+    pub(crate) fn new(
+        did: String,
+        db: deadpool_diesel::Pool<
+            deadpool_diesel::Manager<SqliteConnection>,
+            deadpool_diesel::sqlite::Object,
+        >,
+    ) -> Self {
         Self { did, db }
     }
 
···
 
         let other_did = self.did.clone();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 let res: i64 = record.filter(did.eq(&other_did)).count().get_result(conn)?;
                 Ok(res)
···
 
         let other_did = self.did.clone();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 let collections = record
                     .filter(did.eq(&other_did))
···
         }
         let res: Vec<(Record, RepoBlock)> = self
             .db
+            .get()
+            .await?
             .interact(move |conn| builder.load(conn))
             .await
             .expect("Failed to load records")?;
···
         }
         let record: Option<(Record, RepoBlock)> = self
             .db
+            .get()
+            .await?
             .interact(move |conn| builder.first(conn).optional())
             .await
             .expect("Failed to load record")?;
···
         }
         let record_uri = self
             .db
+            .get()
+            .await?
             .interact(move |conn| builder.first::<String>(conn).optional())
             .await
             .expect("Failed to check record")?;
···
 
         let res = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 RecordSchema::record
                     .select(RecordSchema::takedownRef)
···
 
         let res = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 RecordSchema::record
                     .select(RecordSchema::cid)
···
 
         let res = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 RecordSchema::record
                     .inner_join(
···
         // Track current version of record
         let (record, uri) = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 insert_into(RecordSchema::record)
                     .values(row)
···
         use rsky_pds::schema::pds::record::dsl as RecordSchema;
         let uri = uri.to_string();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 delete(RecordSchema::record)
                     .filter(RecordSchema::uri.eq(&uri))
···
         use rsky_pds::schema::pds::backlink::dsl as BacklinkSchema;
         let uri = uri.to_string();
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 delete(BacklinkSchema::backlink)
                     .filter(BacklinkSchema::uri.eq(uri))
···
         } else {
             use rsky_pds::schema::pds::backlink::dsl as BacklinkSchema;
             self.db
+                .get()
+                .await?
                 .interact(move |conn| {
                     insert_or_ignore_into(BacklinkSchema::backlink)
                         .values(&backlinks)
···
         let uri_string = uri.to_string();
 
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 update(RecordSchema::record)
                     .filter(RecordSchema::uri.eq(uri_string))

src/actor_store/sql_blob.rs

···
 /// SQL-based implementation of blob storage
 pub struct BlobStoreSql {
     /// Database connection for metadata
-    pub db: deadpool_diesel::Connection<SqliteConnection>,
+    pub db: deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
     /// DID of the actor
     pub did: String,
 }
···
 
 impl BlobStoreSql {
     /// Create a new SQL-based blob store for the given DID
-    pub fn new(did: String, db: deadpool_diesel::Connection<SqliteConnection>) -> Self {
+    pub fn new(
+        did: String,
+        db: deadpool_diesel::Pool<
+            deadpool_diesel::Manager<SqliteConnection>,
+            deadpool_diesel::sqlite::Object,
+        >,
+    ) -> Self {
         BlobStoreSql { db, did }
     }
 
     // /// Create a factory function for blob stores
     // pub fn creator(
-    //     db: deadpool_diesel::Connection<SqliteConnection>,
+    //     db: deadpool_diesel::Pool<
+    //     deadpool_diesel::Manager<SqliteConnection>,
+    //     deadpool_diesel::sqlite::Object,
+    // >,
     // ) -> Box<dyn Fn(String) -> BlobStoreSql> {
     //     let db_clone = db.clone();
     //     Box::new(move |did: String| BlobStoreSql::new(did, db_clone.clone()))
···
 
         // Store directly in the database
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 let data_clone = bytes.clone();
                 let entry = BlobEntry {
···
 
         // Update the quarantine flag in the database
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 diesel::update(blobs::table)
                     .filter(blobs::cid.eq(&cid_str))
···
 
         // Update the quarantine flag in the database
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 diesel::update(blobs::table)
                     .filter(blobs::cid.eq(&cid_str))
···
         // Get the blob data from the database
         let blob_data = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 blobs
                     .filter(self::blobs::cid.eq(&cid_str))
···
 
         // Delete from database
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 diesel::delete(blobs)
                     .filter(self::blobs::cid.eq(&blob_cid))
···
 
         // Delete all blobs in one operation
         self.db
+            .get()
+            .await?
             .interact(move |conn| {
                 diesel::delete(blobs)
                     .filter(self::blobs::cid.eq_any(cid_strings))
···
 
         let exists = self
             .db
+            .get()
+            .await?
             .interact(move |conn| {
                 diesel::select(diesel::dsl::exists(
                     blobs

src/actor_store/sql_repo.rs

···
 
 pub struct SqlRepoReader {
     pub cache: Arc<RwLock<BlockMap>>,
-    pub db: deadpool_diesel::Connection<SqliteConnection>,
+    pub db: deadpool_diesel::sqlite::Object,
     pub root: Option<Cid>,
     pub rev: Option<String>,
     pub now: String,
···
 
 // Basically handles getting ipld blocks from db
 impl SqlRepoReader {
-    pub fn new(
-        did: String,
-        now: Option<String>,
-        db: deadpool_diesel::Connection<SqliteConnection>,
-    ) -> Self {
+    pub fn new(did: String, now: Option<String>, db: deadpool_diesel::sqlite::Object) -> Self {
         let now = now.unwrap_or_else(rsky_common::now);
         SqlRepoReader {
             cache: Arc::new(RwLock::new(BlockMap::new())),

src/endpoints/repo/apply_writes.rs

···
         let actor_db = db_actors
             .get(did)
             .ok_or_else(|| anyhow!("Actor DB not found"))?;
+        let conn = actor_db
+            .repo
+            .get()
+            .await
+            .context("Failed to get actor db connection")?;
         let mut actor_store = ActorStore::new(
             did.clone(),
             BlobStoreSql::new(did.clone(), actor_db.blob),
             actor_db.repo,
+            conn,
         );
 
         let commit = actor_store