prototype actor_store

.sqlx/query-22c1e98ac038509ad16ce437e6670a59d3fc97a05ea8b0f1f80dba0157c53e13.json

···
+{
+  "db_name": "SQLite",
+  "query": "SELECT name FROM actor_migration",
+  "describe": {
+    "columns": [
+      {
+        "name": "name",
+        "ordinal": 0,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "22c1e98ac038509ad16ce437e6670a59d3fc97a05ea8b0f1f80dba0157c53e13"
+}

.sqlx/query-5ea8376fbbe3077b2fc62187cc29a2d03eda91fa468c7fe63306f04e160ecb5d.json

···
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO actor_migration (name, appliedAt) VALUES (?, ?)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "5ea8376fbbe3077b2fc62187cc29a2d03eda91fa468c7fe63306f04e160ecb5d"
+}

src/actor_store/blob/reader.rs

···
 use atrium_repo::Cid;
 use sqlx::{Row, SqlitePool};
 
+use crate::repo::types::{BlobStore, BlobStoreTrait};
+
 /// Reader for blob data in the actor store.
 pub(crate) struct BlobReader {
     /// Database connection.
···
     /// Get a blob's full data and metadata.
     pub(crate) async fn get_blob(&self, cid: &Cid) -> Result<Option<BlobData>> {
         let metadata = self.get_blob_metadata(cid).await?;
-        let blob_stream = self.blobstore.get_stream(cid).await?;
+        let blob_stream = self.blobstore.get_stream(*cid)?;
         if blob_stream.is_none() {
             return Err(anyhow::anyhow!("Blob not found")); // InvalidRequestError('Blob not found')
         }
···
     pub size: u64,
     /// The MIME type of the blob.
     pub mime_type: Option<String>,
-    pub stream: BlobStream,
+    pub stream: BlobStream, // stream.Readable,
 }
 
 /// Options for listing blobs.

src/actor_store/blob/transactor.rs

···
 use uuid::Uuid;
 
 use super::BlobReader;
-use crate::repo::types::{PreparedBlobRef, PreparedWrite, WriteOpAction};
+use crate::repo::types::{BlobStore, PreparedBlobRef, PreparedWrite, WriteOpAction};
 
 /// Blob metadata for a newly uploaded blob.
 #[derive(Debug, Clone)]
···
 
     /// Process blobs for a repository write operation.
     pub(crate) async fn process_write_blobs(&self, writes: Vec<PreparedWrite>) -> Result<()> {
-        self.delete_dereferenced_blobs(writes)
+        self.delete_dereferenced_blobs(writes, false)
             .await
             .context("failed to delete dereferenced blobs")?;
         for write in writes {
-            if write.action == WriteOpAction::Create || write.action == WriteOpAction::Update {
-                for blob in &write.blobs {
+            if write.action() == &WriteOpAction::Create || write.action() == &WriteOpAction::Update
+            {
+                for blob in &write.blobs() {
                     self.verify_blob_and_make_permanent(blob)
                         .await
                         .context("failed to verify and make blob permanent")?;
-                    self.associate_blob(&blob.r#ref.0.to_string(), &write.uri)
+                    self.associate_blob(&blob.r#ref.0.to_string(), &write.uri())
                         .await
                         .context("failed to associate blob with record")?;
                 }

src/actor_store/db/migrations.rs

···
 //! Database migrations for the actor store.
 use anyhow::{Context as _, Result};
-use sqlx::SqlitePool;
+use sqlx::{Executor, SqlitePool};
 
 use super::ActorDb;
 
···
     /// Run all migrations
     pub(crate) async fn migrate_to_latest(&self) -> Result<()> {
         let past_migrations = sqlx::query!("SELECT name FROM actor_migration")
-            .fetch_all(&self.db.db)
+            .fetch_all(&self.db.pool)
             .await?;
         let mut past_migration_names = past_migrations
             .iter()
···
             let name = format!("{:p}", migration);
             if !past_migration_names.contains(&name) {
                 migration(&self.db)?;
+                let now = chrono::Utc::now().to_rfc3339();
                 sqlx::query!(
                     "INSERT INTO actor_migration (name, appliedAt) VALUES (?, ?)",
                     name,
-                    chrono::Utc::now().to_rfc3339()
+                    now,
                 )
-                .execute(&self.db.db)
+                .execute(&self.db.pool)
                 .await
                 .context("failed to insert migration record")?;
             }
···
 fn _001_init(db: &ActorDb) -> Result<()> {
     tokio::task::block_in_place(|| {
         tokio::runtime::Handle::current()
-            .block_on(create_tables(&db.db))
+            .block_on(create_tables(&db.pool))
             .context("failed to create initial tables")
     })
 }

src/actor_store/mod.rs

···
 mod actor_store_transactor;
 mod actor_store_writer;
 mod blob;
-// mod db;
+mod db;
 mod preference;
 mod record;
 mod repo;
···
 // pub(crate) use actor_store_reader::ActorStoreReader;
 pub(crate) use actor_store_transactor::ActorStoreTransactor;
 pub(crate) use actor_store_writer::ActorStoreWriter;
-// pub(crate) use db::ActorDb;
+pub(crate) use db::ActorDb;
 pub(crate) use resources::ActorStoreResources;

src/actor_store/preference/mod.rs

···
 //! Preference handling for actor store.
 
-// mod reader;
-// mod transactor;
+mod reader;
+mod transactor;
 
-// pub(crate) use reader::PreferenceReader;
-// pub(crate) use transactor::PreferenceTransactor;
+pub(crate) use reader::PreferenceReader;
+pub(crate) use transactor::PreferenceTransactor;

src/actor_store/preference/reader.rs

···
         scope: &str,
     ) -> Result<Vec<AccountPreference>> {
         let prefs_res = sqlx::query!("SELECT * FROM account_pref ORDER BY id")
-            .fetch_all(&self.db.db)
+            .fetch_all(&self.db.pool)
             .await?;
 
         let prefs = prefs_res

src/actor_store/preference/transactor.rs

···
         let mut tx = self
             .reader
             .db
-            .db
+            .pool
             .begin()
             .await
             .context("failed to begin transaction")?;

src/actor_store/record/mod.rs

···
 //! Record storage and retrieval for the actor store.
 
-// mod reader;
-// mod transactor;
+mod reader;
+mod transactor;
 
-// pub(crate) use reader::RecordReader;
-// pub(crate) use transactor::RecordTransactor;
+pub(crate) use reader::RecordReader;
+pub(crate) use transactor::RecordTransactor;

src/actor_store/record/reader.rs

···
 use anyhow::{Context as _, Result};
 use atrium_repo::Cid;
 use rsky_syntax::aturi::AtUri;
-use sqlx::{Row, SqlitePool};
+use sqlx::Row;
 use std::str::FromStr;
 
-use crate::actor_store::db::schema::Backlink;
+use crate::actor_store::{ActorDb, db::schema::Backlink};
 
 /// Reader for record data.
 pub(crate) struct RecordReader {
     /// Database connection.
-    pub db: SqlitePool,
-    /// DID of the repository owner.
-    pub did: String,
+    pub db: ActorDb,
 }
 
 /// Record descriptor containing URI, path, and CID.
···
 
 impl RecordReader {
     /// Create a new record reader.
-    pub(crate) fn new(db: SqlitePool, did: String) -> Self {
-        Self { db, did }
+    pub(crate) fn new(db: ActorDb) -> Self {
+        Self { db }
     }
 
     /// Count the total number of records.
     pub(crate) async fn record_count(&self) -> Result<i64> {
         let result = sqlx::query!(r#"SELECT COUNT(*) as count FROM record"#)
-            .fetch_one(&self.db)
+            .fetch_one(&self.db.pool)
             .await
             .context("failed to count records")?;
 
···
                 "SELECT uri, cid FROM record WHERE uri > ? ORDER BY uri ASC LIMIT 1000",
                 current_cursor
             )
-            .fetch_all(&self.db)
+            .fetch_all(&self.db.pool)
             .await
             .context("failed to fetch records")?;
 
···
     /// List all collections in the repository.
     pub(crate) async fn list_collections(&self) -> Result<Vec<String>> {
         let rows = sqlx::query!("SELECT collection FROM record GROUP BY collection")
-            .fetch_all(&self.db)
+            .fetch_all(&self.db.pool)
             .await
             .context("failed to list collections")?;
 
···
 
         let rows = query
             .build()
-            .fetch_all(&self.db)
+            .fetch_all(&self.db.pool)
             .await
             .context("failed to list records")?;
 
···
 
         let row = query
             .build()
-            .fetch_optional(&self.db)
+            .fetch_optional(&self.db.pool)
             .await
             .context("failed to fetch record")?;
 
···
 
         let result = query
             .build()
-            .fetch_optional(&self.db)
+            .fetch_optional(&self.db.pool)
             .await
             .context("failed to check record existence")?;
 
···
     /// Get the takedown status of a record.
     pub(crate) async fn get_record_takedown_status(&self, uri: &str) -> Result<Option<StatusAttr>> {
         let result = sqlx::query!("SELECT takedownRef FROM record WHERE uri = ?", uri)
-            .fetch_optional(&self.db)
+            .fetch_optional(&self.db.pool)
             .await
             .context("failed to fetch takedown status")?;
 
···
     /// Get the current CID for a record URI.
     pub(crate) async fn get_current_record_cid(&self, uri: &str) -> Result<Option<Cid>> {
         let result = sqlx::query!("SELECT cid FROM record WHERE uri = ?", uri)
-            .fetch_optional(&self.db)
+            .fetch_optional(&self.db.pool)
             .await
             .context("failed to fetch record CID")?;
 
···
             link_to,
             collection
         )
-        .fetch_all(&self.db)
+        .fetch_all(&self.db.pool)
         .await
         .context("failed to fetch record backlinks")?;
 
···
                 repo_rev: Some(row.repoRev),
                 indexed_at: row.indexedAt,
                 takedown_ref: row.takedownRef,
-                did: self.did.clone(),
             });
         }
 
···
                 backlink.link_to,
                 uri_collection
             )
-            .fetch_all(&self.db)
+            .fetch_all(&self.db.pool)
             .await
             .context("failed to fetch backlink conflicts")?;
 
···
                 "SELECT cid FROM repo_block WHERE cid > ? ORDER BY cid ASC LIMIT 1000",
                 current_cursor
             )
-            .fetch_all(&self.db)
+            .fetch_all(&self.db.pool)
             .await
             .context("failed to fetch blocks")?;
 
···
             LIMIT 1
             "#
         )
-        .fetch_optional(&self.db)
+        .fetch_optional(&self.db.pool)
         .await
         .context("failed to fetch profile record")?;
 
···
             r#"SELECT repoRev FROM record WHERE repoRev <= ? LIMIT 1"#,
             rev
         )
-        .fetch_optional(&self.db)
+        .fetch_optional(&self.db.pool)
         .await
         .context("failed to check revision existence")?;
 
···
             "#,
             rev
         )
-        .fetch_all(&self.db)
+        .fetch_all(&self.db.pool)
         .await
         .context("failed to fetch records since revision")?;
 
···
     pub indexed_at: String,
     /// Reference for takedown, if any.
     pub takedown_ref: Option<String>,
-    /// DID of the repository owner.
-    pub did: String,
 }
 
 /// Status attribute for takedowns

src/actor_store/record/transactor.rs

···
 use atrium_repo::Cid;
 use rsky_repo::types::WriteOpAction;
 use rsky_syntax::aturi::AtUri;
-use sqlx::SqlitePool;
 
+use crate::actor_store::ActorDb;
 use crate::actor_store::db::schema::Backlink;
 use crate::actor_store::record::reader::{RecordReader, StatusAttr, get_backlinks};
+use crate::repo::types::BlobStore as BlobStore;
 
 /// Transaction handler for record operations.
 pub(crate) struct RecordTransactor {
     /// The record reader.
     pub reader: RecordReader,
     /// The blob store.
-    pub blobstore: SqlitePool, // This will be replaced with proper BlobStore type
+    pub blobstore: BlobStore,
 }
 
 impl RecordTransactor {
     /// Create a new record transactor.
-    pub(crate) fn new(db: SqlitePool, blobstore: SqlitePool, did: String) -> Self {
+    pub(crate) fn new(db: ActorDb, blobstore: BlobStore) -> Self {
         Self {
-            reader: RecordReader::new(db, did),
+            reader: RecordReader::new(db),
             blobstore,
         }
     }
···
             repo_rev,
             now
         )
-        .execute(&self.reader.db)
+        .execute(&self.reader.db.pool)
         .await
         .context("failed to index record")?;
 
···
         tracing::debug!("Deleting indexed record {}", uri_str);
 
         // Delete the record and its backlinks in a transaction
-        let mut tx = self.reader.db.begin().await?;
+        let mut tx = self.reader.db.pool.begin().await?;
 
         // Delete from record table
         sqlx::query!("DELETE FROM record WHERE uri = ?", uri_str)
···
     /// Remove backlinks for a URI.
     pub(crate) async fn remove_backlinks_by_uri(&self, uri: &str) -> Result<()> {
         sqlx::query!("DELETE FROM backlink WHERE uri = ?", uri)
-            .execute(&self.reader.db)
+            .execute(&self.reader.db.pool)
             .await
             .context("failed to remove backlinks")?;
 
···
 
         query
             .build()
-            .execute(&self.reader.db)
+            .execute(&self.reader.db.pool)
             .await
             .context("failed to add backlinks")?;
 
···
             takedown_ref,
             uri_str
         )
-        .execute(&self.reader.db)
+        .execute(&self.reader.db.pool)
         .await
         .context("failed to update record takedown status")?;
 

src/db/db.rs

···
 #[derive(Clone)]
 pub(crate) struct Database {
     /// SQLite connection pool.
-    pub(crate) db: Arc<SqlitePool>,
+    pub(crate) pool: SqlitePool,
     /// Flag indicating if the database is destroyed.
-    destroyed: Arc<Mutex<bool>>,
+    pub(crate) destroyed: Arc<Mutex<bool>>,
     /// Queue of commit hooks.
-    commit_hooks: Arc<AsyncMutex<VecDeque<Box<dyn FnOnce() + Send>>>>,
+    pub(crate) commit_hooks: Arc<AsyncMutex<VecDeque<Box<dyn FnOnce() + Send>>>>,
 }
 
 impl Database {
···
 
         let pool = SqlitePool::connect_with(options).await?;
         Ok(Self {
-            db: Arc::new(pool),
+            pool,
             destroyed: Arc::new(Mutex::new(false)),
             commit_hooks: Arc::new(AsyncMutex::new(VecDeque::new())),
         })
···
 
     /// Ensures the database is using Write-Ahead Logging (WAL) mode.
     pub async fn ensure_wal(&self) -> sqlx::Result<()> {
-        let mut conn = self.db.acquire().await?;
+        let mut conn = self.pool.acquire().await?;
         sqlx::query("PRAGMA journal_mode = WAL")
             .execute(&mut *conn)
             .await?;
···
     where
         F: FnOnce(&mut Transaction<'_, Sqlite>) -> sqlx::Result<T>,
     {
-        let mut tx = self.db.begin().await?;
+        let mut tx = self.pool.begin().await?;
         let result = func(&mut tx)?;
         tx.commit().await?;
         self.run_commit_hooks().await;
···
             return Ok(());
         }
         *destroyed = true;
-        drop(self.db.clone()); // Drop the pool to close connections.
+        drop(self.pool.clone()); // Drop the pool to close connections.
         Ok(())
     }
 

src/repo/types.rs

···
     pub rkey: String,
     pub cid: Option<Cid>,
 }
+
+pub(crate) struct BlobStore;
+pub(crate) trait BlobStoreTrait {
+    fn put_temp(&self, bytes: &[u8]) -> Result<String>; // bytes: Uint8Array | stream.Readable
+    fn make_permanent(&self, key: &str, cid: Cid) -> Result<()>;
+    fn put_permanent(&self, cid: Cid, bytes: &[u8]) -> Result<()>;
+    fn quarantine(&self, cid: Cid) -> Result<()>;
+    fn unquarantine(&self, cid: Cid) -> Result<()>;
+    fn get_bytes(&self, cid: Cid) -> Result<Vec<u8>>;
+    fn get_stream(&self, cid: Cid) -> Result<Vec<u8>>; // Promise<stream.Readable>
+    fn has_temp(&self, key: &str) -> Result<bool>;
+    fn has_stored(&self, cid: Cid) -> Result<bool>;
+    fn delete(&self, cid: Cid) -> Result<()>;
+    fn delete_many(&self, cids: Vec<Cid>) -> Result<()>;
+}
+impl BlobStoreTrait for BlobStore {
+    fn put_temp(&self, bytes: &[u8]) -> Result<String> {
+        // Implementation here
+        Ok("temp_key".to_string())
+    }
+
+    fn make_permanent(&self, key: &str, cid: Cid) -> Result<()> {
+        // Implementation here
+        Ok(())
+    }
+
+    fn put_permanent(&self, cid: Cid, bytes: &[u8]) -> Result<()> {
+        // Implementation here
+        Ok(())
+    }
+
+    fn quarantine(&self, cid: Cid) -> Result<()> {
+        // Implementation here
+        Ok(())
+    }
+
+    fn unquarantine(&self, cid: Cid) -> Result<()> {
+        // Implementation here
+        Ok(())
+    }
+
+    fn get_bytes(&self, cid: Cid) -> Result<Vec<u8>> {
+        // Implementation here
+        Ok(vec![])
+    }
+
+    fn get_stream(&self, cid: Cid) -> Result<Vec<u8>> {
+        // Implementation here
+        Ok(vec![])
+    }
+
+    fn has_temp(&self, key: &str) -> Result<bool> {
+        // Implementation here
+        Ok(true)
+    }
+
+    fn has_stored(&self, cid: Cid) -> Result<bool> {
+        // Implementation here
+        Ok(true)
+    }
+    fn delete(&self, cid: Cid) -> Result<()> {
+        // Implementation here
+        Ok(())
+    }
+    fn delete_many(&self, cids: Vec<Cid>) -> Result<()> {
+        // Implementation here
+        Ok(())
+    }
+}