move account data to per-actor db

src/account_manager/helpers/account.rs

···
         deadpool_diesel::Manager<SqliteConnection>,
         deadpool_diesel::sqlite::Object,
     >,
+    actor_db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
+    use crate::schema::actor_store::repo_root::dsl as RepoRootSchema;
     use crate::schema::pds::email_token::dsl as EmailTokenSchema;
     use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
-    use crate::schema::pds::repo_root::dsl as RepoRootSchema;
 
+    let did_clone = did.to_owned();
+    _ = actor_db
+        .get()
+        .await?
+        .interact(move |conn| {
+            delete(RepoRootSchema::repo_root)
+                .filter(RepoRootSchema::did.eq(&did_clone))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to delete actor")?;
     let did = did.to_owned();
     _ = db
         .get()
         .await?
         .interact(move |conn| {
-            _ = delete(RepoRootSchema::repo_root)
-                .filter(RepoRootSchema::did.eq(&did))
-                .execute(conn)?;
             _ = delete(EmailTokenSchema::email_token)
                 .filter(EmailTokenSchema::did.eq(&did))
                 .execute(conn)?;

src/account_manager/helpers/repo.rs

···
 //! Modified for SQLite backend
 use anyhow::Result;
 use cidv10::Cid;
+use deadpool_diesel::{Manager, Pool, sqlite::Object};
 use diesel::*;
 
 pub async fn update_root(
     did: String,
     cid: Cid,
     rev: String,
-    db: &deadpool_diesel::Pool<
-        deadpool_diesel::Manager<SqliteConnection>,
-        deadpool_diesel::sqlite::Object,
-    >,
+    db: &Pool<Manager<SqliteConnection>, Object>,
 ) -> Result<()> {
     // @TODO balance risk of a race in the case of a long retry
-    use crate::schema::pds::repo_root::dsl as RepoRootSchema;
+    use crate::schema::actor_store::repo_root::dsl as RepoRootSchema;
 
     let now = rsky_common::now();
 

src/account_manager/mod.rs

···
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
+use crate::ActorPools;
 use crate::account_manager::helpers::account::{
     AccountStatus, ActorAccount, AvailabilityFlags, GetAccountAdminStatusOutput,
 };
···
 use chrono::DateTime;
 use chrono::offset::Utc as UtcOffset;
 use cidv10::Cid;
+use deadpool_diesel::sqlite::Pool;
 use diesel::*;
 use futures::try_join;
 use helpers::{account, auth, email_token, invite, password, repo};
···
         }
     }
 
-    pub async fn create_account(&self, opts: CreateAccountOpts) -> Result<(String, String)> {
+    pub async fn create_account(
+        &self,
+        opts: CreateAccountOpts,
+        actor_pools: &mut std::collections::HashMap<String, ActorPools>,
+    ) -> Result<(String, String)> {
         let CreateAccountOpts {
             did,
             handle,
···
         }
         invite::record_invite_use(did.clone(), invite_code, now, &self.db).await?;
         auth::store_refresh_token(refresh_payload, None, &self.db).await?;
-        repo::update_root(did, repo_cid, repo_rev, &self.db).await?;
+
+        let did_path = did
+            .strip_prefix("did:plc:")
+            .ok_or_else(|| anyhow::anyhow!("Invalid DID"))?;
+        let path_repo = format!("sqlite://{}", did_path);
+        let actor_repo_pool =
+            crate::establish_pool(path_repo.as_str()).expect("Failed to establish pool");
+        let path_blob = path_repo.replace("repo", "blob");
+        let actor_blob_pool = crate::establish_pool(&path_blob).expect("Failed to establish pool");
+        let actor_pool = ActorPools {
+            repo: actor_repo_pool,
+            blob: actor_blob_pool,
+        };
+        actor_pools
+            .insert(did.clone(), actor_pool)
+            .expect("Failed to insert actor pools");
+        let db = actor_pools
+            .get(&did)
+            .ok_or_else(|| anyhow::anyhow!("Actor not found"))?
+            .repo
+            .clone();
+        repo::update_root(did, repo_cid, repo_rev, &db).await?;
         Ok((access_jwt, refresh_jwt))
     }
 
···
         account::get_account_admin_status(did, &self.db).await
     }
 
-    pub async fn update_repo_root(&self, did: String, cid: Cid, rev: String) -> Result<()> {
-        repo::update_root(did, cid, rev, &self.db).await
+    pub async fn update_repo_root(
+        &self,
+        did: String,
+        cid: Cid,
+        rev: String,
+        actor_pools: &std::collections::HashMap<String, ActorPools>,
+    ) -> Result<()> {
+        let db = actor_pools
+            .get(&did)
+            .ok_or_else(|| anyhow::anyhow!("Actor not found"))?
+            .repo
+            .clone();
+        repo::update_root(did, cid, rev, &db).await
     }
 
-    pub async fn delete_account(&self, did: &str) -> Result<()> {
-        account::delete_account(did, &self.db).await
+    pub async fn delete_account(
+        &self,
+        did: &str,
+        actor_pools: &std::collections::HashMap<String, ActorPools>,
+    ) -> Result<()> {
+        let db = actor_pools
+            .get(did)
+            .ok_or_else(|| anyhow::anyhow!("Actor not found"))?
+            .repo
+            .clone();
+        account::delete_account(did, &self.db, &db).await
     }
 
     pub async fn takedown_account(&self, did: &str, takedown: StatusAttr) -> Result<()> {

src/actor_store/blob.rs

···
 //!
 //! Modified for SQLite backend
 
-use crate::models::pds as models;
+use crate::models::actor_store as models;
 use anyhow::{Result, bail};
 use cidv10::Cid;
 use diesel::dsl::{count_distinct, exists, not};
···
 
     /// Get metadata for a blob by CID
     pub async fn get_blob_metadata(&self, cid: Cid) -> Result<GetBlobMetadataOutput> {
-        use crate::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::actor_store::blob::dsl as BlobSchema;
 
         let did = self.did.clone();
         let found = self
···
 
     /// Get all records that reference a specific blob
     pub async fn get_records_for_blob(&self, cid: Cid) -> Result<Vec<String>> {
-        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::actor_store::record_blob::dsl as RecordBlobSchema;
 
         let did = self.did.clone();
         let res = self
···
 
     /// Track a blob that hasn't been associated with any records yet
     pub async fn track_untethered_blob(&self, metadata: BlobMetadata) -> Result<BlobRef> {
-        use crate::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::actor_store::blob::dsl as BlobSchema;
 
         let did = self.did.clone();
         self.db.get().await?.interact(move |conn| {
···
 
     /// Delete blobs that are no longer referenced by any records
     pub async fn delete_dereferenced_blobs(&self, writes: Vec<PreparedWrite>) -> Result<()> {
-        use crate::schema::pds::blob::dsl as BlobSchema;
-        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::actor_store::blob::dsl as BlobSchema;
+        use crate::schema::actor_store::record_blob::dsl as RecordBlobSchema;
 
         // Extract URIs
         let uris: Vec<String> = writes
···
 
     /// Verify a blob and make it permanent
     pub async fn verify_blob_and_make_permanent(&self, blob: PreparedBlobRef) -> Result<()> {
-        use crate::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::actor_store::blob::dsl as BlobSchema;
 
         let found = self
             .db
···
 
     /// Associate a blob with a record
     pub async fn associate_blob(&self, blob: PreparedBlobRef, record_uri: String) -> Result<()> {
-        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::actor_store::record_blob::dsl as RecordBlobSchema;
 
         let cid = blob.cid.to_string();
         let did = self.did.clone();
···
 
     /// Count all blobs for this actor
     pub async fn blob_count(&self) -> Result<i64> {
-        use crate::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::actor_store::blob::dsl as BlobSchema;
 
         let did = self.did.clone();
         self.db
···
 
     /// Count blobs associated with records
     pub async fn record_blob_count(&self) -> Result<i64> {
-        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::actor_store::record_blob::dsl as RecordBlobSchema;
 
         let did = self.did.clone();
         self.db
···
         &self,
         opts: ListMissingBlobsOpts,
     ) -> Result<Vec<ListMissingBlobsRefRecordBlob>> {
-        use crate::schema::pds::blob::dsl as BlobSchema;
-        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::actor_store::blob::dsl as BlobSchema;
+        use crate::schema::actor_store::record_blob::dsl as RecordBlobSchema;
 
         let did = self.did.clone();
         self.db
···
 
     /// List all blobs with optional filtering
     pub async fn list_blobs(&self, opts: ListBlobsOpts) -> Result<Vec<String>> {
-        use crate::schema::pds::record::dsl as RecordSchema;
-        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
+        use crate::schema::actor_store::record_blob::dsl as RecordBlobSchema;
 
         let ListBlobsOpts {
             since,
···
 
     /// Get the takedown status of a blob
     pub async fn get_blob_takedown_status(&self, cid: Cid) -> Result<Option<StatusAttr>> {
-        use crate::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::actor_store::blob::dsl as BlobSchema;
 
         self.db
             .get()
···
 
     /// Update the takedown status of a blob
     pub async fn update_blob_takedown_status(&self, blob: Cid, takedown: StatusAttr) -> Result<()> {
-        use crate::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::actor_store::blob::dsl as BlobSchema;
 
         let takedown_ref: Option<String> = match takedown.applied {
             true => takedown.r#ref.map_or_else(|| Some(now()), Some),

src/actor_store/mod.rs

···
 
     pub async fn destroy(&mut self) -> Result<()> {
         let did: String = self.did.clone();
-        use crate::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::actor_store::blob::dsl as BlobSchema;
 
         let blob_rows: Vec<String> = self
             .storage
···
             return Ok(vec![]);
         }
         let did: String = self.did.clone();
-        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::actor_store::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();

src/actor_store/preference.rs

···
 //!
 //! Modified for SQLite backend
 
-use crate::models::pds::AccountPref;
+use crate::models::actor_store::AccountPref;
 use anyhow::{Result, bail};
 use diesel::*;
 use rsky_lexicon::app::bsky::actor::RefPreferences;
···
         namespace: Option<String>,
         scope: AuthScope,
     ) -> Result<Vec<RefPreferences>> {
-        use crate::schema::pds::account_pref::dsl as AccountPrefSchema;
+        use crate::schema::actor_store::account_pref::dsl as AccountPrefSchema;
 
         let did = self.did.clone();
         self.db
···
                             bail!("Do not have authorization to set preferences.");
                         }
                         // get all current prefs for user and prep new pref rows
-                        use crate::schema::pds::account_pref::dsl as AccountPrefSchema;
+                        use crate::schema::actor_store::account_pref::dsl as AccountPrefSchema;
                         let all_prefs = AccountPrefSchema::account_pref
                             .filter(AccountPrefSchema::did.eq(&did))
                             .select(AccountPref::as_select())

src/actor_store/record.rs

···
 //!
 //! Modified for SQLite backend
 
-use crate::models::pds::{Backlink, Record, RepoBlock};
+use crate::models::actor_store::{Backlink, Record, RepoBlock};
 use anyhow::{Result, bail};
 use cidv10::Cid;
 use diesel::result::Error;
···
 
     /// Count the total number of records.
     pub(crate) async fn record_count(&mut self) -> Result<i64> {
-        use crate::schema::pds::record::dsl::*;
+        use crate::schema::actor_store::record::dsl::*;
 
         let other_did = self.did.clone();
         self.db
···
 
     /// List all collections in the repository.
     pub(crate) async fn list_collections(&self) -> Result<Vec<String>> {
-        use crate::schema::pds::record::dsl::*;
+        use crate::schema::actor_store::record::dsl::*;
 
         let other_did = self.did.clone();
         self.db
···
         rkey_end: Option<String>,
         include_soft_deleted: Option<bool>,
     ) -> Result<Vec<RecordsForCollection>> {
-        use crate::schema::pds::record::dsl as RecordSchema;
-        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
+        use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
 
         let include_soft_deleted: bool = include_soft_deleted.unwrap_or(false);
         let mut builder = RecordSchema::record
···
         cid: Option<String>,
         include_soft_deleted: Option<bool>,
     ) -> Result<Option<GetRecord>> {
-        use crate::schema::pds::record::dsl as RecordSchema;
-        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
+        use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
 
         let include_soft_deleted: bool = include_soft_deleted.unwrap_or(false);
         let mut builder = RecordSchema::record
···
         cid: Option<String>,
         include_soft_deleted: Option<bool>,
     ) -> Result<bool> {
-        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
 
         let include_soft_deleted: bool = include_soft_deleted.unwrap_or(false);
         let mut builder = RecordSchema::record
···
         &self,
         uri: String,
     ) -> Result<Option<StatusAttr>> {
-        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
 
         let res = self
             .db
···
 
     /// Get the current CID for a record URI.
     pub(crate) async fn get_current_record_cid(&self, uri: String) -> Result<Option<Cid>> {
-        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
 
         let res = self
             .db
···
         path: String,
         link_to: String,
     ) -> Result<Vec<Record>> {
-        use crate::schema::pds::backlink::dsl as BacklinkSchema;
-        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::actor_store::backlink::dsl as BacklinkSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
 
         let res = self
             .db
···
             bail!("Expected indexed URI to contain a record key")
         }
 
-        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
 
         // Track current version of record
         let (record, uri) = self
···
     #[tracing::instrument(skip_all)]
     pub(crate) async fn delete_record(&self, uri: &AtUri) -> Result<()> {
         tracing::debug!("@LOG DEBUG RecordReader::delete_record, deleting indexed record {uri}");
-        use crate::schema::pds::backlink::dsl as BacklinkSchema;
-        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::actor_store::backlink::dsl as BacklinkSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
         let uri = uri.to_string();
         self.db
             .get()
···
 
     /// Remove backlinks for a URI.
     pub(crate) async fn remove_backlinks_by_uri(&self, uri: &AtUri) -> Result<()> {
-        use crate::schema::pds::backlink::dsl as BacklinkSchema;
+        use crate::schema::actor_store::backlink::dsl as BacklinkSchema;
         let uri = uri.to_string();
         self.db
             .get()
···
         if backlinks.is_empty() {
             Ok(())
         } else {
-            use crate::schema::pds::backlink::dsl as BacklinkSchema;
+            use crate::schema::actor_store::backlink::dsl as BacklinkSchema;
             self.db
                 .get()
                 .await?
···
         uri: &AtUri,
         takedown: StatusAttr,
     ) -> Result<()> {
-        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::actor_store::record::dsl as RecordSchema;
 
         let takedown_ref: Option<String> = match takedown.applied {
             true => takedown

src/actor_store/sql_repo.rs

···
 //!
 //! Modified for SQLite backend
 
-use crate::models::pds as models;
-use crate::models::pds::RepoBlock;
+use crate::models::actor_store as models;
+use crate::models::actor_store::RepoBlock;
 use anyhow::Result;
 use cidv10::Cid;
 use diesel::dsl::sql;
···
         let cid = *cid;
 
         Box::pin(async move {
-            use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+            use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
             let cached = {
                 let cache_guard = self.cache.read().await;
                 cache_guard.get(cid).cloned()
···
         let did: String = self.did.clone();
 
         Box::pin(async move {
-            use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+            use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
             let cached = {
                 let mut cache_guard = self.cache.write().await;
                 cache_guard.get_many(cids)?
···
         let did: String = self.did.clone();
         let bytes_cloned = bytes.clone();
         Box::pin(async move {
-            use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+            use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
 
             _ = self
                 .db
···
         let did: String = self.did.clone();
 
         Box::pin(async move {
-            use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+            use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
 
             let blocks: Vec<RepoBlock> = to_put
                 .map
···
         let now: String = self.now.clone();
 
         Box::pin(async move {
-            use crate::schema::pds::repo_root::dsl as RepoRootSchema;
+            use crate::schema::actor_store::repo_root::dsl as RepoRootSchema;
 
             let is_create = is_create.unwrap_or(false);
             if is_create {
···
         let did: String = self.did.clone();
         let since = since.clone();
         let cursor = cursor.clone();
-        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
 
         Ok(self
             .db
···
 
     pub async fn count_blocks(&self) -> Result<i64> {
         let did: String = self.did.clone();
-        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
 
         let res = self
             .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();
-        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
 
         let result: Vec<(String, Vec<u8>)> = self
             .db
···
             return Ok(());
         }
         let did: String = self.did.clone();
-        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::actor_store::repo_block::dsl as RepoBlockSchema;
 
         let cid_strings: Vec<String> = cids.into_iter().map(|c| c.to_string()).collect();
         _ = self
···
 
     pub async fn get_root_detailed(&self) -> Result<CidAndRev> {
         let did: String = self.did.clone();
-        use crate::schema::pds::repo_root::dsl as RepoRootSchema;
+        use crate::schema::actor_store::repo_root::dsl as RepoRootSchema;
 
         let res = self
             .db

src/apis/com/atproto/repo/apply_writes.rs

···
                 did.to_string(),
                 commit.commit_data.cid,
                 commit.commit_data.rev,
+                &actor_pools,
             )
             .await?;
         Ok(())

src/models.rs

···
     #[derive(
         Queryable,
         Identifiable,
-        Insertable,
-        Selectable,
-        Clone,
-        Debug,
-        PartialEq,
-        Default,
-        Serialize,
-        Deserialize,
-    )]
-    #[diesel(table_name = crate::schema::pds::account_pref)]
-    #[diesel(check_for_backend(Sqlite))]
-    pub struct AccountPref {
-        pub id: i32,
-        pub name: String,
-        #[diesel(column_name = valueJson)]
-        #[serde(rename = "valueJson")]
-        pub value_json: Option<String>,
-    }
-
-    #[derive(
-        Queryable,
-        Identifiable,
         Selectable,
         Clone,
         Debug,
···
         #[diesel(column_name = createdAt)]
         #[serde(rename = "createdAt")]
         pub created_at: String,
-    }
-
-    #[derive(
-        Queryable,
-        Identifiable,
-        Insertable,
-        Selectable,
-        Clone,
-        Debug,
-        PartialEq,
-        Default,
-        Serialize,
-        Deserialize,
-    )]
-    #[diesel(primary_key(uri, path))]
-    #[diesel(table_name = crate::schema::pds::backlink)]
-    #[diesel(check_for_backend(Sqlite))]
-    pub struct Backlink {
-        pub uri: String,
-        pub path: String,
-        #[diesel(column_name = linkTo)]
-        #[serde(rename = "linkTo")]
-        pub link_to: String,
-    }
-
-    #[derive(
-        Queryable,
-        Identifiable,
-        Selectable,
-        Clone,
-        Debug,
-        PartialEq,
-        Default,
-        Serialize,
-        Deserialize,
-    )]
-    #[diesel(treat_none_as_null = true)]
-    #[diesel(primary_key(cid))]
-    #[diesel(table_name = crate::schema::pds::blob)]
-    #[diesel(check_for_backend(Sqlite))]
-    pub struct Blob {
-        pub cid: String,
-        pub did: String,
-        #[diesel(column_name = mimeType)]
-        #[serde(rename = "mimeType")]
-        pub mime_type: String,
-        pub size: i32,
-        #[diesel(column_name = tempKey)]
-        #[serde(rename = "tempKey")]
-        pub temp_key: Option<String>,
-        pub width: Option<i32>,
-        pub height: Option<i32>,
-        #[diesel(column_name = createdAt)]
-        #[serde(rename = "createdAt")]
-        pub created_at: String,
-        #[diesel(column_name = takedownRef)]
-        #[serde(rename = "takedownRef")]
-        pub takedown_ref: Option<String>,
     }
 
     #[derive(
···
     #[derive(
         Queryable,
         Identifiable,
-        Insertable,
-        Selectable,
-        Clone,
-        Debug,
-        PartialEq,
-        Default,
-        Serialize,
-        Deserialize,
-    )]
-    #[diesel(primary_key(uri))]
-    #[diesel(table_name = crate::schema::pds::record)]
-    #[diesel(check_for_backend(Sqlite))]
-    pub struct Record {
-        pub uri: String,
-        pub cid: String,
-        pub did: String,
-        pub collection: String,
-        pub rkey: String,
-        #[diesel(column_name = repoRev)]
-        #[serde(rename = "repoRev")]
-        pub repo_rev: Option<String>,
-        #[diesel(column_name = indexedAt)]
-        #[serde(rename = "indexedAt")]
-        pub indexed_at: String,
-        #[diesel(column_name = takedownRef)]
-        #[serde(rename = "takedownRef")]
-        pub takedown_ref: Option<String>,
-    }
-
-    #[derive(
-        QueryableByName,
-        Queryable,
-        Identifiable,
-        Selectable,
-        Clone,
-        Debug,
-        PartialEq,
-        Default,
-        Serialize,
-        Deserialize,
-    )]
-    #[diesel(primary_key(blobCid, recordUri))]
-    #[diesel(table_name = crate::schema::pds::record_blob)]
-    #[diesel(check_for_backend(Sqlite))]
-    pub struct RecordBlob {
-        #[diesel(column_name = blobCid, sql_type = Text)]
-        #[serde(rename = "blobCid")]
-        pub blob_cid: String,
-        #[diesel(column_name = recordUri, sql_type = Text)]
-        #[serde(rename = "recordUri")]
-        pub record_uri: String,
-        #[diesel(sql_type = Text)]
-        pub did: String,
-    }
-
-    #[derive(
-        Queryable,
-        Identifiable,
         Selectable,
         Clone,
         Debug,
···
         #[diesel(column_name = appPasswordName)]
         #[serde(rename = "appPasswordName")]
         pub app_password_name: Option<String>,
-    }
-
-    #[derive(
-        Queryable,
-        Identifiable,
-        Selectable,
-        Insertable,
-        Clone,
-        Debug,
-        PartialEq,
-        Default,
-        Serialize,
-        Deserialize,
-    )]
-    #[diesel(primary_key(cid))]
-    #[diesel(table_name = crate::schema::pds::repo_block)]
-    #[diesel(check_for_backend(Sqlite))]
-    pub struct RepoBlock {
-        #[diesel(sql_type = Text)]
-        pub cid: String,
-        pub did: String,
-        #[diesel(column_name = repoRev)]
-        #[serde(rename = "repoRev")]
-        pub repo_rev: String,
-        pub size: i32,
-        #[diesel(sql_type = Bytea)]
-        pub content: Vec<u8>,
-    }
-
-    #[derive(
-        Queryable,
-        Identifiable,
-        Selectable,
-        Clone,
-        Debug,
-        PartialEq,
-        Default,
-        Serialize,
-        Deserialize,
-    )]
-    #[diesel(primary_key(did))]
-    #[diesel(table_name = crate::schema::pds::repo_root)]
-    #[diesel(check_for_backend(Sqlite))]
-    pub struct RepoRoot {
-        pub did: String,
-        pub cid: String,
-        pub rev: String,
-        #[diesel(column_name = indexedAt)]
-        #[serde(rename = "indexedAt")]
-        pub indexed_at: String,
     }
 
     #[derive(
···
         pub refresh_token: String,
     }
 }
+
+pub mod actor_store {
+
+    #![allow(unnameable_types, unused_qualifications)]
+    use anyhow::{Result, bail};
+    use chrono::DateTime;
+    use chrono::offset::Utc;
+    use diesel::backend::Backend;
+    use diesel::deserialize::FromSql;
+    use diesel::prelude::*;
+    use diesel::serialize::{Output, ToSql};
+    use diesel::sql_types::Text;
+    use diesel::sqlite::Sqlite;
+    use diesel::*;
+    use serde::{Deserialize, Serialize};
+
+    #[derive(
+        Queryable,
+        Identifiable,
+        Insertable,
+        Selectable,
+        Clone,
+        Debug,
+        PartialEq,
+        Default,
+        Serialize,
+        Deserialize,
+    )]
+    #[diesel(table_name = crate::schema::actor_store::account_pref)]
+    #[diesel(check_for_backend(Sqlite))]
+    pub struct AccountPref {
+        pub id: i32,
+        pub name: String,
+        #[diesel(column_name = valueJson)]
+        #[serde(rename = "valueJson")]
+        pub value_json: Option<String>,
+    }
+
+    #[derive(
+        Queryable,
+        Identifiable,
+        Insertable,
+        Selectable,
+        Clone,
+        Debug,
+        PartialEq,
+        Default,
+        Serialize,
+        Deserialize,
+    )]
+    #[diesel(primary_key(uri, path))]
+    #[diesel(table_name = crate::schema::actor_store::backlink)]
+    #[diesel(check_for_backend(Sqlite))]
+    pub struct Backlink {
+        pub uri: String,
+        pub path: String,
+        #[diesel(column_name = linkTo)]
+        #[serde(rename = "linkTo")]
+        pub link_to: String,
+    }
+
+    #[derive(
+        Queryable,
+        Identifiable,
+        Selectable,
+        Clone,
+        Debug,
+        PartialEq,
+        Default,
+        Serialize,
+        Deserialize,
+    )]
+    #[diesel(treat_none_as_null = true)]
+    #[diesel(primary_key(cid))]
+    #[diesel(table_name = crate::schema::actor_store::blob)]
+    #[diesel(check_for_backend(Sqlite))]
+    pub struct Blob {
+        pub cid: String,
+        pub did: String,
+        #[diesel(column_name = mimeType)]
+        #[serde(rename = "mimeType")]
+        pub mime_type: String,
+        pub size: i32,
+        #[diesel(column_name = tempKey)]
+        #[serde(rename = "tempKey")]
+        pub temp_key: Option<String>,
+        pub width: Option<i32>,
+        pub height: Option<i32>,
+        #[diesel(column_name = createdAt)]
+        #[serde(rename = "createdAt")]
+        pub created_at: String,
+        #[diesel(column_name = takedownRef)]
+        #[serde(rename = "takedownRef")]
+        pub takedown_ref: Option<String>,
+    }
+
+    #[derive(
+        Queryable,
+        Identifiable,
+        Insertable,
+        Selectable,
+        Clone,
+        Debug,
+        PartialEq,
+        Default,
+        Serialize,
+        Deserialize,
+    )]
+    #[diesel(primary_key(uri))]
+    #[diesel(table_name = crate::schema::actor_store::record)]
+    #[diesel(check_for_backend(Sqlite))]
+    pub struct Record {
+        pub uri: String,
+        pub cid: String,
+        pub did: String,
+        pub collection: String,
+        pub rkey: String,
+        #[diesel(column_name = repoRev)]
+        #[serde(rename = "repoRev")]
+        pub repo_rev: Option<String>,
+        #[diesel(column_name = indexedAt)]
+        #[serde(rename = "indexedAt")]
+        pub indexed_at: String,
+        #[diesel(column_name = takedownRef)]
+        #[serde(rename = "takedownRef")]
+        pub takedown_ref: Option<String>,
+    }
+
+    #[derive(
+        QueryableByName,
+        Queryable,
+        Identifiable,
+        Selectable,
+        Clone,
+        Debug,
+        PartialEq,
+        Default,
+        Serialize,
+        Deserialize,
+    )]
+    #[diesel(primary_key(blobCid, recordUri))]
+    #[diesel(table_name = crate::schema::actor_store::record_blob)]
+    #[diesel(check_for_backend(Sqlite))]
+    pub struct RecordBlob {
+        #[diesel(column_name = blobCid, sql_type = Text)]
+        #[serde(rename = "blobCid")]
+        pub blob_cid: String,
+        #[diesel(column_name = recordUri, sql_type = Text)]
+        #[serde(rename = "recordUri")]
+        pub record_uri: String,
+        #[diesel(sql_type = Text)]
+        pub did: String,
+    }
+
+    #[derive(
+        Queryable,
+        Identifiable,
+        Selectable,
+        Insertable,
+        Clone,
+        Debug,
+        PartialEq,
+        Default,
+        Serialize,
+        Deserialize,
+    )]
+    #[diesel(primary_key(cid))]
+    #[diesel(table_name = crate::schema::actor_store::repo_block)]
+    #[diesel(check_for_backend(Sqlite))]
+    pub struct RepoBlock {
+        #[diesel(sql_type = Text)]
+        pub cid: String,
+        pub did: String,
+        #[diesel(column_name = repoRev)]
+        #[serde(rename = "repoRev")]
+        pub repo_rev: String,
+        pub size: i32,
+        #[diesel(sql_type = Bytea)]
+        pub content: Vec<u8>,
+    }
+
+    #[derive(
+        Queryable,
+        Identifiable,
+        Selectable,
+        Clone,
+        Debug,
+        PartialEq,
+        Default,
+        Serialize,
+        Deserialize,
+    )]
+    #[diesel(primary_key(did))]
+    #[diesel(table_name = crate::schema::actor_store::repo_root)]
+    #[diesel(check_for_backend(Sqlite))]
+    pub struct RepoRoot {
+        pub did: String,
+        pub cid: String,
+        pub rev: String,
+        #[diesel(column_name = indexedAt)]
+        #[serde(rename = "indexedAt")]
+        pub indexed_at: String,
+    }
+}

src/schema.rs

···
     }
 
     diesel::table! {
-        account_pref (id) {
-            id -> Int4,
-            did -> Varchar,
-            name -> Varchar,
-            valueJson -> Nullable<Text>,
-        }
-    }
-
-    diesel::table! {
         actor (did) {
             did -> Varchar,
             handle -> Nullable<Varchar>,
···
     }
 
     diesel::table! {
-        backlink (uri, path) {
-            uri -> Varchar,
-            path -> Varchar,
-            linkTo -> Varchar,
-        }
-    }
-
-    diesel::table! {
-        blob (cid, did) {
-            cid -> Varchar,
-            did -> Varchar,
-            mimeType -> Varchar,
-            size -> Int4,
-            tempKey -> Nullable<Varchar>,
-            width -> Nullable<Int4>,
-            height -> Nullable<Int4>,
-            createdAt -> Varchar,
-            takedownRef -> Nullable<Varchar>,
-        }
-    }
-
-    diesel::table! {
         device (id) {
             id -> Varchar,
             sessionId -> Nullable<Varchar>,
···
     }
 
     diesel::table! {
-        record (uri) {
-            uri -> Varchar,
-            cid -> Varchar,
-            did -> Varchar,
-            collection -> Varchar,
-            rkey -> Varchar,
-            repoRev -> Nullable<Varchar>,
-            indexedAt -> Varchar,
-            takedownRef -> Nullable<Varchar>,
-        }
-    }
-
-    diesel::table! {
-        record_blob (blobCid, recordUri) {
-            blobCid -> Varchar,
-            recordUri -> Varchar,
-            did -> Varchar,
-        }
-    }
-
-    diesel::table! {
         refresh_token (id) {
             id -> Varchar,
             did -> Varchar,
···
     }
 
     diesel::table! {
-        repo_block (cid, did) {
-            cid -> Varchar,
-            did -> Varchar,
-            repoRev -> Varchar,
-            size -> Int4,
-            content -> Bytea,
-        }
-    }
-
-    diesel::table! {
-        repo_root (did) {
-            did -> Varchar,
-            cid -> Varchar,
-            rev -> Varchar,
-            indexedAt -> Varchar,
-        }
-    }
-
-    diesel::table! {
         repo_seq (seq) {
             seq -> Int8,
             did -> Varchar,
···
 
     diesel::allow_tables_to_appear_in_same_query!(
         account,
-        account_pref,
         actor,
         app_password,
         authorization_request,
-        backlink,
-        blob,
         device,
         device_account,
         did_doc,
         email_token,
         invite_code,
         invite_code_use,
-        record,
-        record_blob,
         refresh_token,
-        repo_block,
-        repo_root,
         repo_seq,
         token,
         used_refresh_token,
     );
 }
+
+pub mod actor_store {
+    // Actor Store
+
+    // Blob
+    diesel::table! {
+        blob (cid, did) {
+            cid -> Varchar,
+            did -> Varchar,
+            mimeType -> Varchar,
+            size -> Int4,
+            tempKey -> Nullable<Varchar>,
+            width -> Nullable<Int4>,
+            height -> Nullable<Int4>,
+            createdAt -> Varchar,
+            takedownRef -> Nullable<Varchar>,
+        }
+    }
+
+    diesel::table! {
+        record_blob (blobCid, recordUri) {
+            blobCid -> Varchar,
+            recordUri -> Varchar,
+            did -> Varchar,
+        }
+    }
+
+    // Preference
+
+    diesel::table! {
+        account_pref (id) {
+            id -> Int4,
+            did -> Varchar,
+            name -> Varchar,
+            valueJson -> Nullable<Text>,
+        }
+    }
+    // Record
+
+    diesel::table! {
+        record (uri) {
+            uri -> Varchar,
+            cid -> Varchar,
+            did -> Varchar,
+            collection -> Varchar,
+            rkey -> Varchar,
+            repoRev -> Nullable<Varchar>,
+            indexedAt -> Varchar,
+            takedownRef -> Nullable<Varchar>,
+        }
+    }
+
+    diesel::table! {
+        repo_block (cid, did) {
+            cid -> Varchar,
+            did -> Varchar,
+            repoRev -> Varchar,
+            size -> Int4,
+            content -> Bytea,
+        }
+    }
+
+    diesel::table! {
+        backlink (uri, path) {
+            uri -> Varchar,
+            path -> Varchar,
+            linkTo -> Varchar,
+        }
+    }
+    // sql_repo
+
+    diesel::table! {
+        repo_root (did) {
+            did -> Varchar,
+            cid -> Varchar,
+            rev -> Varchar,
+            indexedAt -> Varchar,
+        }
+    }
+
+    diesel::allow_tables_to_appear_in_same_query!(
+        account_pref,
+        backlink,
+        blob,
+        record,
+        record_blob,
+        repo_block,
+        repo_root,
+    );
+}