prototype actor_store

src/actor_store/actor_store_reader.rs

···
+use anyhow::Result;
 use std::sync::Arc;
 
 use super::{
-    ActorStoreTransactor, db::ActorDb, preference::PreferenceReader, record::RecordReader,
-    repo::RepoReader, resources::ActorStoreResources,
+    ActorStoreTransactor, actor_store_resources::ActorStoreResources, db::ActorDb,
+    preference::PreferenceReader, record::RecordReader,
 };
 use crate::SigningKey;
 
+/// Reader for the actor store.
 pub(crate) struct ActorStoreReader {
-    pub(crate) repo: RepoReader,
+    /// DID of the actor.
+    pub(crate) did: String,
+    /// Record reader.
     pub(crate) record: RecordReader,
+    /// Preference reader.
     pub(crate) pref: PreferenceReader,
+    /// RepoReader placeholder - will be implemented later.
+    pub(crate) repo: (), // Placeholder for RepoReader
+    /// Function to get keypair.
+    keypair_fn: Box<dyn Fn() -> Result<Arc<SigningKey>> + Send + Sync>,
+    /// Database connection.
+    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>>,
+        keypair: impl Fn() -> Result<Arc<SigningKey>> + Send + Sync + 'static,
     ) -> Self {
-        let blobstore = resources.blobstore(&did);
+        // Create readers
+        let record = RecordReader::new(db.clone());
+        let pref = PreferenceReader::new(db.clone());
 
-        let repo = RepoReader::new(db.clone(), blobstore);
-        let record = RecordReader::new(db.clone());
-        let pref = PreferenceReader::new(db);
-        keypair();
+        // Store keypair function for later use
+        let keypair_fn = Box::new(keypair);
+
+        // Initial keypair call as in TypeScript implementation
+        let _ = keypair_fn();
+
+        // For now, we use a placeholder for RepoReader
+        // Real implementation will need to be added later
+        let repo = ();
+
+        Self {
+            did,
+            repo,
+            record,
+            pref,
+            keypair_fn,
+            db,
+            resources,
+        }
+    }
 
-        Self { repo, record, pref }
+    /// 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>,
     {
-        let keypair = self.keypair();
-        let db_txn = self.db.transaction().await?;
-        let store =
-            ActorStoreTransactor::new(self.did.clone(), db_txn, keypair, self.resources.clone());
-        f(store)
+        let keypair = self.keypair()?;
+        let did = self.did.clone();
+        let resources = self.resources.clone();
+
+        self.db
+            .transaction_no_retry(move |tx| {
+                // Create a transactor with the transaction
+                let store = ActorStoreTransactor::new_with_transaction(
+                    did,
+                    tx, // Pass the transaction directly
+                    keypair.clone(),
+                    resources,
+                );
+
+                // Execute user function
+                f(store).map_err(|e| sqlx::Error::Custom(Box::new(e))) // Convert anyhow::Error to sqlx::Error
+            })
+            .await
+            .map_err(|e| anyhow::anyhow!("Transaction error: {:?}", e))
     }
 }

src/actor_store/actor_store_resources.rs

···
+use crate::repo::types::BlobStore;
+
+use super::blob::BackgroundQueue;
+pub(crate) struct ActorStoreResources {
+    pub(crate) blobstore: fn(did: String) -> BlobStore,
+    pub(crate) background_queue: BackgroundQueue,
+    pub(crate) reserved_key_dir: Option<String>,
+}
+impl ActorStoreResources {
+    pub(crate) fn new(
+        blobstore: fn(did: String) -> BlobStore,
+        background_queue: BackgroundQueue,
+        reserved_key_dir: Option<String>,
+    ) -> Self {
+        Self {
+            blobstore,
+            background_queue,
+            reserved_key_dir,
+        }
+    }
+
+    pub(crate) fn blobstore(&self, did: String) -> BlobStore {
+        (self.blobstore)(did)
+    }
+}

src/actor_store/actor_store_transactor.rs

···
 use std::sync::Arc;
 
-use super::{ActorDb, resources::ActorStoreResources};
+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 {
-    pub(crate) did: String,
-    pub(crate) db: ActorDb,
-    pub(crate) keypair: Arc<SigningKey>,
-    pub(crate) resources: ActorStoreResources,
+    /// 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 {
-        Self {
-            did,
-            db,
-            keypair,
-            resources,
-        }
+        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);
+
+        Self { record, repo, pref }
     }
 }

src/actor_store/blob/transactor.rs

···
     }
 
     /// Process blobs for a repository write operation.
