refactor: run database encryption as an operational worker

Dockerfile

@@ -40,7 +40,7 @@ COPY crates/ ./crates/
 COPY .cargo/ ./.cargo/
 
 # Build the application in release mode
-RUN cargo build --release --locked --bin api
+RUN cargo build --release --locked --bin api --bin database_encryption_worker
 
 
 # Runtime stage
@@ -86,6 +86,7 @@ WORKDIR /app
 
 # Copy the built binary
 COPY --from=builder /app/target/release/api /app/api
+COPY --from=builder /app/target/release/database_encryption_worker /app/database_encryption_worker
 
 # Copy the migration SQL files
 RUN mkdir -p /app/crates/database/src/migrations/sql

crates/api/Cargo.toml

@@ -16,7 +16,9 @@ tokio-stream = "0.1"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 tokio = { version = "1", features = ["full"] }
+tokio-postgres = "0.7.18"
 anyhow = "1.0"
+clap = { version = "4.5", features = ["derive"] }
 config = { path = "../config" }
 database = { path = "../database" }
 services = { path = "../services" }
@@ -73,7 +75,6 @@ dotenvy = "0.15.7"
 k256 = { version = "0.13", features = ["ecdsa", "arithmetic"] }
 sha3 = "0.12"
 hmac = "0.13"
-tokio-postgres = "0.7.18"
 deadpool-postgres = { version = "0.14", features = ["rt_tokio_1"] }
 ed25519-dalek = { version = "2.1", features = ["rand_core"] }
 rand = "0.10"

crates/api/src/bin/database_encryption_worker.rs

@@ -0,0 +1,107 @@
+use anyhow::{Context, Result};
+use api::database_encryption::{
+    operational_migrate, operational_scan, operational_verify, DatabaseEncryptionState,
+};
+use clap::{Parser, Subcommand};
+use database::Database;
+use uuid::Uuid;
+
+#[derive(Parser)]
+#[command(about = "One-off database encryption backfill worker")]
+struct Cli {
+    #[command(subcommand)]
+    command: Command,
+}
+
+#[derive(Subcommand)]
+enum Command {
+    Scan {
+        #[arg(long, value_delimiter = ',')]
+        scope: Vec<String>,
+    },
+    Migrate {
+        #[arg(long, required = true, value_delimiter = ',')]
+        scope: Vec<String>,
+        #[arg(long, default_value_t = 500)]
+        batch_size: i64,
+        #[arg(long)]
+        max_rows: Option<i64>,
+        #[arg(long)]
+        resume: Option<Uuid>,
+        #[arg(long)]
+        operator: String,
+    },
+    Verify {
+        #[arg(long, value_delimiter = ',')]
+        scope: Vec<String>,
+    },
+}
+
+#[tokio::main]
+async fn main() {
+    if run().await.is_err() {
+        println!(
+            "DATABASE_ENCRYPTION_WORKER_RESULT {}",
+            serde_json::json!({"status":"failed","error_class":"worker_failed"})
+        );
+        std::process::exit(1);
+    }
+}
+
+async fn run() -> Result<()> {
+    let cli = Cli::parse();
+    let database_config = config::DatabaseConfig::from_env()
+        .map_err(anyhow::Error::msg)
+        .context("invalid database configuration")?;
+    let key = read_encryption_key()?;
+    let database = Database::from_config(&database_config).await?;
+    let state = DatabaseEncryptionState::new(database.pool().clone(), &key)?;
+
+    match cli.command {
+        Command::Scan { scope } => {
+            println!(
+                "{}",
+                serde_json::to_string_pretty(&operational_scan(&state, scope).await?)?
+            );
+            print_success(None);
+        }
+        Command::Verify { scope } => {
+            let report = operational_verify(&state, scope).await?;
+            println!("{}", serde_json::to_string_pretty(&report)?);
+            if report["pass"] != true {
+                anyhow::bail!("verification found plaintext or invalid envelopes");
+            }
+            print_success(None);
+        }
+        Command::Migrate {
+            scope,
+            batch_size,
+            max_rows,
+            resume,
+            operator,
+        } => {
+            let id =
+                operational_migrate(&state, scope, batch_size, max_rows, resume, &operator).await?;
+            print_success(Some(id));
+        }
+    }
+    Ok(())
+}
+
+fn print_success(job_id: Option<Uuid>) {
+    println!(
+        "DATABASE_ENCRYPTION_WORKER_RESULT {}",
+        serde_json::json!({"status":"completed","job_id":job_id})
+    );
+}
+
+fn read_encryption_key() -> Result<String> {
+    if let Ok(key) = std::env::var("S3_ENCRYPTION_KEY") {
+        return Ok(key);
+    }
+    let path = std::env::var("S3_ENCRYPTION_KEY_FILE")
+        .context("S3_ENCRYPTION_KEY or S3_ENCRYPTION_KEY_FILE is required")?;
+    std::fs::read_to_string(path)
+        .context("failed to read S3_ENCRYPTION_KEY_FILE")
+        .map(|key| key.trim().to_string())
+}