prototype actor_store; reorganize

src/actor_store/actor_store.rs

···
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use anyhow::Result;
+use sqlx::SqlitePool;
+
+use super::actor_store_reader::ActorStoreReader;
+use super::actor_store_transactor::ActorStoreTransactor;
+use super::actor_store_writer::ActorStoreWriter;
+use super::resources::ActorStoreResources;
+use crate::SigningKey;
+
+pub(crate) struct ActorStore {
+    pub(crate) directory: PathBuf,
+    reserved_key_dir: PathBuf,
+    resources: ActorStoreResources,
+}
+
+struct ActorLocation {
+    directory: PathBuf,
+    db_location: PathBuf,
+    key_location: PathBuf,
+}
+
+impl ActorStore {
+    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,
+            resources,
+        }
+    }
+
+    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!()
+    }
+
+    pub(crate) async fn exists(&self, did: &str) -> Result<bool> {
+        // const location = await this.getLocation(did)
+        // return await fileExists(location.dbLocation)
+        todo!()
+    }
+
+    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!()
+    }
+
+    pub(crate) async fn open_db(&self, did: &str) -> Result<SqlitePool> {
+        // const { dbLocation } = await this.getLocation(did)
+        // const exists = await fileExists(dbLocation)
+        // if (!exists) {
+        //   throw new InvalidRequestError('Repo not found', 'NotFound')
+        // }
+
+        // const db = getDb(dbLocation, this.cfg.disableWalAutoCheckpoint)
+
+        // // 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
+        // }
+
+        // return db
+        todo!()
+    }
+
+    pub(crate) async fn read<T, F>(&self, did: &str, f: F) -> Result<T>
+    where
+        F: FnOnce(ActorStoreReader) -> 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!()
+    }
+
+    pub(crate) async fn transact<T, F>(&self, did: &str, f: F) -> Result<T>
+    where
+        F: FnOnce(ActorStoreTransactor) -> 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!()
+    }
+
+    pub(crate) async fn write_no_transaction<T, F>(&self, did: &str, f: F) -> Result<T>
+    where
+        F: FnOnce(ActorStoreWriter) -> 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!()
+    }
+
+    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)
+
+        // const db: ActorDb = getDb(dbLocation, this.cfg.disableWalAutoCheckpoint)
+        // try {
+        //   await db.ensureWal()
+        //   const migrator = getMigrator(db)
+        //   await migrator.migrateToLatestOrThrow()
+        // } finally {
+        //   db.close()
+        // }
+        todo!()
+    }
+
+    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)),
+        //   )
+        // }
+
+        // const { directory } = await this.getLocation(did)
+        // await rmIfExists(directory, true)
+        todo!()
+    }
+
+    // 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
+    // }
+
+    // async getReservedKeypair(
+    //   signingKeyOrDid: string,
+    // ): Promise<ExportableKeypair | undefined> {
+    //   return loadKey(path.join(this.reservedKeyDir, signingKeyOrDid))
+    // }
+
+    // async clearReservedKeypair(keyDid: string, did?: string) {
+    //   await rmIfExists(path.join(this.reservedKeyDir, keyDid))
+    //   if (did) {
+    //     await rmIfExists(path.join(this.reservedKeyDir, 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)
+    // }
+
+    // async getPlcOp(did: string): Promise<Uint8Array> {
+    //   const { directory } = await this.getLocation(did)
+    //   const opLoc = path.join(directory, `did-op`)
+    //   return await fs.readFile(opLoc)
+    // }
+
+    // async clearPlcOp(did: string) {
+    //   const { directory } = await this.getLocation(did)
+    //   const opLoc = path.join(directory, `did-op`)
+    //   await rmIfExists(opLoc)
+    // }
+}
+
+// const loadKey = async (loc: string): Promise<ExportableKeypair | undefined> => {
+//   const privKey = await readIfExists(loc)
+//   if (!privKey) return undefined
+//   return crypto.Secp256k1Keypair.import(privKey, { exportable: true })
+// }
+
+// function assertSafePathPart(part: string) {
+//   const normalized = path.normalize(part)
+//   assert(
+//     part === normalized &&
+//       !part.startsWith('.') &&
+//       !part.includes('/') &&
+//       !part.includes('\\'),
+//     `unsafe path part: ${part}`,
+//   )
+// }

src/actor_store/actor_store_reader.rs

···
+use std::sync::Arc;
+
+use sqlx::SqlitePool;
+
+use super::{
+    ActorStoreTransactor, db::ActorDb, preference::PreferenceReader, record::RecordReader,
+    repo::RepoReader, resources::ActorStoreResources,
+};
+use crate::SigningKey;
+
+pub(crate) struct ActorStoreReader {
+    pub(crate) repo: RepoReader,
+    pub(crate) record: RecordReader,
+    pub(crate) pref: PreferenceReader,
+}
+
+impl ActorStoreReader {
+    pub(crate) fn new(
+        did: String,
+        db: ActorDb,
+        resources: ActorStoreResources,
+        keypair: impl Fn() -> Result<Arc<SigningKey>>,
+    ) -> Self {
+        let blobstore = resources.blobstore(&did);
+
+        let repo = RepoReader::new(db.clone(), blobstore);
+        let record = RecordReader::new(db.clone());
+        let pref = PreferenceReader::new(db);
+        keypair();
+
+        Self { repo, record, pref }
+    }
+
+    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)
+    }
+}

src/actor_store/actor_store_transactor.rs

···
+use std::sync::Arc;
+
+use sqlx::SqlitePool;
+
+use super::resources::ActorStoreResources;
+use crate::SigningKey;
+
+pub(crate) struct ActorStoreTransactor {
+    pub(crate) did: String,
+    pub(crate) db: SqlitePool,
+    pub(crate) keypair: Arc<SigningKey>,
+    pub(crate) resources: ActorStoreResources,
+}
+
+impl ActorStoreTransactor {
+    pub(crate) fn new(
+        did: String,
+        db: SqlitePool,
+        keypair: Arc<SigningKey>,
+        resources: ActorStoreResources,
+    ) -> Self {
+        Self {
+            did,
+            db,
+            keypair,
+            resources,
+        }
+    }
+}

src/actor_store/actor_store_writer.rs

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

src/actor_store/blob/mod.rs

···
 
 mod reader;
 mod transactor;
+
+pub(crate) use reader::BlobReader;
+pub(crate) use transactor::BlobTransactor;

src/actor_store/blob/reader.rs

···
 //! Blob reading functionality.
 
+use std::str::FromStr;
+
 use anyhow::{Context as _, Result};
+use atrium_api::com::atproto::admin::defs::StatusAttrData;
 use atrium_repo::Cid;
 use sqlx::{Row, SqlitePool};
 
-use crate::config::BlobConfig;
-
 /// Reader for blob data in the actor store.
