prototype actor_store

src/actor_store/blob.rs

···
-//! Blob storage and retrieval for the actor store.
-//! Based on https://github.com/blacksky-algorithms/rsky/blob/main/rsky-pds/src/actor_store/blob/mod.rs
-//! blacksky-algorithms/rsky is licensed under the Apache License 2.0
-//!
-//! Modified for SQLite backend
-
 use std::sync::Arc;
 
-use anyhow::{Error, Result, bail};
+use anyhow::{Result, bail};
 use cidv10::Cid;
 use diesel::dsl::{count_distinct, exists, not};
 use diesel::sql_types::{Integer, Nullable, Text};
 use diesel::*;
-use futures::stream::{self, StreamExt};
-use futures::try_join;
-use rsky_pds::actor_store::blob::sha256_stream;
+use futures::{
+    stream::{self, StreamExt},
+    try_join,
+};
 // use rocket::data::{Data, ToByteUnit};
-// use rocket::form::validate::Contains;
 use rsky_common::ipld::sha256_raw_to_cid;
 use rsky_common::now;
 use rsky_lexicon::blob_refs::BlobRef;
 use rsky_lexicon::com::atproto::admin::StatusAttr;
 use rsky_lexicon::com::atproto::repo::ListMissingBlobsRefRecordBlob;
 use rsky_pds::actor_store::blob::{
-    BlobMetadata, GetBlobMetadataOutput, GetBlobOutput, ListBlobsOpts, ListMissingBlobsOpts,
+    BlobMetadata, GetBlobMetadataOutput, ListBlobsOpts, ListMissingBlobsOpts, sha256_stream,
     verify_blob,
 };
 use rsky_pds::image;
 use rsky_pds::models::models;
 use rsky_repo::error::BlobError;
 use rsky_repo::types::{PreparedBlobRef, PreparedWrite};
-use sha2::Digest;
 
-use super::sql_blob::BlobStoreSql;
+use super::sql_blob::{BlobStoreSql, ByteStream};
 use crate::db::DbConn;
 
+pub struct GetBlobOutput {
+    pub size: i32,
+    pub mime_type: Option<String>,
+    pub stream: ByteStream,
+}
+
+/// Handles blob operations for an actor store
 pub struct BlobReader {
+    /// SQL-based blob storage
     pub blobstore: BlobStoreSql,
+    /// DID of the actor
     pub did: String,
+    /// Database connection
     pub db: Arc<DbConn>,
 }
 
-// Basically handles getting blob records from db
 impl BlobReader {
+    /// Create a new blob reader
     pub fn new(blobstore: BlobStoreSql, db: Arc<DbConn>) -> Self {
-        // BlobReader {
-        //     did: blobstore.bucket.clone(),
-        //     blobstore,
-        //     db,
-        // }
-        todo!();
+        BlobReader {
+            did: blobstore.did.clone(),
+            blobstore,
+            db,
+        }
     }
 
+    /// Get metadata for a blob by CID
     pub async fn get_blob_metadata(&self, cid: Cid) -> Result<GetBlobMetadataOutput> {
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
···
         }
     }
 
+    /// Get a blob by CID with metadata and content
     pub async fn get_blob(&self, cid: Cid) -> Result<GetBlobOutput> {
         let metadata = self.get_blob_metadata(cid).await?;
-        // let blob_stream = match self.blobstore.get_stream(cid).await {
-        //     Ok(res) => res,
-        //     Err(e) => {
-        //         return match e.downcast_ref() {
-        //             Some(GetObjectError::NoSuchKey(key)) => {
-        //                 Err(anyhow::Error::new(GetObjectError::NoSuchKey(key.clone())))
-        //             }
-        //             _ => bail!(e.to_string()),
-        //         };
-        //     }
-        // };
-        // Ok(GetBlobOutput {
-        //     size: metadata.size,
-        //     mime_type: metadata.mime_type,
-        //     stream: blob_stream,
-        // })
-        todo!();
+        let blob_stream = match self.blobstore.get_stream(cid).await {
+            Ok(stream) => stream,
+            Err(e) => bail!("Failed to get blob: {}", e),
+        };
+
+        Ok(GetBlobOutput {
+            size: metadata.size,
+            mime_type: metadata.mime_type,
+            stream: blob_stream,
+        })
     }
 
+    /// Get all records that reference a specific blob
     pub async fn get_records_for_blob(&self, cid: Cid) -> Result<Vec<String>> {
         use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
 
···
                     .filter(RecordBlobSchema::did.eq(did))
                     .select(models::RecordBlob::as_select())
                     .get_results(conn)?;
-                Ok::<_, Error>(results.into_iter().map(|row| row.record_uri))
+                Ok::<_, result::Error>(results.into_iter().map(|row| row.record_uri))
             })
             .await?
             .collect::<Vec<String>>();
···
         Ok(res)
     }
 
