blobs proof of concept

.gitignore

···
 /target
 .idea
-.env
+.env
+.DS_Store

Cargo.lock

···
  "log",
  "reqwest",
  "rust-s3",
+ "serde",
+ "serde_json",
  "tokio",
  "tokio-stream",
  "tokio-util",

Cargo.toml

···
 tokio-util = { version = "0.7.16", features = ["compat"] }
 tokio-stream = { version = "0.1.17", features = ["io-util"] }
 rust-s3 = "0.37.0"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"

src/main.rs

···
 use tokio_stream::wrappers::LinesStream;
 use tokio_util::compat::{FuturesAsyncReadCompatExt, TokioAsyncReadCompatExt};
 use tokio_util::io::StreamReader;
+use serde::Deserialize;
 
 #[derive(Parser, Debug)]
 #[command(author, version, about = "PDS Whatsit Compress Test CLI", long_about = None)]
···
 #[derive(Subcommand, Debug)]
 enum Commands {
     /// Export data from a PDS instance
-    Export {
+    Backup {
         /// Output directory path (defaults to ./export)
         // #[arg(short, long)]
         // out: Option<PathBuf>,
···
     },
 
     /// Import: download from S3, decompress zstd, and save locally
-    Import {
+    Restore {
         /// The DID to import
         #[arg(long)]
         did: String,
···
     )?;
 
     match cli.command {
-        Commands::Export { did, pds_url } => {
+        Commands::Backup { did, pds_url } => {
             info!("Export requested: did={}, pds_url={}", did, pds_url,);
 
-            match do_work(pds_url, did, bucket).await {
+            match do_backup(pds_url, did, bucket).await {
                 Ok(_) => {
                     info!("Export completed");
                     Ok(())
···
                 }
             }
         }
-        Commands::Import { did } => {
+        Commands::Restore { did } => {
             info!("Import requested: did={}", did);
-            match do_import(did, bucket).await {
+            match do_restore(did, bucket).await {
                 Ok(path) => {
                     info!("Import completed, wrote {}", path.display());
                     Ok(())
···
     }
 }
 
-async fn do_work(pds_url: String, did: String, bucket: Box<Bucket>) -> anyhow::Result<()> {
+async fn do_backup(pds_url: String, did: String, bucket: Box<Bucket>) -> anyhow::Result<()> {
     use futures::TryStreamExt;
     let atproto_client = reqwest::Client::new();
     let back_up_path = format!("users/{did}");
 
+    // 1) Backup the full repo CAR (compressed)
     let response_reader = atproto_client
         .get(format!("{pds_url}/xrpc/com.atproto.sync.getRepo?did={did}"))
         .header(ACCEPT, "application/vnd.ipld.car")
···
         .map_err(|e| futures::io::Error::new(futures::io::ErrorKind::Other, e))
         .into_async_read();
 
-    // Wrap the response in a buffered reader to satisfy AsyncBufRead for the encoder
     let buf_reader = BufReader::new(response_reader);
-
-    // Create a Zstd encoder that compresses on-the-fly while being read
     let zstd_encoder = ZstdEncoder::new(buf_reader);
-
-    // Convert futures::AsyncRead to tokio::io::AsyncRead for the S3 client
     let mut zstd_tokio_reader = zstd_encoder.compat();
 
-    // Stream the compressed data to S3 with appropriate headers
     bucket
         .put_object_stream_builder(format!("/{back_up_path}/{did}.car.zst").as_str())
         .with_content_type("application/vnd.ipld.car")
···
         .execute_stream(&mut zstd_tokio_reader)
         .await?;
 
+    // 2) Paginate listBlobs and upload each as zstd-compressed
+    #[derive(Deserialize)]
+    struct ListBlobsResponse {
+        #[allow(dead_code)]
+        cursor: Option<String>,
+        cids: Vec<String>,
+    }
+
+    let mut cursor: Option<String> = None;
+    let limit = 1000u32;
+
+    loop {
+        let mut url = format!(
+            "{}/xrpc/com.atproto.sync.listBlobs?did={}&limit={}",
+            pds_url, did, limit
+        );
+        if let Some(ref c) = cursor {
+            if !c.is_empty() {
+                url.push_str("&cursor=");
+                url.push_str(c);
+            }
+        }
+
+        info!("Listing blobs: {}", url);
+        let resp = atproto_client
+            .get(url)
+            .header(ACCEPT, "application/json")
+            .send()
+            .await?
+            .error_for_status()?;
+        let bytes = resp.bytes().await?;
+        let page: ListBlobsResponse = serde_json::from_slice(&bytes)?;
+
+        if page.cids.is_empty() {
+            if cursor.is_none() || cursor.as_deref() == Some("") {
+                break;
+            }
+        }
+
+        for cid in page.cids {
+            let blob_url = format!(
+                "{}/xrpc/com.atproto.sync.getBlob?did={}&cid={}",
+                pds_url, did, cid
+            );
+            info!("Downloading blob {}", cid);
+            let blob_reader = atproto_client
+                .get(blob_url)
+                .header(ACCEPT, "*/*")
+                .send()
+                .await?
+                .error_for_status()?
+                .bytes_stream()
+                .map_err(|e| futures::io::Error::new(futures::io::ErrorKind::Other, e))
+                .into_async_read();
+
+            let blob_buf = BufReader::new(blob_reader);
+            let blob_zstd = ZstdEncoder::new(blob_buf);
+            let mut blob_tokio_reader = blob_zstd.compat();
+
+            let object_key = format!("/{}/blobs/{}.zst", back_up_path, cid);
+            bucket
+                .put_object_stream_builder(&object_key)
+                .with_content_type("application/octet-stream")
+                .with_content_encoding("zstd")?
+                .execute_stream(&mut blob_tokio_reader)
+                .await?;
+        }
+
+        // Update or finish based on cursor
+        match page.cursor {
+            Some(c) if !c.is_empty() => {
+                cursor = Some(c);
+            }
+            _ => break,
+        }
+    }
+
     Ok(())
 }
 
-async fn do_import(did: String, bucket: Box<Bucket>) -> anyhow::Result<PathBuf> {
+async fn do_restore(did: String, bucket: Box<Bucket>) -> anyhow::Result<PathBuf> {
     use futures::StreamExt;
 
     let back_up_path = format!("users/{did}");