prototype actor_store; merge handlers

src/actor_store/actor_store.rs

···
 use std::path::PathBuf;
+use std::str::FromStr;
 use std::sync::Arc;
 
-use anyhow::Result;
+use anyhow::{Context as _, Result, anyhow, bail};
+use atrium_crypto::keypair::{Did as _, Export as _, Secp256k1Keypair};
+use atrium_repo::Cid;
+use diesel::prelude::*;
+use sha2::Digest as _;
+use tokio::fs;
 
-use super::ActorDb;
-use super::actor_store_reader::ActorStoreReader;
+use super::PreparedWrite;
+use super::actor_store_handler::ActorStoreHandler;
 use super::actor_store_resources::ActorStoreResources;
-use super::actor_store_transactor::ActorStoreTransactor;
-use super::actor_store_writer::ActorStoreWriter;
+use super::blob::{BlobStore as _, BlobStorePlaceholder};
+use super::db::{ActorDb, get_db};
 use crate::SigningKey;
 
+/// Central manager for actor stores
 pub(crate) struct ActorStore {
-    pub(crate) directory: PathBuf,
-    reserved_key_dir: PathBuf,
-    resources: ActorStoreResources,
+    /// Base directory for actor data
+    pub directory: PathBuf,
+    /// Resources shared between actor stores
+    pub resources: ActorStoreResources,
 }
 
 struct ActorLocation {
+    /// Actor's directory path
     directory: PathBuf,
+    /// Database path
     db_location: PathBuf,
+    /// Key path
     key_location: PathBuf,
 }
 
 impl ActorStore {
+    /// Create a new actor store manager
     pub(crate) fn new(directory: impl Into<PathBuf>, resources: ActorStoreResources) -> Self {
-        let directory = directory.into();
-        let reserved_key_dir = directory.join("reserved_keys");
         Self {
-            directory,
-            reserved_key_dir,
+            directory: directory.into(),
             resources,
         }
     }
 
+    /// Get the location information for an actor
     pub(crate) async fn get_location(&self, did: &str) -> Result<ActorLocation> {
-        // const didHash = await crypto.sha256Hex(did)
-        // const directory = path.join(this.cfg.directory, didHash.slice(0, 2), did)
-        // const dbLocation = path.join(directory, `store.sqlite`)
-        // const keyLocation = path.join(directory, `key`)
-        // return { directory, dbLocation, keyLocation }
-        todo!()
+        // Hash the DID for directory organization
+        let did_hash = sha2::Sha256::digest(did.as_bytes());
+        let hash_prefix = format!("{:02x}", did_hash[0]);
+
+        // Create paths
+        let directory = self.directory.join(hash_prefix).join(did);
+        let db_location = directory.join("store.sqlite");
+        let key_location = directory.join("key");
+
+        Ok(ActorLocation {
+            directory,
+            db_location,
+            key_location,
+        })
     }
 
+    /// Check if an actor store exists
     pub(crate) async fn exists(&self, did: &str) -> Result<bool> {
-        // const location = await this.getLocation(did)
-        // return await fileExists(location.dbLocation)
-        todo!()
+        let location = self.get_location(did).await?;
+        Ok(location.db_location.exists())
     }
 
+    /// Get the signing keypair for an actor
     pub(crate) async fn keypair(&self, did: &str) -> Result<Arc<SigningKey>> {
-        // const { keyLocation } = await this.getLocation(did)
-        // const privKey = await fs.readFile(keyLocation)
-        // return crypto.Secp256k1Keypair.import(privKey)
-        todo!()
+        let location = self.get_location(did).await?;
+        let priv_key = fs::read(&location.key_location)
+            .await
+            .context("Failed to read key file")?;
+
+        let keypair = SigningKey::import(&priv_key).context("Failed to import signing key")?;
+
+        Ok(Arc::new(keypair))
     }
 
+    /// Open the database for an actor
     pub(crate) async fn open_db(&self, did: &str) -> Result<ActorDb> {
-        // const { dbLocation } = await this.getLocation(did)
-        // const exists = await fileExists(dbLocation)
-        // if (!exists) {
-        //   throw new InvalidRequestError('Repo not found', 'NotFound')
-        // }
+        let location = self.get_location(did).await?;
+
+        if !location.db_location.exists() {
+            bail!("Repo not found");
+        }
+
+        // Convert path to string for SQLite connection
+        let db_path = location
+            .db_location
+            .to_str()
+            .ok_or_else(|| anyhow!("Invalid path encoding"))?;
 
-        // const db = getDb(dbLocation, this.cfg.disableWalAutoCheckpoint)
+        // Open database with WAL mode enabled
+        let db = get_db(db_path, false)
+            .await
+            .context("Failed to open actor database")?;
 
-        // // run a simple select with retry logic to ensure the db is ready (not in wal recovery mode)
-        // try {
-        //   await retrySqlite(() =>
-        //     db.db.selectFrom('repo_root').selectAll().execute(),
-        //   )
-        // } catch (err) {
-        //   db.close()
-        //   throw err
-        // }
+        // Run a simple query to ensure the database is ready
+        db.run(|conn| diesel::sql_query("SELECT 1 FROM repo_root LIMIT 1").execute(conn))
+            .await
+            .context("Database not ready")?;
 
-        // return db
-        todo!()
+        Ok(db)
     }
 
+    /// Execute read operations on an actor store
     pub(crate) async fn read<T, F>(&self, did: &str, f: F) -> Result<T>
     where
-        F: FnOnce(ActorStoreReader) -> Result<T>,
+        F: FnOnce(ActorStoreHandler) -> Result<T>,
     {
-        // const db = await this.openDb(did)
-        // try {
-        //   const getKeypair = () => this.keypair(did)
-        //   return await fn(new ActorStoreReader(did, db, this.resources, getKeypair))
-        // } finally {
-        //   db.close()
-        // }
-        todo!()
+        let db = self.open_db(did).await?;
+        let blobstore = self.resources.blobstore(did.to_string());
+
+        // Create a read-only handler
+        let handler = ActorStoreHandler::new_reader(db.clone(), did.to_string(), blobstore);
+
+        // Execute the function
+        f(handler)
     }
 
+    /// Execute read-write operations with a transaction
     pub(crate) async fn transact<T, F>(&self, did: &str, f: F) -> Result<T>
     where
-        F: FnOnce(ActorStoreTransactor) -> Result<T>,
+        F: FnOnce(ActorStoreHandler) -> Result<T>,
     {
-        // const keypair = await this.keypair(did)
-        // const db = await this.openDb(did)
-        // try {
-        //   return await db.transaction((dbTxn) => {
-        //     return fn(new ActorStoreTransactor(did, dbTxn, keypair, this.resources))
-        //   })
-        // } finally {
-        //   db.close()
-        // }
-        todo!()
+        let db = self.open_db(did).await?;
+        let keypair = self.keypair(did).await?;
+        let blobstore = self.resources.blobstore(did.to_string());
+        let background_queue = self.resources.background_queue();
+
+        // Create a read-write handler with transaction support
+        let handler = ActorStoreHandler::new_writer(
+            db,
+            did.to_string(),
+            blobstore,
+            keypair,
+            background_queue.as_ref().clone(),
+        );
+
+        // Execute the function (will handle transactions internally)
+        f(handler)
     }
 
+    /// Execute read-write operations without a transaction
     pub(crate) async fn write_no_transaction<T, F>(&self, did: &str, f: F) -> Result<T>
     where
-        F: FnOnce(ActorStoreWriter) -> Result<T>,
+        F: FnOnce(ActorStoreHandler) -> Result<T>,
     {
-        // const keypair = await this.keypair(did)
-        // const db = await this.openDb(did)
-        // try {
-        //   return await fn(new ActorStoreWriter(did, db, keypair, this.resources))
-        // } finally {
-        //   db.close()
-        // }
-        todo!()
+        let db = self.open_db(did).await?;
+        let keypair = self.keypair(did).await?;
+        let blobstore = self.resources.blobstore(did.to_string());
+        let background_queue = self.resources.background_queue();
+
+        // Create a read-write handler without automatic transaction
+        let handler = ActorStoreHandler::new_writer(
+            db,
+            did.to_string(),
+            blobstore,
+            keypair,
+            background_queue.as_ref().clone(),
+        );
+
+        // Execute the function
+        f(handler)
     }
 
+    /// Create a new actor store
     pub(crate) async fn create(&self, did: &str, keypair: SigningKey) -> Result<()> {
-        // const { directory, dbLocation, keyLocation } = await this.getLocation(did)
-        // // ensure subdir exists
-        // await mkdir(directory, { recursive: true })
-        // const exists = await fileExists(dbLocation)
-        // if (exists) {
-        //   throw new InvalidRequestError('Repo already exists', 'AlreadyExists')
-        // }
-        // const privKey = await keypair.export()
-        // await fs.writeFile(keyLocation, privKey)
+        let location = self.get_location(did).await?;
+
+        // Ensure directory exists
+        fs::create_dir_all(&location.directory)
+            .await
+            .context("Failed to create directory")?;
+
+        // Check if repo already exists
+        if location.db_location.exists() {
+            bail!("Repo already exists");
+        }
+
+        // Export and save private key
+        let priv_key = keypair.export();
+        fs::write(&location.key_location, priv_key)
+            .await
+            .context("Failed to write key file")?;
 
-        // const db: ActorDb = getDb(dbLocation, this.cfg.disableWalAutoCheckpoint)
-        // try {
-        //   await db.ensureWal()
-        //   const migrator = getMigrator(db)
-        //   await migrator.migrateToLatestOrThrow()
-        // } finally {
-        //   db.close()
-        // }
-        todo!()
+        // Initialize the database
+        let db_path = location
+            .db_location
+            .to_str()
+            .ok_or_else(|| anyhow!("Invalid path encoding"))?;
+
+        let db = get_db(db_path, false)
+            .await
+            .context("Failed to create actor database")?;
+
+        // Ensure WAL mode and run migrations
+        db.ensure_wal().await?;
+        db.run_migrations()?;
+
+        Ok(())
     }
 
+    /// Destroy an actor store
     pub(crate) async fn destroy(&self, did: &str) -> Result<()> {
-        // const blobstore = this.resources.blobstore(did)
-        // if (blobstore instanceof DiskBlobStore) {
-        //   await blobstore.deleteAll()
-        // } else {
-        //   const cids = await this.read(did, async (store) =>
-        //     store.repo.blob.getBlobCids(),
-        //   )
-        //   await Promise.allSettled(
-        //     chunkArray(cids, 500).map((chunk) => blobstore.deleteMany(chunk)),
-        //   )
-        // }
+        // Get all blob CIDs first
+        let cids = self
+            .read(did, |handler| async move {
+                handler.repo.blob.get_blob_cids().await
+            })
+            .await?;
+
+        // Delete all blobs
+        let blobstore = self.resources.blobstore(did.to_string());
+        if !cids.is_empty() {
+            // Process in chunks of 500
+            for chunk in cids.chunks(500) {
+                let _ = blobstore.delete_many(chunk.to_vec()).await;
+            }
+        }
+
+        // Remove directory and all files
+        let location = self.get_location(did).await?;
+        if location.directory.exists() {
+            fs::remove_dir_all(&location.directory)
+                .await
+                .context("Failed to remove actor directory")?;
+        }
 
-        // const { directory } = await this.getLocation(did)
-        // await rmIfExists(directory, true)
-        todo!()
+        Ok(())
     }
 
-    // async reserveKeypair(did?: string): Promise<string> {
-    //   let keyLoc: string | undefined
-    //   if (did) {
-    //     assertSafePathPart(did)
-    //     keyLoc = path.join(this.reservedKeyDir, did)
-    //     const maybeKey = await loadKey(keyLoc)
-    //     if (maybeKey) {
-    //       return maybeKey.did()
-    //     }
-    //   }
-    //   const keypair = await crypto.Secp256k1Keypair.create({ exportable: true })
-    //   const keyDid = keypair.did()
-    //   keyLoc = keyLoc ?? path.join(this.reservedKeyDir, keyDid)
-    //   await mkdir(this.reservedKeyDir, { recursive: true })
-    //   await fs.writeFile(keyLoc, await keypair.export())
-    //   return keyDid
-    // }
+    /// Reserve a keypair for future use
+    pub(crate) async fn reserve_keypair(&self, did: Option<&str>) -> Result<String> {
+        let reserved_dir = self
+            .resources
+            .reserved_key_dir()
+            .ok_or_else(|| anyhow!("No reserved key directory configured"))?;
 
-    // async getReservedKeypair(
-    //   signingKeyOrDid: string,
-    // ): Promise<ExportableKeypair | undefined> {
-    //   return loadKey(path.join(this.reservedKeyDir, signingKeyOrDid))
-    // }
+        // If DID is provided, check if key already exists
+        let mut key_path = None;
+        if let Some(did_str) = did {
+            assert_safe_path_part(did_str)?;
+            key_path = Some(reserved_dir.join(did_str));
 
-    // async clearReservedKeypair(keyDid: string, did?: string) {
-    //   await rmIfExists(path.join(this.reservedKeyDir, keyDid))
-    //   if (did) {
-    //     await rmIfExists(path.join(this.reservedKeyDir, did))
-    //   }
-    // }
+            if key_path.as_ref().unwrap().exists() {
+                let key_data = fs::read(key_path.as_ref().unwrap()).await?;
+                let keypair = Secp256k1Keypair::import(&key_data)
+                    .context("Failed to import existing reserved key")?;
+                return Ok(keypair.did());
+            }
+        }
 
-    // async storePlcOp(did: string, op: Uint8Array) {
-    //   const { directory } = await this.getLocation(did)
-    //   const opLoc = path.join(directory, `did-op`)
-    //   await fs.writeFile(opLoc, op)
-    // }
+        // Create a new keypair
+        let keypair = Secp256k1Keypair::create(&mut rand::thread_rng());
+        let key_did = keypair.did();
 
-    // async getPlcOp(did: string): Promise<Uint8Array> {
-    //   const { directory } = await this.getLocation(did)
-    //   const opLoc = path.join(directory, `did-op`)
-    //   return await fs.readFile(opLoc)
-    // }
+        // Set path if not already set
+        let final_path = key_path.unwrap_or_else(|| reserved_dir.join(&key_did));
 
-    // async clearPlcOp(did: string) {
-    //   const { directory } = await this.getLocation(did)
-    //   const opLoc = path.join(directory, `did-op`)
-    //   await rmIfExists(opLoc)
-    // }
+        // Ensure directory exists
+        fs::create_dir_all(reserved_dir).await?;
+
+        // Save key
+        fs::write(&final_path, keypair.export()).await?;
+
+        Ok(key_did)
+    }
+
+    /// Get a reserved keypair
+    pub(crate) async fn get_reserved_keypair(
+        &self,
+        key_did: &str,
+    ) -> Result<Option<Arc<SigningKey>>> {
+        let reserved_dir = self
+            .resources
+            .reserved_key_dir()
+            .ok_or_else(|| anyhow!("No reserved key directory configured"))?;
+
+        let key_path = reserved_dir.join(key_did);
+        if !key_path.exists() {
+            return Ok(None);
+        }
+
+        let key_data = fs::read(key_path).await?;
+        let keypair = SigningKey::import(&key_data).context("Failed to import reserved key")?;
+
+        Ok(Some(Arc::new(keypair)))
+    }
+
+    /// Clear a reserved keypair
+    pub(crate) async fn clear_reserved_keypair(
+        &self,
+        key_did: &str,
+        did: Option<&str>,
+    ) -> Result<()> {
+        let reserved_dir = self
+            .resources
+            .reserved_key_dir()
+            .ok_or_else(|| anyhow!("No reserved key directory configured"))?;
+
+        // Remove key by DID
+        let key_path = reserved_dir.join(key_did);
+        if key_path.exists() {
+            fs::remove_file(key_path).await?;
+        }
+
+        // If DID mapping provided, remove that too
+        if let Some(did_str) = did {
+            let did_path = reserved_dir.join(did_str);
+            if did_path.exists() {
+                fs::remove_file(did_path).await?;
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Store a PLC operation
+    pub(crate) async fn store_plc_op(&self, did: &str, op: &[u8]) -> Result<()> {
+        let location = self.get_location(did).await?;
+        let op_path = location.directory.join("did-op");
+
+        fs::write(op_path, op).await?;
+        Ok(())
+    }
+
+    /// Get a stored PLC operation
+    pub(crate) async fn get_plc_op(&self, did: &str) -> Result<Vec<u8>> {
+        let location = self.get_location(did).await?;
+        let op_path = location.directory.join("did-op");
+
+        let data = fs::read(op_path).await?;
+        Ok(data)
+    }
+
+    /// Clear a stored PLC operation
+    pub(crate) async fn clear_plc_op(&self, did: &str) -> Result<()> {
+        let location = self.get_location(did).await?;
+        let op_path = location.directory.join("did-op");
+
+        if op_path.exists() {
+            fs::remove_file(op_path).await?;
+        }
+
+        Ok(())
+    }
 }
 
-// const loadKey = async (loc: string): Promise<ExportableKeypair | undefined> => {
-//   const privKey = await readIfExists(loc)
-//   if (!privKey) return undefined
-//   return crypto.Secp256k1Keypair.import(privKey, { exportable: true })
-// }
+/// Ensure a path part is safe to use in a filename
+fn assert_safe_path_part(part: &str) -> Result<()> {
+    let normalized = std::path::Path::new(part)
+        .file_name()
+        .and_then(|s| s.to_str())
+        .ok_or_else(|| anyhow!("Invalid path"))?;
 
-// function assertSafePathPart(part: string) {
-//   const normalized = path.normalize(part)
-//   assert(
-//     part === normalized &&
-//       !part.startsWith('.') &&
-//       !part.includes('/') &&
-//       !part.includes('\\'),
-//     `unsafe path part: ${part}`,
-//   )
-// }
+    if part != normalized || part.starts_with('.') || part.contains('/') || part.contains('\\') {
+        bail!("Unsafe path part: {}", part);
+    }
+
+    Ok(())
+}

src/actor_store/actor_store_handler.rs

···
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use anyhow::{Context as _, Result, anyhow};
+use futures::TryStreamExt;
+use rsky_repo::repo::Repo;
+use rsky_repo::types::{CommitData, CommitDataWithOps, PreparedWrite as RskyPreparedWrite};
+
+use super::PreparedWrite;
+use super::blob::{BackgroundQueue, BlobStorePlaceholder};
+use super::db::ActorDb;
+use super::preference::PreferenceHandler;
+use super::record::RecordHandler;
+use super::repo::RepoHandler;
+use crate::SigningKey;
+
+/// Unified handler for actor store operations.
+pub(crate) struct ActorStoreHandler {
+    /// Actor DID
+    pub did: String,
+    /// Database connection
+    pub db: ActorDb,
+    /// Repository handler
+    pub repo: RepoHandler,
+    /// Record handler
+    pub record: RecordHandler,
+    /// Preference handler
+    pub pref: PreferenceHandler,
+    /// Background queue for async operations
+    pub background_queue: Option<BackgroundQueue>,
+    /// Signing keypair (required for write operations)
+    pub signing_key: Option<Arc<SigningKey>>,
+}
+
+impl ActorStoreHandler {
+    /// Create a new actor store handler with read-only capabilities
+    pub(crate) fn new_reader(db: ActorDb, did: String, blobstore: BlobStorePlaceholder) -> Self {
+        let record = RecordHandler::new(db.clone(), did.clone());
+        let pref = PreferenceHandler::new(db.clone(), did.clone());
+        let repo = RepoHandler::new_reader(db.clone(), blobstore, did.clone());
+
+        Self {
+            did,
+            db,
+            repo,
+            record,
+            pref,
+            background_queue: None,
+            signing_key: None,
+        }
+    }
+
+    /// Create a new actor store handler with read/write capabilities
+    pub(crate) fn new_writer(
+        db: ActorDb,
+        did: String,
+        blobstore: BlobStorePlaceholder,
+        signing_key: Arc<SigningKey>,
+        background_queue: BackgroundQueue,
+    ) -> Self {
+        let record = RecordHandler::new_with_blobstore(db.clone(), blobstore.clone(), did.clone());
+        let pref = PreferenceHandler::new(db.clone(), did.clone());
+        let repo = RepoHandler::new_writer(
+            db.clone(),
+            blobstore,
+            did.clone(),
+            signing_key.clone(),
+            background_queue.clone(),
+        );
+
+        Self {
+            did,
+            db,
+            repo,
+            record,
+            pref,
+            background_queue: Some(background_queue),
+            signing_key: Some(signing_key),
+        }
+    }
+
+    /// Set signing key (needed for write operations)
+    pub(crate) fn with_signing_key(mut self, signing_key: Arc<SigningKey>) -> Self {
+        self.signing_key = Some(signing_key);
+        self
+    }
+
+    /// Set background queue (needed for async operations)
+    pub(crate) fn with_background_queue(mut self, queue: BackgroundQueue) -> Self {
+        self.background_queue = Some(queue);
+        self
+    }
+
+    // Repository Operations
+    // --------------------
+
+    /// Try to load repository
+    pub(crate) async fn maybe_load_repo(&self) -> Result<Option<Repo>> {
+        self.repo.maybe_load_repo().await
+    }
+
+    /// Get repository root CID
+    pub(crate) async fn get_repo_root(&self) -> Result<Option<atrium_repo::Cid>> {
+        self.repo.get_repo_root().await
+    }
+
+    /// Create a new repository with prepared writes
+    pub(crate) async fn create_repo(
+        &self,
+        writes: Vec<PreparedWrite>,
+    ) -> Result<CommitDataWithOps> {
+        if self.signing_key.is_none() {
+            return Err(anyhow!(
+                "No signing key available for create_repo operation"
+            ));
+        }
+
+        let rsky_writes = writes
+            .into_iter()
+            .map(|w| RskyPreparedWrite::from(w))
+            .collect::<Vec<_>>();
+
+        self.repo.create_repo(rsky_writes).await
+    }
+
+    /// Process writes to the repository
+    pub(crate) async fn process_writes(
+        &self,
+        writes: Vec<PreparedWrite>,
+        swap_commit_cid: Option<atrium_repo::Cid>,
+    ) -> Result<CommitDataWithOps> {
+        if self.signing_key.is_none() {
+            return Err(anyhow!(
+                "No signing key available for process_writes operation"
+            ));
+        }
+
+        let rsky_writes = writes
+            .into_iter()
+            .map(|w| RskyPreparedWrite::from(w))
+            .collect::<Vec<_>>();
+
+        self.repo.process_writes(rsky_writes, swap_commit_cid).await
+    }
+
+    /// Import a repository from external data
+    pub(crate) async fn process_import_repo(
+        &self,
+        commit: CommitData,
+        writes: Vec<PreparedWrite>,
+    ) -> Result<()> {
+        let rsky_writes = writes
+            .into_iter()
+            .map(|w| RskyPreparedWrite::from(w))
+            .collect::<Vec<_>>();
+
+        // First index the records
+        self.repo.index_writes(&rsky_writes, &commit.rev).await?;
+
+        // Then process the commit
+        self.repo.storage.apply_commit(commit.clone(), None).await?;
+
+        // Finally process any blobs
+        if let Some(bg_queue) = &self.background_queue {
+            self.repo
+                .blob_transactor
+                .process_write_blobs(&commit.rev, rsky_writes)
+                .await?;
+        } else {
+            return Err(anyhow!(
+                "Background queue required for process_import_repo operation"
+            ));
+        }
+
+        Ok(())
+    }
+
+    /// Get sync event data for replication
+    pub(crate) async fn get_sync_event_data(&self) -> Result<super::repo::SyncEventData> {
+        self.repo.get_sync_event_data().await
+    }
+
+    /// Destroy the repository and all associated data
+    pub(crate) async fn destroy(&self) -> Result<()> {
+        // Get all blob CIDs
+        let blob_cids = self.repo.blob.get_blob_cids().await?;
+
+        // Delete all blobs
+        if !blob_cids.is_empty() {
+            self.repo
+                .blob_transactor
+                .blobstore
+                .delete_many(blob_cids.clone())
+                .await?;
+        }
+
+        Ok(())
+    }
+
+    // Record Operations
+    // ----------------
+
+    /// Get a specific record
+    pub(crate) async fn get_record(
+        &self,
+        uri: &rsky_syntax::aturi::AtUri,
+        cid: Option<&str>,
+        include_soft_deleted: bool,
+    ) -> Result<Option<super::record::RecordData>> {
+        self.record.get_record(uri, cid, include_soft_deleted).await
+    }
+
+    /// List collections in the repository
+    pub(crate) async fn list_collections(&self) -> Result<Vec<String>> {
+        self.record.list_collections().await
+    }
+
+    /// List records in a collection
+    pub(crate) async fn list_records_for_collection(
+        &self,
+        opts: super::record::ListRecordsOptions,
+    ) -> Result<Vec<super::record::RecordData>> {
+        self.record.list_records_for_collection(opts).await
+    }
+
+    /// Get record count
+    pub(crate) async fn record_count(&self) -> Result<i64> {
+        self.record.record_count().await
+    }
+
+    /// Update record takedown status
+    pub(crate) async fn update_record_takedown_status(
+        &self,
+        uri: &rsky_syntax::aturi::AtUri,
+        takedown: atrium_api::com::atproto::admin::defs::StatusAttr,
+    ) -> Result<()> {
+        self.record
+            .update_record_takedown_status(uri, takedown)
+            .await
+    }
+
+    // Preference Operations
+    // -------------------
+
+    /// Get preferences for a namespace
+    pub(crate) async fn get_preferences(
+        &self,
+        namespace: Option<&str>,
+        scope: &str,
+    ) -> Result<Vec<super::preference::AccountPreference>> {
+        self.pref.get_preferences(namespace, scope).await
+    }
+
+    /// Put preferences for a namespace
+    pub(crate) async fn put_preferences(
+        &self,
+        values: Vec<super::preference::AccountPreference>,
+        namespace: &str,
+        scope: &str,
+    ) -> Result<()> {
+        self.pref.put_preferences(values, namespace, scope).await
+    }
+
+    // Blob Operations
+    // --------------
+
+    /// Get blob metadata
+    pub(crate) async fn get_blob_metadata(
+        &self,
+        cid: &atrium_repo::Cid,
+    ) -> Result<super::blob::BlobMetadata> {
+        self.repo.blob.get_blob_metadata(cid).await
+    }
+
+    /// Get blob data
+    pub(crate) async fn get_blob(&self, cid: &atrium_repo::Cid) -> Result<super::blob::BlobData> {
+        self.repo.blob.get_blob(cid).await
+    }
+
+    /// Update blob takedown status
+    pub(crate) async fn update_blob_takedown_status(
+        &self,
+        cid: atrium_repo::Cid,
+        takedown: atrium_api::com::atproto::admin::defs::StatusAttr,
+    ) -> Result<()> {
+        self.repo
+            .blob
+            .update_blob_takedown_status(cid, takedown)
+            .await
+    }
+
+    /// Upload blob and get metadata
+    pub(crate) async fn upload_blob_and_get_metadata(
+        &self,
+        user_suggested_mime: &str,
+        blob_bytes: &[u8],
+    ) -> Result<super::blob::BlobMetadata> {
+        self.repo
+            .blob
+            .upload_blob_and_get_metadata(user_suggested_mime, blob_bytes)
+            .await
+    }
+
+    /// Count blobs
+    pub(crate) async fn blob_count(&self) -> Result<i64> {
+        self.repo.blob.blob_count().await
+    }
+
+    // Transaction Support
+    // -----------------
+
+    /// Execute a transaction
+    pub(crate) async fn transaction<T, F>(&self, f: F) -> Result<T>
+    where
+        F: FnOnce(&mut diesel::SqliteConnection) -> Result<T> + Send,
+        T: Send + 'static,
+    {
+        self.db.transaction(f).await
+    }
+
+    /// Execute a database operation with retries
+    pub(crate) async fn run<F, T>(&self, operation: F) -> Result<T>
+    where
+        F: FnOnce(&mut diesel::SqliteConnection) -> diesel::QueryResult<T> + Send,
+        T: Send + 'static,
+    {
+        self.db.run(operation).await
+    }
+}

src/actor_store/actor_store_reader.rs

···
-use anyhow::Result;
-use diesel::prelude::*;
-use std::sync::Arc;
-
-use super::{
-    ActorStoreTransactor, actor_store_resources::ActorStoreResources, db::ActorDb,
-    preference::PreferenceReader, record::RecordReader, repo::RepoReader,
-};
-use crate::SigningKey;
-
-/// Reader for the actor store.
-pub(crate) struct ActorStoreReader {
-    /// DID of the actor.
-    pub(crate) did: String,
-    /// Record reader.
-    pub(crate) record: RecordReader,
-    /// Preference reader.
-    pub(crate) pref: PreferenceReader,
-    /// RepoReader
-    pub(crate) repo: RepoReader,
-    /// Function to get keypair.
-    keypair_fn: Box<dyn Fn() -> Result<Arc<SigningKey>> + Send + Sync>,
-    /// Database connection.
-    pub(crate) db: ActorDb,
-    /// Actor store resources.
-    resources: ActorStoreResources,
-}
-
-impl ActorStoreReader {
-    /// Create a new actor store reader.
-    pub(crate) fn new(
-        did: String,
-        db: ActorDb,
-        resources: ActorStoreResources,
-        keypair: impl Fn() -> Result<Arc<SigningKey>> + Send + Sync + 'static,
-    ) -> Self {
-        // Create readers
-        let record = RecordReader::new(db.clone(), did.clone());
-        let pref = PreferenceReader::new(db.clone(), did.clone());
-
-        // Store keypair function for later use
-        let keypair_fn = Box::new(keypair);
-
-        // Initial keypair call as in TypeScript implementation
-        let _ = keypair_fn();
-
-        // Create repo reader
-        let repo = RepoReader::new(db.clone(), resources.blobstore(did.clone()), did.clone());
-
-        Self {
-            did,
-            repo,
-            record,
-            pref,
-            keypair_fn,
-            db,
-            resources,
-        }
-    }
-
-    /// Get the keypair for this actor.
-    pub(crate) fn keypair(&self) -> Result<Arc<SigningKey>> {
-        (self.keypair_fn)()
-    }
-
-    /// Execute a transaction with the actor store.
-    pub(crate) async fn transact<T, F>(&self, f: F) -> Result<T>
-    where
-        F: FnOnce(ActorStoreTransactor) -> Result<T> + Send,
-        T: Send + 'static,
-    {
-        let keypair = self.keypair()?;
-        let did = self.did.clone();
-        let resources = self.resources.clone();
-
-        self.db
-            .transaction(move |conn| {
-                // Create a transactor with the transaction
-                // We'll create a temporary ActorDb with the same pool
-                let db = ActorDb {
-                    pool: self.db.pool.clone(),
-                };
-
-                let store = ActorStoreTransactor::new(did, db, keypair.clone(), resources);
-
-                // Execute user function
-                f(store)
-            })
-            .await
-    }
-}

src/actor_store/actor_store_resources.rs

···
+use std::path::PathBuf;
+use std::sync::Arc;
+
 use super::blob::{BackgroundQueue, BlobStorePlaceholder};
+
 pub(crate) struct ActorStoreResources {
-    pub(crate) blobstore: fn(did: String) -> BlobStorePlaceholder,
-    pub(crate) background_queue: BackgroundQueue,
-    pub(crate) reserved_key_dir: Option<String>,
+    // Factory function to create blobstore instances
+    blobstore_factory: Arc<dyn Fn(String) -> BlobStorePlaceholder + Send + Sync>,
+    // Shared background queue
+    background_queue: Arc<BackgroundQueue>,
+    // Optional directory for reserved keys
+    reserved_key_dir: Option<PathBuf>,
 }
+
 impl ActorStoreResources {
+    // Simple constructor with minimal parameters
     pub(crate) fn new(
-        blobstore: fn(did: String) -> BlobStorePlaceholder,
-        background_queue: BackgroundQueue,
-        reserved_key_dir: Option<String>,
+        blobstore_factory: impl Fn(String) -> BlobStorePlaceholder + Send + Sync + 'static,
+        concurrency: usize,
     ) -> Self {
         Self {
-            blobstore,
-            background_queue,
-            reserved_key_dir,
+            blobstore_factory: Arc::new(blobstore_factory),
+            background_queue: Arc::new(BackgroundQueue::new(concurrency)),
+            reserved_key_dir: None,
         }
     }
 
+    // Set reserved key directory
+    pub(crate) fn with_reserved_key_dir(mut self, dir: impl Into<PathBuf>) -> Self {
+        self.reserved_key_dir = Some(dir.into());
+        self
+    }
+
+    // Get a blobstore for a DID
     pub(crate) fn blobstore(&self, did: String) -> BlobStorePlaceholder {
-        (self.blobstore)(did)
+        (self.blobstore_factory)(did)
+    }
+
+    // Get the background queue
+    pub(crate) fn background_queue(&self) -> Arc<BackgroundQueue> {
+        self.background_queue.clone()
+    }
+
+    // Get the reserved key directory
+    pub(crate) fn reserved_key_dir(&self) -> Option<&PathBuf> {
+        self.reserved_key_dir.as_ref()
     }
 }

src/actor_store/actor_store_transactor.rs

···
-use std::sync::Arc;
-
-use super::{
-    ActorDb, actor_store_resources::ActorStoreResources, preference::PreferenceTransactor,
-    record::RecordTransactor, repo::RepoTransactor,
-};
-use crate::SigningKey;
-
-/// Actor store transactor for managing actor-related transactions.
-pub(crate) struct ActorStoreTransactor {
-    /// Record transactor.
-    pub(crate) record: RecordTransactor,
-    /// Repo transactor.
-    pub(crate) repo: RepoTransactor,
-    /// Preference transactor.
-    pub(crate) pref: PreferenceTransactor,
-}
-impl ActorStoreTransactor {
-    /// Creates a new actor store transactor.
-    ///
-    /// # Arguments
-    ///
-    /// * `did` - The DID of the actor.
-    /// * `db` - The database connection.
-    /// * `keypair` - The signing keypair.
-    /// * `resources` - The actor store resources.
-    pub(crate) fn new(
-        did: String,
-        db: ActorDb,
-        keypair: Arc<SigningKey>,
-        resources: ActorStoreResources,
-    ) -> Self {
-        let blobstore = resources.blobstore(did.clone());
-
-        let record = RecordTransactor::new(db.clone(), blobstore.clone());
-        let pref = PreferenceTransactor::new(db.clone());
-        let repo = RepoTransactor::new(
-            db,
-            blobstore,
-            did,
-            keypair,
-            resources.background_queue,
-            None,
-        );
-
-        Self { record, repo, pref }
-    }
-}

src/actor_store/actor_store_writer.rs

···
-use std::sync::Arc;
-
-use super::{ActorDb, ActorStoreResources};
-use crate::SigningKey;
-
-pub(crate) struct ActorStoreWriter {
-    pub(crate) did: String,
-    pub(crate) db: ActorDb,
-    pub(crate) keypair: Arc<SigningKey>,
-    pub(crate) resources: ActorStoreResources,
-}

src/actor_store/blob/mod.rs

···
 }
 
 impl BlobHandler {
-    /// Create a new blob handler for read operations
-    pub fn new_reader(db: ActorDb, blobstore: impl BlobStore + 'static, did: String) -> Self {
-        Self {
-            db,
-            did,
-            blobstore: Box::new(blobstore),
-            background_queue: None,
-        }
-    }
-
     /// Create a new blob handler with background queue for write operations
-    pub fn new_writer(
+    pub fn new(
         db: ActorDb,
         blobstore: impl BlobStore + 'static,
         background_queue: background::BackgroundQueue,

src/actor_store/mod.rs

···
 //! Actor store implementation for ATProto PDS.
 
 mod actor_store;
-mod actor_store_reader;
+mod actor_store_handler;
 mod actor_store_resources;
-mod actor_store_transactor;
-mod actor_store_writer;
 mod blob;
 mod db;
 mod preference;
···
 mod sql_repo;
 
 pub(crate) use actor_store::ActorStore;
-pub(crate) use actor_store_reader::ActorStoreReader;
+pub(crate) use actor_store_handler::ActorStoreHandler;
 pub(crate) use actor_store_resources::ActorStoreResources;
-pub(crate) use actor_store_transactor::ActorStoreTransactor;
-pub(crate) use actor_store_writer::ActorStoreWriter;
 pub(crate) use db::ActorDb;
 pub(crate) use prepared_write::PreparedWrite;

src/actor_store/repo.rs

···
     block_map::BlockMap,
     cid_set::CidSet,
     repo::Repo,
-    storage::{readable_blockstore::ReadableBlockstore as _, types::RepoStorage as _},
+    storage::{readable_blockstore::ReadableBlockstore as _, types::RepoStorage},
     types::{
         CommitAction, CommitData, CommitDataWithOps, CommitOp, PreparedBlobRef, PreparedWrite,
         WriteOpAction, write_to_op,
···
     util::format_data_key,
 };
 use rsky_syntax::aturi::AtUri;
+use tokio::sync::RwLock;
 
 use super::{
     ActorDb,
-    blob::{BackgroundQueue, BlobReader, BlobStorePlaceholder, BlobTransactor},
+    blob::{BackgroundQueue, BlobHandler, BlobStorePlaceholder},
     record::RecordHandler,
 };
 use crate::SigningKey;
···
     /// Actor DID
     pub did: String,
     /// Backend storage
-    pub storage: SqlRepoStorage,
+    pub storage: Arc<RwLock<dyn RepoStorage>>,
     /// BlobReader for handling blob operations
-    pub blob: BlobReader,
+    pub blob: BlobHandler,
     /// RecordHandler for handling record operations
     pub record: RecordHandler,
     /// BlobTransactor for handling blob writes
-    pub blob_transactor: BlobTransactor,
+    pub blob_transactor: BlobHandler,
     /// RecordHandler for handling record writes
     pub record_transactor: RecordHandler,
     /// Signing keypair
···
         background_queue: BackgroundQueue,
     ) -> Self {
         // Create readers
-        let blob = BlobReader::new(db.clone(), blobstore.clone());
+        let blob = BlobHandler::new(db.clone(), blobstore.clone());
         let record = RecordHandler::new(db.clone(), did.clone());
 
         // Create storage backend with current timestamp
···
 
         // Create transactors
         let blob_transactor =
-            BlobTransactor::new(db.clone(), blobstore.clone(), background_queue.clone());
+            BlobHandler::new(db.clone(), blobstore.clone(), background_queue.clone());
         let record_transactor = RecordHandler::new(db.clone(), blobstore);
 
         Self {

src/db/mod.rs

···
             .pool
             .get()
             .context("Failed to get connection for migrations")?;
+
         conn.run_pending_migrations(MIGRATIONS)
-            .context("Failed to run migrations")?;
+            .map_err(|e| anyhow::anyhow!("Failed to run migrations: {}", e))?;
+
         Ok(())
     }
 
···
             let mut conn = self.pool.get().context("Failed to get connection")?;
             match operation(&mut conn) {
                 Ok(result) => return Ok(result),
-                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(std::time::Duration::from_millis(backoff_ms)).await;
-                }
+                // 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(diesel::result::Error::RollbackTransaction(Box::new(
-                "Unknown error"
-            )))
+            last_error.unwrap_or_else(|| diesel::result::Error::RollbackTransaction)
         ))
     }
 
···
         T: Send + 'static,
     {
         self.run(|conn| {
-            conn.transaction(|tx| {
-                f(tx).map_err(|e| diesel::result::Error::RollbackTransaction(Box::new(e)))
-            })
+            conn.transaction(|tx| f(tx).map_err(|e| diesel::result::Error::RollbackTransaction))
         })
         .await
     }
···
         F: FnOnce(&mut SqliteConnection) -> std::result::Result<T, diesel::result::Error> + Send,
         T: Send + 'static,
     {
-        let conn = &mut self
+        let mut conn = self
             .pool
             .get()
             .context("Failed to get connection for transaction")?;
 
-        conn.transaction(f)
+        conn.transaction(|tx| f(tx))
             .map_err(|e| anyhow::anyhow!("Transaction error: {:?}", e))
     }
 }