+    /// Upload a blob and get its metadata
     pub async fn upload_blob_and_get_metadata(
         &self,
         user_suggested_mime: String,
-        blob: Data<'_>, // Type representing the body data of a request.
+        blob: Vec<u8>,
     ) -> Result<BlobMetadata> {
-        todo!();
-        let blob_stream = blob.open(100.mebibytes());
-        let bytes = blob_stream.into_bytes().await?;
-        let size = bytes.n.written;
-        let bytes = bytes.into_inner();
+        let bytes = blob;
+        let size = bytes.len() as i64;
+
         let (temp_key, sha256, img_info, sniffed_mime) = try_join!(
             self.blobstore.put_temp(bytes.clone()),
             sha256_stream(bytes.clone()),
···
 
         Ok(BlobMetadata {
             temp_key,
-            size: size as i64,
+            size,
             cid,
             mime_type,
             width: if let Some(ref info) = img_info {
···
         })
     }
 
+    /// Track a blob that hasn't been associated with any records yet
     pub async fn track_untethered_blob(&self, metadata: BlobMetadata) -> Result<BlobRef> {
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
···
         ON CONFLICT (cid, did) DO UPDATE \
         SET \"tempKey\" = EXCLUDED.\"tempKey\" \
             WHERE pds.blob.\"tempKey\" is not null;");
+            #[expect(trivial_casts)]
             upsert
                 .bind::<Text, _>(&cid.to_string())
                 .bind::<Text, _>(&did)
···
         }).await
     }
 
+    /// Process blobs associated with writes
     pub async fn process_write_blobs(&self, writes: Vec<PreparedWrite>) -> Result<()> {
         self.delete_dereferenced_blobs(writes.clone()).await?;
+
         let _ = stream::iter(writes)
             .then(|write| async move {
                 Ok::<(), anyhow::Error>(match write {
···
             .await
             .into_iter()
             .collect::<Result<Vec<_>, _>>()?;
+
         Ok(())
     }
 
+    /// Delete blobs that are no longer referenced by any records
     pub async fn delete_dereferenced_blobs(&self, writes: Vec<PreparedWrite>) -> Result<()> {
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
         use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
···
                 _ => None,
             })
             .collect();
+
         if uris.is_empty() {
             return Ok(());
         }
···
             .await?
             .into_iter()
             .collect::<Vec<models::RecordBlob>>();
+
         if deleted_repo_blobs.is_empty() {
             return Ok(());
         }
···
             .into_iter()
             .flat_map(|v: Vec<PreparedBlobRef>| v.into_iter().map(|b| b.cid.to_string()))
             .collect();
+
         let mut cids_to_keep = Vec::new();
         cids_to_keep.append(&mut new_blob_cids);
         cids_to_keep.append(&mut duplicated_cids);
···
         let cids_to_delete = deleted_repo_blob_cids
             .into_iter()
             .filter_map(|cid: String| match cids_to_keep.contains(&cid) {
-                true => Some(cid),
-                false => None,
+                true => None,
+                false => Some(cid),
             })
             .collect::<Vec<String>>();
+
         if cids_to_delete.is_empty() {
             return Ok(());
         }
···
             })
             .await?;
 
-        // Original code queues a background job to delete by CID from S3 compatible blobstore
+        // Delete from blob storage
         let _ = stream::iter(cids_to_delete)
             .then(|cid| async { self.blobstore.delete(cid).await })
             .collect::<Vec<_>>()
             .await
             .into_iter()
             .collect::<Result<Vec<_>, _>>()?;
+
         Ok(())
     }
 
+    /// Verify a blob and make it permanent
     pub async fn verify_blob_and_make_permanent(&self, blob: PreparedBlobRef) -> Result<()> {
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
···
                     .optional()
             })
             .await?;
+
         if let Some(found) = found {
             verify_blob(&blob, &found).await?;
             if let Some(ref temp_key) = found.temp_key {
···
                 .await?;
             Ok(())
         } else {
-            bail!("Cound not find blob: {:?}", blob.cid.to_string())
+            bail!("Could not find blob: {:?}", blob.cid.to_string())
         }
     }
 
-    pub async fn associate_blob(&self, blob: PreparedBlobRef, _record_uri: String) -> Result<()> {
+    /// Associate a blob with a record
+    pub async fn associate_blob(&self, blob: PreparedBlobRef, record_uri: String) -> Result<()> {
         use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
 
         let cid = blob.cid.to_string();
-        let record_uri = _record_uri;
         let did = self.did.clone();
+
         self.db
             .run(move |conn| {
                 insert_into(RecordBlobSchema::record_blob)
···
                     .execute(conn)
             })
             .await?;
+
         Ok(())
     }
 
+    /// Count all blobs for this actor
     pub async fn blob_count(&self) -> Result<i64> {
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
···
             .await
     }
 
+    /// Count blobs associated with records
     pub async fn record_blob_count(&self) -> Result<i64> {
         use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
 
···
             .await
     }
 
+    /// List blobs that are referenced but missing
     pub async fn list_missing_blobs(
         &self,
         opts: ListMissingBlobsOpts,
···
             .await
     }
 
+    /// List all blobs with optional filtering
     pub async fn list_blobs(&self, opts: ListBlobsOpts) -> Result<Vec<String>> {
         use rsky_pds::schema::pds::record::dsl as RecordSchema;
         use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
+
         let ListBlobsOpts {
             since,
             cursor,
···
             }
             self.db.run(move |conn| builder.load(conn)).await?
         };
+
         Ok(res)
     }
 
