wip

Cargo.lock

···
  "lexicon_types_crate",
  "log",
  "reqwest",
+ "rust-s3",
+ "shared",
+ "sqlx",
  "tokio",
 ]
 

admin_cli/Cargo.toml

···
 jacquard.workspace = true
 lexicon_types_crate.workspace = true
 reqwest.workspace = true
-base64 = "0.22"
+base64 = "0.22"
+shared = { path = "../shared" }
+sqlx = { version = "0.8.6", features = ["runtime-tokio", "postgres"] }
+rust-s3.workspace = true

admin_cli/src/main.rs

···
 use log;
 use reqwest::header;
 use reqwest::header::{HeaderMap, HeaderValue};
+use s3::creds::Credentials;
+use s3::{Bucket, Region};
+use sqlx::PgPool;
 use std::env;
 
 fn init_logging() {
···
     },
     /// Trigger an instance-wide backup job (no parameters)
     RequestInstanceBackup,
+    /// Verify all backups in S3 against the database
+    VerifyBackups,
 }
 
 #[derive(Debug, Subcommand)]
···
                 }
                 Err(err) => {
                     log::error!("Instance backup request failed: {}", err);
+                }
+            }
+        }
+        Commands::VerifyBackups => {
+            log::info!("Verifying backups in S3...");
+
+            // Get database URL from environment
+            let database_url =
+                env::var("DATABASE_URL").context("DATABASE_URL environment variable not set")?;
+
+            // Connect to database
+            let pool = PgPool::connect(&database_url)
+                .await
+                .context("Failed to connect to database")?;
+
+            // Setup S3 client
+            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,
+                )?,
+            )?;
+
+            // Call the verify_backups function
+            match shared::jobs::verify_backups::verify_backups(&pool, &bucket).await {
+                Ok(missing_blobs) => {
+                    if missing_blobs.is_empty() {
+                        log::info!("✓ All backups verified successfully! No missing blobs found.");
+                    } else {
+                        log::error!("✗ Found {} missing blobs:", missing_blobs.len());
+                        for missing in &missing_blobs {
+                            println!(
+                                "Missing: DID={}, CID/REV={}, TYPE={:?}, PATH={}",
+                                missing.did, missing.cid_or_rev, missing.blob_type, missing.s3_path
+                            );
+                        }
+                    }
+                }
+                Err(err) => {
+                    log::error!("Failed to verify backups: {}", err);
                 }
             }
         }

shared/src/db/models.rs

···
     pub created_at: DateTime<Utc>,
     pub account_did: String,
     pub size: i64,
-    pub blob_type: BlobType,
+    pub r#type: BlobType,
     pub cid_or_rev: String,
 }
 

shared/src/jobs/mod.rs

···
 pub mod scheduled_back_up_start;
 pub mod start_all_backup;
 pub mod upload_blob;
+pub mod verify_backups;
 
 use crate::db::models;
 use crate::db::models::BlobModel;

shared/src/jobs/verify_backups.rs

···
+use crate::db::models::{BlobModel, BlobType};
+use crate::storage::{blob_backup_path, repo_backup_path};
+use anyhow::Result;
+use s3::Bucket;
+use sqlx::{Pool, Postgres};
+
+/// Verifies that all blobs in the database exist in S3.
+/// Returns a Vec of missing blob information (did, cid_or_rev, blob_type).
+pub async fn verify_backups(pool: &Pool<Postgres>, s3_bucket: &Bucket) -> Result<Vec<MissingBlob>> {
+    // Get all blobs from the database
+    let blobs = sqlx::query_as::<_, BlobModel>("SELECT * FROM blobs ORDER BY created_at")
+        .fetch_all(pool)
+        .await?;
+
+    let total_blobs = blobs.len();
+    log::info!("Checking {} blobs in S3...", total_blobs);
+
+    let mut missing_blobs = Vec::new();
+    let mut checked = 0;
+
+    for blob in blobs {
+        checked += 1;
+        if checked % 100 == 0 {
+            log::info!("Checked {}/{} blobs...", checked, total_blobs);
+        }
+
+        let s3_path = match blob.r#type {
+            BlobType::Repo => repo_backup_path(blob.account_did.clone()),
+            BlobType::Blob => blob_backup_path(blob.account_did.clone(), blob.cid_or_rev.clone()),
+            BlobType::Prefs => {
+                // Handle prefs if needed - for now skip
+                log::debug!("Skipping prefs blob: {:?}", blob);
+                continue;
+            }
+        };
+
+        // Check if the object exists in S3
+        match s3_bucket.head_object(&s3_path).await {
+            Ok(_) => {
+                // Object exists, all good
+                log::debug!("✓ Found: {}", s3_path);
+            }
+            Err(e) => {
+                // Check if it's a 404 error (not found)
+                if e.to_string().contains("404") {
+                    log::warn!("✗ Missing: {}", s3_path);
+                    missing_blobs.push(MissingBlob {
+                        did: blob.account_did.clone(),
+                        cid_or_rev: blob.cid_or_rev.clone(),
+                        blob_type: blob.r#type.clone(),
+                        s3_path,
+                    });
+                } else {
+                    // Some other error - log it but don't count as missing
+                    log::error!("Error checking {}: {}", s3_path, e);
+                }
+            }
+        }
+    }
+
+    log::info!(
+        "Verification complete. Checked {} blobs, found {} missing.",
+        checked,
+        missing_blobs.len()
+    );
+
+    Ok(missing_blobs)
+}
+
+#[derive(Debug, Clone)]
+pub struct MissingBlob {
+    pub did: String,
+    pub cid_or_rev: String,
+    pub blob_type: BlobType,
+    pub s3_path: String,
+}