owpurge 0.2.0

//! Purge several agents from one host in a single invocation.
//!
//! Each agent is a separate wallet with its own `.env`. `batch` runs `owpurge <command>`
//! once per env file, one after another, as a child process with a cleared environment.
//! A child process is the point: configuration is read from process-wide environment
//! variables, so running agents in the same process would let one agent's TENANT_ID or
//! DRY_RUN leak into the next. A fresh process per agent cannot.
//!
//! Each agent keeps its own safety posture — DRY_RUN, TTL_DAYS, BULK_CONFIRM and the
//! rest all come from that agent's env file, so a batch of one dry-run agent and one live
//! agent does exactly that. A failure on one agent does not stop the others (use
//! `--stop-on-error` to change that); the exit status is non-zero if any agent failed.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Instant;

use anyhow::{bail, Context, Result};

/// Subcommands that make sense to repeat per agent. `env-apply` and `batch` itself are
/// deliberately absent.
pub const RUNNABLE: &[&str] = &[
    "purge",
    "census",
    "bulk-purge",
    "orphan-sweep",
    "credentials",
    "proofs",
    "oob",
    "basic-messages",
    "question-answer",
    "tenants",
];

pub struct Args<'a> {
    pub envs: &'a [PathBuf],
    pub dir: Option<&'a Path>,
    pub run: &'a str,
    pub stop_on_error: bool,
    pub server_side_delete: bool,
    pub no_audit: bool,
}

/// Every `*.env` file directly inside `dir`, sorted so the run order is stable.
fn env_files_in(dir: &Path) -> Result<Vec<PathBuf>> {
    let mut files: Vec<PathBuf> = std::fs::read_dir(dir)
        .with_context(|| format!("could not read {}", dir.display()))?
        .filter_map(|e| e.ok().map(|e| e.path()))
        .filter(|p| p.is_file() && p.extension().is_some_and(|x| x == "env"))
        .collect();
    files.sort();
    Ok(files)
}

/// WALLET_ID and DB_HOST from an env file, without touching the process environment.
fn identity(path: &Path) -> Result<(String, String)> {
    let vars: HashMap<String, String> = dotenvy::from_path_iter(path)
        .with_context(|| format!("could not read env file {}", path.display()))?
        .collect::<std::result::Result<_, _>>()
        .with_context(|| format!("could not parse env file {}", path.display()))?;
    let get = |k: &str| vars.get(k).cloned().filter(|v| !v.is_empty());
    let Some(wallet) = get("WALLET_ID") else {
        bail!("{} has no WALLET_ID", path.display());
    };
    Ok((wallet, get("DB_HOST").unwrap_or_default()))
}

struct Outcome {
    file: PathBuf,
    wallet: String,
    status: String,
    secs: u64,
    ok: bool,
}

pub fn run(args: Args<'_>) -> Result<std::process::ExitCode> {
    if !RUNNABLE.contains(&args.run) {
        bail!("`batch --run {}` is not supported; choose one of: {}", args.run, RUNNABLE.join(", "));
    }

    let mut files: Vec<PathBuf> = args.envs.to_vec();
    if let Some(dir) = args.dir {
        files.extend(env_files_in(dir)?);
    }
    if files.is_empty() {
        bail!("no env files: pass --env <FILE>... and/or --dir <DIR> (a directory of *.env files)");
    }

    // Everything is validated up front, before any agent is touched: a typo in the fourth
    // path should not be discovered after three agents have already been drained.
    let mut seen: HashMap<(String, String), PathBuf> = HashMap::new();
    let mut agents = Vec::new();
    for f in &files {
        let (wallet, host) = identity(f)?;
        if let Some(first) = seen.insert((wallet.clone(), host), f.clone()) {
            bail!(
                "{} and {} point at the same wallet ({wallet}) on the same database host; \
                 refusing to drain one agent twice",
                first.display(),
                f.display()
            );
        }
        agents.push((f.clone(), wallet));
    }

    let exe = std::env::current_exe().context("could not locate the owpurge binary")?;
    println!("\nbatch: `{}` across {} agent(s)", args.run, agents.len());

    let mut outcomes: Vec<Outcome> = Vec::new();
    for (i, (file, wallet)) in agents.iter().enumerate() {
        println!("\n{}", "#".repeat(60));
        println!("# agent {}/{}: {}  (wallet {wallet})", i + 1, agents.len(), file.display());
        println!("{}", "#".repeat(60));

        let mut cmd = Command::new(&exe);
        // env_clear: nothing from this shell or a previous agent may reach the child; its
        // whole configuration is the env file.
        cmd.env_clear();
        if args.server_side_delete {
            cmd.arg("--server-side-delete");
        }
        if args.no_audit {
            cmd.arg("--no-audit");
        }
        cmd.arg("--env-file").arg(file).arg(args.run);

        let started = Instant::now();
        let (ok, status) = match cmd.status() {
            Ok(s) if s.success() => (true, "ok".to_string()),
            Ok(s) => (false, format!("failed ({s})")),
            Err(e) => (false, format!("could not start: {e}")),
        };
        outcomes.push(Outcome {
            file: file.clone(),
            wallet: wallet.clone(),
            status,
            secs: started.elapsed().as_secs(),
            ok,
        });
        if !ok && args.stop_on_error {
            eprintln!("\nstopping: --stop-on-error and this agent failed");
            break;
        }
    }

    println!("\n{}", "=".repeat(60));
    println!("batch summary ({})", args.run);
    println!("{}", "=".repeat(60));
    for o in &outcomes {
        println!(
            "  {:<8} {:>5}s  {}  [{}]",
            if o.ok { "ok" } else { "FAILED" },
            o.secs,
            o.file.display(),
            o.wallet
        );
        if !o.ok {
            println!("           {}", o.status);
        }
    }
    let not_run = agents.len() - outcomes.len();
    if not_run > 0 {
        println!("  {not_run} agent(s) not run (stopped early)");
    }

    let failed = outcomes.iter().filter(|o| !o.ok).count();
    if failed > 0 || not_run > 0 {
        eprintln!("\n{failed} agent(s) failed.");
        return Ok(std::process::ExitCode::FAILURE);
    }
    println!("\nAll {} agent(s) completed.", outcomes.len());
    Ok(std::process::ExitCode::SUCCESS)
}