-pub(super) struct BlobReader {
+pub(crate) struct BlobReader {
     /// Database connection.
     pub db: SqlitePool,
-    /// Configuration for blob storage.
-    pub config: BlobConfig,
-    /// DID of the repository owner.
-    pub did: String,
+    /// BlobStore.
+    pub blobstore: BlobStore,
 }
 
 impl BlobReader {
     /// Create a new blob reader.
-    pub(super) fn new(db: SqlitePool, config: BlobConfig, did: String) -> Self {
-        Self { db, config, did }
+    pub(crate) fn new(db: SqlitePool, blobstore: BlobStore) -> Self {
+        Self { db, blobstore }
     }
 
     /// Get metadata for a blob.
-    pub(super) async fn get_blob_metadata(&self, cid: &Cid) -> Result<Option<BlobMetadata>> {
+    pub(crate) async fn get_blob_metadata(&self, cid: &Cid) -> Result<Option<BlobMetadata>> {
         let cid_str = cid.to_string();
-        let result = sqlx::query!(
-            r#"SELECT size, mimeType, takedownRef FROM blob WHERE cid = ?"#,
+        let found = sqlx::query!(
+            r#"
+            SELECT mimeType, size, takedownRef
+            FROM blob
+            WHERE cid = ? AND takedownRef IS NULL
+            "#,
             cid_str
         )
         .fetch_optional(&self.db)
         .await
         .context("failed to fetch blob metadata")?;
-
-        match result {
-            Some(row) => Ok(Some(BlobMetadata {
-                cid: cid.clone(),
-                size: row.size as u64,
-                mime_type: row.mimeType,
-                takedown_ref: row.takedownRef,
-            })),
-            None => Ok(None),
+        if found.is_none() {
+            return Err(anyhow::anyhow!("Blob not found")); // InvalidRequestError('Blob not found')
         }
+        let found = found.unwrap();
+        let size = found.size as u64;
+        let mime_type = found.mimeType;
+        return Ok(Some(BlobMetadata { size, mime_type }));
     }
 
     /// Get a blob's full data and metadata.
-    pub(super) async fn get_blob(&self, cid: &Cid) -> Result<Option<BlobData>> {
-        // First check the metadata
-        let metadata = match self.get_blob_metadata(cid).await? {
-            Some(meta) => meta,
-            None => return Ok(None),
-        };
-
-        // If there's a takedown, return metadata only with no content
-        if metadata.takedown_ref.is_some() {
-            return Ok(Some(BlobData {
-                metadata,
-                content: None,
-            }));
-        }
-
-        // Get the blob file path
-        let blob_path = self.config.path.join(format!("{}.blob", cid));
-
-        // Check if file exists
-        if !blob_path.exists() {
-            return Ok(None);
+    pub(crate) async fn get_blob(&self, cid: &Cid) -> Result<Option<BlobData>> {
+        let metadata = self.get_blob_metadata(cid).await?;
+        let blob_stream = self.blobstore.get_stream(cid).await?;
+        if blob_stream.is_none() {
+            return Err(anyhow::anyhow!("Blob not found")); // InvalidRequestError('Blob not found')
         }
-
-        // Read the file
-        let content = tokio::fs::read(&blob_path)
-            .await
-            .context("failed to read blob file")?;
-
+        let metadata = metadata.unwrap();
         Ok(Some(BlobData {
-            metadata,
-            content: Some(content),
+            size: metadata.size,
+            mime_type: Some(metadata.mime_type),
+            stream: blob_stream.unwrap(),
         }))
     }
 
     /// List blobs for a repository.
-    pub(super) async fn list_blobs(&self, opts: ListBlobsOptions) -> Result<Vec<String>> {
-        let mut query = sqlx::QueryBuilder::new("SELECT cid FROM blob");
-
-        // Add filters for since revision
+    pub(crate) async fn list_blobs(&self, opts: ListBlobsOptions) -> Result<Vec<String>> {
+        let mut query = sqlx::QueryBuilder::new(
+            "SELECT rb.blobCid FROM record_blob rb
+             INNER JOIN record r ON r.uri = rb.recordUri",
+        );
         if let Some(since) = &opts.since {
-            query
-                .push(" WHERE EXISTS (")
-                .push("SELECT 1 FROM record_blob rb JOIN record r ON rb.recordUri = r.uri")
-                .push(" WHERE rb.blobCid = blob.cid AND r.repoRev > ")
-                .push_bind(since)
-                .push(")");
+            query.push(" WHERE r.repoRev > ").push_bind(since);
         }
-
-        // Add cursor pagination
         if let Some(cursor) = &opts.cursor {
-            if opts.since.is_some() {
-                query.push(" AND");
-            } else {
-                query.push(" WHERE");
-            }
-            query.push(" cid > ").push_bind(cursor);
+            query.push(" AND rb.blobCid > ").push_bind(cursor);
         }
-
-        // Add order and limit
         query
-            .push(" ORDER BY cid ASC")
+            .push(" ORDER BY rb.blobCid ASC")
             .push(" LIMIT ")
             .push_bind(opts.limit);
-
-        // Execute query
         let blobs = query
             .build()
             .map(|row: sqlx::sqlite::SqliteRow| row.get::<String, _>(0))
             .fetch_all(&self.db)
             .await
-            .context("failed to list blobs")?;
-
+            .context("failed to fetch blobs")?;
         Ok(blobs)
     }
 
     /// Get takedown status for a blob.
-    pub(super) async fn get_blob_takedown_status(&self, cid: &Cid) -> Result<Option<String>> {
+    pub(crate) async fn get_blob_takedown_status(
+        &self,
+        cid: &Cid,
+    ) -> Result<Option<StatusAttrData>> {
+        // const res = await this.db.db
+        //   .selectFrom('blob')
+        //   .select('takedownRef')
+        //   .where('cid', '=', cid.toString())
+        //   .executeTakeFirst()
+        // if (!res) return null
+        // return res.takedownRef
+        //   ? { applied: true, ref: res.takedownRef }
+        //   : { applied: false }
         let cid_str = cid.to_string();
         let result = sqlx::query!(r#"SELECT takedownRef FROM blob WHERE cid = ?"#, cid_str)
             .fetch_optional(&self.db)
             .await
             .context("failed to fetch blob takedown status")?;
 
-        Ok(result.and_then(|row| row.takedownRef))
+        Ok({
+            if result.is_none() {
+                None
+            } else {
+                let takedown_ref = result.unwrap().takedownRef.unwrap();
+                if takedown_ref == "false" {
+                    Some(StatusAttrData {
+                        applied: false,
+                        r#ref: None,
+                    })
+                } else {
+                    Some(StatusAttrData {
+                        applied: true,
+                        r#ref: Some(takedown_ref),
+                    })
+                }
+            }
+        })
     }
 
     /// Get records that reference a blob.
-    pub(super) async fn get_records_for_blob(&self, cid: &Cid) -> Result<Vec<String>> {
+    pub(crate) async fn get_records_for_blob(&self, cid: &Cid) -> Result<Vec<String>> {
         let cid_str = cid.to_string();
         let records = sqlx::query!(
             r#"SELECT recordUri FROM record_blob WHERE blobCid = ?"#,
···
     }
 
     /// Get blobs referenced by a record.
-    pub(super) async fn get_blobs_for_record(&self, record_uri: &str) -> Result<Vec<String>> {
+    pub(crate) async fn get_blobs_for_record(&self, record_uri: &str) -> Result<Vec<String>> {
+        // const res = await this.db.db
+        //   .selectFrom('blob')
+        //   .innerJoin('record_blob', 'record_blob.blobCid', 'blob.cid')
+        //   .where('recordUri', '=', recordUri)
+        //   .select('blob.cid')
+        //   .execute()
+        // return res.map((row) => row.cid)
         let blobs = sqlx::query!(
-            r#"SELECT blobCid FROM record_blob WHERE recordUri = ?"#,
+            r#"SELECT blob.cid FROM blob INNER JOIN record_blob ON record_blob.blobCid = blob.cid WHERE recordUri = ?"#,
             record_uri
         )
         .fetch_all(&self.db)
         .await
         .context("failed to fetch blobs for record")?;
 
-        Ok(blobs.into_iter().map(|r| r.blobCid).collect())
+        Ok(blobs.into_iter().map(|blob| blob.cid).collect())
     }
 
     /// Count total blobs.
-    pub(super) async fn blob_count(&self) -> Result<i64> {
+    pub(crate) async fn blob_count(&self) -> Result<i64> {
         let result = sqlx::query!(r#"SELECT COUNT(*) as count FROM blob"#)
             .fetch_one(&self.db)
             .await
···
     }
 
     /// Count distinct blobs referenced by records.
-    pub(super) async fn record_blob_count(&self) -> Result<i64> {
+    pub(crate) async fn record_blob_count(&self) -> Result<i64> {
         let result = sqlx::query!(r#"SELECT COUNT(DISTINCT blobCid) as count FROM record_blob"#)
             .fetch_one(&self.db)
             .await
···
     }
 
     /// List blobs that are referenced but missing from storage.
-    pub(super) async fn list_missing_blobs(
+    pub(crate) async fn list_missing_blobs(
         &self,
         opts: ListMissingBlobsOptions,
     ) -> Result<Vec<MissingBlob>> {
···
         Ok(missing)
     }
 
-    /// Register a new blob in the database (without file storage)
-    pub(super) async fn register_blob(
-        &self,
-        cid: String,
-        mime_type: String,
-        size: i64,
-    ) -> Result<()> {
-        let now = chrono::Utc::now().to_rfc3339();
-        sqlx::query!(
-            r#"
-            INSERT INTO blob (cid, mimeType, size, createdAt)
-            VALUES (?, ?, ?, ?)
-            ON CONFLICT DO NOTHING
-            "#,
-            cid,
-            mime_type,
-            size,
-            now
-        )
-        .execute(&self.db)
-        .await
-        .context("failed to register blob")?;
-
-        Ok(())
+    pub(crate) async fn get_blod_cids(&self) -> Result<Vec<Cid>> {
+        let rows = sqlx::query!("SELECT cid FROM blob")
+            .fetch_all(&self.db)
+            .await
+            .context("failed to fetch blob CIDs")?;
+        Ok(rows
+            .into_iter()
+            .map(|row| Cid::from_str(&row.cid).unwrap())
+            .collect())
     }
 }
 
 /// Metadata about a blob.
 #[derive(Debug, Clone)]
-pub(super) struct BlobMetadata {
-    /// The CID of the blob.
-    pub cid: Cid,
+pub(crate) struct BlobMetadata {
     /// The size of the blob in bytes.
     pub size: u64,
     /// The MIME type of the blob.
     pub mime_type: String,
-    /// Reference for takedown, if any.
-    pub takedown_ref: Option<String>,
 }
 
 /// Complete blob data with content.
 #[derive(Debug)]
-pub(super) struct BlobData {
-    /// Metadata about the blob.
-    pub metadata: BlobMetadata,
-    /// The actual content of the blob, if available.
-    pub content: Option<Vec<u8>>,
+pub(crate) struct BlobData {
+    /// The size of the blob.
+    pub size: u64,
+    /// The MIME type of the blob.
+    pub mime_type: Option<String>,
+    pub stream: BlobStream,
 }
 
 /// Options for listing blobs.
 #[derive(Debug, Clone)]
-pub(super) struct ListBlobsOptions {
+pub(crate) struct ListBlobsOptions {
     /// Optional revision to list blobs since.
     pub since: Option<String>,
     /// Optional cursor for pagination.
···
 
 /// Options for listing missing blobs.
 #[derive(Debug, Clone)]
-pub(super) struct ListMissingBlobsOptions {
+pub(crate) struct ListMissingBlobsOptions {
     /// Optional cursor for pagination.
     pub cursor: Option<String>,
     /// Maximum number of missing blobs to return.
···
 
 /// Information about a missing blob.
 #[derive(Debug, Clone)]
-pub(super) struct MissingBlob {
+pub(crate) struct MissingBlob {
     /// CID of the missing blob.
     pub cid: String,
     /// URI of the record referencing the missing blob.

src/actor_store/blob/transactor.rs

···
 //! Blob transaction functionality.
 
 use anyhow::{Context as _, Result};
-use atrium_api::types::{Blob, CidLink};
+use atrium_api::{
+    com::atproto::admin::defs::StatusAttr,
+    types::{Blob, CidLink},
+};
 use atrium_repo::Cid;
 use sha2::Digest;
 use sqlx::{Row, SqlitePool};
···
 use tokio::fs;
 use uuid::Uuid;
 
-use super::reader::BlobReader;
-use crate::config::BlobConfig;
+use super::BlobReader;
+use crate::repo::types::{PreparedBlobRef, PreparedWrite, WriteOpAction};
 
 /// Blob metadata for a newly uploaded blob.
 #[derive(Debug, Clone)]
-pub(super) struct BlobMetadata {
+pub(crate) struct BlobMetadata {
     /// Temporary key for the blob during upload.
     pub temp_key: String,
     /// Size of the blob in bytes.
···
 }
 
 /// Transactor for blob operations.
-pub(super) struct BlobTransactor {
+pub(crate) struct BlobTransactor {
     /// The blob reader.
     pub reader: BlobReader,
-    /// The blob storage directory.
-    pub blobs_dir: PathBuf,
-    /// Temporary directory for blob uploads.
-    pub temp_dir: PathBuf,
+    pub background_queue: BackgroundQueue,
 }
 
 impl BlobTransactor {
     /// Create a new blob transactor.
-    pub(super) fn new(db: SqlitePool, config: BlobConfig, did: String) -> Self {
+    pub(crate) fn new(
+        db: SqlitePool,
+        blob_store: BlobStore,
+        background_queue: BackgroundQueue,
+    ) -> Self {
         Self {
-            reader: BlobReader::new(db, config.clone(), did),
-            blobs_dir: config.path.clone(),
-            temp_dir: config.path.join("temp"),
+            reader: BlobReader::new(db, blob_store),
+            background_queue,
         }
     }
 
     /// Register blob associations with records.
-    pub(super) async fn insert_blobs(&self, record_uri: &str, blobs: &[Blob]) -> Result<()> {
+    pub(crate) async fn insert_blobs(&self, record_uri: &str, blobs: &[Blob]) -> Result<()> {
         if blobs.is_empty() {
             return Ok(());
         }
···
     }
 
     /// Upload a blob and get its metadata.
-    pub(super) async fn upload_blob_and_get_metadata(
+    pub(crate) async fn upload_blob_and_get_metadata(
         &self,
-        mime_type: &str,
-        data: &[u8],
+        user_suggested_mime: &str,
+        blob_stream: &[u8],
     ) -> Result<BlobMetadata> {
-        // Ensure temp directory exists
-        fs::create_dir_all(&self.temp_dir)
-            .await
-            .context("failed to create temp directory")?;
-
-        // Generate a temporary key
-        let temp_key = format!("temp-{}", Uuid::new_v4());
-        let temp_path = self.temp_dir.join(&temp_key);
-
-        // Write data to temp file
-        fs::write(&temp_path, data)
-            .await
-            .context("failed to write blob to temp file")?;
-
-        // Calculate SHA-256 hash for CID
-        let digest = sha2::Sha256::digest(data);
-
-        // Create CID from hash (using raw codec 0x55)
-        let multihash = atrium_repo::Multihash::wrap(atrium_repo::blockstore::SHA2_256, &digest)
-            .context("failed to create multihash")?;
-        let cid = Cid::new_v1(0x55, multihash);
-
-        // For now, we're not detecting image dimensions
-        let width = None;
-        let height = None;
-
+        let temp_key = self.reader.blobstore.put_temp(blob_stream).await?;
+        let size = stream_size(blob_stream).await?;
+        let sha256 = sha256_stream(blob_stream).await?;
+        let img_info = img::maybe_get_info(blob_stream).await?;
+        let sniffed_mime = mime_type_from_stream(blob_stream).await?;
+        let cid = sha256_raw_to_cid(sha256);
+        let mime_type = sniffed_mime.unwrap_or_else(|| user_suggested_mime.to_string());
         Ok(BlobMetadata {
             temp_key,
-            size: data.len() as u64,
+            size,
             cid,
-            mime_type: mime_type.to_string(),
-            width,
-            height,
+            mime_type,
+            width: img_info.map(|info| info.width),
+            height: img_info.map(|info| info.height),
         })
     }
 
     /// Track a new blob that's not yet associated with a record.
-    pub(super) async fn track_untethered_blob(&self, metadata: &BlobMetadata) -> Result<Blob> {
+    pub(crate) async fn track_untethered_blob(&self, metadata: &BlobMetadata) -> Result<Blob> {
         let cid_str = metadata.cid.to_string();
 
         // Check if blob exists and is taken down
···
     }
 
     /// Process blobs for a repository write operation.
-    pub(super) async fn process_write_blobs(
+    pub(crate) async fn process_write_blobs(&self, writes: Vec<PreparedWrite>) -> Result<()> {
+        self.delete_dereferenced_blobs(writes)
+            .await
+            .context("failed to delete dereferenced blobs")?;
+        for write in writes {
+            if write.action == WriteOpAction::Create || write.action == WriteOpAction::Update {
+                for blob in &write.blobs {
+                    self.verify_blob_and_make_permanent(blob)
+                        .await
+                        .context("failed to verify and make blob permanent")?;
+                    self.associate_blob(&blob.r#ref.0.to_string(), &write.uri)
+                        .await
+                        .context("failed to associate blob with record")?;
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Update the takedown status of a blob.
+    pub(crate) async fn update_blob_takedown_status(
         &self,
-        _rev: &str,
-        blobs: &[Blob],
-        uris: &[String],
+        blob: &Blob,
+        takedown: &StatusAttr,
     ) -> Result<()> {
-        // Handle deleted/updated records
-        self.delete_dereferenced_blobs(uris).await?;
+        let takedown_ref = if takedown.applied {
+            Some(
+                takedown
+                    .r#ref
+                    .clone()
+                    .unwrap_or_else(|| Uuid::new_v4().to_string()),
+            )
+        } else {
+            None
+        };
 
-        // Process each blob
-        for blob in blobs {
-            // Verify and make permanent
-            self.verify_blob_and_make_permanent(blob).await?;
-        }
+        let cid_str = blob.r#ref.0.to_string();
+        sqlx::query!(
+            r#"UPDATE blob SET takedownRef = ? WHERE cid = ?"#,
+            takedown_ref,
+            cid_str
+        )
+        .execute(&self.reader.db)
+        .await
+        .context("failed to update blob takedown status")?;
 
         Ok(())
     }
 
     /// Delete blobs that are no longer referenced by any record.
-    pub(super) async fn delete_dereferenced_blobs(&self, updated_uris: &[String]) -> Result<()> {
-        if updated_uris.is_empty() {
-            return Ok(());
-        }
-
-        // Find blobs that were referenced by the updated URIs
-        let placeholders = (0..updated_uris.len())
-            .map(|_| "?")
-            .collect::<Vec<_>>()
-            .join(",");
-
-        let query = format!(
-            "DELETE FROM record_blob WHERE recordUri IN ({}) RETURNING blobCid",
-            placeholders
-        );
+    pub(crate) async fn delete_dereferenced_blobs(
+        &self,
+        writes: Vec<PreparedWrite>,
+        skip_blob_store: bool,
+    ) -> Result<()> {
+        let deletes = writes
+            .iter()
+            .filter(|w| w.action() == &WriteOpAction::Delete)
+            .collect::<Vec<_>>();
+        let updates = writes
+            .iter()
+            .filter(|w| w.action() == &WriteOpAction::Update)
+            .collect::<Vec<_>>();
+        let uris: Vec<String> = deletes
+            .iter()
+            .chain(updates.iter())
+            .map(|w| w.uri().to_string())
+            .collect();
 
-        let mut query_builder = sqlx::query(&query);
-        for uri in updated_uris {
-            query_builder = query_builder.bind(uri);
+        if uris.is_empty() {
+            return Ok(());
         }
 
-        let deleted_blobs = query_builder
-            .map(|row: sqlx::sqlite::SqliteRow| row.get::<String, _>(0))
-            .fetch_all(&self.reader.db)
-            .await
-            .context("failed to delete dereferenced blobs")?;
+        // Delete blobs from record_blob table
+        let deleted_repo_blobs = sqlx::query!(
+            r#"DELETE FROM record_blob WHERE recordUri IN (?1) RETURNING *"#,
+            uris.join(",")
+        )
+        .fetch_all(&self.reader.db)
+        .await
+        .context("failed to delete dereferenced blobs")?;
 
-        if deleted_blobs.is_empty() {
+        if deleted_repo_blobs.is_empty() {
             return Ok(());
         }
 
-        // Find blobs that are still referenced elsewhere
-        let still_referenced = sqlx::query!(
-            r#"SELECT DISTINCT blobCid FROM record_blob WHERE blobCid IN (SELECT blobCid FROM record_blob)"#
+        // Get the CIDs of the deleted blobs
+        let deleted_repo_blob_cids: Vec<String> = deleted_repo_blobs
+            .iter()
+            .map(|row| row.blobCid.clone())
+            .collect();
+
+        // Check for duplicates in the record_blob table
+        let duplicate_cids = sqlx::query!(
+            r#"SELECT blobCid FROM record_blob WHERE blobCid IN (?1)"#,
+            deleted_repo_blob_cids.join(",")
         )
         .fetch_all(&self.reader.db)
         .await
-        .context("failed to find referenced blobs")?;
+        .context("failed to fetch duplicate CIDs")?;
 
-        let referenced_set: std::collections::HashSet<String> = still_referenced
-            .into_iter()
-            .map(|row| row.blobCid)
+        // Get new blob CIDs from the writes
+        let new_blob_cids: Vec<String> = writes
+            .iter()
+            .filter_map(|w| {
+                if w.action() == &WriteOpAction::Create || w.action() == &WriteOpAction::Update {
+                    todo!()
+                    // Some(w.blobs().iter().map(|b| b.r#ref.0.to_string()).collect())
+                } else {
+                    None
+                }
+            })
+            .flatten()
             .collect();
 
-        // Delete blobs that are no longer referenced anywhere
-        let blobs_to_delete: Vec<String> = deleted_blobs
+        // Determine which CIDs to keep and which to delete
+        let cids_to_keep: Vec<String> = new_blob_cids
             .into_iter()
-            .filter(|cid| !referenced_set.contains(cid))
+            .chain(duplicate_cids.into_iter().map(|row| row.blobCid))
             .collect();
-
-        if !blobs_to_delete.is_empty() {
-            // Delete from database
-            let placeholders = (0..blobs_to_delete.len())
-                .map(|_| "?")
-                .collect::<Vec<_>>()
-                .join(",");
-
-            let query = format!("DELETE FROM blob WHERE cid IN ({})", placeholders);
-
-            let mut query_builder = sqlx::query(&query);
-            for cid in &blobs_to_delete {
-                query_builder = query_builder.bind(cid);
-            }
-
-            query_builder
-                .execute(&self.reader.db)
-                .await
-                .context("failed to delete blob records")?;
-
-            // Delete files from disk
-            for cid in blobs_to_delete {
-                let path = self.blobs_dir.join(format!("{}.blob", cid));
-                if let Err(e) = fs::remove_file(&path).await {
-                    tracing::warn!("Failed to delete blob file: {:?}", e);
-                }
-            }
+        let cids_to_delete: Vec<String> = deleted_repo_blob_cids
+            .into_iter()
+            .filter(|cid| !cids_to_keep.contains(cid))
+            .collect();
+        if cids_to_delete.is_empty() {
+            return Ok(());
+        }
+        // Delete blobs from the blob table
+        sqlx::query!(
+            r#"DELETE FROM blob WHERE cid IN (?1)"#,
+            cids_to_delete.join(",")
+        )
+        .execute(&self.reader.db)
+        .await
+        .context("failed to delete dereferenced blobs from blob table")?;
+        // Optionally delete blobs from the blob store
+        if !skip_blob_store {
+            todo!();
         }
-
         Ok(())
     }
 
     /// Verify a blob's integrity and move it from temporary to permanent storage.
-    pub(super) async fn verify_blob_and_make_permanent(&self, blob: &Blob) -> Result<()> {
-        let cid_str = blob.r#ref.0.to_string();
-
-        // Get blob from database
-        let row = sqlx::query!(
-            r#"SELECT * FROM blob WHERE cid = ? AND takedownRef IS NULL"#,
-            cid_str
-        )
-        .fetch_optional(&self.reader.db)
-        .await
-        .context("failed to find blob")?;
-
-        let row = match row {
-            Some(r) => r,
-            None => return Err(anyhow::anyhow!("Could not find blob: {}", cid_str)),
-        };
-
-        // Verify size constraint
-        if (row.size as u64) > self.reader.config.limit {
-            return Err(anyhow::anyhow!(
-                "Blob is too large. Size is {} bytes but maximum allowed is {} bytes",
-                row.size,
-                self.reader.config.limit
-            ));
+    pub(crate) async fn verify_blob_and_make_permanent(
+        &self,
+        blob: &PreparedBlobRef,
+    ) -> Result<()> {
+        let cid_str = blob.cid.to_string();
+        let found = sqlx::query!(r#"SELECT * FROM blob WHERE cid = ?"#, cid_str)
+            .fetch_optional(&self.reader.db)
+            .await
+            .context("failed to fetch blob")?;
+        if found.is_none() {
+            return Err(anyhow::anyhow!("Blob not found"));
         }
-
-        // Check MIME type
-        if row.mimeType != blob.mime_type {
-            return Err(anyhow::anyhow!(
-                "MIME type does not match. Expected: {}, got: {}",
-                row.mimeType,
-                blob.mime_type
-            ));
+        let found = found.unwrap();
+        if found.takedownRef.is_some() {
+            return Err(anyhow::anyhow!("Blob has been taken down"));
         }
-
-        // If temp key exists, move to permanent storage
-        if let Some(temp_key) = &row.tempKey {
-            let temp_path = self.temp_dir.join(temp_key);
-            let blob_path = self.blobs_dir.join(format!("{}.blob", cid_str));
-
-            // Only move if temp file exists and permanent file doesn't
-            if fs::try_exists(&temp_path).await.unwrap_or(false)
-                && !fs::try_exists(&blob_path).await.unwrap_or(false)
-            {
-                // Create parent directories if needed
-                if let Some(parent) = blob_path.parent() {
-                    fs::create_dir_all(parent)
-                        .await
-                        .context("failed to create blob directory")?;
-                }
-
-                // Move file from temp to permanent location
-                fs::copy(&temp_path, &blob_path)
-                    .await
-                    .context("failed to copy blob to permanent storage")?;
-                fs::remove_file(&temp_path)
-                    .await
-                    .context("failed to remove temporary blob file")?;
-            }
-
-            // Update database
-            sqlx::query!(r#"UPDATE blob SET tempKey = NULL WHERE cid = ?"#, cid_str)
-                .execute(&self.reader.db)
-                .await
-                .context("failed to update blob record after making permanent")?;
+        if found.tempKey.is_some() {
+            todo!("verify_blob"); // verify_blob(blob, found);
+            self.reader
+                .blobstore
+                .make_permanent(found.tempKey.unwrap(), cid_str)
+                .await?;
+            sqlx::query!(
+                r#"UPDATE blob SET tempKey = NULL WHERE tempKey = ?"#,
+                found.tempKey
+            )
+            .execute(&self.reader.db)
+            .await
+            .context("failed to update blob temp key")?;
         }
-
         Ok(())
     }
 
-    /// Register a blob in the database
-    pub(super) async fn register_blob(
+    /// Associate a blob with a record
+    pub(crate) async fn associate_blob(
         &self,
-        cid: String,
-        mime_type: String,
-        size: u64,
+        blob: &PreparedBlobRef,
+        record_uri: &str,
     ) -> Result<()> {
-        self.reader.register_blob(cid, mime_type, size as i64).await
-    }
-
-    /// Associate a blob with a record
-    pub(super) async fn associate_blob(&self, cid: &str, record_uri: &str) -> Result<()> {
+        let cid = blob.cid.to_string();
         sqlx::query!(
             r#"
             INSERT INTO record_blob (blobCid, recordUri)

src/actor_store/mod.rs

···
 //! Actor store implementation for ATProto PDS.
 
+mod actor_store;
+mod actor_store_reader;
+mod actor_store_transactor;
+mod actor_store_writer;
 mod blob;
 mod db;
 mod preference;
 mod record;
 mod repo;
-
-use std::path::PathBuf;
-use std::sync::Arc;
-
-use anyhow::{Context as _, Result};
-use atrium_crypto::keypair::Export as _;
-use sqlx::SqlitePool;
-
-use crate::SigningKey;
-use crate::config::RepoConfig;
-
-/// Resources required by the actor store.
-pub(crate) struct ActorStoreResources {
-    /// Configuration for the repo.
-    pub(crate) config: RepoConfig,
-    /// Configuration for the blob store.
-    pub(crate) blob_config: crate::config::BlobConfig,
-    /// Background task queue (we'll need to implement this later).
-    pub(crate) background_queue: Arc<()>, // TODO: Placeholder until we implement a proper queue
-}
-
-/// The location of an actor's data.
-pub(crate) struct ActorLocation {
-    /// The directory for the actor's data.
-    pub(crate) directory: PathBuf,
-    /// The database location for the actor.
-    pub(crate) db_location: PathBuf,
-    /// The keypair location for the actor.
-    pub(crate) key_location: PathBuf,
-}
-
-/// The actor store for repository data.
-pub(crate) struct ActorStore {
-    /// The directory for actor data.
-    pub(crate) directory: PathBuf,
-    /// The directory for reserved keys.
-    reserved_key_dir: PathBuf,
-    /// Resources used by the actor store.
-    resources: ActorStoreResources,
-}
-
-/// Reader for actor data.
-pub(crate) struct ActorStoreReader {
-    /// The DID of the actor.
-    pub(crate) did: String,
-    /// The database connection.
-    pub(crate) db: SqlitePool,
-    /// The actor's keypair.
-    keypair: Arc<SigningKey>,
-    /// Resources for the actor store.
-    pub(crate) resources: Arc<ActorStoreResources>,
-    /// Repository reader
-    pub(crate) repo: repo::RepoReader,
-    /// Record reader
-    pub(crate) record: record::RecordReader,
-    /// Preference reader
-    pub(crate) pref: preference::PreferenceReader,
-}
+mod resources;
 
-/// Writer for actor data with transaction support.
-pub(crate) struct ActorStoreWriter {
-    /// The DID of the actor.
-    pub(crate) did: String,
-    /// The database connection.
-    pub(crate) db: SqlitePool,
-    /// The actor's keypair.
-    keypair: Arc<SigningKey>,
-    /// Resources for the actor store.
-    pub(crate) resources: Arc<ActorStoreResources>,
-    /// Repository access
-    pub(crate) repo: repo::RepoTransactor,
-    /// Record access
-    pub(crate) record: record::RecordTransactor,
-    /// Preference access
-    pub(crate) pref: preference::PreferenceTransactor,
-}
-
-/// Transactor for actor data.
-pub(crate) struct ActorStoreTransactor {
-    /// The DID of the actor.
-    pub(crate) did: String,
-    /// The database connection.
-    pub(crate) db: SqlitePool,
-    /// The actor's keypair.
-    keypair: Arc<SigningKey>,
-    /// Resources for the actor store.
-    pub(crate) resources: Arc<ActorStoreResources>,
-    /// Repository access
-    pub(crate) repo: repo::RepoTransactor,
-    /// Record access
-    pub(crate) record: record::RecordTransactor,
-    /// Preference access
-    pub(crate) pref: preference::PreferenceTransactor,
-}
-
-impl ActorStore {
-    /// Create a new actor store.
-    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,
-            resources,
-        }
-    }
-
-    /// Load an actor store based on a given DID.
-    pub(crate) async fn load(did: &str, resources: ActorStoreResources) -> Result<Self> {
-        let did_hash = sha256_hex(did).await?;
-        let directory = resources.config.path.join(&did_hash[0..2]).join(did);
-        let reserved_key_dir = directory.join("reserved_keys");
-
-        Ok(Self {
-            directory,
-            reserved_key_dir,
-            resources,
-        })
-    }
-
-    /// Get the location of a DID's data.
-    pub(crate) async fn get_location(&self, did: &str) -> Result<ActorLocation> {
-        let did_hash = sha256_hex(did).await?;
-        let directory = self.directory.join(&did_hash[0..2]).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 exists.
-    pub(crate) async fn exists(&self, did: &str) -> Result<bool> {
-        let location = self.get_location(did).await?;
-        Ok(tokio::fs::try_exists(&location.db_location).await?)
-    }
-
-    /// Get the keypair for an actor.
-    pub(crate) async fn keypair(&self, did: &str) -> Result<Arc<SigningKey>> {
-        let location = self.get_location(did).await?;
-        let priv_key = tokio::fs::read(&location.key_location).await?;
-        let keypair = SigningKey::import(&priv_key)?;
-        Ok(Arc::new(keypair))
-    }
-
-    /// Open the database for an actor.
-    pub(crate) async fn open_db(&self, did: &str) -> Result<SqlitePool> {
-        let location = self.get_location(did).await?;
-
-        if !tokio::fs::try_exists(&location.db_location).await? {
-            return Err(anyhow::anyhow!("Repo not found"));
-        }
-
-        let db = sqlx::sqlite::SqlitePoolOptions::new()
-            .max_connections(5)
-            .connect_with(
-                sqlx::sqlite::SqliteConnectOptions::new()
-                    .filename(&location.db_location)
-                    .create_if_missing(false),
-            )
-            .await
-            .context("failed to connect to SQLite database")?;
-
-        Ok(db)
-    }
-
-    /// Read from an actor store.
-    pub(crate) async fn read<T, F>(&self, did: &str, f: F) -> Result<T>
-    where
-        F: FnOnce(ActorStoreReader) -> Result<T>,
-    {
-        let db = self.open_db(did).await?;
-        let keypair = self.keypair(did).await?;
-        let resources = Arc::new(self.resources.clone());
-        let did_str = did.to_string();
-
-        let reader = ActorStoreReader {
-            did: did_str.clone(),
-            repo: repo::RepoReader::new(
-                db.clone(),
-                did_str.clone(),
-                self.resources.blob_config.clone(),
-            ),
-            record: record::RecordReader::new(db.clone(), did_str.clone()),
-            pref: preference::PreferenceReader::new(db.clone(), did_str),
-            db,
-            keypair,
-            resources,
-        };
-
-        f(reader)
-    }
-
-    /// Transact against an actor store with full transaction support.
-    pub(crate) async fn transact<T, F>(&self, did: &str, f: F) -> Result<T>
-    where
-        F: FnOnce(ActorStoreTransactor) -> Result<T>,
-    {
-        let keypair = self.keypair(did).await?;
-        let db = self.open_db(did).await?;
-        let resources = Arc::new(self.resources.clone());
-        let did_str = did.to_string();
-
-        let transactor = ActorStoreTransactor {
-            did: did_str.clone(),
-            repo: repo::RepoTransactor::new(
-                db.clone(),
-                did_str.clone(),
-                (*keypair).clone(),
-                self.resources.blob_config.clone(),
-            ),
-            record: record::RecordTransactor::new(
-                db.clone(),
-                db.clone(), // Using db as placeholder for blobstore
-                did_str.clone(),
-            ),
-            pref: preference::PreferenceTransactor::new(db.clone(), did_str),
-            db,
-            keypair,
-            resources,
-        };
-
-        f(transactor)
-    }
-
-    /// Write to an actor store without transaction support.
-    pub(crate) async fn write_no_transaction<T, F>(&self, did: &str, f: F) -> Result<T>
-    where
-        F: FnOnce(ActorStoreWriter) -> Result<T>,
-    {
-        let db = self.open_db(did).await?;
-        let keypair = self.keypair(did).await?;
-        let resources = Arc::new(self.resources.clone());
-        let did_str = did.to_string();
-
-        let writer = ActorStoreWriter {
-            did: did_str.clone(),
-            repo: repo::RepoTransactor::new(
-                db.clone(),
-                did_str.clone(),
-                (*keypair).clone(),
-                self.resources.blob_config.clone(),
-            ),
-            record: record::RecordTransactor::new(
-                db.clone(),
-                db.clone(), // Using db as placeholder for blobstore
-                did_str.clone(),
-            ),
-            pref: preference::PreferenceTransactor::new(db.clone(), did_str),
-            db,
-            keypair,
-            resources,
-        };
-
-        f(writer)
-    }
-
-    /// Create a new actor repository.
-    pub(crate) async fn create(&self, did: &str, keypair: SigningKey) -> Result<()> {
-        let location = self.get_location(did).await?;
-
-        // Create directory if it doesn't exist
-        tokio::fs::create_dir_all(&location.directory).await?;
-
-        // Check if repo already exists
-        if tokio::fs::try_exists(&location.db_location).await? {
-            return Err(anyhow::anyhow!("Repo already exists"));
-        }
-
-        // Export and save keypair
-        let priv_key = keypair.export();
-        tokio::fs::write(&location.key_location, priv_key).await?;
-
-        // Create database
-        let db = sqlx::sqlite::SqlitePoolOptions::new()
-            .connect_with(
-                sqlx::sqlite::SqliteConnectOptions::new()
-                    .filename(&location.db_location)
-                    .create_if_missing(true),
-            )
-            .await
-            .context("failed to create SQLite database")?;
-
-        // Create database schema
-        db::create_tables(&db).await?;
-
-        Ok(())
-    }
-
-    /// Destroy an actor's repository and associated data.
-    pub(crate) async fn destroy(&self, did: &str) -> Result<()> {
-        // TODO: Implement repository destruction
-        // - Delete all blobs for the repository
-        // - Remove the repository directory
-        todo!("Implement repository destruction")
-    }
-
-    /// Reserve a keypair for a DID.
-    pub(crate) async fn reserve_keypair(&self, did: Option<&str>) -> Result<String> {
-        // TODO: Implement keypair reservation
-        // - Generate a keypair if one doesn't exist
-        // - Store the keypair in the reserved_key_dir
-        // - Return the DID of the keypair
-        todo!("Implement keypair reservation")
-    }
-
-    /// Get a reserved keypair.
-    pub(crate) async fn get_reserved_keypair(
-        &self,
-        signing_key_or_did: &str,
-    ) -> Result<Option<()>> {
-        // TODO: Implement getting a reserved keypair
-        // - Load the keypair from the reserved_key_dir
-        todo!("Implement getting a reserved keypair")
-    }
-
-    /// Clear a reserved keypair.
-    pub(crate) async fn clear_reserved_keypair(
-        &self,
-        key_did: &str,
-        did: Option<&str>,
-    ) -> Result<()> {
-        // TODO: Implement clearing a reserved keypair
-        // - Remove the keypair file from the reserved_key_dir
-        todo!("Implement clearing a reserved keypair")
-    }
-
-    /// Store a PLC operation.
-    pub(crate) async fn store_plc_op(&self, did: &str, op: &[u8]) -> Result<()> {
-        // TODO: Implement storing a PLC operation
-        // - Store the operation in the actor's directory
-        todo!("Implement storing a PLC operation")
-    }
-
-    /// Get a stored PLC operation.
-    pub(crate) async fn get_plc_op(&self, did: &str) -> Result<Vec<u8>> {
-        // TODO: Implement getting a PLC operation
-        // - Retrieve the operation from the actor's directory
-        todo!("Implement getting a PLC operation")
-    }
-
-    /// Clear a stored PLC operation.
-    pub(crate) async fn clear_plc_op(&self, did: &str) -> Result<()> {
-        // TODO: Implement clearing a PLC operation
-        // - Remove the operation file from the actor's directory
-        todo!("Implement clearing a PLC operation")
-    }
-}
-
-impl ActorStoreWriter {
-    /// Transact with the writer.
-    pub(crate) async fn transact<T, F>(&self, f: F) -> Result<T>
-    where
-        F: FnOnce(ActorStoreTransactor) -> Result<T>,
-    {
-        todo!("Implement transact method for ActorStoreWriter")
-    }
-}
-
-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(),
-        }
-    }
-}
-
-// Helper function for SHA-256 hashing
-async fn sha256_hex(input: &str) -> Result<String> {
-    use sha2::{Digest, Sha256};
-    let mut hasher = Sha256::new();
-    hasher.update(input.as_bytes());
-    let result = hasher.finalize();
-    Ok(hex::encode(result))
-}
+pub(crate) use actor_store::ActorStore;
+pub(crate) use actor_store_reader::ActorStoreReader;
+pub(crate) use actor_store_transactor::ActorStoreTransactor;
+pub(crate) use actor_store_writer::ActorStoreWriter;
+pub(crate) use resources::ActorStoreResources;

src/actor_store/repo/reader.rs

···
 use sqlx::SqlitePool;
 
 use super::sql_repo_reader::SqlRepoReader;
-use crate::{config::BlobConfig, repo::block_map::BlockMap};
+use crate::{actor_store::record::RecordReader, config::BlobConfig, repo::block_map::BlockMap};
 
 /// Reader for repository data in the actor store.
 pub(crate) struct RepoReader {
+    blob: BlobReader,
+    record: RecordReader,
     /// The SQL repository reader.
-    pub storage: SqlRepoReader,
-    /// The database connection.
-    pub db: SqlitePool,
-    /// The DID of the repository owner.
-    pub did: String,
+    storage: SqlRepoReader,
 }
 
 impl RepoReader {
     /// Create a new repository reader.
-    pub(crate) fn new(db: SqlitePool, did: String, blob_config: BlobConfig) -> Self {
-        Self {
-            storage: SqlRepoReader::new(db.clone(), did.clone()),
-            db,
-            did,
-        }
-    }
+    // pub(crate) fn new(db: SqlitePool, did: String, blob_config: BlobConfig) -> Self {
+    //     Self {
+    //         storage: SqlRepoReader::new(db.clone(), did.clone()),
+    //         db,
+    //         did,
+    //     }
+    // }
 
     /// Get event data for synchronization.
     pub(crate) async fn get_sync_event_data(&self) -> Result<SyncEventData> {

src/actor_store/repo/sql_repo_transactor.rs

···
 use sha2::Digest;
 use sqlx::SqlitePool;
 
-use crate::repo::block_map::{BlockMap, CommitData};
+use crate::repo::{block_map::BlockMap, types::CommitData};
 
-use super::sql_repo_reader::SqlRepoReader;
+use super::sql_repo_reader::{RootInfo, SqlRepoReader};
 
 /// SQL-based repository transactor that extends the reader.
 pub(crate) struct SqlRepoTransactor {
···
             cache: BlockMap::new(),
             now,
         }
+    }
+
+    /// Get the root CID and revision of the repository.
+    pub(crate) async fn get_root_detailed(&self) -> Result<RootInfo> {
+        let row = sqlx::query!(
+            r#"
+                SELECT cid, rev
+                FROM repo_root
+                WHERE did = ?
+                LIMIT 1
+            "#,
+            self.reader.did
+        )
+        .fetch_one(&self.reader.db)
+        .await?;
+
+        let cid = Cid::from_str(&row.cid)?;
+        Ok(RootInfo { cid, rev: row.rev })
     }
 
     /// Proactively cache all blocks from a particular commit.

src/actor_store/repo/transactor.rs

···
 //! Repository transactor for the actor store.
 
+use std::str::FromStr;
+
 use anyhow::Result;
 use atrium_repo::Cid;
-use rsky_repo::types::{CommitDataWithOps, PreparedWrite};
 use rsky_syntax::aturi::AtUri;
 use sqlx::SqlitePool;
 
-use crate::SigningKey;
+use crate::{
+    SigningKey,
+    repo::types::{CommitAction, CommitDataWithOps, CommitOp, PreparedWrite, WriteOpAction},
+};
 
 use super::{reader::RepoReader, sql_repo_transactor::SqlRepoTransactor};
 
 /// Transactor for repository operations.
 pub(crate) struct RepoTransactor {
-    /// The repository reader.
-    pub reader: RepoReader,
-    /// The SQL repository transactor.
-    pub storage: SqlRepoTransactor,
-    /// The DID of the repository owner.
-    pub did: String,
-    /// The signing key for the repository.
-    pub signing_key: SigningKey,
+    /// The inner reader.
+    pub(crate) reader: RepoReader,
+    ///
 }
+
 
 impl RepoTransactor {
     /// Create a new repository transactor.
···
         writes: Vec<PreparedWrite>,
         swap_commit_cid: Option<Cid>,
     ) -> Result<CommitDataWithOps> {
-        todo!("Implement process_writes")
+        // Validate parameters
+        if writes.len() > 200 {
+            return Err(anyhow::anyhow!("Too many writes. Max: 200").into());
+        }
+
+        // Format the commit (creates blocks and structures the operations)
+        let commit = self.format_commit(writes.clone(), 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());
+        }
+
+        // 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),
+        )?;
+
+        Ok(commit)
     }
 
     /// Format a commit for writing.
···
         writes: Vec<PreparedWrite>,
         swap_commit: Option<Cid>,
     ) -> Result<CommitDataWithOps> {
-        todo!("Implement format_commit")
+        // Get current repository 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());
+        }
+
+        // Check swap commit if provided
+        if let Some(swap_cid) = swap_commit {
+            if !curr_root.cid.equals(swap_cid) {
+                return Err(anyhow::anyhow!(
+                    "Bad commit swap: expected {}, got {}",
+                    swap_cid,
+                    curr_root.cid
+                )
+                .into());
+            }
+        }
+
+        // Cache last commit for performance
+        self.storage.cache_rev(&curr_root.rev).await?;
+
+        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())?;
+
+            match write.action() {
+                WriteOpAction::Create | WriteOpAction::Update => {
+                    if let Some(cid) = write.cid() {
+                        new_record_cids.push(cid);
+                    }
+                }
+                _ => {}
+            }
+
+            if write.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();
+
+            // Create the operation
+            let path = format!("{}/{}", uri.get_collection(), uri.get_rkey());
+            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(),
+                },
+                prev: curr_record.clone(),
+            };
+
+            // Validate swap consistency
+            if let Some(swap_cid) = write.swap_cid() {
+                match write.action() {
+                    WriteOpAction::Create if swap_cid.is_some() => {
+                        return Err(anyhow::anyhow!(
+                            "Bad record swap: cannot provide swap CID for create"
+                        )
+                        .into());
+                    }
+                    WriteOpAction::Update | WriteOpAction::Delete if swap_cid.is_none() => {
+                        return Err(anyhow::anyhow!(
+                            "Bad record swap: must provide swap CID for update/delete"
+                        )
+                        .into());
+                    }
+                    _ => {}
+                }
+
+                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());
+                }
+            }
+
+            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;
+
+        // Convert PreparedWrites to RecordWriteOps
+        let write_ops = writes
+            .iter()
+            .map(|w| write_to_op(w.clone()))
+            .collect::<Result<Vec<_>>>()?;
+
+        // Format the new commit
+        let commit = repo.format_commit(write_ops, &self.signing_key).await?;
+
+        // Find blocks referenced by other records
+        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);
+        }
+
+        // Ensure all necessary blocks are included
+        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?;
+            commit.relevant_blocks.add_map(missing_blocks.blocks)?;
+        }
+
+        Ok(CommitDataWithOps {
+            commit_data: commit,
+            ops: commit_ops,
+            prev_data: Some(prev_data),
+        })
     }
 
     /// Index writes in the repository.
     pub(crate) async fn index_writes(&self, writes: Vec<PreparedWrite>, rev: &str) -> Result<()> {
-        todo!("Implement index_writes")
+        // Process each write for indexing
+        for write in writes {
+            let uri_str = write.uri().clone();
+            let uri = AtUri::try_from(uri_str.as_str())?;
+
+            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?;
+                    }
+                }
+                WriteOpAction::Delete => {
+                    self.record.delete_record(&uri).await?;
+                }
+            }
+        }
+
+        Ok(())
     }
 
     /// Get duplicate record CIDs.
···
         cids: &[Cid],
         touched_uris: &[AtUri],
     ) -> Result<Vec<Cid>> {
-        todo!("Implement get_duplicate_record_cids")
+        if touched_uris.is_empty() || cids.is_empty() {
+            return Ok(Vec::new());
+        }
+
+        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
+        )
+        .fetch_all(&self.reader.db)
+        .await?;
+
+        // Convert string CIDs back to Cid objects
+        let dupe_cids = duplicates
+            .into_iter()
+            .filter_map(|cid_str| Cid::from_str(&cid_str).ok())
+            .collect();
+
+        Ok(dupe_cids)
+    }
+
+    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?;
+        }
+
+        // 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?;
+                }
+            }
+        }
+
+        Ok(())
     }
 }

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/block_map.rs

···
 use std::collections::{BTreeMap, HashSet};
 use std::str::FromStr;
 
-/// Ref: https://github.com/blacksky-algorithms/rsky/blob/main/rsky-repo/src/types.rs#L341C1-L350C2
-#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
-pub(crate) struct CommitData {
-    pub cid: Cid,
-    pub rev: String,
-    pub since: Option<String>,
-    pub prev: Option<Cid>,
-    pub new_blocks: BlockMap,
-    pub relevant_blocks: BlockMap,
-    pub removed_cids: CidSet,
-}
-
-#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
-pub enum CommitAction {
-    Create,
-    Update,
-    Delete,
-}
-
-#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
-pub struct CommitOp {
-    pub action: CommitAction,
-    pub path: String,
-    pub cid: Option<Cid>,
-    pub prev: Option<Cid>,
-}
-
-#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
-pub struct CommitDataWithOps {
-    #[serde(flatten)]
-    pub commit_data: CommitData,
-    pub ops: Vec<CommitOp>,
-    pub prev_data: Option<Cid>,
-}
-
 /// Ref: https://github.com/blacksky-algorithms/rsky/blob/main/rsky-repo/src/cid_set.rs
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
 pub(crate) struct CidSet {

src/repo/types.rs

···
             PreparedWrite::Delete(w) => &w.action,
         }
     }
+
+    /// TEQ: Add blobs() impl
+    pub fn blobs(&self) -> Option<&Vec<PreparedBlobRef>> {
+        match self {
+            PreparedWrite::Create(w) => Some(&w.blobs),
+            PreparedWrite::Update(w) => Some(&w.blobs),
+            PreparedWrite::Delete(_) => None,
+        }
+    }
 }
 
 impl From<&PreparedWrite> for CommitAction {