-    pub(crate) async fn process_write_blobs(&self, writes: Vec<PreparedWrite>) -> Result<()> {
+    pub(crate) async fn process_write_blobs(
+        &self,
+        _rev: &String, // Typescript impl declares rev but never uses it
+        writes: Vec<PreparedWrite>,
+    ) -> Result<()> {
         self.delete_dereferenced_blobs(writes.clone(), false)
             .await
             .context("failed to delete dereferenced blobs")?;

src/actor_store/mod.rs

···
 
 mod actor_store;
 mod actor_store_reader;
+mod actor_store_resources;
 mod actor_store_transactor;
 mod actor_store_writer;
 mod blob;
···
 mod preference;
 mod record;
 mod repo;
-mod resources;
 
 pub(crate) use actor_store::ActorStore;
 pub(crate) use actor_store_reader::ActorStoreReader;
+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 resources::ActorStoreResources;

src/actor_store/repo/reader.rs

···
 
 use super::sql_repo_reader::SqlRepoReader;
 use crate::{
-    actor_store::{blob::BlobReader, record::RecordReader},
-    repo::block_map::BlockMap,
+    actor_store::{ActorDb, blob::BlobReader, record::RecordReader},
+    repo::{block_map::BlockMap, types::BlobStore},
 };
 
 /// Reader for repository data in the actor store.
···
 
 impl RepoReader {
     /// Create a new repository reader.
-    // pub(crate) fn new(db: ActorDb, did: String, blob_config: BlobConfig) -> Self {
-    //     Self {
-    //         storage: SqlRepoReader::new(db.clone(), did.clone()),
-    //         db,
-    //         did,
-    //     }
-    // }
+    pub(crate) fn new(db: ActorDb, blobstore: BlobStore) -> Self {
+        let blob = BlobReader::new(db.clone(), blobstore);
+        let record = RecordReader::new(db.clone());
+        let storage = SqlRepoReader::new(db);
+
+        Self {
+            blob,
+            record,
+            storage,
+        }
+    }
 
     /// Get event data for synchronization.
     pub(crate) async fn get_sync_event_data(&self) -> Result<SyncEventData> {

src/actor_store/repo/sql_repo_reader.rs

···
     Cid,
     blockstore::{AsyncBlockStoreRead, Error as BlockstoreError},
 };
+use rsky_repo::storage::readable_blockstore::ReadableBlockstore;
 use sqlx::Row;
 use std::str::FromStr;
 use std::sync::Arc;
···
     pub cache: Arc<RwLock<BlockMap>>,
     /// Database connection.
     pub db: ActorDb,
-    /// DID of the repository owner.
-    pub did: String,
 }
 
 /// Repository root with CID and revision.
···
 
 impl SqlRepoReader {
     /// Create a new SQL repository reader.
-    pub(crate) fn new(db: ActorDb, did: String) -> Self {
+    pub(crate) fn new(db: ActorDb) -> Self {
         Self {
             cache: Arc::new(RwLock::new(BlockMap::new())),
             db,
-            did,
         }
     }
-
     // async getRoot(): Promise<CID> {
-    // async has(cid: CID): Promise<boolean> {
     // async getCarStream(since?: string) {
     // async *iterateCarBlocks(since?: string): AsyncIterable<CarBlock> {
     // async getBlockRange(since?: string, cursor?: RevCursor) {
     // async countBlocks(): Promise<number> {
     // async destroy(): Promise<void> {
 
-    pub(crate) async fn get_bytes(&self, cid: &Cid) -> Result<Option<Vec<u8>>> {
+    /// Get the detailed root information.
+    pub(crate) async fn get_root_detailed(&self) -> Result<RootInfo> {
+        let row = sqlx::query!(r#"SELECT cid, rev FROM repo_root"#)
+            .fetch_one(&self.db.pool)
+            .await
+            .context("failed to fetch repo root")?;
+
+        Ok(RootInfo {
+            cid: Cid::from_str(&row.cid)?,
+            rev: row.rev,
+        })
+    }
+}
+
+impl ReadableBlockstore for SqlRepoReader {
+    async fn get_bytes(&self, cid: &Cid) -> Result<Option<Vec<u8>>> {
         // First check the cache
         {
             let cache_guard = self.cache.read().await;
···
         Ok(content)
     }
 
-    /// Get the detailed root information.
-    pub(crate) async fn get_root_detailed(&self) -> Result<RootInfo> {
-        let did = self.did.clone();
-        let row = sqlx::query!(r#"SELECT cid, rev FROM repo_root WHERE did = ?"#, did)
-            .fetch_one(&self.db.pool)
-            .await
-            .context("failed to fetch repo root")?;
-
-        Ok(RootInfo {
-            cid: Cid::from_str(&row.cid)?,
-            rev: row.rev,
-        })
+    async fn has(&self, cid: &Cid) -> Result<bool> {
+        self.get_bytes(cid).await.map(|bytes| bytes.is_some())
     }
 
     /// Get blocks from the database.
-    pub(crate) async fn get_blocks(&self, cids: Vec<Cid>) -> Result<BlocksAndMissing> {
+    async fn get_blocks(&self, cids: Vec<Cid>) -> Result<BlocksAndMissing> {
         let cached = { self.cache.write().await.get_many(cids)? }; // TODO: use read lock?
 
         if cached.missing.is_empty() {
···
 
             let query = format!(
                 "SELECT cid, content FROM repo_block
-                     WHERE did = ? AND cid IN ({})
+                     WHERE cid IN ({})
                      ORDER BY cid",
                 placeholders
             );
 
             let mut query_builder = sqlx::query(&query);
-            query_builder = query_builder.bind(&self.did);
             for cid in chunk {
                 query_builder = query_builder.bind(cid);
             }

src/actor_store/repo/sql_repo_transactor.rs

···
     /// Cache for blocks.
     pub cache: BlockMap,
     /// Current timestamp.
-    pub now: String,
+    pub now: Option<String>,
 }
 
 impl SqlRepoTransactor {
     /// Create a new SQL repository transactor.
-    pub(crate) fn new(db: ActorDb, did: String) -> Self {
-        let now = chrono::Utc::now().to_rfc3339();
+    pub(crate) fn new(db: ActorDb, did: String, now: Option<String>) -> Self {
         Self {
             reader: SqlRepoReader::new(db, did),
             cache: BlockMap::new(),
···
     }
 
     /// Apply a commit to the repository.
-    pub(crate) async fn apply_commit(&self, commit: CommitData, is_create: bool) -> Result<()> {
+    pub(crate) async fn apply_commit(&self, commit: &CommitData, is_create: bool) -> Result<()> {
         let is_create = is_create || false;
         let removed_cids_list = commit.removed_cids.to_list();
 

src/actor_store/repo/transactor.rs

···
 //! Repository transactor for the actor store.
 
-use std::str::FromStr;
-
-use anyhow::Result;
+use anyhow::{Context as _, Result};
 use atrium_repo::Cid;
 use rsky_syntax::aturi::AtUri;
+use std::sync::Arc;
 
 use crate::{
-    actor_store::ActorDb, repo::types::{CommitAction, CommitDataWithOps, CommitOp, PreparedWrite, WriteOpAction}, SigningKey
+    SigningKey,
+    actor_store::{
+        ActorDb,
+        blob::{BackgroundQueue, BlobTransactor},
+        record::RecordTransactor,
+        repo::{reader::RepoReader, sql_repo_transactor::SqlRepoTransactor},
+        resources::ActorStoreResources,
+    },
+    repo::{
+        Repo,
+        block_map::BlockMap,
+        types::{
+            BlobStore, CommitAction, CommitDataWithOps, CommitOp, PreparedCreate, PreparedWrite,
+            WriteOpAction, format_data_key,
+        },
+    },
 };
 
-use super::{reader::RepoReader, sql_repo_transactor::SqlRepoTransactor};
-
-/// Transactor for repository operations.
+/// Repository transactor for the actor store.
 pub(crate) struct RepoTransactor {
     /// The inner reader.
-    pub(crate) reader: RepoReader,
-    ///
+    pub reader: RepoReader,
+    /// BlobTransactor for handling blobs.
+    pub blob: BlobTransactor,
+    /// RecordTransactor for handling records.
+    pub record: RecordTransactor,
+    /// SQL repository transactor.
+    pub storage: SqlRepoTransactor,
 }
-
 
 impl RepoTransactor {
     /// Create a new repository transactor.
     pub(crate) fn new(
         db: ActorDb,
+        blobstore: BlobStore,
         did: String,
-        signing_key: SigningKey,
-        blob_config: crate::config::BlobConfig,
+        signing_key: Arc<SigningKey>,
+        background_queue: BackgroundQueue,
+        now: Option<String>,
     ) -> Self {
+        // Create a new RepoReader
+        let reader = RepoReader::new(db.clone(), blobstore.clone(), did.clone(), signing_key);
+
+        // Create a new BlobTransactor
+        let blob = BlobTransactor::new(db.clone(), blobstore.clone(), background_queue);
+
+        // Create a new RecordTransactor
+        let record = RecordTransactor::new(db.clone(), blobstore);
+
+        // Create a new SQL repository transactor
+        let storage = SqlRepoTransactor::new(db, did.clone(), now);
+
         Self {
-            reader: RepoReader::new(db.clone(), did.clone(), blob_config),
-            storage: SqlRepoTransactor::new(db, did.clone()),
-            did,
-            signing_key,
+            reader,
+            blob,
+            record,
+            storage,
         }
     }
 
-    /// Load the repository if it exists.
-    // pub async fn maybe_load_repo(
-    //     &self,
-    // ) -> Result<Option<Repository<impl AsyncBlockStoreRead + AsyncBlockStoreWrite>>> {
-    //     todo!("Implement maybe_load_repo")
-    // }
+    /// Try to load a repository.
+    pub(crate) async fn maybe_load_repo(&self) -> Result<Option<Repo>> {
+        // Query the repository root
+        let root = sqlx::query!("SELECT cid FROM repo_root LIMIT 1")
+            .fetch_optional(&self.db.pool)
+            .await
+            .context("failed to query repo root")?;
+
+        // If found, load the repo
+        if let Some(row) = root {
+            let cid = Cid::try_from(&row.cid)?;
+            let repo = Repo::load(&self.storage, cid).await?;
+            Ok(Some(repo))
+        } else {
+            Ok(None)
+        }
+    }
 
-    /// Create a new repository.
+    /// Create a new repository with the given writes.
     pub(crate) async fn create_repo(
         &self,
-        writes: Vec<PreparedWrite>,
+        writes: Vec<PreparedCreate>,
     ) -> Result<CommitDataWithOps> {
-        todo!("Implement create_repo")
+        // Assert we're in a transaction
+        self.db.assert_transaction()?;
+
+        // Convert writes to operations
+        let ops = writes.iter().map(|w| create_write_to_op(w)).collect();
+
+        // Format the initial commit
+        let commit =
+            Repo::format_init_commit(&self.storage, &self.did, &self.signing_key, ops).await?;
+
+        // Apply the commit, index the writes, and process blobs in parallel
+        let results = futures::future::join3(
+            self.storage.apply_commit(&commit, true),
+            self.index_writes(&writes, &commit.rev),
+            self.blob.process_write_blobs(&commit.rev, &writes),
+        )
+        .await;
+
+        // Check for errors in each parallel task
+        results.0?;
+        results.1?;
+        results.2?;
+
+        // Create commit operations
+        let ops = writes
+            .iter()
+            .map(|w| CommitOp {
+                action: CommitAction::Create,
+                path: format_data_key(&w.uri.collection, &w.uri.rkey),
+                cid: Some(w.cid),
+                prev: None,
+            })
+            .collect();
+
+        // Return the commit data with operations
+        Ok(CommitDataWithOps {
+            commit_data: commit,
+            ops,
+            prev_data: None,
+        })
     }
 
     /// Process writes to the repository.
···
         writes: Vec<PreparedWrite>,
         swap_commit_cid: Option<Cid>,
     ) -> Result<CommitDataWithOps> {
-        // Validate parameters
+        // Assert we're in a transaction
+        self.db.assert_transaction()?;
+
+        // Check write limit
         if writes.len() > 200 {
-            return Err(anyhow::anyhow!("Too many writes. Max: 200").into());
+            return Err(anyhow::anyhow!("Too many writes. Max: 200"));
         }
 
-        // Format the commit (creates blocks and structures the operations)
-        let commit = self.format_commit(writes.clone(), swap_commit_cid).await?;
+        // Format the commit
+        let commit = self.format_commit(writes, swap_commit_cid).await?;
 
-        // Check commit size (prevent large commits)
-        if commit.commit_data.relevant_blocks.byte_size()? > 2000000 {
-            return Err(anyhow::anyhow!("Too many writes. Max event size: 2MB").into());
+        // Check commit size limit (2MB)
+        if commit.commit_data.relevant_blocks.byte_size()? > 2_000_000 {
+            return Err(anyhow::anyhow!("Too many writes. Max event size: 2MB"));
         }
 
-        // Execute these operations in parallel for better performance
-        tokio::try_join!(
-            // Persist the commit to repo storage
-            self.storage.apply_commit(commit.commit_data.clone(), true),
-            // Send to indexing
-            self.index_writes(writes.clone(), &commit.commit_data.rev),
-            // Process blobs from writes
-            self.process_write_blobs(&commit.commit_data.rev, &writes),
-        )?;
+        // Apply the commit, index the writes, and process blobs in parallel
+        let results = futures::future::join3(
+            self.storage.apply_commit(&commit.commit_data, false),
+            self.index_writes(writes, &commit.commit_data.rev),
+            self.blob
+                .process_write_blobs(&commit.commit_data.rev, writes),
+        )
+        .await;
+
+        // Check for errors in each parallel task
+        results.0?;
+        results.1?;
+        results.2?;
 
         Ok(commit)
     }
 
-    /// Format a commit for writing.
+    /// Format a commit for the given writes.
     pub(crate) async fn format_commit(
         &self,
         writes: Vec<PreparedWrite>,
         swap_commit: Option<Cid>,
     ) -> Result<CommitDataWithOps> {
-        // Get current repository root
+        // Get the current root
         let curr_root = self.storage.get_root_detailed().await?;
-        if curr_root.cid.is_nil() {
-            return Err(anyhow::anyhow!("No repo root found for {}", self.did).into());
+        if curr_root.is_none() {
+            return Err(anyhow::anyhow!("No repo root found for {}", self.did));
         }
 
-        // Check swap commit if provided
-        if let Some(swap_cid) = swap_commit {
-            if !curr_root.cid.equals(swap_cid) {
+        let curr_root = curr_root.unwrap();
+
+        // Check commit swap if requested
+        if let Some(swap) = swap_commit {
+            if curr_root.cid != swap {
                 return Err(anyhow::anyhow!(
-                    "Bad commit swap: expected {}, got {}",
-                    swap_cid,
-                    curr_root.cid
-                )
-                .into());
+                    "Bad commit swap: current={}, expected={}",
+                    curr_root.cid,
+                    swap
+                ));
             }
         }
 
-        // Cache last commit for performance
+        // Cache the revision for better performance
         self.storage.cache_rev(&curr_root.rev).await?;
 
+        // Prepare collections for tracking changes
         let mut new_record_cids = Vec::new();
         let mut del_and_update_uris = Vec::new();
         let mut commit_ops = Vec::new();
 
-        // Process each write to create commit operations
-        for write in &writes {
-            let uri_str = write.uri().clone();
-            let uri = AtUri::try_from(uri_str.as_str())?;
+        // Process each write
+        for write in writes {
+            let action = &write.action;
+            let uri = &write.uri;
+            let swap_cid = write.swap_cid;
 
-            match write.action() {
-                WriteOpAction::Create | WriteOpAction::Update => {
-                    if let Some(cid) = write.cid() {
-                        new_record_cids.push(cid);
-                    }
-                }
-                _ => {}
+            // Track new record CIDs
+            if *action != WriteOpAction::Delete {
+                new_record_cids.push(write.cid);
             }
 
-            if write.action() != &WriteOpAction::Create {
+            // Track deleted/updated URIs
+            if *action != WriteOpAction::Create {
                 del_and_update_uris.push(uri.clone());
             }
 
-            // Get current record if it exists
-            let record = self.record.get_record(&uri.to_string(), None, true).await?;
-            let curr_record = record.map(|r| Cid::from_str(&r.cid).ok()).flatten();
+            // Get the current record if it exists
+            let record = self.record.get_record(uri, None, true).await?;
+            let curr_record = record.as_ref().map(|r| Cid::try_from(&r.cid).unwrap());
 
-            // Create the operation
-            let path = format!("{}/{}", uri.get_collection(), uri.get_rkey());
+            // Create commit operation
             let mut op = CommitOp {
-                action: match write.action() {
-                    WriteOpAction::Create => CommitAction::Create,
-                    WriteOpAction::Update => CommitAction::Update,
-                    WriteOpAction::Delete => CommitAction::Delete,
-                },
-                path: path.clone(),
-                cid: match write.action() {
-                    WriteOpAction::Delete => None,
-                    _ => write.cid(),
+                action: action.clone(),
+                path: format_data_key(&uri.collection, &uri.rkey),
+                cid: if *action == WriteOpAction::Delete {
+                    None
+                } else {
+                    Some(write.cid)
                 },
-                prev: curr_record.clone(),
+                prev: curr_record,
             };
 
-            // Validate swap consistency
-            if let Some(swap_cid) = write.swap_cid() {
-                match write.action() {
-                    WriteOpAction::Create if swap_cid.is_some() => {
+            commit_ops.push(op);
+
+            // Validate swap_cid conditions
+            if swap_cid.is_some() {
+                match action {
+                    WriteOpAction::Create if swap_cid != Some(None) => {
+                        return Err(anyhow::anyhow!(
+                            "Bad record swap: there should be no current record for a create"
+                        ));
+                    }
+                    WriteOpAction::Update if swap_cid == Some(None) => {
                         return Err(anyhow::anyhow!(
-                            "Bad record swap: cannot provide swap CID for create"
-                        )
-                        .into());
+                            "Bad record swap: there should be a current record for an update"
+                        ));
                     }
-                    WriteOpAction::Update | WriteOpAction::Delete if swap_cid.is_none() => {
+                    WriteOpAction::Delete if swap_cid == Some(None) => {
                         return Err(anyhow::anyhow!(
-                            "Bad record swap: must provide swap CID for update/delete"
-                        )
-                        .into());
+                            "Bad record swap: there should be a current record for a delete"
+                        ));
                     }
                     _ => {}
                 }
 
-                if swap_cid.is_some()
-                    && curr_record.is_some()
-                    && !curr_record.unwrap().equals(swap_cid.unwrap())
-                {
-                    return Err(anyhow::anyhow!(
-                        "Bad record swap: expected {}, got {:?}",
-                        swap_cid.unwrap(),
-                        curr_record
-                    )
-                    .into());
+                if let Some(Some(swap)) = swap_cid {
+                    if curr_record.is_some() && curr_record != Some(swap) {
+                        return Err(anyhow::anyhow!(
+                            "Bad record swap: current={:?}, expected={}",
+                            curr_record,
+                            swap
+                        ));
+                    }
                 }
             }
-
-            commit_ops.push(op);
         }
 
-        // Load the current repository and prepare for modification
-        let repo = crate::storage::open_repo(&self.storage.reader.config, &self.did, curr_root.cid)
-            .await?;
-        let prev_data = repo.commit().data;
+        // Load the repo
+        let repo = Repo::load(&self.storage, curr_root.cid).await?;
+        let prev_data = repo.commit.data.clone();
 
-        // Convert PreparedWrites to RecordWriteOps
-        let write_ops = writes
-            .iter()
-            .map(|w| write_to_op(w.clone()))
-            .collect::<Result<Vec<_>>>()?;
+        // Convert writes to ops
+        let write_ops = writes.iter().map(|w| write_to_op(w)).collect();
 
-        // Format the new commit
+        // Format the commit
         let commit = repo.format_commit(write_ops, &self.signing_key).await?;
 
-        // Find blocks referenced by other records
+        // Find blocks that would be deleted but are referenced by another record
         let dupe_record_cids = self
             .get_duplicate_record_cids(&commit.removed_cids.to_list(), &del_and_update_uris)
             .await?;
 
-        // Remove duplicated CIDs from the removal list
-        for cid in dupe_record_cids {
-            commit.removed_cids.delete(cid);
+        // Remove duplicates from removed_cids
+        for cid in &dupe_record_cids {
+            commit.removed_cids.delete(*cid);
         }
 
-        // Ensure all necessary blocks are included
-        let new_record_blocks = commit.relevant_blocks.get_many(new_record_cids)?;
+        // Find blocks that are relevant to ops but not included in diff
+        let new_record_blocks = commit.relevant_blocks.get_many(&new_record_cids)?;
         if !new_record_blocks.missing.is_empty() {
-            let missing_blocks = self.storage.get_blocks(new_record_blocks.missing).await?;
+            let missing_blocks = self
+                .storage
+                .reader
+                .get_blocks(&new_record_blocks.missing)
+                .await?;
             commit.relevant_blocks.add_map(missing_blocks.blocks)?;
         }
 
···
         })
     }
 
-    /// Index writes in the repository.
-    pub(crate) async fn index_writes(&self, writes: Vec<PreparedWrite>, rev: &str) -> Result<()> {
-        // Process each write for indexing
-        for write in writes {
-            let uri_str = write.uri().clone();
-            let uri = AtUri::try_from(uri_str.as_str())?;
+    /// Index writes to the database.
+    pub(crate) async fn index_writes(&self, writes: &[PreparedWrite], rev: &str) -> Result<()> {
+        // Assert we're in a transaction
+        self.db.assert_transaction()?;
 
-            match write.action() {
+        // Process each write in parallel
+        let futures = writes.iter().map(|write| async move {
+            match write.action {
                 WriteOpAction::Create | WriteOpAction::Update => {
-                    if let PreparedWrite::Create(w) | PreparedWrite::Update(w) = write {
-                        self.record
-                            .index_record(
-                                uri,
-                                w.cid,
-                                Some(w.record),
-                                write.action().clone(),
-                                rev,
-                                None, // Use current timestamp
-                            )
-                            .await?;
-                    }
+                    self.record
+                        .index_record(
+                            &write.uri,
+                            &write.cid,
+                            &write.record,
+                            &write.action,
+                            rev,
+                            &self.now,
+                        )
+                        .await
                 }
-                WriteOpAction::Delete => {
-                    self.record.delete_record(&uri).await?;
-                }
+                WriteOpAction::Delete => self.record.delete_record(&write.uri).await,
             }
-        }
+        });
+
+        // Wait for all indexing operations to complete
+        futures::future::try_join_all(futures).await?;
 
         Ok(())
     }
 
-    /// Get duplicate record CIDs.
+    /// Get record CIDs that are duplicated elsewhere in the repository.
     pub(crate) async fn get_duplicate_record_cids(
         &self,
         cids: &[Cid],
···
             return Ok(Vec::new());
         }
 
+        // Convert CIDs and URIs to strings
         let cid_strs: Vec<String> = cids.iter().map(|c| c.to_string()).collect();
         let uri_strs: Vec<String> = touched_uris.iter().map(|u| u.to_string()).collect();
 
-        // Find records that have the same CIDs but weren't touched in this operation
-        let duplicates = sqlx::query_scalar!(
-            r#"
-            SELECT cid FROM record
-            WHERE cid IN (SELECT unnest($1::text[]))
-            AND uri NOT IN (SELECT unnest($2::text[]))
-            "#,
-            &cid_strs,
-            &uri_strs
+        // Query the database for duplicates
+        let rows = sqlx::query!(
+            "SELECT cid FROM record WHERE cid IN (?) AND uri NOT IN (?)",
+            cid_strs.join(","),
+            uri_strs.join(",")
         )
-        .fetch_all(&self.reader.db)
-        .await?;
+        .fetch_all(&self.db.pool)
+        .await
+        .context("failed to query duplicate record CIDs")?;
 
-        // Convert string CIDs back to Cid objects
-        let dupe_cids = duplicates
+        // Convert to CIDs
+        let result = rows
             .into_iter()
-            .filter_map(|cid_str| Cid::from_str(&cid_str).ok())
-            .collect();
+            .map(|row| Cid::try_from(&row.cid))
+            .collect::<Result<Vec<_>, _>>()?;
 
-        Ok(dupe_cids)
+        Ok(result)
     }
-
-    pub(crate) async fn process_write_blobs(
-        &self,
-        rev: &str,
-        writes: &[PreparedWrite],
-    ) -> Result<()> {
-        // First handle deletions or updates
-        let uris: Vec<String> = writes
-            .iter()
-            .filter(|w| matches!(w.action(), WriteOpAction::Delete | WriteOpAction::Update))
-            .map(|w| w.uri().clone())
-            .collect();
+}
 
-        if !uris.is_empty() {
-            // Remove blob references for deleted/updated records
-            self.blob.delete_dereferenced_blobs(&uris).await?;
-        }
+// Helper functions
 
-        // Process each blob in each write
-        for write in writes {
-            if let PreparedWrite::Create(w) | PreparedWrite::Update(w) = write {
-                for blob in &w.blobs {
-                    // Verify and make permanent
-                    self.blob.verify_blob_and_make_permanent(blob).await?;
-
-                    // Associate blob with record
-                    self.blob
-                        .associate_blob(&blob.cid.to_string(), write.uri())
-                        .await?;
-                }
-            }
-        }
+/// Convert a PreparedCreate to an operation.
+fn create_write_to_op(write: &PreparedCreate) -> WriteOp {
+    WriteOp {
+        action: WriteOpAction::Create,
+        collection: write.uri.collection.clone(),
+        rkey: write.uri.rkey.clone(),
+        record: write.record.clone(),
+    }
+}
 
-        Ok(())
+/// Convert a PreparedWrite to an operation.
+fn write_to_op(write: &PreparedWrite) -> WriteOp {
+    match write.action {
+        WriteOpAction::Create => WriteOp {
+            action: WriteOpAction::Create,
+            collection: write.uri.collection.clone(),
+            rkey: write.uri.rkey.clone(),
+            record: write.record.clone(),
+        },
+        WriteOpAction::Update => WriteOp {
+            action: WriteOpAction::Update,
+            collection: write.uri.collection.clone(),
+            rkey: write.uri.rkey.clone(),
+            record: write.record.clone(),
+        },
+        WriteOpAction::Delete => WriteOp {
+            action: WriteOpAction::Delete,
+            collection: write.uri.collection.clone(),
+            rkey: write.uri.rkey.clone(),
+            record: None,
+        },
     }
 }

src/actor_store/resources.rs

···
-
-use std::sync::Arc;
-
-use crate::config::{BlobConfig, RepoConfig};
-
-pub(crate) struct ActorStoreResources {
-    pub(crate) config: RepoConfig,
-    pub(crate) blob_config: BlobConfig,
-    pub(crate) background_queue: Arc<()>, // Placeholder
-}
-
-impl Clone for ActorStoreResources {
-    fn clone(&self) -> Self {
-        Self {
-            config: self.config.clone(),
-            blob_config: self.blob_config.clone(),
-            background_queue: self.background_queue.clone(),
-        }
-    }
-}

src/repo/mod.rs

···
 pub(crate) mod block_map;
+pub(crate) mod storage;
 pub(crate) mod types;

src/repo/storage/mod.rs

···
+//! ReadableBlockstore for managing blocks in the repository.
+
+use crate::repo::block_map::{BlockMap, MissingBlockError};
+use crate::repo::util::{cbor_to_lex_record, parse_obj_by_def};
+use anyhow::{Context, Result};
+use atrium_repo::Cid;
+use std::collections::HashMap;
+use std::sync::Arc;
+
+/// Trait for a readable blockstore.
+pub(crate) trait ReadableBlockstore {
+    /// Get the raw bytes for a given CID.
+    fn get_bytes(&self, cid: &Cid) -> Result<Option<Vec<u8>>>;
+
+    /// Check if a block exists for a given CID.
+    fn has(&self, cid: &Cid) -> Result<bool>;
+
+    /// Get multiple blocks for a list of CIDs.
+    fn get_blocks(&self, cids: &[Cid]) -> Result<(BlockMap, Vec<Cid>)>;
+
+    /// Attempt to read an object by CID and definition.
+    fn attempt_read<T>(&self, cid: &Cid, def: &str) -> Result<Option<(T, Vec<u8>)>>
+    where
+        T: serde::de::DeserializeOwned;
+
+    /// Read an object and its raw bytes by CID and definition.
+    fn read_obj_and_bytes<T>(&self, cid: &Cid, def: &str) -> Result<(T, Vec<u8>)>
+    where
+        T: serde::de::DeserializeOwned;
+
+    /// Read an object by CID and definition.
+    fn read_obj<T>(&self, cid: &Cid, def: &str) -> Result<T>
+    where
+        T: serde::de::DeserializeOwned;
+
+    /// Attempt to read a record by CID.
+    fn attempt_read_record(&self, cid: &Cid) -> Result<Option<RepoRecord>>;
+
+    /// Read a record by CID.
+    fn read_record(&self, cid: &Cid) -> Result<RepoRecord>;
+}
+
+/// Concrete implementation of the ReadableBlockstore.
+pub(crate) struct InMemoryBlockstore {
+    blocks: HashMap<Cid, Vec<u8>>,
+}
+
+impl InMemoryBlockstore {
+    /// Create a new in-memory blockstore.
+    pub fn new() -> Self {
+        Self {
+            blocks: HashMap::new(),
+        }
+    }
+
+    /// Add a block to the blockstore.
+    pub fn add_block(&mut self, cid: Cid, bytes: Vec<u8>) {
+        self.blocks.insert(cid, bytes);
+    }
+}
+
+impl ReadableBlockstore for InMemoryBlockstore {
+    fn get_bytes(&self, cid: &Cid) -> Result<Option<Vec<u8>>> {
+        Ok(self.blocks.get(cid).cloned())
+    }
+
+    fn has(&self, cid: &Cid) -> Result<bool> {
+        Ok(self.blocks.contains_key(cid))
+    }
+
+    fn get_blocks(&self, cids: &[Cid]) -> Result<(BlockMap, Vec<Cid>)> {
+        let mut blocks = BlockMap::new();
+        let mut missing = Vec::new();
+
+        for cid in cids {
+            if let Some(bytes) = self.blocks.get(cid) {
+                blocks.insert(*cid, bytes.clone());
+            } else {
+                missing.push(*cid);
+            }
+        }
+
+        Ok((blocks, missing))
+    }
+
+    fn attempt_read<T>(&self, cid: &Cid, def: &str) -> Result<Option<(T, Vec<u8>)>>
+    where
+        T: serde::de::DeserializeOwned,
+    {
+        if let Some(bytes) = self.get_bytes(cid)? {
+            let obj = parse_obj_by_def(&bytes, cid, def)?;
+            Ok(Some((obj, bytes)))
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn read_obj_and_bytes<T>(&self, cid: &Cid, def: &str) -> Result<(T, Vec<u8>)>
+    where
+        T: serde::de::DeserializeOwned,
+    {
+        if let Some((obj, bytes)) = self.attempt_read(cid, def)? {
+            Ok((obj, bytes))
+        } else {
+            Err(MissingBlockError::new(*cid, def).into())
+        }
+    }
+
+    fn read_obj<T>(&self, cid: &Cid, def: &str) -> Result<T>
+    where
+        T: serde::de::DeserializeOwned,
+    {
+        let (obj, _) = self.read_obj_and_bytes(cid, def)?;
+        Ok(obj)
+    }
+
+    fn attempt_read_record(&self, cid: &Cid) -> Result<Option<RepoRecord>> {
+        match self.read_record(cid) {
+            Ok(record) => Ok(Some(record)),
+            Err(_) => Ok(None),
+        }
+    }
+
+    fn read_record(&self, cid: &Cid) -> Result<RepoRecord> {
+        if let Some(bytes) = self.get_bytes(cid)? {
+            cbor_to_lex_record(&bytes)
+        } else {
+            Err(MissingBlockError::new(*cid, "RepoRecord").into())
+        }
+    }
+}