Prototype SQLite repo storage

.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-2e13e052dfc64f29d9da1bce2bf844cbb918ad3bb01e386801d3b0d3be246573.json

···
+{
+  "db_name": "SQLite",
+  "query": "SELECT COUNT(*) FROM oauth_refresh_tokens WHERE dpop_thumbprint = ? AND client_id = ?",
+  "describe": {
+    "columns": [
+      {
+        "name": "COUNT(*)",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "2e13e052dfc64f29d9da1bce2bf844cbb918ad3bb01e386801d3b0d3be246573"
+}

.sqlx/query-73fd3e30b7694c92cf9309751d186fe622fa7d99fdf56dde7e60c3696581116c.json

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

.sqlx/query-9ae396483f5c0f338003dde167870448e01664d6666b64878f87777416b82773.json

···
-{
-  "db_name": "SQLite",
-  "query": "\n                UPDATE invites\n                    SET count = count - 1\n                    WHERE id = ?\n                    AND count > 0\n                    RETURNING id\n                ",
-  "describe": {
-    "columns": [
-      {
-        "name": "id",
-        "ordinal": 0,
-        "type_info": "Text"
-      }
-    ],
-    "parameters": {
-      "Right": 1
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "9ae396483f5c0f338003dde167870448e01664d6666b64878f87777416b82773"
-}

.sqlx/query-c51b4c9de70b5be51a6e0a5fd744387ae804e8ba978b61c4d04d74b1f8de2614.json

···
+{
+  "db_name": "SQLite",
+  "query": "UPDATE oauth_refresh_tokens SET revoked = TRUE\n                        WHERE client_id = ? AND dpop_thumbprint = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "c51b4c9de70b5be51a6e0a5fd744387ae804e8ba978b61c4d04d74b1f8de2614"
+}

.sqlx/query-e26b7c36a34130e350f3f3e06b3200c56a0e3330ac0b658de6bbdb39b5497fab.json

···
+{
+  "db_name": "SQLite",
+  "query": "\n                    UPDATE invites\n                        SET count = count - 1\n                        WHERE id = ?\n                        AND count > 0\n                        RETURNING id\n                    ",
+  "describe": {
+    "columns": [
+      {
+        "name": "id",
+        "ordinal": 0,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "e26b7c36a34130e350f3f3e06b3200c56a0e3330ac0b658de6bbdb39b5497fab"
+}

Cargo.lock

···
 dependencies = [
  "anyhow",
  "argon2",
+ "async-trait",
  "atrium-api",
  "atrium-crypto",
  "atrium-repo",

Cargo.toml

···
 url = "2.5.4"
 uuid = { version = "1.14.0", features = ["v4"] }
 urlencoding = "2.1.3"
+async-trait = "0.1.88"

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);

src/config.rs

···
 pub(crate) struct RepoConfig {
     /// The path to the repository storage.
     pub path: PathBuf,
+    /// Use SQLite for repository storage instead of CAR files.
+    #[serde(default)]
+    pub use_sqlite: bool,
 }
 
 #[derive(Deserialize, Debug, Clone)]

src/endpoints/server.rs

···
     firehose::{Commit, FirehoseProducer},
     metrics::AUTH_FAILED,
     plc::{self, PlcOperation, PlcService},
+    storage,
 };
 
 /// This is a dummy password that can be used in absence of a real password.
···
     // Write out an initial commit for the user.
     // https://atproto.com/guides/account-lifecycle
     let (cid, rev, store) = async {
-        let file = tokio::fs::File::create_new(config.repo.path.join(format!("{did_hash}.car")))
-            .await
-            .context("failed to create repo file")?;
-        let mut store = CarStore::create(file)
+        let store = storage::create_storage_for_did(&config.repo, &did_hash)
             .await
-            .context("failed to create carstore")?;
+            .context("failed to create storage")?;
 
+        // Initialize the repository with the storage
         let repo_builder = Repository::create(
-            &mut store,
+            store,
             Did::from_str(&did).expect("should be valid DID format"),
         )
         .await
···
         let root = repo.root();
         let rev = repo.commit().rev();
 
+        // Create a temporary CAR store for firehose events
         let mut mem = Vec::new();
         let mut firehose_store =
             CarStore::create_with_roots(std::io::Cursor::new(&mut mem), [repo.root()])

src/storage.rs

···
-//! `ATProto` user repository datastore functionality.
-
-use std::str::FromStr as _;
-
-use anyhow::{Context as _, Result};
-use atrium_repo::{
-    Cid, Repository,
-    blockstore::{AsyncBlockStoreRead, AsyncBlockStoreWrite, CarStore},
-};
-
-use crate::{Db, config::RepoConfig, mmap::MappedFile};
-
-/// Open a block store for a given DID.
-pub(crate) async fn open_store(
-    config: &RepoConfig,
-    did: impl Into<String>,
-) -> Result<impl AsyncBlockStoreRead + AsyncBlockStoreWrite> {
-    let did = did.into();
-    let id = did
-        .strip_prefix("did:plc:")
-        .context("did in unknown format")?;
-
-    let p = config.path.join(id).with_extension("car");
-
-    let f = std::fs::File::options()
-        .read(true)
-        .write(true)
-        .open(p)
-        .context("failed to open repository file")?;
-    let f = MappedFile::new(f).context("failed to map repo")?;
-
-    CarStore::open(f).await.context("failed to open car store")
-}
-
-/// Open a repository for a given DID.
-pub(crate) async fn open_repo_db(
-    config: &RepoConfig,
-    db: &Db,
-    did: impl Into<String>,
-) -> Result<Repository<impl AsyncBlockStoreRead + AsyncBlockStoreWrite>> {
-    let did = did.into();
-    let cid = sqlx::query_scalar!(
-        r#"
-        SELECT root FROM accounts
-        WHERE did = ?
-        "#,
-        did
-    )
-    .fetch_one(db)
-    .await
-    .context("failed to query database")?;
-
-    open_repo(
-        config,
-        did,
-        Cid::from_str(&cid).context("should be valid CID")?,
-    )
-    .await
-}
-
-/// Open a repository for a given DID and CID.
-pub(crate) async fn open_repo(
-    config: &RepoConfig,
-    did: impl Into<String>,
-    cid: Cid,
-) -> Result<Repository<impl AsyncBlockStoreRead + AsyncBlockStoreWrite>> {
-    let did = did.into();
-    let store = open_store(config, did)
-        .await
-        .context("failed to open storage")?;
-
-    Repository::open(store, cid)
-        .await
-        .context("failed to open repo")
-}

src/storage/car.rs

···
+//! CAR file-based repository storage
+
+use anyhow::{Context as _, Result};
+use atrium_repo::blockstore::{AsyncBlockStoreRead, AsyncBlockStoreWrite, CarStore};
+
+use crate::{config::RepoConfig, mmap::MappedFile};
+
+/// Open a CAR block store for a given DID.
+pub(crate) async fn open_car_store(
+    config: &RepoConfig,
+    did: impl AsRef<str>,
+) -> Result<impl AsyncBlockStoreRead + AsyncBlockStoreWrite> {
+    let id = did
+        .as_ref()
+        .strip_prefix("did:plc:")
+        .context("did in unknown format")?;
+
+    let p = config.path.join(id).with_extension("car");
+
+    let f = std::fs::File::options()
+        .read(true)
+        .write(true)
+        .open(p)
+        .context("failed to open repository file")?;
+    let f = MappedFile::new(f).context("failed to map repo")?;
+
+    CarStore::open(f).await.context("failed to open car store")
+}

src/storage/mod.rs

···
+//! `ATProto` user repository datastore functionality.
+
+mod car;
+mod sqlite;
+
+use anyhow::{Context as _, Result};
+use atrium_repo::{
+    Cid, Repository,
+    blockstore::{AsyncBlockStoreRead, AsyncBlockStoreWrite},
+};
+use std::str::FromStr as _;
+
+use crate::{Db, config::RepoConfig};
+
+// Re-export public items
+pub(crate) use car::open_car_store;
+pub(crate) use sqlite::{SQLiteStore, open_sqlite_store};
+
+/// Open a repository for a given DID.
+pub(crate) async fn open_repo_db(
+    config: &RepoConfig,
+    db: &Db,
+    did: impl Into<String>,
+) -> Result<Repository<impl AsyncBlockStoreRead + AsyncBlockStoreWrite>> {
+    let did = did.into();
+    let cid = sqlx::query_scalar!(
+        r#"
+        SELECT root FROM accounts
+        WHERE did = ?
+        "#,
+        did
+    )
+    .fetch_one(db)
+    .await
+    .context("failed to query database")?;
+
+    open_repo(
+        config,
+        did,
+        Cid::from_str(&cid).context("should be valid CID")?,
+    )
+    .await
+}
+
+/// Open a repository for a given DID and CID.
+pub(crate) async fn open_repo(
+    config: &RepoConfig,
+    did: impl Into<String>,
+    cid: Cid,
+) -> Result<Repository<impl AsyncBlockStoreRead + AsyncBlockStoreWrite>> {
+    let did = did.into();
+
+    // if config.use_sqlite {
+    let store = open_sqlite_store(config, did.clone()).await?;
+    tracing::info!("Opening SQLite store for DID: {}, CID: {}", did, cid);
+    return Repository::open(store, cid)
+        .await
+        .context("failed to open repo");
+    // }
+    // let store = open_car_store(config, &did).await?;
+    // Repository::open(store, cid)
+    //     .await
+    //     .context("failed to open repo")
+}
+
+/// Open a block store for a given DID.
+pub(crate) async fn open_store(
+    config: &RepoConfig,
+    did: impl Into<String>,
+) -> Result<impl AsyncBlockStoreRead + AsyncBlockStoreWrite> {
+    let did = did.into();
+
+    // if config.use_sqlite {
+    return open_sqlite_store(config, did.clone()).await;
+    // }
+    // Default to CAR store
+    // open_car_store(config, &did).await
+}
+
+/// Create a storage backend for a DID
+pub(crate) async fn create_storage_for_did(
+    config: &RepoConfig,
+    did_hash: &str,
+) -> Result<impl AsyncBlockStoreRead + AsyncBlockStoreWrite> {
+    // Use standard file structure but change extension based on type
+    // if config.use_sqlite {
+    // For SQLite, create a new database file
+    let db_path = config.path.join(format!("{}.db", did_hash));
+
+    // Ensure parent directory exists
+    if let Some(parent) = db_path.parent() {
+        tokio::fs::create_dir_all(parent)
+            .await
+            .context("failed to create directory")?;
+    }
+
+    // Create SQLite store
+    let pool = sqlx::sqlite::SqlitePoolOptions::new()
+        .max_connections(5)
+        .connect_with(
+            sqlx::sqlite::SqliteConnectOptions::new()
+                .filename(&db_path)
+                .create_if_missing(true),
+        )
+        .await
+        .context("failed to connect to SQLite database")?;
+
+    // Initialize tables
+    sqlx::query(
+        "
+            CREATE TABLE IF NOT EXISTS blocks (
+                cid TEXT PRIMARY KEY NOT NULL,
+                data BLOB NOT NULL,
+                multicodec INTEGER NOT NULL,
+                multihash INTEGER NOT NULL
+            );
+            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 IF NOT EXISTS idx_blocks_cid ON blocks(cid);
+            CREATE INDEX IF NOT EXISTS idx_tree_nodes_repo ON tree_nodes(repo_did);
+        ",
+    )
+    .execute(&pool)
+    .await
+    .context("failed to create tables")?;
+
+    Ok(SQLiteStore {
+        pool,
+        did: format!("did:plc:{}", did_hash),
+    })
+    // } else {
+    //     // For CAR files, create a new file
+    //     let file_path = config.path.join(format!("{}.car", did_hash));
+
+    //     // Ensure parent directory exists
+    //     if let Some(parent) = file_path.parent() {
+    //         tokio::fs::create_dir_all(parent)
+    //             .await
+    //             .context("failed to create directory")?;
+    //     }
+
+    //     let file = tokio::fs::File::create_new(file_path)
+    //         .await
+    //         .context("failed to create repo file")?;
+
+    //     CarStore::create(file)
+    //         .await
+    //         .context("failed to create carstore")
+    // }
+}

src/storage/sqlite.rs

···
+//! SQLite-based repository storage implementation.
+
+use anyhow::{Context as _, Result};
+use atrium_repo::{
+    Cid, Multihash,
+    blockstore::{AsyncBlockStoreRead, AsyncBlockStoreWrite, Error as BlockstoreError},
+};
+use sha2::Digest;
+use sqlx::SqlitePool;
+
+use crate::config::RepoConfig;
+
+/// SQLite-based implementation of block storage.
+pub(crate) struct SQLiteStore {
+    pub did: String,
+    pub pool: SqlitePool,
+}
+
+impl AsyncBlockStoreRead for SQLiteStore {
+    async fn read_block(&mut self, cid: Cid) -> Result<Vec<u8>, BlockstoreError> {
+        tracing::info!("Reading block with CID: {}", cid);
+        let mut contents = Vec::new();
+        self.read_block_into(cid, &mut contents).await?;
+        Ok(contents)
+    }
+    fn read_block_into(
+        &mut self,
+        cid: Cid,
+        contents: &mut Vec<u8>,
+    ) -> impl Future<Output = Result<(), BlockstoreError>> + Send {
+        tracing::info!("Reading block into buffer with CID: {}", cid);
+        let cid_str = cid.to_string();
+        let pool = self.pool.clone();
+
+        tracing::info!("Async moving block read");
+        async move {
+            tracing::info!("Async move block read");
+            // let record = sqlx::query!(r#"SELECT data FROM blocks WHERE cid = ?"#, cid_str)
+            //     .fetch_optional(&pool)
+            //     .await
+            //     .map_err(|e| BlockstoreError::Other(Box::new(e)))?
+            //     .ok_or(BlockstoreError::CidNotFound)?;
+            let record = sqlx::query!(r#"SELECT data FROM blocks WHERE cid = ?"#, cid_str)
+                .fetch_optional(&pool)
+                .await;
+            tracing::info!("Record fetched: {:?}", record);
+            let record = match record {
+                Ok(Some(record)) => record,
+                Ok(None) => return Err(BlockstoreError::CidNotFound),
+                Err(e) => return Err(BlockstoreError::Other(Box::new(e))),
+            };
+            tracing::info!("Block read successful");
+
+            contents.clear();
+            tracing::info!("Contents cleared");
+            contents.extend_from_slice(&record.data);
+            tracing::info!("Contents extended");
+            Ok(())
+        }
+    }
+}
+
+impl AsyncBlockStoreWrite for SQLiteStore {
+    fn write_block(
+        &mut self,
+        codec: u64,
+        hash: u64,
+        contents: &[u8],
+    ) -> impl Future<Output = Result<Cid, BlockstoreError>> + Send {
+        let contents = contents.to_vec(); // Clone the data
+        let pool = self.pool.clone();
+
+        async move {
+            let digest = match hash {
+                atrium_repo::blockstore::SHA2_256 => sha2::Sha256::digest(&contents),
+                _ => return Err(BlockstoreError::UnsupportedHash(hash)),
+            };
+
+            let multihash = Multihash::wrap(hash, digest.as_slice())
+                .map_err(|_| BlockstoreError::UnsupportedHash(hash))?;
+
+            let cid = Cid::new_v1(codec, multihash);
+            let cid_str = cid.to_string();
+
+            // Use a transaction for atomicity
+            let mut tx = pool
+                .begin()
+                .await
+                .map_err(|e| BlockstoreError::Other(Box::new(e)))?;
+
+            // Check if block already exists
+            let exists =
+                sqlx::query_scalar!(r#"SELECT COUNT(*) FROM blocks WHERE cid = ?"#, cid_str)
+                    .fetch_one(&mut *tx)
+                    .await
+                    .map_err(|e| BlockstoreError::Other(Box::new(e)))?;
+
+            // Only insert if block doesn't exist
+            let codec = codec as i64;
+            let hash = hash as i64;
+            if exists == 0 {
+                _ = sqlx::query!(
+                    r#"INSERT INTO blocks (cid, data, multicodec, multihash) VALUES (?, ?, ?, ?)"#,
+                    cid_str,
+                    contents,
+                    codec,
+                    hash
+                )
+                .execute(&mut *tx)
+                .await
+                .map_err(|e| BlockstoreError::Other(Box::new(e)))?;
+            }
+
+            tx.commit()
+                .await
+                .map_err(|e| BlockstoreError::Other(Box::new(e)))?;
+
+            Ok(cid)
+        }
+    }
+}
+
+/// Open a SQLite store for the given DID.
+pub(crate) async fn open_sqlite_store(
+    config: &RepoConfig,
+    did: impl Into<String>,
+) -> Result<SQLiteStore> {
+    tracing::info!("Opening SQLite store for DID");
+    let did_str = did.into();
+
+    // Extract the PLC ID from the DID
+    let id = did_str
+        .strip_prefix("did:plc:")
+        .context("DID in unknown format")?;
+
+    // Create database connection pool
+    let db_path = config.path.join(format!("{id}.db"));
+
+    let pool = sqlx::sqlite::SqlitePoolOptions::new()
+        .max_connections(5)
+        .connect_with(
+            sqlx::sqlite::SqliteConnectOptions::new()
+                .filename(&db_path)
+                .create_if_missing(true),
+        )
+        .await
+        .context("failed to connect to SQLite database")?;
+
+    // Ensure tables exist
+    _ = sqlx::query(
+        "
+        CREATE TABLE IF NOT EXISTS blocks (
+            cid TEXT PRIMARY KEY NOT NULL,
+            data BLOB NOT NULL,
+            multicodec INTEGER NOT NULL,
+            multihash INTEGER NOT NULL
+        );
+        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 IF NOT EXISTS idx_blocks_cid ON blocks(cid);
+        CREATE INDEX IF NOT EXISTS idx_tree_nodes_repo ON tree_nodes(repo_did);
+        ",
+    )
+    .execute(&pool)
+    .await
+    .context("failed to create tables")?;
+
+    Ok(SQLiteStore { pool, did: did_str })
+}

src/tests.rs

···
 
     #[tokio::test]
     async fn test_create_account() -> Result<()> {
-        return Ok(());
-        #[expect(unreachable_code, reason = "Disabled")]
+        // return Ok(());
+        // #[expect(unreachable_code, reason = "Disabled")]
         let state = init_test_state().await?;
         let account = state.create_test_account().await?;
 
···
 
     #[tokio::test]
     async fn test_create_record_benchmark() -> Result<()> {
-        return Ok(());
-        #[expect(unreachable_code, reason = "Disabled")]
+        // return Ok(());
+        // #[expect(unreachable_code, reason = "Disabled")]
         let duration = create_record_benchmark(100, 1).await?;
 
         println!("Created 100 records in {duration:?}");