experiment with actor_store from reference impl

.sqlx/query-0c112e544ad7f92f3699ee79d2bd674ff72041c4dd6050c00a1fa3eac38553e6.json

···
+{
+  "db_name": "SQLite",
+  "query": "UPDATE repo_root SET cid = ?, rev = ?, indexedAt = ? WHERE did = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 4
+    },
+    "nullable": []
+  },
+  "hash": "0c112e544ad7f92f3699ee79d2bd674ff72041c4dd6050c00a1fa3eac38553e6"
+}

.sqlx/query-1d9681bf4929b3f6dcfbb55fd5e15fb2593615b6e796c7d63731b78b0c43738c.json

···
+{
+  "db_name": "SQLite",
+  "query": "SELECT COUNT(*) FROM repo_block WHERE cid = ?",
+  "describe": {
+    "columns": [
+      {
+        "name": "COUNT(*)",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "1d9681bf4929b3f6dcfbb55fd5e15fb2593615b6e796c7d63731b78b0c43738c"
+}

.sqlx/query-1db52857493a1e8a7004872eaff6e8fe5dec41579dd57d696008385b8d23788d.json

···
-{
-  "db_name": "SQLite",
-  "query": "SELECT data FROM blocks WHERE cid = ?",
-  "describe": {
-    "columns": [
-      {
-        "name": "data",
-        "ordinal": 0,
-        "type_info": "Blob"
-      }
-    ],
-    "parameters": {
-      "Right": 1
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "1db52857493a1e8a7004872eaff6e8fe5dec41579dd57d696008385b8d23788d"
-}

.sqlx/query-2918ecf03675a789568c777904966911ca63e991dede42a2d7d87e174799ea46.json

···
-{
-  "db_name": "SQLite",
-  "query": "INSERT INTO blocks (cid, data, multicodec, multihash) VALUES (?, ?, ?, ?)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Right": 4
-    },
-    "nullable": []
-  },
-  "hash": "2918ecf03675a789568c777904966911ca63e991dede42a2d7d87e174799ea46"
-}

.sqlx/query-3b4745208f268678a84401e522c3836e0632ca34a0f23bbae5297d076610f0ab.json

···
+{
+  "db_name": "SQLite",
+  "query": "SELECT content FROM repo_block WHERE cid = ?",
+  "describe": {
+    "columns": [
+      {
+        "name": "content",
+        "ordinal": 0,
+        "type_info": "Blob"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "3b4745208f268678a84401e522c3836e0632ca34a0f23bbae5297d076610f0ab"
+}

.sqlx/query-73fd3e30b7694c92cf9309751d186fe622fa7d99fdf56dde7e60c3696581116c.json .sqlx/query-1d91d3f785f8f3e86baa2af1e643364b242bb6e8dbc0d50b86602022c0000ed4.json

···
 {
   "db_name": "SQLite",
-  "query": "SELECT COUNT(*) FROM blocks WHERE cid = ?",
+  "query": "SELECT COUNT(*) FROM repo_root WHERE did = ?",
   "describe": {
     "columns": [
       {
···
       false
     ]
   },
-  "hash": "73fd3e30b7694c92cf9309751d186fe622fa7d99fdf56dde7e60c3696581116c"
+  "hash": "1d91d3f785f8f3e86baa2af1e643364b242bb6e8dbc0d50b86602022c0000ed4"
 }

.sqlx/query-aeeee4d0a717e2207e5e810db35b21728f6473795b025a25050cec30a4f5ba55.json

···
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO repo_block (cid, repoRev, size, content) VALUES (?, ?, ?, ?)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 4
+    },
+    "nullable": []
+  },
+  "hash": "aeeee4d0a717e2207e5e810db35b21728f6473795b025a25050cec30a4f5ba55"
+}

.sqlx/query-fcd6cb9fdec21590db0645925b4ceeb9e457bd52807f1e60b829bbf304106037.json

···
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO repo_root (did, cid, rev, indexedAt) VALUES (?, ?, ?, ?)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 4
+    },
+    "nullable": []
+  },
+  "hash": "fcd6cb9fdec21590db0645925b4ceeb9e457bd52807f1e60b829bbf304106037"
+}

Cargo.lock

···
  "figment",
  "futures",
  "http-cache-reqwest",
+ "k256",
  "memmap2",
  "metrics",
  "metrics-exporter-prometheus",
···
  "cfg-if",
  "ecdsa",
  "elliptic-curve",
+ "once_cell",
  "sha2",
+ "signature",
 ]
 
 [[package]]

Cargo.toml

···
 # pattern_type_mismatch = "allow"
 # Style Allows
 implicit_return = "allow"
-# self_named_module_files = "allow" # Choose one of self_named_module_files or mod_module_files
+self_named_module_files = "allow"           # Choose one of self_named_module_files or mod_module_files
 mod_module_files = "allow"
 else_if_without_else = "allow"
 std_instead_of_alloc = "allow"
···
 uuid = { version = "1.14.0", features = ["v4"] }
 urlencoding = "2.1.3"
 async-trait = "0.1.88"
+k256 = "0.13.4"

migrations/20250507035800_sqlite_blockstore.down.sql

···
-DROP INDEX IF EXISTS idx_tree_nodes_repo;
-DROP INDEX IF EXISTS idx_blocks_cid;
-DROP TABLE IF EXISTS tree_nodes;
-DROP TABLE IF EXISTS blocks;

migrations/20250507035800_sqlite_blockstore.up.sql

···
--- Store raw blocks with their CIDs
-CREATE TABLE IF NOT EXISTS blocks (
-    cid TEXT PRIMARY KEY NOT NULL,
-    data BLOB NOT NULL,
-    multicodec INTEGER NOT NULL,
-    multihash INTEGER NOT NULL
-);
-
--- Store the repository tree structure
-CREATE TABLE IF NOT EXISTS tree_nodes (
-    repo_did TEXT NOT NULL,
-    key TEXT NOT NULL,
-    value_cid TEXT NOT NULL,
-    PRIMARY KEY (repo_did, key),
-    FOREIGN KEY (value_cid) REFERENCES blocks(cid)
-);
-
--- Create index for faster lookups
-CREATE INDEX IF NOT EXISTS idx_blocks_cid ON blocks(cid);
-CREATE INDEX IF NOT EXISTS idx_tree_nodes_repo ON tree_nodes(repo_did);

migrations/20250508251242_actor_store.down.sql

···
+-- Drop indexes
+DROP INDEX IF EXISTS idx_backlink_link_to;
+DROP INDEX IF EXISTS idx_blob_tempkey;
+DROP INDEX IF EXISTS idx_record_repo_rev;
+DROP INDEX IF EXISTS idx_record_collection;
+DROP INDEX IF EXISTS idx_record_cid;
+DROP INDEX IF EXISTS idx_repo_block_repo_rev;
+
+-- Drop tables
+DROP TABLE IF EXISTS account_pref;
+DROP TABLE IF EXISTS backlink;
+DROP TABLE IF EXISTS record_blob;
+DROP TABLE IF EXISTS blob;
+DROP TABLE IF EXISTS record;
+DROP TABLE IF EXISTS repo_block;
+DROP TABLE IF EXISTS repo_root;

migrations/20250508251242_actor_store.up.sql

···
+-- Actor store schema matching TypeScript implementation
+
+-- Repository root information
+CREATE TABLE IF NOT EXISTS repo_root (
+    did TEXT PRIMARY KEY NOT NULL,
+    cid TEXT NOT NULL,
+    rev TEXT NOT NULL,
+    indexedAt TEXT NOT NULL
+);
+
+-- Repository blocks (IPLD blocks)
+CREATE TABLE IF NOT EXISTS repo_block (
+    cid TEXT PRIMARY KEY NOT NULL,
+    repoRev TEXT NOT NULL,
+    size INTEGER NOT NULL,
+    content BLOB NOT NULL
+);
+
+-- Record index
+CREATE TABLE IF NOT EXISTS record (
+    uri TEXT PRIMARY KEY NOT NULL,
+    cid TEXT NOT NULL,
+    collection TEXT NOT NULL,
+    rkey TEXT NOT NULL,
+    repoRev TEXT NOT NULL,
+    indexedAt TEXT NOT NULL,
+    takedownRef TEXT
+);
+
+-- Blob storage metadata
+CREATE TABLE IF NOT EXISTS blob (
+    cid TEXT PRIMARY KEY NOT NULL,
+    mimeType TEXT NOT NULL,
+    size INTEGER NOT NULL,
+    tempKey TEXT,
+    width INTEGER,
+    height INTEGER,
+    createdAt TEXT NOT NULL,
+    takedownRef TEXT
+);
+
+-- Record-blob associations
+CREATE TABLE IF NOT EXISTS record_blob (
+    blobCid TEXT NOT NULL,
+    recordUri TEXT NOT NULL,
+    PRIMARY KEY (blobCid, recordUri)
+);
+
+-- Backlinks between records
+CREATE TABLE IF NOT EXISTS backlink (
+    uri TEXT NOT NULL,
+    path TEXT NOT NULL,
+    linkTo TEXT NOT NULL,
+    PRIMARY KEY (uri, path)
+);
+
+-- User preferences
+CREATE TABLE IF NOT EXISTS account_pref (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    name TEXT NOT NULL,
+    valueJson TEXT NOT NULL
+);
+
+-- Create indexes
+CREATE INDEX IF NOT EXISTS idx_repo_block_repo_rev ON repo_block(repoRev, cid);
+CREATE INDEX IF NOT EXISTS idx_record_cid ON record(cid);
+CREATE INDEX IF NOT EXISTS idx_record_collection ON record(collection);
+CREATE INDEX IF NOT EXISTS idx_record_repo_rev ON record(repoRev);
+CREATE INDEX IF NOT EXISTS idx_blob_tempkey ON blob(tempKey);
+CREATE INDEX IF NOT EXISTS idx_backlink_link_to ON backlink(path, linkTo);

src/actor_store/blob.rs