+    /// Get the takedown status of a blob
     pub async fn get_blob_takedown_status(&self, cid: Cid) -> Result<Option<StatusAttr>> {
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
···
                     .select(models::Blob::as_select())
                     .first(conn)
                     .optional()?;
+
                 match res {
                     None => Ok(None),
                     Some(res) => match res.takedown_ref {
···
             .await
     }
 
-    // Transactors
-    // -------------------
-
+    /// Update the takedown status of a blob
     pub async fn update_blob_takedown_status(&self, blob: Cid, takedown: StatusAttr) -> Result<()> {
         use rsky_pds::schema::pds::blob::dsl as BlobSchema;
 
···
             false => None,
         };
 
-        let blob = self
-            .db
+        let blob_cid = blob.to_string();
+        let did_clone = self.did.clone();
+
+        self.db
             .run(move |conn| {
                 update(BlobSchema::blob)
-                    .filter(BlobSchema::cid.eq(blob.to_string()))
+                    .filter(BlobSchema::cid.eq(blob_cid))
+                    .filter(BlobSchema::did.eq(did_clone))
                     .set(BlobSchema::takedownRef.eq(takedown_ref))
                     .execute(conn)?;
-                Ok::<_, Error>(blob)
+                Ok::<_, result::Error>(blob)
             })
             .await?;
 
···
             true => self.blobstore.quarantine(blob).await,
             false => self.blobstore.unquarantine(blob).await,
         };
+
         match res {
             Ok(_) => Ok(()),
             Err(e) => match e.downcast_ref() {

src/actor_store/sql_blob.rs

···
+#![expect(
+    clippy::pub_use,
+    clippy::single_char_lifetime_names,
+    unused_qualifications
+)]
 use std::{path::PathBuf, str::FromStr as _, sync::Arc};
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 use cidv10::Cid;
+use diesel::*;
 use rsky_common::get_random_str;
+use tokio::fs;
 
 use crate::db::DbConn;
 
 /// Type for stream of blob data
 pub type BlobStream = Box<dyn std::io::Read + Send>;
 
-/// Placeholder implementation for blob store
+/// ByteStream implementation for blob data
+pub struct ByteStream {
+    pub bytes: Vec<u8>,
+}
+
+impl ByteStream {
+    pub fn new(bytes: Vec<u8>) -> Self {
+        Self { bytes }
+    }
+
+    pub async fn collect(self) -> Result<Vec<u8>> {
+        Ok(self.bytes)
+    }
+}
+
+/// SQL-based implementation of blob storage
 #[derive(Clone)]
-pub(crate) struct BlobStoreSql {
-    client: Arc<DbConn>,
-    path: PathBuf,
+pub struct BlobStoreSql {
+    /// Database connection for metadata
+    pub db: Arc<DbConn>,
+    /// DID of the actor
+    pub did: String,
+    /// Path for blob storage
+    pub path: PathBuf,
 }
 
 /// Configuration for the blob store
-/// TODO: Implement this placeholder
-pub(crate) struct BlobConfig {
-    pub(crate) path: PathBuf,
+pub struct BlobConfig {
+    /// Base path for blob storage
+    pub path: PathBuf,
 }
 
-/// ByteStream
-/// TODO: Implement this placeholder
-pub(crate) struct ByteStream {
-    pub(crate) bytes: Vec<u8>,
+/// Represents a move operation for blobs
+struct MoveObject {
+    from: String,
+    to: String,
 }
-impl ByteStream {
-    pub async fn collect(self) -> Result<Vec<u8>> {
-        Ok(self.bytes)
+
+/// Blob table structure for SQL operations
+#[derive(Queryable, Insertable, Debug)]
+#[diesel(table_name = blobs)]
+struct BlobEntry {
+    cid: String,
+    did: String,
+    path: String,
+    size: i32,
+    mime_type: String,
+    quarantined: bool,
+    temp: bool,
+}
+
+// Table definition for blobs
+table! {
+    blobs (cid, did) {
+        cid -> Text,
+        did -> Text,
+        path -> Text,
+        size -> Integer,
+        mime_type -> Text,
+        quarantined -> Bool,
+        temp -> Bool,
     }
 }
 
 impl BlobStoreSql {
-    pub fn new(did: String, cfg: &BlobConfig) -> Self {
-        // let client = aws_sdk_s3::Client::new(cfg);
-        // BlobStorePlaceholder {
-        //     client,
-        //     bucket: did,
-        // }
-        todo!();
+    /// Create a new SQL-based blob store for the given DID
+    pub fn new(did: String, cfg: &BlobConfig, db: Arc<DbConn>) -> Self {
+        let actor_path = cfg.path.join(&did);
+
+        // Create actor directory if it doesn't exist
+        if !actor_path.exists() {
+            // Use blocking to avoid complicating this constructor
+            std::fs::create_dir_all(&actor_path).unwrap_or_else(|_| {
+                panic!("Failed to create blob directory: {}", actor_path.display())
+            });
+        }
+
+        BlobStoreSql {
+            db,
+            did,
+            path: actor_path,
+        }
     }
 
-    pub fn creator(cfg: &BlobConfig) -> Box<dyn Fn(String) -> BlobStoreSql + '_> {
-        Box::new(move |did: String| BlobStoreSql::new(did, cfg))
+    /// Create a factory function for blob stores
+    pub fn creator(cfg: &BlobConfig, db: Arc<DbConn>) -> Box<dyn Fn(String) -> BlobStoreSql + '_> {
+        let db_clone = db.clone();
+        Box::new(move |did: String| BlobStoreSql::new(did, cfg, db_clone.clone()))
     }
 
+    /// Generate a random key for temporary blobs
     fn gen_key(&self) -> String {
         get_random_str()
     }
 
-    fn get_tmp_path(&self, key: &String) -> String {
-        // format!("tmp/{0}/{1}", self.bucket, key)
-        todo!();
+    /// Get the path for a temporary blob
+    fn get_tmp_path(&self, key: &str) -> PathBuf {
+        self.path.join("tmp").join(key)
     }
 
-    fn get_stored_path(&self, cid: Cid) -> String {
-        // format!("blocks/{0}/{1}", self.bucket, cid)
-        todo!();
+    /// Get the filesystem path for a stored blob
+    fn get_stored_path(&self, cid: &Cid) -> PathBuf {
+        self.path.join("blocks").join(cid.to_string())
     }
 
-    fn get_quarantined_path(&self, cid: Cid) -> String {
-        // format!("quarantine/{0}/{1}", self.bucket, cid)
-        todo!();
+    /// Get the filesystem path for a quarantined blob
+    fn get_quarantined_path(&self, cid: &Cid) -> PathBuf {
+        self.path.join("quarantine").join(cid.to_string())
     }
 
+    /// Store a blob temporarily
     pub async fn put_temp(&self, bytes: Vec<u8>) -> Result<String> {
         let key = self.gen_key();
-        // let body = ByteStream::from(bytes);
-        // self.client
-        //     .put_object()
-        //     .body(body)
-        //     .bucket(&self.bucket)
-        //     .key(self.get_tmp_path(&key))
-        //     .acl(ObjectCannedAcl::PublicRead)
-        //     .send()
-        //     .await?;
-        // Ok(key)
-        todo!();
+        let tmp_path = self.get_tmp_path(&key);
+
+        // Ensure the directory exists
+        if let Some(parent) = tmp_path.parent() {
+            fs::create_dir_all(parent)
+                .await
+                .context("Failed to create temp directory")?;
+        }
+
+        // Write the blob data to the file
+        fs::write(&tmp_path, &bytes)
+            .await
+            .context("Failed to write temporary blob")?;
+
+        // Clone values to be used in the closure
+        let did_clone = self.did.clone();
+        let tmp_path_str = tmp_path.to_string_lossy().to_string();
+        let bytes_len = bytes.len() as i32;
+
+        // Store metadata in the database (will be updated when made permanent)
+        self.db
+            .run(move |conn| {
+                let entry = BlobEntry {
+                    cid: "temp".to_string(), // Will be updated when made permanent
+                    did: did_clone,
+                    path: tmp_path_str,
+                    size: bytes_len,
+                    mime_type: "application/octet-stream".to_string(), // Will be updated when made permanent
+                    quarantined: false,
+                    temp: true,
+                };
+
+                diesel::insert_into(blobs::table)
+                    .values(&entry)
+                    .execute(conn)
+                    .context("Failed to insert temporary blob metadata")
+            })
+            .await?;
+
+        Ok(key)
     }
 
+    /// Make a temporary blob permanent
     pub async fn make_permanent(&self, key: String, cid: Cid) -> Result<()> {
-        // let already_has = self.has_stored(cid).await?;
-        // if !already_has {
-        //     Ok(self
-        //         .move_object(MoveObject {
-        //             from: self.get_tmp_path(&key),
-        //             to: self.get_stored_path(cid),
-        //         })
-        //         .await?)
-        // } else {
-        //     // already saved, so we no-op & just delete the temp
-        //     Ok(self.delete_key(self.get_tmp_path(&key)).await?)
-        // }
-        todo!();
+        let already_has = self.has_stored(cid).await?;
+        if !already_has {
+            let tmp_path = self.get_tmp_path(&key);
+            let stored_path = self.get_stored_path(&cid);
+
+            // Ensure parent directory exists
+            if let Some(parent) = stored_path.parent() {
+                fs::create_dir_all(parent)
+                    .await
+                    .context("Failed to create blocks directory")?;
+            }
+
+            // Read the bytes
+            let bytes = fs::read(&tmp_path)
+                .await
+                .context("Failed to read temporary blob")?;
+
+            // Write to permanent location
+            fs::write(&stored_path, &bytes)
+                .await
+                .context("Failed to write permanent blob")?;
+
+            // Update database metadata
+            let tmp_path_clone = tmp_path.clone();
+            self.db
+                .run(move |conn| {
+                    // Update the entry with the correct CID and path
+                    diesel::update(blobs::table)
+                        .filter(blobs::path.eq(tmp_path_clone.to_string_lossy().to_string()))
+                        .set((
+                            blobs::cid.eq(cid.to_string()),
+                            blobs::path.eq(stored_path.to_string_lossy().to_string()),
+                            blobs::temp.eq(false),
+                        ))
+                        .execute(conn)
+                        .context("Failed to update blob metadata")
+                })
+                .await?;
+
+            // Remove the temporary file
+            fs::remove_file(tmp_path)
+                .await
+                .context("Failed to remove temporary blob")?;
+
+            Ok(())
+        } else {
+            // Already saved, so delete the temp file
+            let tmp_path = self.get_tmp_path(&key);
+            if tmp_path.exists() {
+                fs::remove_file(tmp_path)
+                    .await
+                    .context("Failed to remove existing temporary blob")?;
+            }
+            Ok(())
+        }
     }
 
+    /// Store a blob directly as permanent
     pub async fn put_permanent(&self, cid: Cid, bytes: Vec<u8>) -> Result<()> {
-        // let body = ByteStream::from(bytes);
-        // self.client
-        //     .put_object()
-        //     .body(body)
-        //     .bucket(&self.bucket)
-        //     .key(self.get_stored_path(cid))
-        //     .acl(ObjectCannedAcl::PublicRead)
-        //     .send()
-        //     .await?;
-        // Ok(())
-        todo!();
+        let stored_path = self.get_stored_path(&cid);
+
+        // Ensure parent directory exists
+        if let Some(parent) = stored_path.parent() {
+            fs::create_dir_all(parent)
+                .await
+                .context("Failed to create blocks directory")?;
+        }
+
+        // Write to permanent location
+        fs::write(&stored_path, &bytes)
+            .await
+            .context("Failed to write permanent blob")?;
+
+        let stored_path_str = stored_path.to_string_lossy().to_string();
+        let cid_str = cid.to_string();
+        let did_clone = self.did.clone();
+        let bytes_len = bytes.len() as i32;
+
+        // Update database metadata
+        self.db
+            .run(move |conn| {
+                let entry = BlobEntry {
+                    cid: cid_str,
+                    did: did_clone,
+                    path: stored_path_str.clone(),
+                    size: bytes_len,
+                    mime_type: "application/octet-stream".to_string(), // Could be improved with MIME detection
+                    quarantined: false,
+                    temp: false,
+                };
+
+                diesel::insert_into(blobs::table)
+                    .values(&entry)
+                    .on_conflict((blobs::cid, blobs::did))
+                    .do_update()
+                    .set(blobs::path.eq(stored_path_str))
+                    .execute(conn)
+                    .context("Failed to insert permanent blob metadata")
+            })
+            .await?;
+
+        Ok(())
     }
 
+    /// Quarantine a blob
     pub async fn quarantine(&self, cid: Cid) -> Result<()> {
-        // self.move_object(MoveObject {
-        //     from: self.get_stored_path(cid),
-        //     to: self.get_quarantined_path(cid),
-        // })
-        // .await
-        todo!();
+        let stored_path = self.get_stored_path(&cid);
+        let quarantined_path = self.get_quarantined_path(&cid);
+
+        // Ensure parent directory exists
+        if let Some(parent) = quarantined_path.parent() {
+            fs::create_dir_all(parent)
+                .await
+                .context("Failed to create quarantine directory")?;
+        }
+
+        // Move the blob if it exists
+        if stored_path.exists() {
+            // Read the bytes
+            let bytes = fs::read(&stored_path)
+                .await
+                .context("Failed to read stored blob")?;
+
+            // Write to quarantine location
+            fs::write(&quarantined_path, &bytes)
+                .await
+                .context("Failed to write quarantined blob")?;
+
+            // Update database metadata
+            let cid_str = cid.to_string();
+            let did_clone = self.did.clone();
+            let quarantined_path_str = quarantined_path.to_string_lossy().to_string();
+
+            self.db
+                .run(move |conn| {
+                    diesel::update(blobs::table)
+                        .filter(blobs::cid.eq(cid_str))
+                        .filter(blobs::did.eq(did_clone))
+                        .set((
+                            blobs::path.eq(quarantined_path_str),
+                            blobs::quarantined.eq(true),
+                        ))
+                        .execute(conn)
+                        .context("Failed to update blob metadata for quarantine")
+                })
+                .await?;
+
+            // Remove the original file
+            fs::remove_file(stored_path)
+                .await
+                .context("Failed to remove quarantined blob")?;
+        }
+
+        Ok(())
     }
 
+    /// Unquarantine a blob
     pub async fn unquarantine(&self, cid: Cid) -> Result<()> {
-        // self.move_object(MoveObject {
-        //     from: self.get_quarantined_path(cid),
-        //     to: self.get_stored_path(cid),
-        // })
-        // .await
-        todo!();
+        let quarantined_path = self.get_quarantined_path(&cid);
+        let stored_path = self.get_stored_path(&cid);
+
+        // Ensure parent directory exists
+        if let Some(parent) = stored_path.parent() {
+            fs::create_dir_all(parent)
+                .await
+                .context("Failed to create blocks directory")?;
+        }
+
+        // Move the blob if it exists
+        if quarantined_path.exists() {
+            // Read the bytes
+            let bytes = fs::read(&quarantined_path)
+                .await
+                .context("Failed to read quarantined blob")?;
+
+            // Write to normal location
+            fs::write(&stored_path, &bytes)
+                .await
+                .context("Failed to write unquarantined blob")?;
+
+            // Update database metadata
+            let stored_path_str = stored_path.to_string_lossy().to_string();
+            let cid_str = cid.to_string();
+            let did_clone = self.did.clone();
+
+            self.db
+                .run(move |conn| {
+                    diesel::update(blobs::table)
+                        .filter(blobs::cid.eq(cid_str))
+                        .filter(blobs::did.eq(did_clone))
+                        .set((
+                            blobs::path.eq(stored_path_str),
+                            blobs::quarantined.eq(false),
+                        ))
+                        .execute(conn)
+                        .context("Failed to update blob metadata for unquarantine")
+                })
+                .await?;
+
+            // Remove the quarantined file
+            fs::remove_file(quarantined_path)
+                .await
+                .context("Failed to remove from quarantine")?;
+        }
+
+        Ok(())
     }
 
-    async fn get_object(&self, cid: Cid) -> Result<ByteStream> {
-        // let res = self
-        //     .client
-        //     .get_object()
-        //     .bucket(&self.bucket)
-        //     .key(self.get_stored_path(cid))
-        //     .send()
-        //     .await;
-        // match res {
-        //     Ok(res) => Ok(res.body),
-        //     Err(SdkError::ServiceError(s)) => Err(anyhow::Error::new(s.into_err())),
-        //     Err(e) => Err(anyhow::Error::new(e.into_service_error())),
-        // }
-        todo!();
+    /// Get a blob as a stream
+    pub async fn get_object(&self, cid_param: Cid) -> Result<ByteStream> {
+        use self::blobs::dsl::*;
+
+        // Get the blob path from the database
+        let cid_string = cid_param.to_string();
+        let did_clone = self.did.clone();
+
+        let blob_record = self
+            .db
+            .run(move |conn| {
+                blobs
+                    .filter(cid.eq(&cid_string))
+                    .filter(did.eq(&did_clone))
+                    .filter(quarantined.eq(false))
+                    .select(path)
+                    .first::<String>(conn)
+                    .optional()
+                    .context("Failed to query blob metadata")
+            })
+            .await?;
+
+        if let Some(blob_path) = blob_record {
+            // Read the blob data
+            let bytes = fs::read(blob_path)
+                .await
+                .context("Failed to read blob data")?;
+            Ok(ByteStream::new(bytes))
+        } else {
+            anyhow::bail!("Blob not found: {:?}", cid)
+        }
     }
 
+    /// Get blob bytes
     pub async fn get_bytes(&self, cid: Cid) -> Result<Vec<u8>> {
-        let res = self.get_object(cid).await?;
-        // let bytes = res.collect().await.map(|data| data.into_bytes())?;
-        // Ok(bytes.to_vec())
-        todo!();
+        let stream = self.get_object(cid).await?;
+        stream.collect().await
     }
 
+    /// Get a blob as a stream
     pub async fn get_stream(&self, cid: Cid) -> Result<ByteStream> {
         self.get_object(cid).await
     }
 
+    /// Delete a blob by CID string
     pub async fn delete(&self, cid: String) -> Result<()> {
-        self.delete_key(self.get_stored_path(Cid::from_str(&cid)?))
+        self.delete_key(self.get_stored_path(&Cid::from_str(&cid)?))
             .await
     }
 
+    /// Delete multiple blobs by CID
     pub async fn delete_many(&self, cids: Vec<Cid>) -> Result<()> {
-        let keys: Vec<String> = cids
-            .into_iter()
-            .map(|cid| self.get_stored_path(cid))
-            .collect();
-        self.delete_many_keys(keys).await
+        for cid in cids {
+            self.delete_key(self.get_stored_path(&cid)).await?;
+        }
+        Ok(())
     }
 
-    pub async fn has_stored(&self, cid: Cid) -> Result<bool> {
-        Ok(self.has_key(self.get_stored_path(cid)).await)
+    /// Check if a blob is stored
+    pub async fn has_stored(&self, cid_param: Cid) -> Result<bool> {
+        use self::blobs::dsl::*;
+
+        let cid_string = cid_param.to_string();
+        let did_clone = self.did.clone();
+
+        let exists = self
+            .db
+            .run(move |conn| {
+                diesel::select(diesel::dsl::exists(
+                    blobs
+                        .filter(cid.eq(&cid_string))
+                        .filter(did.eq(&did_clone))
+                        .filter(temp.eq(false)),
+                ))
+                .get_result::<bool>(conn)
+                .context("Failed to check if blob exists")
+            })
+            .await?;
+
+        Ok(exists)
     }
 
+    /// Check if a temporary blob exists
     pub async fn has_temp(&self, key: String) -> Result<bool> {
-        Ok(self.has_key(self.get_tmp_path(&key)).await)
+        let tmp_path = self.get_tmp_path(&key);
+        Ok(tmp_path.exists())
     }
 
-    async fn has_key(&self, key: String) -> bool {
-        // let res = self
-        //     .client
-        //     .head_object()
-        //     .bucket(&self.bucket)
-        //     .key(key)
-        //     .send()
-        //     .await;
-        // res.is_ok()
-        todo!();
+    /// Check if a blob exists by key
+    async fn has_key(&self, key_path: PathBuf) -> bool {
+        key_path.exists()
     }
 
-    async fn delete_key(&self, key: String) -> Result<()> {
-        // self.client
-        //     .delete_object()
-        //     .bucket(&self.bucket)
-        //     .key(key)
-        //     .send()
-        //     .await?;
-        // Ok(())
-        todo!();
+    /// Delete a blob by its key path
+    async fn delete_key(&self, key_path: PathBuf) -> Result<()> {
+        use self::blobs::dsl::*;
+
+        // Delete from database first
+        let key_path_clone = key_path.clone();
+        self.db
+            .run(move |conn| {
+                diesel::delete(blobs)
+                    .filter(path.eq(key_path_clone.to_string_lossy().to_string()))
+                    .execute(conn)
+                    .context("Failed to delete blob metadata")
+            })
+            .await?;
+
+        // Then delete the file if it exists
+        if key_path.exists() {
+            fs::remove_file(key_path)
+                .await
+                .context("Failed to delete blob file")?;
+        }
+
+        Ok(())
     }
 
+    /// Delete multiple blobs by key path
     async fn delete_many_keys(&self, keys: Vec<String>) -> Result<()> {
-        // let objects: Vec<ObjectIdentifier> = keys
-        //     .into_iter()
-        //     .map(|key| Ok(ObjectIdentifier::builder().key(key).build()?))
-        //     .collect::<Result<Vec<ObjectIdentifier>>>()?;
-        // let deletes = Delete::builder().set_objects(Some(objects)).build()?;
-        // self.client
-        //     .delete_objects()
-        //     .bucket(&self.bucket)
-        //     .delete(deletes)
-        //     .send()
-        //     .await?;
-        // Ok(())
-        todo!();
+        for key in keys {
+            self.delete_key(PathBuf::from(key)).await?;
+        }
+        Ok(())
     }
 
+    /// Move a blob from one location to another
     async fn move_object(&self, keys: MoveObject) -> Result<()> {
-        // self.client
-        //     .copy_object()
-        //     .bucket(&self.bucket)
-        //     .copy_source(format!(
-        //         "{0}/{1}/{2}",
-        //         env_str("AWS_ENDPOINT_BUCKET").unwrap(),
-        //         self.bucket,
-        //         keys.from
-        //     ))
-        //     .key(keys.to)
-        //     .acl(ObjectCannedAcl::PublicRead)
-        //     .send()
-        //     .await?;
-        // self.client
-        //     .delete_object()
-        //     .bucket(&self.bucket)
-        //     .key(keys.from)
-        //     .send()
-        //     .await?;
-        // Ok(())
-        todo!();
+        let from_path = PathBuf::from(&keys.from);
+        let to_path = PathBuf::from(&keys.to);
+
+        // Ensure parent directory exists
+        if let Some(parent) = to_path.parent() {
+            fs::create_dir_all(parent)
+                .await
+                .context("Failed to create directory")?;
+        }
+
+        // Only move if the source exists
+        if from_path.exists() {
+            // Read the data
+            let data = fs::read(&from_path)
+                .await
+                .context("Failed to read source blob")?;
+
+            // Write to the destination
+            fs::write(&to_path, data)
+                .await
+                .context("Failed to write destination blob")?;
+
+            // Update the database record
+            let from_path_clone = from_path.clone();
+            self.db
+                .run(move |conn| {
+                    diesel::update(blobs::table)
+                        .filter(blobs::path.eq(from_path_clone.to_string_lossy().to_string()))
+                        .set(blobs::path.eq(to_path.to_string_lossy().to_string()))
+                        .execute(conn)
+                        .context("Failed to update blob path")
+                })
+                .await?;
+
+            // Delete the source file
+            fs::remove_file(from_path)
+                .await
+                .context("Failed to remove source blob")?;
+        }
+
+        Ok(())
     }
 }
-
-struct MoveObject {
-    from: String,
-    to: String,
-}

src/actor_store/sql_repo.rs

···
 }
 
 impl ReadableBlockstore for SqlRepoReader {
-    fn get_bytes<'a>(
-        &'a self,
-        cid: &'a Cid,
-    ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>>> + Send + Sync + 'a>> {
+    fn get_bytes<'life>(
+        &'life self,
+        cid: &'life Cid,
+    ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>>> + Send + Sync + 'life>> {
         let did: String = self.did.clone();
         let db: Arc<DbConn> = self.db.clone();
         let cid = cid.clone();
···
         })
     }
 
-    fn has<'a>(
-        &'a self,
+    fn has<'life>(
+        &'life self,
         cid: Cid,
-    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + Sync + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + Sync + 'life>> {
         Box::pin(async move {
             let got = <Self as ReadableBlockstore>::get_bytes(self, &cid).await?;
             Ok(got.is_some())
         })
     }
 
-    fn get_blocks<'a>(
-        &'a self,
+    fn get_blocks<'life>(
+        &'life self,
         cids: Vec<Cid>,
-    ) -> Pin<Box<dyn Future<Output = Result<BlocksAndMissing>> + Send + Sync + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Result<BlocksAndMissing>> + Send + Sync + 'life>> {
         let did: String = self.did.clone();
         let db: Arc<DbConn> = self.db.clone();
 
···
 }
 
 impl RepoStorage for SqlRepoReader {
-    fn get_root<'a>(&'a self) -> Pin<Box<dyn Future<Output = Option<Cid>> + Send + Sync + 'a>> {
+    fn get_root<'life>(
+        &'life self,
+    ) -> Pin<Box<dyn Future<Output = Option<Cid>> + Send + Sync + 'life>> {
         Box::pin(async move {
             match self.get_root_detailed().await {
                 Ok(root) => Some(root.cid),
···
         })
     }
 
-    fn put_block<'a>(
-        &'a self,
+    fn put_block<'life>(
+        &'life self,
         cid: Cid,
         bytes: Vec<u8>,
         rev: String,
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'life>> {
         let did: String = self.did.clone();
         let db: Arc<DbConn> = self.db.clone();
         let bytes_cloned = bytes.clone();
···
         })
     }
 
-    fn put_many<'a>(
-        &'a self,
+    fn put_many<'life>(
+        &'life self,
         to_put: BlockMap,
         rev: String,
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'life>> {
         let did: String = self.did.clone();
         let db: Arc<DbConn> = self.db.clone();
 
···
             Ok(())
         })
     }
-    fn update_root<'a>(
-        &'a self,
+    fn update_root<'life>(
+        &'life self,
         cid: Cid,
         rev: String,
         is_create: Option<bool>,
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'life>> {
         let did: String = self.did.clone();
         let db: Arc<DbConn> = self.db.clone();
         let now: String = self.now.clone();
···
         })
     }
 
-    fn apply_commit<'a>(
-        &'a self,
+    fn apply_commit<'life>(
+        &'life self,
         commit: CommitData,
         is_create: Option<bool>,
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'life>> {
         Box::pin(async move {
             self.update_root(commit.cid, commit.rev.clone(), is_create)
                 .await?;

src/endpoints/mod.rs

···
 //! Root module for all endpoints.
 mod identity;
-// mod repo;
+mod repo;
 mod server;
 mod sync;
 
···
     Router::new()
         .route("/_health", get(health))
         .merge(identity::routes()) // com.atproto.identity
-        // .merge(repo::routes()) // com.atproto.repo
+        .merge(repo::routes()) // com.atproto.repo
         .merge(server::routes()) // com.atproto.server
         .merge(sync::routes()) // com.atproto.sync
 }