wip

src/main.rs

···
 use log::{error, info};
 use reqwest::header::{ACCEPT, ACCEPT_ENCODING};
 use anyhow;
-use tokio_util::compat::FuturesAsyncReadCompatExt;
+use tokio_util::compat::{FuturesAsyncReadCompatExt, TokioAsyncReadCompatExt};
+use tokio_util::io::StreamReader;
 use tokio_stream::wrappers::LinesStream;
+use tokio::io::AsyncWriteExt;
 use std::future::Future;
-use async_compression::futures::bufread::ZstdEncoder;
+use async_compression::futures::bufread::{ZstdEncoder, ZstdDecoder};
 use s3::{Bucket, Region};
 use s3::creds::Credentials;
 use futures::io::BufReader;
···
         /// The PDS base URL
         #[arg(long, value_name = "URL")]
         pds_url: String,
+    },
+
+    /// Import: download from S3, decompress zstd, and save locally
+    Import {
+        /// The DID to import
+        #[arg(long)]
+        did: String,
     },
 }
 
···
                     error!("Export failed: {}", err);
                 }
             }
-            // Placeholder for actual export logic.
-            // Implement your export functionality here.
-            // println!(
-            //     "Export would run here with did={} pds_url={} out_dir={}",
-            //     did,
-            //     pds_url,
-            //     out_dir.display()
-            // );
+        }
+        Commands::Import { did } => {
+            info!("Import requested: did={}", did);
+            match do_import(did).await {
+                Ok(path) => info!("Import completed, wrote {}", path.display()),
+                Err(err) => error!("Import failed: {}", err),
+            }
         }
     }
 }
···
         .await?;
 
     Ok(())
+}
 
-    
+async fn do_import(did: String) -> anyhow::Result<PathBuf> {
+    use futures::StreamExt;
+
+    let back_up_path = format!("users/{did}");
+
+    // S3 setup from env
+    let region_name = env::var("S3_REGION")?;
+    let endpoint = env::var("S3_ENDPOINT")?;
+    let region = Region::Custom { region: region_name, endpoint };
+
+    let bucket = Bucket::new(
+        env::var("S3_BUCKET_NAME")?.as_str(),
+        region,
+        Credentials::new(Some(env::var("S3_ACCESS_KEY")?.as_str()), Some(env::var("S3_SECRET_KEY")?.as_str()), None, None, None)?,
+    )?;
+
+    // Stream download from S3
+    let mut s3_stream = bucket
+        .get_object_stream(format!("/{back_up_path}/{did}.car.zst"))
+        .await?;
+
+    // Convert the stream of Bytes into a tokio AsyncRead
+    let byte_stream = s3_stream
+        .bytes()
+        .map(|res| res.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)));
+
+    let tokio_reader = StreamReader::new(byte_stream);
+
+    // Convert tokio AsyncRead -> futures AsyncRead, then buffer for decoder
+    let futures_reader = tokio_reader.compat();
+    let futures_buf = futures::io::BufReader::new(futures_reader);
+
+    // Zstd decode
+    let decoder = ZstdDecoder::new(futures_buf);
+
+    // Convert back to tokio AsyncRead for writing
+    let mut decoded_tokio_reader = decoder.compat();
+
+    // Prepare local output path, labeled as decompressed
+    let out_path: PathBuf = ["export", "users", did.as_str(), &format!("{}-decompressed.car", did)].iter().collect();
+    ensure_dir(&out_path)?;
+
+    let mut out_file = tokio::fs::File::create(&out_path).await?;
+
+    // Stream copy decoded content to file
+    tokio::io::copy(&mut decoded_tokio_reader, &mut out_file).await?;
+    out_file.flush().await?;
+
+    Ok(out_path)
 }