···
+// src/actor_store/blob.rs
+use anyhow::Result;
+use atrium_api::types::string::Cid;
+use sqlx::Row;
+
+use crate::actor_store::db::ActorDb;
+
+#[derive(Clone)]
+pub struct BlobReader {
+    db: ActorDb,
+}
+
+impl BlobReader {
+    pub fn new(db: ActorDb) -> Self {
+        Self { db }
+    }
+
+    pub async fn get_blob_metadata(&self, cid: &Cid) -> Result<Option<BlobMetadata>> {
+        let cid_str = format!("{:?}", cid);
+        let row = sqlx::query(
+            "SELECT mime_type, size, width, height
+             FROM blob
+             WHERE cid = ? AND takedown_ref IS NULL",
+        )
+        .bind(&cid_str)
+        .fetch_optional(&self.db.pool)
+        .await?;
+
+        match row {
+            Some(row) => Ok(Some(BlobMetadata {
+                size: row.get::<i64, _>("size") as u64,
+                mime_type: row.get("mime_type"),
+            })),
+            None => Ok(None),
+        }
+    }
+
+    pub async fn get_blobs_for_record(&self, record_uri: &str) -> Result<Vec<String>> {
+        let rows = sqlx::query(
+            "SELECT blob.cid
+             FROM blob
+             INNER JOIN record_blob ON record_blob.blob_cid = blob.cid
+             WHERE record_uri = ?",
+        )
+        .bind(record_uri)
+        .fetch_all(&self.db.pool)
+        .await?;
+
+        Ok(rows.into_iter().map(|row| row.get("cid")).collect())
+    }
+}
+
+#[derive(Clone)]
+pub struct BlobTransactor {
+    db: ActorDb,
+}
+
+impl BlobTransactor {
+    pub fn new(db: ActorDb) -> Self {
+        Self { db }
+    }
+
+    pub async fn associate_blob(&self, blob_cid: &str, record_uri: &str) -> Result<()> {
+        sqlx::query(
+            "INSERT INTO record_blob (blob_cid, record_uri)
+             VALUES (?, ?)
+             ON CONFLICT DO NOTHING",
+        )
+        .bind(blob_cid)
+        .bind(record_uri)
+        .execute(&self.db.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    pub async fn register_blob(&self, cid: &str, mime_type: &str, size: u64) -> Result<()> {
+        let now = chrono::Utc::now().to_rfc3339();
+
+        sqlx::query(
+            "INSERT INTO blob (cid, mime_type, size, created_at)
+             VALUES (?, ?, ?, ?)
+             ON CONFLICT DO NOTHING",
+        )
+        .bind(cid)
+        .bind(mime_type)
+        .bind(size as i64)
+        .bind(now)
+        .execute(&self.db.pool)
+        .await?;
+
+        Ok(())
+    }
+}
+
+pub struct BlobMetadata {
+    pub size: u64,
+    pub mime_type: String,
+}

src/actor_store/db/migrations.rs

···
+// src/actor_store/db/migrations.rs
+use anyhow::{Context as _, Result};
+use sqlx::{Pool, Sqlite};
+
+pub async fn run(pool: &Pool<Sqlite>) -> Result<()> {
+    sqlx::query(
+        "
+        CREATE TABLE IF NOT EXISTS repo_root (
+            did TEXT PRIMARY KEY NOT NULL,
+            cid TEXT NOT NULL,
+            rev TEXT NOT NULL,
+            indexed_at TEXT NOT NULL
+        );
+
+        CREATE TABLE IF NOT EXISTS repo_block (
+            cid TEXT PRIMARY KEY NOT NULL,
+            repo_rev TEXT NOT NULL,
+            size INTEGER NOT NULL,
+            content BLOB NOT NULL
+        );
+
+        CREATE TABLE IF NOT EXISTS record (
+            uri TEXT PRIMARY KEY NOT NULL,
+            cid TEXT NOT NULL,
+            collection TEXT NOT NULL,
+            rkey TEXT NOT NULL,
+            repo_rev TEXT NOT NULL,
+            indexed_at TEXT NOT NULL,
+            takedown_ref TEXT
+        );
+
+        CREATE TABLE IF NOT EXISTS blob (
+            cid TEXT PRIMARY KEY NOT NULL,
+            mime_type TEXT NOT NULL,
+            size INTEGER NOT NULL,
+            temp_key TEXT,
+            width INTEGER,
+            height INTEGER,
+            created_at TEXT NOT NULL,
+            takedown_ref TEXT
+        );
+
+        CREATE TABLE IF NOT EXISTS record_blob (
+            blob_cid TEXT NOT NULL,
+            record_uri TEXT NOT NULL,
+            PRIMARY KEY (blob_cid, record_uri)
+        );
+
+        CREATE TABLE IF NOT EXISTS backlink (
+            uri TEXT NOT NULL,
+            path TEXT NOT NULL,
+            link_to TEXT NOT NULL,
+            PRIMARY KEY (uri, path)
+        );
+
+        CREATE TABLE IF NOT EXISTS account_pref (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            name TEXT NOT NULL,
+            value_json TEXT NOT NULL
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_record_cid ON record(cid);
+        CREATE INDEX IF NOT EXISTS idx_record_collection ON record(collection);
+        CREATE INDEX IF NOT EXISTS idx_record_repo_rev ON record(repo_rev);
+        CREATE INDEX IF NOT EXISTS idx_blob_temp_key ON blob(temp_key);
+        CREATE INDEX IF NOT EXISTS idx_repo_block_repo_rev ON repo_block(repo_rev, cid);
+        CREATE INDEX IF NOT EXISTS idx_backlink_link_to ON backlink(path, link_to);
+        ",
+    )
+    .execute(pool)
+    .await
+    .context("Failed to run migrations")?;
+
+    Ok(())
+}

src/actor_store/db/mod.rs

···
+// src/actor_store/db/mod.rs
+pub mod migrations;
+
+use anyhow::Result;
+use sqlx::{Pool, Sqlite, Transaction};
+
+#[derive(Clone)]
+pub struct ActorDb {
+    pub pool: Pool<Sqlite>,
+}
+
+impl ActorDb {
+    pub fn new(pool: Pool<Sqlite>) -> Self {
+        Self { pool }
+    }
+
+    pub async fn migrate(&self) -> Result<()> {
+        migrations::run(&self.pool).await
+    }
+
+    pub async fn transaction<F, R>(&self, f: F) -> Result<R>
+    where
+        F: for<'txn> FnOnce(&'txn mut Transaction<'_, Sqlite>) -> Result<R> + Send,
+        R: Send + 'static,
+    {
+        self.pool.begin().await?.commit().await?;
+        let mut tx = self.pool.begin().await?;
+        let result = f(&mut tx)?;
+        tx.commit().await?;
+        Ok(result)
+    }
+
+    pub fn assert_transaction(&self) {
+        // In a real implementation, we might want to check if we're in a transaction,
+        // but SQLite allows nested transactions so it's not strictly necessary
+    }
+}

src/actor_store/mod.rs

···
+// src/actor_store/mod.rs
+pub mod blob;
+pub mod db;
+pub mod preference;
+pub mod record;
+pub mod repo;
+
+use anyhow::{Context as _, Result};
+use atrium_crypto::keypair::{Export as _, Keypair};
+use k256::Secp256k1;
+use sqlx::sqlite::SqliteConnectOptions;
+use std::path::PathBuf;
+use std::str::FromStr;
+use std::sync::Arc;
+
+use crate::config::AppConfig;
+use blob::{BlobReader, BlobTransactor};
+use db::ActorDb;
+use preference::{PreferenceReader, PreferenceTransactor};
+use record::{RecordReader, RecordTransactor};
+use repo::{RepoReader, RepoTransactor};
+
+pub struct ActorStore {
+    pub config: Arc<AppConfig>,
+}
+
+impl ActorStore {
+    pub fn new(config: Arc<AppConfig>) -> Self {
+        Self { config }
+    }
+
+    pub async fn exists(&self, did: &str) -> Result<bool> {
+        let path = self.get_db_path(did)?;
+        Ok(tokio::fs::try_exists(&path).await.unwrap_or(false))
+    }
+
+    pub async fn keypair(&self, did: &str) -> Result<Keypair<Secp256k1>> {
+        let path = self.get_keypair_path(did)?;
+        let key_data = tokio::fs::read(&path)
+            .await
+            .context("failed to read keypair")?;
+        Ok(Keypair::<Secp256k1>::import(&key_data)?)
+    }
+
+    async fn get_db(&self, did: &str) -> Result<ActorDb> {
+        let path = self.get_db_path(did)?;
+        let options = SqliteConnectOptions::from_str(&format!("sqlite:{}", path.display()))?
+            .create_if_missing(true);
+        let pool = sqlx::SqlitePool::connect_with(options).await?;
+
+        Ok(ActorDb::new(pool))
+    }
+
+    pub async fn read<F, T>(&self, did: &str, f: F) -> Result<T>
+    where
+        F: FnOnce(ActorStoreReader) -> Result<T>,
+    {
+        let db = self.get_db(did).await?;
+        let keypair = Arc::new(self.keypair(did).await?);
+        let reader = ActorStoreReader::new(did.to_string(), db, self.config.clone(), keypair);
+        f(reader)
+    }
+
+    pub async fn transact<F, T>(&self, did: &str, f: F) -> Result<T>
+    where
+        F: FnOnce(ActorStoreTransactor) -> Result<T> + Send + 'static,
+        T: Send + 'static,
+    {
+        let db = self.get_db(did).await?;
+        let keypair = Arc::new(self.keypair(did).await?);
+
+        db.transaction(|_| {
+            let transactor = ActorStoreTransactor::new(
+                did.to_string(),
+                db.clone(),
+                self.config.clone(),
+                keypair.clone(),
+            );
+            f(transactor)
+        })
+        .await
+    }
+
+    pub async fn create(&self, did: &str, keypair: &Keypair<Secp256k1>) -> Result<()> {
+        // Create directory structure
+        let db_path = self.get_db_path(did)?;
+        if let Some(parent) = db_path.parent() {
+            tokio::fs::create_dir_all(parent).await?;
+        }
+
+        // Export keypair
+        let key_path = self.get_keypair_path(did)?;
+        let key_data = keypair.export();
+        tokio::fs::write(key_path, key_data).await?;
+
+        // Create and initialize DB
+        let db = self.get_db(did).await?;
+        db.migrate().await?;
+
+        Ok(())
+    }
+
+    fn get_db_path(&self, did: &str) -> Result<PathBuf> {
+        let did_hash = did
+            .strip_prefix("did:plc:")
+            .context("DID must be a PLC identifier")?;
+        Ok(self.config.repo.path.join(format!("{}.db", did_hash)))
+    }
+
+    fn get_keypair_path(&self, did: &str) -> Result<PathBuf> {
+        let did_hash = did
+            .strip_prefix("did:plc:")
+            .context("DID must be a PLC identifier")?;
+        Ok(self.config.repo.path.join(format!("{}.key", did_hash)))
+    }
+}
+
+pub struct ActorStoreReader {
+    pub did: String,
+    db: ActorDb,
+    config: Arc<AppConfig>,
+    keypair: Arc<Keypair<Secp256k1>>,
+}
+
+impl ActorStoreReader {
+    fn new(
+        did: String,
+        db: ActorDb,
+        config: Arc<AppConfig>,
+        keypair: Arc<Keypair<Secp256k1>>,
+    ) -> Self {
+        Self {
+            did,
+            db,
+            config,
+            keypair,
+        }
+    }
+
+    pub fn blob(&self) -> BlobReader {
+        BlobReader::new(self.db.clone())
+    }
+
+    pub fn record(&self) -> RecordReader {
+        RecordReader::new(self.db.clone())
+    }
+
+    pub fn repo(&self) -> RepoReader {
+        RepoReader::new(self.db.clone(), self.did.clone())
+    }
+
+    pub fn pref(&self) -> PreferenceReader {
+        PreferenceReader::new(self.db.clone())
+    }
+}
+
+pub struct ActorStoreTransactor {
+    pub did: String,
+    db: ActorDb,
+    config: Arc<AppConfig>,
+    keypair: Arc<Keypair<Secp256k1>>,
+}
+
+impl ActorStoreTransactor {
+    fn new(
+        did: String,
+        db: ActorDb,
+        config: Arc<AppConfig>,
+        keypair: Arc<Keypair<Secp256k1>>,
+    ) -> Self {
+        Self {
+            did,
+            db,
+            config,
+            keypair,
+        }
+    }
+
+    pub fn blob(&self) -> BlobTransactor {
+        BlobTransactor::new(self.db.clone())
+    }
+
+    pub fn record(&self) -> RecordTransactor {
+        RecordTransactor::new(self.db.clone())
+    }
+
+    pub fn repo(&self) -> RepoTransactor {
+        RepoTransactor::new(self.db.clone(), self.keypair.clone(), self.did.clone())
+    }
+
+    pub fn pref(&self) -> PreferenceTransactor {
+        PreferenceTransactor::new(self.db.clone())
+    }
+}

src/actor_store/preference.rs

···
+// src/actor_store/preference.rs
+use anyhow::Result;
+use serde_json::Value;
+use sqlx::Row;
+
+use crate::actor_store::db::ActorDb;
+
+#[derive(Clone)]
+pub struct PreferenceReader {
+    db: ActorDb,
+}
+
+impl PreferenceReader {
+    pub fn new(db: ActorDb) -> Self {
+        Self { db }
+    }
+
+    pub async fn get_preferences(&self, namespace: Option<&str>) -> Result<Vec<Value>> {
+        let rows = match namespace {
+            Some(ns) => {
+                let pattern = format!("{}%", ns);
+                sqlx::query(
+                    "SELECT name, value_json FROM account_pref
+                     WHERE name = ? OR name LIKE ?",
+                )
+                .bind(ns)
+                .bind(&pattern)
+                .fetch_all(&self.db.pool)
+                .await?
+            }
+            None => {
+                sqlx::query("SELECT name, value_json FROM account_pref")
+                    .fetch_all(&self.db.pool)
+                    .await?
+            }
+        };
+
+        let mut prefs = Vec::with_capacity(rows.len());
+        for row in rows {
+            let name: String = row.get("name");
+            let value_json: String = row.get("value_json");
+            if let Ok(value) = serde_json::from_str(&value_json) {
+                let mut obj: Value = value;
+                if let Value::Object(ref mut map) = obj {
+                    map.insert("$type".to_string(), Value::String(name));
+                }
+                prefs.push(obj);
+            }
+        }
+
+        Ok(prefs)
+    }
+}
+
+#[derive(Clone)]
+pub struct PreferenceTransactor {
+    db: ActorDb,
+}
+
+impl PreferenceTransactor {
+    pub fn new(db: ActorDb) -> Self {
+        Self { db }
+    }
+
+    pub async fn put_preferences(&self, values: Vec<Value>, namespace: &str) -> Result<()> {
+        self.db.assert_transaction();
+
+        // Delete existing preferences in namespace
+        let pattern = format!("{}%", namespace);
+        sqlx::query("DELETE FROM account_pref WHERE name = ? OR name LIKE ?")
+            .bind(namespace)
+            .bind(&pattern)
+            .execute(&self.db.pool)
+            .await?;
+
+        // Insert new preferences
+        for value in values {
+            if let Some(type_val) = value.get("$type") {
+                if let Some(name) = type_val.as_str() {
+                    // Check namespace
+                    if name == namespace || name.starts_with(&format!("{}.", namespace)) {
+                        let mut pref_value = value.clone();
+                        if let Value::Object(ref mut map) = pref_value {
+                            map.remove("$type");
+                        }
+
+                        let value_json = serde_json::to_string(&pref_value)?;
+
+                        sqlx::query("INSERT INTO account_pref (name, value_json) VALUES (?, ?)")
+                            .bind(name)
+                            .bind(&value_json)
+                            .execute(&self.db.pool)
+                            .await?;
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+}

src/actor_store/record.rs

···
+// src/actor_store/record.rs
+use anyhow::Result;
+use sqlx::Row;
+
+use crate::actor_store::db::ActorDb;
+
+#[derive(Clone)]
+pub struct RecordReader {
+    db: ActorDb,
+}
+
+impl RecordReader {
+    pub fn new(db: ActorDb) -> Self {
+        Self { db }
+    }
+
+    pub async fn get_record(&self, uri: &str) -> Result<Option<RecordData>> {
+        let row = sqlx::query(
+            "SELECT record.*, repo_block.content
+             FROM record
+             INNER JOIN repo_block ON repo_block.cid = record.cid
+             WHERE record.uri = ? AND record.takedown_ref IS NULL",
+        )
+        .bind(uri)
+        .fetch_optional(&self.db.pool)
+        .await?;
+
+        match row {
+            Some(row) => Ok(Some(RecordData {
+                uri: row.get("uri"),
+                cid: row.get("cid"),
+                collection: row.get("collection"),
+                rkey: row.get("rkey"),
+                content: row.get("content"),
+            })),
+            None => Ok(None),
+        }
+    }
+
+    pub async fn list_collections(&self) -> Result<Vec<String>> {
+        let rows = sqlx::query("SELECT DISTINCT collection FROM record")
+            .fetch_all(&self.db.pool)
+            .await?;
+
+        Ok(rows.into_iter().map(|row| row.get("collection")).collect())
+    }
+}
+
+#[derive(Clone)]
+pub struct RecordTransactor {
+    db: ActorDb,
+}
+
+impl RecordTransactor {
+    pub fn new(db: ActorDb) -> Self {
+        Self { db }
+    }
+
+    pub async fn index_record(
+        &self,
+        uri: &str,
+        cid: &str,
+        collection: &str,
+        rkey: &str,
+        repo_rev: &str,
+    ) -> Result<()> {
+        let now = chrono::Utc::now().to_rfc3339();
+
+        sqlx::query(
+            "INSERT INTO record (uri, cid, collection, rkey, repo_rev, indexed_at)
+             VALUES (?, ?, ?, ?, ?, ?)
+             ON CONFLICT (uri) DO UPDATE SET
+                cid = ?,
+                repo_rev = ?,
+                indexed_at = ?",
+        )
+        .bind(uri)
+        .bind(cid)
+        .bind(collection)
+        .bind(rkey)
+        .bind(repo_rev)
+        .bind(&now)
+        .bind(cid)
+        .bind(repo_rev)
+        .bind(&now)
+        .execute(&self.db.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    pub async fn delete_record(&self, uri: &str) -> Result<()> {
+        sqlx::query("DELETE FROM record WHERE uri = ?")
+            .bind(uri)
+            .execute(&self.db.pool)
+            .await?;
+
+        sqlx::query("DELETE FROM backlink WHERE uri = ?")
+            .bind(uri)
+            .execute(&self.db.pool)
+            .await?;
+
+        Ok(())
+    }
+}
+
+pub struct RecordData {
+    pub uri: String,
+    pub cid: String,
+    pub collection: String,
+    pub rkey: String,
+    pub content: Vec<u8>,
+}

src/actor_store/repo.rs

···
+// src/actor_store/repo.rs
+use anyhow::Result;
+use atrium_crypto::keypair::Keypair;
+use k256::Secp256k1;
+use sqlx::Row;
+use std::sync::Arc;
+
+use crate::actor_store::db::ActorDb;
+
+#[derive(Clone)]
+pub struct RepoReader {
+    db: ActorDb,
+    did: String,
+}
+
+impl RepoReader {
+    pub fn new(db: ActorDb, did: String) -> Self {
+        Self { db, did }
+    }
+
+    pub async fn get_root(&self) -> Result<Option<RepoRoot>> {
+        let row = sqlx::query("SELECT cid, rev, indexed_at FROM repo_root WHERE did = ?")
+            .bind(&self.did)
+            .fetch_optional(&self.db.pool)
+            .await?;
+
+        match row {
+            Some(row) => Ok(Some(RepoRoot {
+                cid: row.get("cid"),
+                rev: row.get("rev"),
+                indexed_at: row.get("indexed_at"),
+            })),
+            None => Ok(None),
+        }
+    }
+
+    pub async fn get_block(&self, cid: &str) -> Result<Option<Vec<u8>>> {
+        let row = sqlx::query("SELECT content FROM repo_block WHERE cid = ?")
+            .bind(cid)
+            .fetch_optional(&self.db.pool)
+            .await?;
+
+        Ok(row.map(|r| r.get("content")))
+    }
+}
+
+#[derive(Clone)]
+pub struct RepoTransactor {
+    db: ActorDb,
+    keypair: Arc<Keypair<Secp256k1>>,
+    did: String,
+}
+
+impl RepoTransactor {
+    pub fn new(db: ActorDb, keypair: Arc<Keypair<Secp256k1>>, did: String) -> Self {
+        Self { db, keypair, did }
+    }
+
+    pub async fn update_root(&self, cid: &str, rev: &str) -> Result<()> {
+        let now = chrono::Utc::now().to_rfc3339();
+
+        sqlx::query(
+            "INSERT INTO repo_root (did, cid, rev, indexed_at)
+             VALUES (?, ?, ?, ?)
+             ON CONFLICT (did) DO UPDATE SET
+                cid = ?,
+                rev = ?,
+                indexed_at = ?",
+        )
+        .bind(&self.did)
+        .bind(cid)
+        .bind(rev)
+        .bind(&now)
+        .bind(cid)
+        .bind(rev)
+        .bind(&now)
+        .execute(&self.db.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    pub async fn put_block(&self, cid: &str, content: &[u8], repo_rev: &str) -> Result<()> {
+        sqlx::query(
+            "INSERT INTO repo_block (cid, repo_rev, size, content)
+             VALUES (?, ?, ?, ?)
+             ON CONFLICT (cid) DO NOTHING",
+        )
+        .bind(cid)
+        .bind(repo_rev)
+        .bind(content.len() as i64)
+        .bind(content)
+        .execute(&self.db.pool)
+        .await?;
+
+        Ok(())
+    }
+}
+
+pub struct RepoRoot {
+    pub cid: String,
+    pub rev: String,
+    pub indexed_at: String,
+}

src/endpoints/repo.rs

···
 //! PDS repository endpoints /xrpc/com.atproto.repo.*)
-use std::{collections::HashSet, str::FromStr as _};
+use std::{collections::HashSet, str::FromStr as _, sync::Arc};
 
 use anyhow::{Context as _, anyhow};
 use atrium_api::{
···
         string::{AtIdentifier, Nsid, Tid},
     },
 };
-use atrium_repo::{Cid, blockstore::CarStore};
+use atrium_repo::Cid;
 use axum::{
     Json, Router,
     body::Body,
-    extract::{Query, Request, State},
+    extract::{Query, State, multipart::Field},
     http::{self, StatusCode},
     routing::{get, post},
 };
 use constcat::concat;
 use futures::TryStreamExt as _;
+use k256::Secp256k1;
 use metrics::counter;
 use serde::Deserialize;
 use sha2::{Digest as _, Sha256};
 use tokio::io::AsyncWriteExt as _;
 
 use crate::{
-    AppState, Db, Error, Result, SigningKey,
+    AppState, Db, Error, Result,
+    actor_store::{ActorStore, ActorStoreTransactor},
     auth::AuthenticatedUser,
     config::AppConfig,
     error::ErrorMessage,
     firehose::{self, FirehoseProducer, RepoOp},
     metrics::{REPO_COMMITS, REPO_OP_CREATE, REPO_OP_DELETE, REPO_OP_UPDATE},
-    storage,
 };
 
-/// IPLD CID raw binary
-const IPLD_RAW: u64 = 0x55;
-/// SHA2-256 mulithash
-const IPLD_MH_SHA2_256: u64 = 0x12;
-
-/// Used in [`scan_blobs`] to identify a blob.
-#[derive(Deserialize, Debug, Clone)]
-struct BlobRef {
-    /// `BlobRef` link. Include `$` when serializing to JSON, since `$` isn't allowed in struct names.
-    #[serde(rename = "$link")]
-    link: String,
-}
-
+/// Parameters for [`list_records`].
 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
-/// Parameters for [`list_records`].
 pub(super) struct ListRecordsParameters {
-    ///The NSID of the record type.
     pub collection: Nsid,
-    /// The cursor to start from.
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
     pub cursor: Option<String>,
-    ///The number of records to return.
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
     pub limit: Option<String>,
-    ///The handle or DID of the repo.
     pub repo: AtIdentifier,
-    ///Flag to reverse the order of the returned records.
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
     pub reverse: Option<bool>,
-    ///DEPRECATED: The highest sort-ordered rkey to stop at (exclusive)
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
     pub rkey_end: Option<String>,
-    ///DEPRECATED: The lowest sort-ordered rkey to start from (exclusive)
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
     pub rkey_start: Option<String>,
 }
 
-async fn swap_commit(
-    db: impl sqlx::Executor<'_, Database = sqlx::Sqlite>,
-    cid: Cid,
-    rev: Tid,
-    old_cid: Option<Cid>,
-    did_str: &str,
-) -> anyhow::Result<bool> {
-    let cid_str = cid.to_string();
-    let rev_str = rev.to_string();
-
-    if let Some(swap) = old_cid {
-        let swap_str = swap.to_string();
-
-        let result = sqlx::query!(
-            r#"UPDATE accounts SET root = ?, rev = ? WHERE did = ? AND root = ?"#,
-            cid_str,
-            rev_str,
-            did_str,
-            swap_str,
-        )
-        .execute(db)
-        .await
-        .context("failed to update root")?;
-
-        // If the swap failed, indicate as such.
-        Ok(result.rows_affected() != 0)
-    } else {
-        _ = sqlx::query!(
-            r#"UPDATE accounts SET root = ?, rev = ? WHERE did = ?"#,
-            cid_str,
-            rev_str,
-            did_str,
-        )
-        .execute(db)
-        .await
-        .context("failed to update root")?;
-
-        Ok(true)
-    }
-}
-
-/// Resolves DID to DID document. Does not bi-directionally verify handle.
-/// - GET /xrpc/com.atproto.repo.resolveDid
-/// ### Query Parameters
-/// - `did`: DID to resolve.
-/// ### Responses
-/// - 200 OK: {`did_doc`: `did_doc`}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `DidNotFound`, `DidDeactivated`]}
+/// Resolves DID from an identifier
 async fn resolve_did(
     db: &Db,
     identifier: &AtIdentifier,
···
 )> {
     let (handle, did) = match *identifier {
         AtIdentifier::Handle(ref handle) => {
-            let handle_as_str = &handle.as_str();
-            (
-                &handle.to_owned(),
-                &atrium_api::types::string::Did::new(
-                    sqlx::query_scalar!(
-                        r#"SELECT did FROM handles WHERE handle = ?"#,
-                        handle_as_str
-                    )
+            let handle_str = handle.as_str();
+            let did_str =
+                sqlx::query_scalar!(r#"SELECT did FROM handles WHERE handle = ?"#, handle_str)
                     .fetch_one(db)
                     .await
-                    .context("failed to query did")?,
-                )
-                .expect("should be valid DID"),
+                    .context("failed to query did")?;
+
+            (
+                handle.clone(),
+                atrium_api::types::string::Did::new(did_str).expect("should be valid DID format"),
             )
         }
         AtIdentifier::Did(ref did) => {
-            let did_as_str = &did.as_str();
+            let did_str = did.as_str();
+            let handle_str =
+                sqlx::query_scalar!(r#"SELECT handle FROM handles WHERE did = ?"#, did_str)
+                    .fetch_one(db)
+                    .await
+                    .context("failed to query handle")?;
+
             (
-                &atrium_api::types::string::Handle::new(
-                    sqlx::query_scalar!(r#"SELECT handle FROM handles WHERE did = ?"#, did_as_str)
-                        .fetch_one(db)
-                        .await
-                        .context("failed to query did")?,
-                )
-                .expect("should be valid handle"),
-                &did.to_owned(),
+                atrium_api::types::string::Handle::new(handle_str).expect("should be valid handle"),
+                did.clone(),
             )
         }
     };
 
-    Ok((did.to_owned(), handle.to_owned()))
+    Ok((did, handle))
 }
 
-/// Used in [`apply_writes`] to scan for blobs in the JSON object and return their CIDs.
-fn scan_blobs(unknown: &Unknown) -> anyhow::Result<Vec<Cid>> {
-    // { "$type": "blob", "ref": { "$link": "bafyrei..." } }
+/// Extract blobs from a record
+fn extract_blobs(record: &Unknown) -> anyhow::Result<Vec<Cid>> {
+    let val = serde_json::Value::try_from_unknown(record.clone())?;
+    let mut cids = Vec::new();
+    let mut stack = vec![val];
 
-    let mut cids = Vec::new();
-    let mut stack = vec![
-        serde_json::Value::try_from_unknown(unknown.clone())
-            .context("failed to convert unknown into json")?,
-    ];
     while let Some(value) = stack.pop() {
         match value {
-            serde_json::Value::Bool(_)
-            | serde_json::Value::Null
-            | serde_json::Value::Number(_)
-            | serde_json::Value::String(_) => (),
-            serde_json::Value::Array(values) => stack.extend(values.into_iter()),
             serde_json::Value::Object(map) => {
-                if let (Some(blob_type), Some(blob_ref)) = (map.get("$type"), map.get("ref")) {
-                    if blob_type == &serde_json::Value::String("blob".to_owned()) {
-                        if let Ok(rf) = serde_json::from_value::<BlobRef>(blob_ref.clone()) {
-                            cids.push(Cid::from_str(&rf.link).context("failed to convert cid")?);
+                if let (Some(typ), Some(blob_ref)) = (map.get("$type"), map.get("ref")) {
+                    if typ == "blob" {
+                        if let Some(link) = blob_ref.get("$link").and_then(|v| v.as_str()) {
+                            cids.push(Cid::from_str(link)?);
                         }
                     }
                 }
-
                 stack.extend(map.values().cloned());
             }
+            serde_json::Value::Array(arr) => {
+                stack.extend(arr);
+            }
+            _ => {}
         }
     }
 
     Ok(cids)
 }
 
-#[test]
-fn test_scan_blobs() {
-    use std::str::FromStr as _;
-
-    let json = serde_json::json!({
-        "test": "a",
-        "blob": {
-            "$type": "blob",
-            "ref": {
-                "$link": "bafkreifzxf2wa6dyakzbdaxkz2wkvfrv3hiuafhxewbn5wahcw6eh3hzji"
-            }
-        }
-    });
-
-    let blob = scan_blobs(&json.try_into_unknown().expect("should be valid JSON"))
-        .expect("should be able to scan blobs");
-    assert_eq!(
-        blob,
-        vec![
-            Cid::from_str("bafkreifzxf2wa6dyakzbdaxkz2wkvfrv3hiuafhxewbn5wahcw6eh3hzji")
-                .expect("should be valid CID")
-        ]
-    );
-}
-
-#[expect(clippy::too_many_lines)]
-/// Apply a batch transaction of repository creates, updates, and deletes. Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.applyWrites
-/// ### Request Body
-/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
-/// - `validate`: `boolean` // Can be set to 'false' to skip Lexicon schema validation of record data across all operations, 'true' to require it, or leave unset to validate only for known Lexicons.
-/// - `writes`: `object[]` // One of:
-/// - - com.atproto.repo.applyWrites.create
-/// - - com.atproto.repo.applyWrites.update
-/// - - com.atproto.repo.applyWrites.delete
-/// - `swap_commit`: `cid` // If provided, the entire operation will fail if the current repo commit CID does not match this value. Used to prevent conflicting repo mutations.
+/// Apply writes to a repository
 async fn apply_writes(
     user: AuthenticatedUser,
-    State(skey): State<SigningKey>,
     State(config): State<AppConfig>,
     State(db): State<Db>,
     State(fhp): State<FirehoseProducer>,
···
 ) -> Result<Json<repo::apply_writes::Output>> {
     use atrium_api::com::atproto::repo::apply_writes::{self, InputWritesItem, OutputResultsItem};
 
-    // TODO: `input.validate`
-
-    let (target_did, _) = resolve_did(&db, &input.repo)
-        .await
-        .context("failed to resolve did")?;
-
-    // Ensure that we are updating the correct repository.
+    // Resolve target DID
+    let (target_did, _) = resolve_did(&db, &input.repo).await?;
     if target_did.as_str() != user.did() {
         return Err(Error::with_status(
             StatusCode::BAD_REQUEST,
···
         ));
     }
 
-    let mut repo = storage::open_repo_db(&config.repo, &db, user.did())
-        .await
-        .context("failed to open user repo")?;
-    let orig_cid = repo.root();
-    let orig_rev = repo.commit().rev();
+    let actor_store = ActorStore::new(Arc::new(config));
+
+    // Convert input writes to internal format
+    let result = actor_store
+        .transact(user.did(), |txn| {
+            let mut results = Vec::new();
+            let mut blob_info = Vec::new();
+            let mut ops = Vec::new();
 
-    let mut blobs = vec![];
-    let mut res = vec![];
-    let mut ops = vec![];
-    let mut keys = vec![];
-    for write in &input.writes {
-        let (builder, key) = match *write {
-            InputWritesItem::Create(ref object) => {
-                let now = Tid::now(LimitedU32::MIN);
-                let key = format!(
-                    "{}/{}",
-                    &object.collection.as_str(),
-                    &object.rkey.as_deref().unwrap_or_else(|| now.as_str())
-                );
-                let uri = format!("at://{}/{}", user.did(), key);
+            for write in &input.writes {
+                match write {
+                    InputWritesItem::Create(create) => {
+                        let collection = create.collection.as_str();
+                        let rkey = create
+                            .rkey
+                            .as_deref()
+                            .unwrap_or_else(|| Tid::now(LimitedU32::MIN).as_str());
+                        let uri = format!("at://{}/{}/{}", user.did(), collection, rkey);
 
-                let (builder, cid) = repo
-                    .add_raw(&key, &object.value)
-                    .await
-                    .context("failed to add record")?;
+                        // Process blobs in record
+                        for blob_cid in extract_blobs(&create.value)? {
+                            blob_info.push((blob_cid, uri.clone()));
+                        }
 
-                if let Ok(new_blobs) = scan_blobs(&object.value) {
-                    blobs.extend(
-                        new_blobs
-                            .into_iter()
-                            .map(|blob_cid| (key.clone(), blob_cid)),
-                    );
-                }
+                        // Create the record
+                        let cid = txn.repo().create_record(collection, rkey, &create.value)?;
 
-                ops.push(RepoOp::Create {
-                    cid,
-                    path: key.clone(),
-                });
+                        ops.push(RepoOp::Create {
+                            cid,
+                            path: format!("{}/{}", collection, rkey),
+                        });
 
-                res.push(OutputResultsItem::CreateResult(Box::new(
-                    apply_writes::CreateResultData {
-                        cid: atrium_api::types::string::Cid::new(cid),
-                        uri,
-                        validation_status: None,
+                        results.push(OutputResultsItem::CreateResult(Box::new(
+                            apply_writes::CreateResultData {
+                                cid: atrium_api::types::string::Cid::new(cid),
+                                uri,
+                                validation_status: None,
+                            }
+                            .into(),
+                        )));
                     }
-                    .into(),
-                )));
+                    InputWritesItem::Update(update) => {
+                        let collection = update.collection.as_str();
+                        let rkey = update.rkey.as_str();
+                        let uri = format!("at://{}/{}/{}", user.did(), collection, rkey);
+                        let record_path = format!("{}/{}", collection, rkey);
 
-                (builder, key)
-            }
-            InputWritesItem::Update(ref object) => {
-                let builder: atrium_repo::repo::CommitBuilder<_>;
-                let key = format!("{}/{}", &object.collection.as_str(), &object.rkey.as_str());
-                let uri = format!("at://{}/{}", user.did(), key);
-
-                let prev = repo
-                    .tree()
-                    .get(&key)
-                    .await
-                    .context("failed to search MST")?;
-                if prev.is_none() {
-                    let (create_builder, cid) = repo
-                        .add_raw(&key, &object.value)
-                        .await
-                        .context("failed to add record")?;
-                    if let Ok(new_blobs) = scan_blobs(&object.value) {
-                        blobs.extend(
-                            new_blobs
-                                .into_iter()
-                                .map(|blod_cid| (key.clone(), blod_cid)),
-                        );
-                    }
-                    ops.push(RepoOp::Create {
-                        cid,
-                        path: key.clone(),
-                    });
-                    res.push(OutputResultsItem::CreateResult(Box::new(
-                        apply_writes::CreateResultData {
-                            cid: atrium_api::types::string::Cid::new(cid),
-                            uri,
-                            validation_status: None,
+                        // Process blobs in record
+                        for blob_cid in extract_blobs(&update.value)? {
+                            blob_info.push((blob_cid, uri.clone()));
                         }
-                        .into(),
-                    )));
-                    builder = create_builder;
-                } else {
-                    let prev = prev.context("should be able to find previous record")?;
-                    let (update_builder, cid) = repo
-                        .update_raw(&key, &object.value)
-                        .await
-                        .context("failed to add record")?;
-                    if let Ok(new_blobs) = scan_blobs(&object.value) {
-                        blobs.extend(
-                            new_blobs
-                                .into_iter()
-                                .map(|blod_cid| (key.clone(), blod_cid)),
-                        );
-                    }
-                    ops.push(RepoOp::Update {
-                        cid,
-                        path: key.clone(),
-                        prev,
-                    });
-                    res.push(OutputResultsItem::UpdateResult(Box::new(
-                        apply_writes::UpdateResultData {
-                            cid: atrium_api::types::string::Cid::new(cid),
-                            uri,
-                            validation_status: None,
-                        }
-                        .into(),
-                    )));
-                    builder = update_builder;
-                }
-                (builder, key)
-            }
-            InputWritesItem::Delete(ref object) => {
-                let key = format!("{}/{}", &object.collection.as_str(), &object.rkey.as_str());
 
-                let prev = repo
-                    .tree()
-                    .get(&key)
-                    .await
-                    .context("failed to search MST")?
-                    .context("previous record does not exist")?;
+                        // Get current record
+                        let record = txn.record().get_record(&record_path)?;
 
-                ops.push(RepoOp::Delete {
-                    path: key.clone(),
-                    prev,
-                });
+                        if let Some(existing) = record {
+                            // Update existing record
+                            let prev = Cid::from_str(&existing.cid)?;
+                            let cid = txn.repo().update_record(collection, rkey, &update.value)?;
 
-                res.push(OutputResultsItem::DeleteResult(Box::new(
-                    apply_writes::DeleteResultData {}.into(),
-                )));
-
-                let builder = repo
-                    .delete_raw(&key)
-                    .await
-                    .context("failed to add record")?;
-
-                (builder, key)
-            }
-        };
+                            ops.push(RepoOp::Update {
+                                cid,
+                                path: record_path,
+                                prev,
+                            });
 
-        let sig = skey
-            .sign(&builder.bytes())
-            .context("failed to sign commit")?;
+                            results.push(OutputResultsItem::UpdateResult(Box::new(
+                                apply_writes::UpdateResultData {
+                                    cid: atrium_api::types::string::Cid::new(cid),
+                                    uri,
+                                    validation_status: None,
+                                }
+                                .into(),
+                            )));
+                        } else {
+                            // Create record if doesn't exist
+                            let cid = txn.repo().create_record(collection, rkey, &update.value)?;
 
-        _ = builder
-            .finalize(sig)
-            .await
-            .context("failed to write signed commit")?;
+                            ops.push(RepoOp::Create {
+                                cid,
+                                path: record_path,
+                            });
 
-        keys.push(key);
-    }
+                            results.push(OutputResultsItem::CreateResult(Box::new(
+                                apply_writes::CreateResultData {
+                                    cid: atrium_api::types::string::Cid::new(cid),
+                                    uri,
+                                    validation_status: None,
+                                }
+                                .into(),
+                            )));
+                        }
+                    }
+                    InputWritesItem::Delete(delete) => {
+                        let collection = delete.collection.as_str();
+                        let rkey = delete.rkey.as_str();
+                        let record_path = format!("{}/{}", collection, rkey);
 
-    // Construct a firehose record.
-    let mut mem = Vec::new();
-    let mut store = CarStore::create_with_roots(std::io::Cursor::new(&mut mem), [repo.root()])
-        .await
-        .context("failed to create temp store")?;
+                        // Get current record
+                        let record = txn.record().get_record(&record_path)?;
 
-    // Extract the records out of the user's repository.
-    for key in keys {
-        repo.extract_raw_into(&key, &mut store)
-            .await
-            .context("failed to extract key")?;
-    }
+                        if let Some(existing) = record {
+                            let prev = Cid::from_str(&existing.cid)?;
 
-    let mut tx = db.begin().await.context("failed to begin transaction")?;
+                            // Delete record
+                            txn.repo().delete_record(collection, rkey)?;
 
-    if !swap_commit(
-        &mut *tx,
-        repo.root(),
-        repo.commit().rev(),
-        input.swap_commit.as_ref().map(|cid| *cid.as_ref()),
-        &user.did(),
-    )
-    .await
-    .context("failed to swap commit")?
-    {
-        // This should always succeed.
-        let old = input
-            .swap_commit
-            .clone()
-            .context("swap_commit should always be Some")?;
+                            ops.push(RepoOp::Delete {
+                                path: record_path,
+                                prev,
+                            });
 
-        // The swap failed. Return the old commit and do not update the repository.
-        return Ok(Json(
-            apply_writes::OutputData {
-                results: None,
-                commit: Some(
-                    CommitMetaData {
-                        cid: old,
-                        rev: orig_rev,
+                            results.push(OutputResultsItem::DeleteResult(Box::new(
+                                apply_writes::DeleteResultData {}.into(),
+                            )));
+                        }
                     }
-                    .into(),
-                ),
+                }
             }
-            .into(),
-        ));
-    }
 
-    let did_str = user.did();
-
-    // For updates and removals, unlink the old/deleted record from the blob_ref table.
-    for op in &ops {
-        match op {
-            &RepoOp::Update { ref path, .. } | &RepoOp::Delete { ref path, .. } => {
-                // FIXME: This may cause issues if a user deletes more than one record referencing the same blob.
-                _ = &sqlx::query!(
-                    r#"UPDATE blob_ref SET record = NULL WHERE did = ? AND record = ?"#,
-                    did_str,
-                    path
-                )
-                .execute(&mut *tx)
-                .await
-                .context("failed to remove blob_ref")?;
+            // Process blob associations
+            for (blob_cid, record_uri) in &blob_info {
+                txn.blob()
+                    .register_blob(blob_cid.to_string(), "application/octet-stream", 0)?;
+                txn.blob()
+                    .associate_blob(&blob_cid.to_string(), record_uri)?;
             }
-            &RepoOp::Create { .. } => {}
-        }
-    }
 
-    for &mut (ref key, cid) in &mut blobs {
-        let cid_str = cid.to_string();
+            // Get updated repo root
+            let repo_root = txn.repo().get_root()?;
 
-        // Handle the case where a new record references an existing blob.
-        if sqlx::query!(
-            r#"UPDATE blob_ref SET record = ? WHERE cid = ? AND did = ? AND record IS NULL"#,
-            key,
-            cid_str,
-            did_str,
-        )
-        .execute(&mut *tx)
-        .await
-        .context("failed to update blob_ref")?
-        .rows_affected()
-            == 0
-        {
-            _ = sqlx::query!(
-                r#"INSERT INTO blob_ref (record, cid, did) VALUES (?, ?, ?)"#,
-                key,
-                cid_str,
-                did_str,
-            )
-            .execute(&mut *tx)
-            .await
-            .context("failed to update blob_ref")?;
-        }
-    }
+            // Update metrics
+            counter!(REPO_COMMITS).increment(1);
+            for op in &ops {
+                match op {
+                    RepoOp::Create { .. } => counter!(REPO_OP_CREATE).increment(1),
+                    RepoOp::Update { .. } => counter!(REPO_OP_UPDATE).increment(1),
+                    RepoOp::Delete { .. } => counter!(REPO_OP_DELETE).increment(1),
+                }
+            }
 
-    tx.commit()
-        .await
-        .context("failed to commit blob ref to database")?;
+            // Send ops to firehose
+            // (In real impl, we'd construct a firehose event here)
 
-    // Update counters.
-    counter!(REPO_COMMITS).increment(1);
-    for op in &ops {
-        match *op {
-            RepoOp::Create { .. } => counter!(REPO_OP_CREATE).increment(1),
-            RepoOp::Update { .. } => counter!(REPO_OP_UPDATE).increment(1),
-            RepoOp::Delete { .. } => counter!(REPO_OP_DELETE).increment(1),
-        }
-    }
-
-    // We've committed the transaction to the database, and the commit is now stored in the user's
-    // canonical repository.
-    // We can now broadcast this on the firehose.
-    fhp.commit(firehose::Commit {
-        car: mem,
-        ops,
-        cid: repo.root(),
-        rev: repo.commit().rev().to_string(),
-        did: atrium_api::types::string::Did::new(user.did()).expect("should be valid DID"),
-        pcid: Some(orig_cid),
-        blobs: blobs.into_iter().map(|(_, cid)| cid).collect::<Vec<_>>(),
-    })
-    .await;
+            Ok(apply_writes::OutputData {
+                results: Some(results),
+                commit: Some(
+                    CommitMetaData {
+                        cid: atrium_api::types::string::Cid::new(Cid::from_str(&repo_root.cid)?),
+                        rev: Tid::from_str(&repo_root.rev)?,
+                    }
+                    .into(),
+                ),
+            })
+        })
+        .await?;
 
-    Ok(Json(
-        apply_writes::OutputData {
-            results: Some(res),
-            commit: Some(
-                CommitMetaData {
-                    cid: atrium_api::types::string::Cid::new(repo.root()),
-                    rev: repo.commit().rev(),
-                }
-                .into(),
-            ),
-        }
-        .into(),
-    ))
+    Ok(Json(result.into()))
 }
 
-/// Create a single new repository record. Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.createRecord
-/// ### Request Body
-/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
-/// - `collection`: `nsid` // The NSID of the record collection.
-/// - `rkey`: `string` // The record key. <= 512 characters.
-/// - `validate`: `boolean` // Can be set to 'false' to skip Lexicon schema validation of record data, 'true' to require it, or leave unset to validate only for known Lexicons.
-/// - `record`
-/// - `swap_commit`: `cid` // Compare and swap with the previous commit by CID.
-/// ### Responses
-/// - 200 OK: {`cid`: `cid`, `uri`: `at-uri`, `commit`: {`cid`: `cid`, `rev`: `tid`}, `validation_status`: [`valid`, `unknown`]}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `InvalidSwap`]}
-/// - 401 Unauthorized
+/// Create a single new repository record
 async fn create_record(
     user: AuthenticatedUser,
-    State(skey): State<SigningKey>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    State(fhp): State<FirehoseProducer>,
+    state: State<AppConfig>,
+    db_state: State<Db>,
+    fhp_state: State<FirehoseProducer>,
     Json(input): Json<repo::create_record::Input>,
 ) -> Result<Json<repo::create_record::Output>> {
-    let write_result = apply_writes(
-        user,
-        State(skey),
-        State(config),
-        State(db),
-        State(fhp),
-        Json(
-            repo::apply_writes::InputData {
-                repo: input.repo.clone(),
-                validate: input.validate,
-                swap_commit: input.swap_commit.clone(),
-                writes: vec![repo::apply_writes::InputWritesItem::Create(Box::new(
-                    repo::apply_writes::CreateData {
-                        collection: input.collection.clone(),
-                        rkey: input.rkey.clone(),
-                        value: input.record.clone(),
-                    }
-                    .into(),
-                ))],
+    // Convert the create_record operation to apply_writes
+    let apply_writes_input = repo::apply_writes::InputData {
+        repo: input.repo.clone(),
+        validate: input.validate,
+        swap_commit: input.swap_commit.clone(),
+        writes: vec![repo::apply_writes::InputWritesItem::Create(Box::new(
+            repo::apply_writes::CreateData {
+                collection: input.collection.clone(),
+                rkey: input.rkey.clone(),
+                value: input.record.clone(),
             }
             .into(),
-        ),
-    )
-    .await
-    .context("failed to apply writes")?;
+        ))],
+    }
+    .into();
+
+    let write_result =
+        apply_writes(user, state, db_state, fhp_state, Json(apply_writes_input)).await?;
 
-    let create_result = if let repo::apply_writes::OutputResultsItem::CreateResult(create_result) =
-        write_result
-            .results
-            .clone()
-            .and_then(|result| result.first().cloned())
-            .context("unexpected output from apply_writes")?
+    // Extract the first result
+    let create_result = match write_result
+        .results
+        .and_then(|results| results.first().cloned())
     {
-        Some(create_result)
-    } else {
-        None
-    }
-    .context("unexpected result from apply_writes")?;
+        Some(repo::apply_writes::OutputResultsItem::CreateResult(res)) => res,
+        _ => {
+            return Err(Error::with_status(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                anyhow!("unexpected result type from apply_writes"),
+            ));
+        }
+    };
 
     Ok(Json(
         repo::create_record::OutputData {
···
     ))
 }
 
-/// Write a repository record, creating or updating it as needed. Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.putRecord
-/// ### Request Body
-/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
-/// - `collection`: `nsid` // The NSID of the record collection.
-/// - `rkey`: `string` // The record key. <= 512 characters.
-/// - `validate`: `boolean` // Can be set to 'false' to skip Lexicon schema validation of record data, 'true' to require it, or leave unset to validate only for known Lexicons.
-/// - `record`
-/// - `swap_record`: `boolean` // Compare and swap with the previous record by CID. WARNING: nullable and optional field; may cause problems with golang implementation
-/// - `swap_commit`: `cid` // Compare and swap with the previous commit by CID.
-/// ### Responses
-/// - 200 OK: {"uri": "string","cid": "string","commit": {"cid": "string","rev": "string"},"validationStatus": "valid | unknown"}
-/// - 400 Bad Request: {error:"`InvalidRequest` | `ExpiredToken` | `InvalidToken` | `InvalidSwap`"}
-/// - 401 Unauthorized
+/// Update or create a repository record
 async fn put_record(
     user: AuthenticatedUser,
-    State(skey): State<SigningKey>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    State(fhp): State<FirehoseProducer>,
+    state: State<AppConfig>,
+    db_state: State<Db>,
+    fhp_state: State<FirehoseProducer>,
     Json(input): Json<repo::put_record::Input>,
 ) -> Result<Json<repo::put_record::Output>> {
-    // TODO: `input.swap_record`
-    // FIXME: "put" implies that we will create the record if it does not exist.
-    // We currently only update existing records and/or throw an error if one doesn't exist.
-    let input = (*input).clone();
-    let input = repo::apply_writes::InputData {
+    // Convert put_record operation to apply_writes
+    let apply_writes_input = repo::apply_writes::InputData {
         repo: input.repo,
         validate: input.validate,
         swap_commit: input.swap_commit,
···
     }
     .into();
 
-    let write_result = apply_writes(
-        user,
-        State(skey),
-        State(config),
-        State(db),
-        State(fhp),
-        Json(input),
-    )
-    .await
-    .context("failed to apply writes")?;
+    let write_result =
+        apply_writes(user, state, db_state, fhp_state, Json(apply_writes_input)).await?;
 
-    let update_result = write_result
-        .results
-        .clone()
-        .and_then(|result| result.first().cloned())
-        .context("unexpected output from apply_writes")?;
-    let (cid, uri) = match update_result {
-        repo::apply_writes::OutputResultsItem::CreateResult(create_result) => (
-            Some(create_result.cid.clone()),
-            Some(create_result.uri.clone()),
-        ),
-        repo::apply_writes::OutputResultsItem::UpdateResult(update_result) => (
-            Some(update_result.cid.clone()),
-            Some(update_result.uri.clone()),
-        ),
-        repo::apply_writes::OutputResultsItem::DeleteResult(_) => (None, None),
+    // Extract result with appropriate handling for create/update
+    let (cid, uri) = match write_result.results.and_then(|r| r.first().cloned()) {
+        Some(repo::apply_writes::OutputResultsItem::CreateResult(create)) => {
+            (create.cid, create.uri)
+        }
+        Some(repo::apply_writes::OutputResultsItem::UpdateResult(update)) => {
+            (update.cid, update.uri)
+        }
+        _ => {
+            return Err(Error::with_status(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                anyhow!("unexpected result type from apply_writes"),
+            ));
+        }
     };
+
     Ok(Json(
         repo::put_record::OutputData {
-            cid: cid.context("missing cid")?,
-            commit: write_result.commit.clone(),
-            uri: uri.context("missing uri")?,
+            cid,
+            commit: write_result.commit,
+            uri,
             validation_status: Some("unknown".to_owned()),
         }
         .into(),
     ))
 }
 
-/// Delete a repository record, or ensure it doesn't exist. Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.deleteRecord
-/// ### Request Body
-/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
-/// - `collection`: `nsid` // The NSID of the record collection.
-/// - `rkey`: `string` // The record key. <= 512 characters.
-/// - `swap_record`: `boolean` // Compare and swap with the previous record by CID.
-/// - `swap_commit`: `cid` // Compare and swap with the previous commit by CID.
-/// ### Responses
-/// - 200 OK: {"commit": {"cid": "string","rev": "string"}}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `InvalidSwap`]}
-/// - 401 Unauthorized
+/// Delete a repository record
 async fn delete_record(
     user: AuthenticatedUser,
-    State(skey): State<SigningKey>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    State(fhp): State<FirehoseProducer>,
+    state: State<AppConfig>,
+    db_state: State<Db>,
+    fhp_state: State<FirehoseProducer>,
     Json(input): Json<repo::delete_record::Input>,
 ) -> Result<Json<repo::delete_record::Output>> {
-    // TODO: `input.swap_record`
+    // Convert delete_record operation to apply_writes
+    let apply_writes_input = repo::apply_writes::InputData {
+        repo: input.repo,
+        validate: None,
+        swap_commit: input.swap_commit,
+        writes: vec![repo::apply_writes::InputWritesItem::Delete(Box::new(
+            repo::apply_writes::DeleteData {
+                collection: input.collection,
+                rkey: input.rkey,
+            }
+            .into(),
+        ))],
+    }
+    .into();
+
+    let write_result =
+        apply_writes(user, state, db_state, fhp_state, Json(apply_writes_input)).await?;
 
     Ok(Json(
         repo::delete_record::OutputData {
-            commit: apply_writes(
-                user,
-                State(skey),
-                State(config),
-                State(db),
-                State(fhp),
-                Json(
-                    repo::apply_writes::InputData {
-                        repo: input.repo.clone(),
-                        swap_commit: input.swap_commit.clone(),
-                        validate: None,
-                        writes: vec![repo::apply_writes::InputWritesItem::Delete(Box::new(
-                            repo::apply_writes::DeleteData {
-                                collection: input.collection.clone(),
-                                rkey: input.rkey.clone(),
-                            }
-                            .into(),
-                        ))],
-                    }
-                    .into(),
-                ),
-            )
-            .await
-            .context("failed to apply writes")?
-            .commit
-            .clone(),
+            commit: write_result.commit,
         }
         .into(),
     ))
 }
 
-/// Get information about an account and repository, including the list of collections. Does not require auth.
-/// - GET /xrpc/com.atproto.repo.describeRepo
-/// ### Query Parameters
-/// - `repo`: `at-identifier` // The handle or DID of the repo.
-/// ### Responses
-/// - 200 OK: {"handle": "string","did": "string","didDoc": {},"collections": [string],"handleIsCorrect": true} \
-///   handeIsCorrect - boolean - Indicates if handle is currently valid (resolves bi-directionally)
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`]}
-/// - 401 Unauthorized
+/// Get information about an account and repository
 async fn describe_repo(
     State(config): State<AppConfig>,
     State(db): State<Db>,
     Query(input): Query<repo::describe_repo::ParametersData>,
 ) -> Result<Json<repo::describe_repo::Output>> {
-    // Lookup the DID by the provided handle.
-    let (did, handle) = resolve_did(&db, &input.repo)
-        .await
-        .context("failed to resolve handle")?;
+    // Resolve the DID
+    let (did, handle) = resolve_did(&db, &input.repo).await?;
 
-    let mut repo = storage::open_repo_db(&config.repo, &db, did.as_str())
-        .await
-        .context("failed to open user repo")?;
-
-    let mut collections = HashSet::new();
-
-    let mut tree = repo.tree();
-    let mut it = Box::pin(tree.keys());
-    while let Some(key) = it.try_next().await.context("failed to iterate repo keys")? {
-        if let Some((collection, _rkey)) = key.split_once('/') {
-            _ = collections.insert(collection.to_owned());
-        }
-    }
+    // Get collections from actor store
+    let actor_store = ActorStore::new(Arc::new(config));
+    let collections = actor_store
+        .read(did.as_str(), |reader| reader.record().list_collections())
+        .await?;
 
     Ok(Json(
         repo::describe_repo::OutputData {
             collections: collections
                 .into_iter()
-                .map(|nsid| Nsid::new(nsid).expect("should be valid NSID"))
-                .collect::<Vec<_>>(),
-            did: did.clone(),
-            did_doc: Unknown::Null, // TODO: Fetch the DID document from the PLC directory
-            handle: handle.clone(),
-            handle_is_correct: true, // TODO
+                .map(|c| Nsid::new(c).expect("valid NSID"))
+                .collect(),
+            did,
+            did_doc: Unknown::Null, // TODO: fetch from PLC directory
+            handle,
+            handle_is_correct: true, // TODO: validate handle resolution
         }
         .into(),
     ))
 }
 
-/// Get a single record from a repository. Does not require auth.
-/// - GET /xrpc/com.atproto.repo.getRecord
-/// ### Query Parameters
-/// - `repo`: `at-identifier` // The handle or DID of the repo.
-/// - `collection`: `nsid` // The NSID of the record collection.
-/// - `rkey`: `string` // The record key. <= 512 characters.
-/// - `cid`: `cid` // The CID of the version of the record. If not specified, then return the most recent version.
-/// ### Responses
-/// - 200 OK: {"uri": "string","cid": "string","value": {}}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `RecordNotFound`]}
-/// - 401 Unauthorized
+/// Get a single record from a repository
 async fn get_record(
     State(config): State<AppConfig>,
     State(db): State<Db>,
···
 ) -> Result<Json<repo::get_record::Output>> {
     if input.cid.is_some() {
         return Err(Error::unimplemented(anyhow!(
-            "looking up old records is unsupported"
+            "fetching specific CID versions not supported"
         )));
     }
 
-    // Lookup the DID by the provided handle.
-    let (did, _handle) = resolve_did(&db, &input.repo)
-        .await
-        .context("failed to resolve handle")?;
-
-    let mut repo = storage::open_repo_db(&config.repo, &db, did.as_str())
-        .await
-        .context("failed to open user repo")?;
+    // Resolve the DID
+    let (did, _) = resolve_did(&db, &input.repo).await?;
 
+    // Construct record path and URI
     let key = format!("{}/{}", input.collection.as_str(), input.rkey.as_str());
-    let uri = format!("at://{}/{}", did.as_str(), &key);
+    let uri = format!("at://{}/{}", did.as_str(), key);
 
-    let cid = repo
-        .tree()
-        .get(&key)
-        .await
-        .context("failed to find record")?;
+    // Get record from actor store
+    let actor_store = ActorStore::new(Arc::new(config));
+    let record_opt = actor_store
+        .read(did.as_str(), |reader| reader.record().get_record(&key))
+        .await?;
 
-    let record: Option<serde_json::Value> =
-        repo.get_raw(&key).await.context("failed to read record")?;
+    match record_opt {
+        Some(record) => {
+            // Parse record content
+            let value: serde_json::Value = serde_json::from_slice(&record.content)?;
+            let cid = Cid::from_str(&record.cid)?;
 
-    record.map_or_else(
-        || {
-            Err(Error::with_message(
-                StatusCode::BAD_REQUEST,
-                anyhow!("could not find the requested record at {}", uri),
-                ErrorMessage::new("RecordNotFound", format!("Could not locate record: {uri}")),
-            ))
-        },
-        |record_value| {
             Ok(Json(
                 repo::get_record::OutputData {
-                    cid: cid.map(atrium_api::types::string::Cid::new),
-                    uri: uri.clone(),
-                    value: record_value
-                        .try_into_unknown()
-                        .context("should be valid JSON")?,
+                    cid: Some(atrium_api::types::string::Cid::new(cid)),
+                    uri,
+                    value: value.try_into_unknown()?,
                 }
                 .into(),
             ))
-        },
-    )
+        }
+        None => Err(Error::with_message(
+            StatusCode::BAD_REQUEST,
+            anyhow!("record not found: {}", uri),
+            ErrorMessage::new("RecordNotFound", format!("Could not locate record: {uri}")),
+        )),
+    }
 }
 
-/// List a range of records in a repository, matching a specific collection. Does not require auth.
-/// - GET /xrpc/com.atproto.repo.listRecords
-/// ### Query Parameters
-/// - `repo`: `at-identifier` // The handle or DID of the repo.
-/// - `collection`: `nsid` // The NSID of the record type.
-/// - `limit`: `integer` // The maximum number of records to return. Default 50, >=1 and <=100.
-/// - `cursor`: `string`
-/// - `reverse`: `boolean` // Flag to reverse the order of the returned records.
-/// ### Responses
-/// - 200 OK: {"cursor": "string","records": [{"uri": "string","cid": "string","value": {}}]}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`]}
-/// - 401 Unauthorized
+/// List records from a repository collection
 async fn list_records(
     State(config): State<AppConfig>,
     State(db): State<Db>,
     Query(input): Query<Object<ListRecordsParameters>>,
 ) -> Result<Json<repo::list_records::Output>> {
-    // TODO: `input.reverse`
-
-    // Lookup the DID by the provided handle.
-    let (did, _handle) = resolve_did(&db, &input.repo)
-        .await
-        .context("failed to resolve handle")?;
+    // Resolve the DID
+    let (did, _) = resolve_did(&db, &input.repo).await?;
 
-    let mut repo = storage::open_repo_db(&config.repo, &db, did.as_str())
-        .await
-        .context("failed to open user repo")?;
+    // Parse limit parameter
+    let limit = input
+        .limit
+        .as_deref()
+        .and_then(|l| l.parse::<u32>().ok())
+        .unwrap_or(50)
+        .min(100);
 
-    let mut keys = Vec::new();
-    let mut tree = repo.tree();
-
-    let mut entry = input.collection.to_string();
-    entry.push('/');
-    let mut iterator = Box::pin(tree.entries_prefixed(entry.as_str()));
-    while let Some((key, cid)) = iterator
-        .try_next()
-        .await
-        .context("failed to iterate keys")?
-    {
-        keys.push((key, cid));
-    }
-
-    drop(iterator);
+    // Get records from actor store
+    let actor_store = ActorStore::new(Arc::new(config));
+    let records = actor_store
+        .read(did.as_str(), |reader| {
+            reader.record().list_collection_records(
+                input.collection.as_str(),
+                limit,
+                input.cursor.as_deref(),
+                input.reverse.unwrap_or(false),
+            )
+        })
+        .await?;
 
-    // TODO: Calculate the view on `keys` using `cursor` and `limit`.
+    // Format records for response
+    let mut result_records = Vec::new();
+    let mut last_cursor = None;
 
-    let mut records = Vec::new();
-    for &(ref key, cid) in &keys {
-        let value: serde_json::Value = repo
-            .get_raw(key)
-            .await
-            .context("failed to get record")?
-            .context("record not found")?;
+    for record in &records {
+        // Parse record
+        let value: serde_json::Value = serde_json::from_slice(&record.content)?;
+        let cid = Cid::from_str(&record.cid)?;
+        last_cursor = Some(record.rkey.clone());
 
-        records.push(
+        result_records.push(
             repo::list_records::RecordData {
                 cid: atrium_api::types::string::Cid::new(cid),
-                uri: format!("at://{}/{}", did.as_str(), key),
-                value: value.try_into_unknown().context("should be valid JSON")?,
+                uri: format!(
+                    "at://{}/{}/{}",
+                    did.as_str(),
+                    input.collection.as_str(),
+                    record.rkey
+                ),
+                value: value.try_into_unknown()?,
             }
             .into(),
         );
     }
 
-    #[expect(clippy::pattern_type_mismatch)]
     Ok(Json(
         repo::list_records::OutputData {
-            cursor: keys.last().map(|(_, cid)| cid.to_string()),
-            records,
+            cursor: last_cursor,
+            records: result_records,
         }
         .into(),
     ))
 }
 
-/// Upload a new blob, to be referenced from a repository record. \
-/// The blob will be deleted if it is not referenced within a time window (eg, minutes). \
-/// Blob restrictions (mimetype, size, etc) are enforced when the reference is created. \
-/// Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.uploadBlob
-/// ### Request Body
-/// ### Responses
-/// - 200 OK: {"blob": "binary"}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`]}
-/// - 401 Unauthorized
+/// Upload a blob
 async fn upload_blob(
     user: AuthenticatedUser,
     State(config): State<AppConfig>,
-    State(db): State<Db>,
     request: Request<Body>,
 ) -> Result<Json<repo::upload_blob::Output>> {
+    // Get content-length and content-type
     let length = request
         .headers()
         .get(http::header::CONTENT_LENGTH)
         .context("no content length provided")?
-        .to_str()
-        .map_err(anyhow::Error::from)
-        .and_then(|content_length| content_length.parse::<u64>().map_err(anyhow::Error::from))
-        .context("invalid content-length header")?;
+        .to_str()?
+        .parse::<u64>()?;
+
     let mime = request
         .headers()
         .get(http::header::CONTENT_TYPE)
         .context("no content-type provided")?
-        .to_str()
-        .context("invalid content-type provided")?
+        .to_str()?
         .to_owned();
 
+    // Check size limit
     if length > config.blob.limit {
         return Err(Error::with_status(
             StatusCode::PAYLOAD_TOO_LARGE,
···
         ));
     }
 
-    // FIXME: Need to make this more robust. This will fail under load.
+    // Create temp file
     let filename = config
         .blob
         .path
         .join(format!("temp-{}.blob", chrono::Utc::now().timestamp()));
-    let mut file = tokio::fs::File::create(&filename)
-        .await
-        .context("failed to create temporary file")?;
+    let mut file = tokio::fs::File::create(&filename).await?;
 
-    let mut len = 0_usize;
+    // Process upload
+    let mut len = 0;
     let mut sha = Sha256::new();
     let mut stream = request.into_body().into_data_stream();
-    while let Some(bytes) = stream.try_next().await.context("failed to receive file")? {
-        len = len.checked_add(bytes.len()).context("size overflow")?;
 
-        // Deal with any sneaky end-users trying to bypass size limitations.
-        let len_u64: u64 = len.try_into().context("failed to convert `len`")?;
-        if len_u64 > config.blob.limit {
-            drop(file);
-            tokio::fs::remove_file(&filename)
-                .await
-                .context("failed to remove temp file")?;
+    while let Some(bytes) = stream.try_next().await? {
+        len += bytes.len();
 
+        if len as u64 > config.blob.limit {
+            drop(file);
+            tokio::fs::remove_file(&filename).await?;
             return Err(Error::with_status(
                 StatusCode::PAYLOAD_TOO_LARGE,
-                anyhow!("size above limit and content-length header was wrong"),
+                anyhow!("actual size exceeds limit"),
             ));
         }
 
         sha.update(&bytes);
-
-        file.write_all(&bytes)
-            .await
-            .context("failed to write blob")?;
+        file.write_all(&bytes).await?;
     }
 
+    // Finalize blob
     drop(file);
     let hash = sha.finalize();
 
     let cid = Cid::new_v1(
-        IPLD_RAW,
-        atrium_repo::Multihash::wrap(IPLD_MH_SHA2_256, hash.as_slice())
-            .context("should be valid hash")?,
+        0x55,                                                 // IPLD RAW
+        atrium_repo::Multihash::wrap(0x12, hash.as_slice())?, // SHA2-256
     );
 
     let cid_str = cid.to_string();
 
-    tokio::fs::rename(&filename, config.blob.path.join(format!("{cid_str}.blob")))
-        .await
-        .context("failed to finalize blob")?;
-
-    let did_str = user.did();
-
-    _ = sqlx::query!(
-        r#"INSERT INTO blob_ref (cid, did, record) VALUES (?, ?, NULL)"#,
-        cid_str,
-        did_str
+    // Move to permanent location
+    tokio::fs::rename(
+        &filename,
+        config.blob.path.join(format!("{}.blob", cid_str)),
     )
-    .execute(&db)
-    .await
-    .context("failed to insert blob into database")?;
+    .await?;
+
+    // Register blob in actor store
+    let actor_store = ActorStore::new(Arc::new(config));
+    actor_store
+        .transact(user.did(), |txn| {
+            txn.blob()
+                .register_blob(cid_str.clone(), mime.clone(), len as u64)
+        })
+        .await?;
 
     Ok(Json(
         repo::upload_blob::OutputData {
···
     Err(Error::unimplemented(anyhow!("not implemented")))
 }
 
-/// These endpoints are part of the atproto PDS repository management APIs. \
-/// Requests usually require authentication (unlike the com.atproto.sync.* endpoints), and are made directly to the user's own PDS instance.
-/// ### Routes
-/// - AP /xrpc/com.atproto.repo.applyWrites     -> [`apply_writes`]
-/// - AP /xrpc/com.atproto.repo.createRecord    -> [`create_record`]
-/// - AP /xrpc/com.atproto.repo.putRecord       -> [`put_record`]
-/// - AP /xrpc/com.atproto.repo.deleteRecord    -> [`delete_record`]
-/// - AP /xrpc/com.atproto.repo.uploadBlob      -> [`upload_blob`]
-/// - UG /xrpc/com.atproto.repo.describeRepo    -> [`describe_repo`]
-/// - UG /xrpc/com.atproto.repo.getRecord       -> [`get_record`]
-/// - UG /xrpc/com.atproto.repo.listRecords     -> [`list_records`]
-///     - [ ] xx /xrpc/com.atproto.repo.importRepo
-// - [ ] xx /xrpc/com.atproto.repo.listMissingBlobs
+/// Register repo endpoints
 pub(super) fn routes() -> Router<AppState> {
     Router::new()
         .route(concat!("/", repo::apply_writes::NSID), post(apply_writes))

src/main.rs

···
 //! PDS implementation.
+mod actor_store;
 mod auth;
 mod config;
 mod did;