feat(node): add --bulk mode to purge-spam for fast set-based cleanup

The per-repo cascade (one transaction per repo) stalls and drops connections
on small DB instances with large match sets (node2/node3, ~70k keep* mirrors).

--bulk deletes all matching repos and their child rows with a handful of
set-based statements in one transaction, then removes the on-disk bare repos
(collected before the delete). Skips Tigris (mirror rows have no archive).
Slug-keyed child tables match both the repo id and the {owner_short}/{name}
slug so canonical and mirror rows are both cleaned.

Tested: bulk_delete_by_name_regex removes both canonical + mirror matches and
their id-/slug-keyed children while leaving non-matching repos intact.

Cargo.lock

@@ -3300,7 +3300,7 @@ dependencies = [
 
 [[package]]
 name = "git-remote-gitlawb"
-version = "0.3.9"
+version = "0.4.0"
 dependencies = [
  "anyhow",
  "gitlawb-core",
@@ -3311,7 +3311,7 @@ dependencies = [
 
 [[package]]
 name = "gitlawb-attest"
-version = "0.3.9"
+version = "0.4.0"
 dependencies = [
  "base64",
  "ed25519-dalek",
@@ -3328,7 +3328,7 @@ dependencies = [
 
 [[package]]
 name = "gitlawb-core"
-version = "0.3.9"
+version = "0.4.0"
 dependencies = [
  "anyhow",
  "base64",
@@ -3355,7 +3355,7 @@ dependencies = [
 
 [[package]]
 name = "gitlawb-node"
-version = "0.3.9"
+version = "0.4.0"
 dependencies = [
  "alloy",
  "anyhow",
@@ -3411,7 +3411,7 @@ dependencies = [
 
 [[package]]
 name = "gl"
-version = "0.3.9"
+version = "0.4.0"
 dependencies = [
  "alloy",
  "anyhow",

crates/gitlawb-node/src/admin.rs

@@ -0,0 +1,130 @@
+//! Admin subcommands (one-off maintenance), invoked as `gitlawb-node <cmd>`.
+//! These run instead of the daemon and exit.
+
+use anyhow::Result;
+
+use crate::config::{Config, PurgeSpamArgs};
+use crate::db::Db;
+use crate::git::tigris::TigrisClient;
+
+/// How many repos to delete per query/commit batch.
+const BATCH: i64 = 500;
+
+/// Bulk-delete repos whose name matches a Postgres regex, removing DB rows,
+/// the on-disk bare repo, and (unless skipped) the Tigris archive.
+///
+/// Defaults to a dry run: it reports the match count and a sample and deletes
+/// nothing. Pass `--execute` to actually delete.
+pub async fn purge_spam(config: &Config, args: &PurgeSpamArgs) -> Result<()> {
+    let db = Db::connect(&config.database_url).await?;
+
+    let total = db.count_repos_by_name_regex(&args.regex).await?;
+    println!("repos matching /{}/ : {total}", args.regex);
+
+    let sample = db.list_repos_by_name_regex(&args.regex, 20).await?;
+    println!("sample (up to 20):");
+    for (id, name, owner, _disk) in &sample {
+        println!("  {name}  owner={owner}  id={id}");
+    }
+
+    if !args.execute {
+        println!(
+            "\nDRY RUN — nothing deleted. Re-run with --execute to delete these {total} repos \
+             ({}).",
+            if args.bulk {
+                "bulk set-based DB delete + on-disk .git"
+            } else if args.skip_tigris {
+                "DB rows + on-disk .git"
+            } else {
+                "DB rows + on-disk .git + Tigris archive"
+            }
+        );
+        return Ok(());
+    }
+
+    // Bulk mode: set-based DB delete (a few statements) + on-disk removal. Far
+    // faster on large match sets / small DB instances than per-repo cascade.
+    if args.bulk {
+        let disk_paths = db.list_disk_paths_by_name_regex(&args.regex).await?;
+        println!(
+            "collected {} on-disk paths; running bulk DB delete…",
+            disk_paths.len()
+        );
+        let deleted = db.bulk_delete_by_name_regex(&args.regex).await?;
+        println!("bulk DB delete done: {deleted} repos removed; removing on-disk repos…");
+        let mut disk_removed = 0u64;
+        for p in &disk_paths {
+            if std::path::Path::new(p).exists() {
+                match std::fs::remove_dir_all(p) {
+                    Ok(()) => disk_removed += 1,
+                    Err(e) => eprintln!("warning: rm {p} failed: {e}"),
+                }
+            }
+        }
+        println!("done: {deleted} repos purged (bulk), {disk_removed} on-disk dirs removed");
+        return Ok(());
+    }
+
+    // Tigris client (best-effort; archive deletion failures are logged, not fatal).
+    let tigris = if !args.skip_tigris && !config.tigris_bucket.is_empty() {
+        match TigrisClient::new(&config.tigris_bucket).await {
+            Ok(t) => Some(t),
+            Err(e) => {
+                eprintln!("warning: could not init Tigris ({e}); skipping archive deletion");
+                None
+            }
+        }
+    } else {
+        None
+    };
+
+    let mut deleted = 0i64;
+    let mut disk_removed = 0i64;
+    let mut tigris_removed = 0i64;
+
+    loop {
+        let take = match args.limit {
+            n if n > 0 => (n - deleted).min(BATCH),
+            _ => BATCH,
+        };
+        if take <= 0 {
+            break;
+        }
+        let batch = db.list_repos_by_name_regex(&args.regex, take).await?;
+        if batch.is_empty() {
+            break;
+        }
+
+        for (id, name, owner_did, disk_path) in &batch {
+            let short = owner_did.rsplit(':').next().unwrap_or(owner_did);
+            let slug = format!("{short}/{name}");
+
+            // DB first: once the row is gone the repo is invisible to all serve/
+            // list paths even if disk/Tigris cleanup lags.
+            db.delete_repo_cascade(id, &slug).await?;
+            deleted += 1;
+
+            if !disk_path.is_empty() && std::path::Path::new(disk_path).exists() {
+                match std::fs::remove_dir_all(disk_path) {
+                    Ok(()) => disk_removed += 1,
+                    Err(e) => eprintln!("warning: rm {disk_path} failed: {e}"),
+                }
+            }
+
+            if let Some(t) = &tigris {
+                let owner_slug = owner_did.replace([':', '/'], "_");
+                match t.delete(&owner_slug, name).await {
+                    Ok(()) => tigris_removed += 1,
+                    Err(e) => eprintln!("warning: tigris delete {owner_slug}/{name} failed: {e}"),
+                }
+            }
+        }
+
+        println!("… deleted {deleted}/{total} (disk {disk_removed}, tigris {tigris_removed})");
+    }
+
+    println!(
+        "done: {deleted} repos purged (disk dirs removed: {disk_removed}, tigris archives removed: {tigris_removed})"
+    );
+    Ok(())
+}

crates/gitlawb-node/src/config.rs

@@ -1,9 +1,48 @@
-use clap::Parser;
+use clap::{Args, Parser, Subcommand};
 use std::path::PathBuf;
 
+/// Admin subcommands. When absent, the binary runs the node daemon as usual.
+#[derive(Subcommand, Debug, Clone)]
+pub enum Command {
+    /// Bulk-delete repos whose name matches a Postgres regex, removing their DB
+    /// rows, on-disk bare repo, and Tigris archive. Intended for spam cleanup.
+    /// Defaults to a dry run; pass --execute to actually delete.
+    PurgeSpam(PurgeSpamArgs),
+}
+
+#[derive(Args, Debug, Clone)]
+pub struct PurgeSpamArgs {
+    /// Postgres regex matched against `repos.name` (e.g. '^keep[0-9]').
+    #[arg(long)]
+    pub regex: String,
+
+    /// Actually delete. Without this flag the command only reports what would
+    /// be removed (dry run).
+    #[arg(long, default_value_t = false)]
+    pub execute: bool,
+
+    /// Stop after processing at most this many repos (0 = no limit).
+    #[arg(long, default_value_t = 0)]
+    pub limit: i64,
+
+    /// Skip deleting the Tigris archive (DB + on-disk only).
+    #[arg(long, default_value_t = false)]
+    pub skip_tigris: bool,
+
+    /// Bulk mode: delete all matching rows with set-based SQL (a handful of
+    /// statements) + remove their on-disk repos, instead of one transaction per
+    /// repo. Far faster on large match sets / small DB instances. Skips Tigris.
+    #[arg(long, default_value_t = false)]
+    pub bulk: bool,
+}
+
 #[derive(Parser, Debug, Clone)]
 #[command(name = "gitlawb-node", about = "gitlawb node daemon", version)]
 pub struct Config {
+    /// Optional admin subcommand. When omitted, runs the node daemon.
+    #[command(subcommand)]
+    pub command: Option<Command>,
+
     /// Directory where bare git repositories are stored
     #[arg(long, env = "GITLAWB_REPOS_DIR", default_value = "./data/repos")]
     pub repos_dir: PathBuf,

crates/gitlawb-node/src/db/mod.rs

@@ -1249,6 +1249,169 @@ impl Db {
         Ok(rows.into_iter().map(|r| r.get::<String, _>("id")).collect())
     }
 
+    /// Count repos whose name matches a Postgres regex. Used by the admin
+    /// `purge-spam` command's dry run.
+    pub async fn count_repos_by_name_regex(&self, regex: &str) -> Result<i64> {
+        let row = sqlx::query("SELECT COUNT(*) AS cnt FROM repos WHERE name ~ $1")
+            .bind(regex)
+            .fetch_one(&self.pool)
+            .await?;
+        Ok(row.get::<i64, _>("cnt"))
+    }
+
+    /// A page of repos whose name matches a Postgres regex, ordered by id.
+    /// Returns `(id, name, owner_did, disk_path)`. `purge-spam` re-queries with
+    /// the same `limit` after each batch is deleted (deleted rows fall out, so
+    /// no offset is needed).
+    pub async fn list_repos_by_name_regex(
+        &self,
+        regex: &str,
+        limit: i64,
+    ) -> Result<Vec<(String, String, String, String)>> {
+        let rows = sqlx::query(
+            "SELECT id, name, owner_did, disk_path FROM repos WHERE name ~ $1 ORDER BY id LIMIT $2",
+        )
+        .bind(regex)
+        .bind(limit)
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows
+            .into_iter()
+            .map(|r| {
+                (
+                    r.get::<String, _>("id"),
+                    r.get::<String, _>("name"),
+                    r.get::<String, _>("owner_did"),
+                    r.get::<String, _>("disk_path"),
+                )
+            })
+            .collect())
+    }
+
+    /// Delete a repo and all of its child rows in one transaction. `id` is the
+    /// `repos.id`; `slug` is the `{owner_short}/{name}` form. The slug-keyed
+    /// tables (`branch_cids`, `sync_queue`, `received_ref_updates`,
+    /// `arweave_anchors`) store either the id or the slug depending on the write
+    /// path, so both are matched. Tables keyed on `repo_id` match the id.
+    pub async fn delete_repo_cascade(&self, id: &str, slug: &str) -> Result<()> {
+        let mut tx = self.pool.begin().await?;
+
+        // Tables keyed on repos.id.
+        for table in [
+            "ref_certificates",
+            "push_events",
+            "webhooks",
+            "agent_tasks",
+            "protected_branches",
+            "repo_stars",
+            "repo_replicas",
+            "repo_labels",
+            "visibility_rules",
+            "encrypted_blobs",
+            "repo_icaptcha_proofs",
+            "pull_requests",
+        ] {
+            sqlx::query(&format!("DELETE FROM {table} WHERE repo_id = $1"))
+                .bind(id)
+                .execute(&mut *tx)
+                .await?;
+        }
+
+        // Tables keyed on the repo slug/id string (`repo` column).
+        for table in [
+            "branch_cids",
+            "sync_queue",
+            "received_ref_updates",
+            "arweave_anchors",
+        ] {
+            sqlx::query(&format!("DELETE FROM {table} WHERE repo = $1 OR repo = $2"))
+                .bind(id)
+                .bind(slug)
+                .execute(&mut *tx)
+                .await?;
+        }
+
+        sqlx::query("DELETE FROM repos WHERE id = $1")
+            .bind(id)
+            .execute(&mut *tx)
+            .await?;
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    /// On-disk bare-repo paths of all repos whose name matches `regex`. Collected
+    /// before a bulk delete so the dirs can be removed afterward.
+    pub async fn list_disk_paths_by_name_regex(&self, regex: &str) -> Result<Vec<String>> {
+        let rows = sqlx::query("SELECT disk_path FROM repos WHERE name ~ $1")
+            .bind(regex)
+            .fetch_all(&self.pool)
+            .await?;
+        Ok(rows
+            .into_iter()
+            .map(|r| r.get::<String, _>("disk_path"))
+            .filter(|p| !p.is_empty())
+            .collect())
+    }
+
+    /// Set-based bulk delete of every repo whose name matches `regex`, plus their
+    /// child rows, in one transaction (a handful of statements total, vs one
+    /// transaction per repo). Returns the number of `repos` rows deleted. Does
+    /// NOT touch on-disk repos or Tigris — the caller handles disk cleanup.
+    pub async fn bulk_delete_by_name_regex(&self, regex: &str) -> Result<u64> {
+        let mut tx = self.pool.begin().await?;
+
+        // Child tables keyed on repos.id.
+        for table in [
+            "ref_certificates",
+            "push_events",
+            "webhooks",
+            "agent_tasks",
+            "protected_branches",
+            "repo_stars",
+            "repo_replicas",
+            "repo_labels",
+            "visibility_rules",
+            "encrypted_blobs",
+            "repo_icaptcha_proofs",
+            "pull_requests",
+        ] {
+            sqlx::query(&format!(
+                "DELETE FROM {table} WHERE repo_id IN (SELECT id FROM repos WHERE name ~ $1)"
+            ))
+            .bind(regex)
+            .execute(&mut *tx)
+            .await?;
+        }
+
+        // Child tables keyed on the repo slug/id string. Match either the repo
+        // id or the `{owner_short}/{name}` slug (owner_short = trailing segment
+        // after the last ':'), covering both canonical and mirror rows.
+        for table in [
+            "branch_cids",
+            "sync_queue",
+            "received_ref_updates",
+            "arweave_anchors",
+        ] {
+            sqlx::query(&format!(
+                "DELETE FROM {table} WHERE repo IN (SELECT id FROM repos WHERE name ~ $1) \
+                 OR repo IN (SELECT substring(owner_did from '[^:]*$') || '/' || name \
+                             FROM repos WHERE name ~ $1)"
+            ))
+            .bind(regex)
+            .execute(&mut *tx)
+            .await?;
+        }
+
+        let result = sqlx::query("DELETE FROM repos WHERE name ~ $1")
+            .bind(regex)
+            .execute(&mut *tx)
+            .await?;
+
+        tx.commit().await?;
+        Ok(result.rows_affected())
+    }
+
     pub async fn get_trust_score(&self, agent_did: &str) -> Result<f64> {
         let row = sqlx::query("SELECT trust_score FROM agents WHERE did = $1")
             .bind(agent_did)
@@ -3909,3 +4072,152 @@ mod icaptcha_quarantine_tests {
         assert!(with_stars.iter().all(|(r, _)| r.name != "spam"));
     }
 }
+
+/// Exercises the admin `purge-spam` data layer: regex matching is precise (only
+/// `^keep[0-9]` spam, not a legit "keeper-*" repo) and cascade delete removes
+/// the repo plus its child rows.
+#[cfg(test)]
+mod purge_spam_tests {
+    use super::{Db, RepoRecord};
+    use chrono::Utc;
+    use sqlx::{PgPool, Row};
+
+    async fn db(pool: PgPool) -> Db {
+        let db = Db::for_testing(pool);
+        db.run_migrations().await.unwrap();
+        db
+    }
+
+    fn rec(id: &str, owner: &str, name: &str) -> RepoRecord {
+        let now = Utc::now();
+        RepoRecord {
+            id: id.to_string(),
+            name: name.to_string(),
+            owner_did: owner.to_string(),
+            description: None,
+            is_public: true,
+            default_branch: "main".to_string(),
+            created_at: now,
+            updated_at: now,
+            disk_path: format!("/srv/{id}"),
+            forked_from: None,
+            machine_id: None,
+        }
+    }
+
+    #[sqlx::test]
+    async fn regex_is_precise_and_cascade_deletes_children(pool: PgPool) {
+        let db = db(pool).await;
+        db.create_repo(&rec(
+            "id-keep-1",
+            "did:key:z6MkSpamA",
+            "keep27x25731x584132399",
+        ))
+        .await
+        .unwrap();
+        db.create_repo(&rec("id-keep-2", "did:key:z6MkSpamB", "keep9x1x2"))
+            .await
+            .unwrap();
+        // Must NOT match ^keep[0-9]: a real word starting with "keep" + a letter.
+        db.create_repo(&rec("id-legit", "did:key:z6MkReal", "keeper-tools"))
+            .await
+            .unwrap();
+
+        // Child rows on a spam repo: one slug/id-keyed (branch_cids), one
+        // repo_id-keyed (icaptcha proof).
+        db.upsert_branch_cid(
+            "id-keep-1",
+            "refs/heads/main",
+            "deadbeef",
+            "cidX",
+            "did:key:zNode",
+        )
+        .await
+        .unwrap();
+        db.record_repo_proof("id-keep-1", "tok.sig", "did:key:z6MkSpamA", 3, "jtiX", 123)
+            .await
+            .unwrap();
+
+        // Regex matches exactly the two keepN* repos, not keeper-tools.
+        assert_eq!(db.count_repos_by_name_regex("^keep[0-9]").await.unwrap(), 2);
+        let matched = db
+            .list_repos_by_name_regex("^keep[0-9]", 100)
+            .await
+            .unwrap();
+        assert_eq!(matched.len(), 2);
+        assert!(matched
+            .iter()
+            .all(|(_, name, _, _)| name.starts_with("keep")));
+
+        // Cascade-delete one spam repo; its child rows go with it.
+        db.delete_repo_cascade("id-keep-1", "z6MkSpamA/keep27x25731x584132399")
+            .await
+            .unwrap();
+        assert_eq!(db.count_repos_by_name_regex("^keep[0-9]").await.unwrap(), 1);
+        assert_eq!(db.get_repo_proof_token("id-keep-1").await.unwrap(), None);
+        let cids = sqlx::query("SELECT COUNT(*) AS c FROM branch_cids WHERE repo = 'id-keep-1'")
+            .fetch_one(&db.pool)
+            .await
+            .unwrap();
+        assert_eq!(cids.get::<i64, _>("c"), 0, "branch_cids child rows removed");
+
+        // The legit repo is untouched.
+        assert!(db
+            .get_repo("z6MkReal", "keeper-tools")
+            .await
+            .unwrap()
+            .is_some());
+    }
+
+    /// Bulk delete removes all matching repos + child rows in one pass (set-based),
+    /// matching child rows keyed by both repo id and `{owner_short}/{name}` slug,
+    /// and leaves non-matching repos intact.
+    #[sqlx::test]
+    async fn bulk_delete_clears_matches_and_children(pool: PgPool) {
+        let db = db(pool).await;
+        // Canonical spam (full DID) + mirror-style spam (bare owner) + a legit repo.
+        db.create_repo(&rec("uuid-1", "did:key:z6MkSpamA", "keep1x2x3"))
+            .await
+            .unwrap();
+        db.create_repo(&rec("z6Mb/keep4x5", "z6Mb", "keep4x5"))
+            .await
+            .unwrap();
+        db.create_repo(&rec("id-legit", "did:key:z6MkReal", "keeper"))
+            .await
+            .unwrap();
+
+        // Child rows: repo_id-keyed proof + slug-keyed branch_cid for the
+        // canonical one (slug = z6MkSpamA/keep1x2x3), and an id-keyed branch_cid
+        // for the mirror one.
+        db.record_repo_proof("uuid-1", "t.s", "did:key:z6MkSpamA", 3, "j1", 1)
+            .await
+            .unwrap();
+        db.upsert_branch_cid(
+            "z6MkSpamA/keep1x2x3",
+            "refs/heads/main",
+            "sha",
+            "cid",
+            "z6N",
+        )
+        .await
+        .unwrap();
+        db.upsert_branch_cid("z6Mb/keep4x5", "refs/heads/main", "sha", "cid", "z6N")
+            .await
+            .unwrap();
+
+        let deleted = db.bulk_delete_by_name_regex("^keep[0-9]").await.unwrap();
+        assert_eq!(deleted, 2, "both keepN* repos deleted, keeper untouched");
+        assert_eq!(db.count_repos_by_name_regex("^keep[0-9]").await.unwrap(), 0);
+        assert_eq!(db.get_repo_proof_token("uuid-1").await.unwrap(), None);
+        let cids = sqlx::query("SELECT COUNT(*) AS c FROM branch_cids")
+            .fetch_one(&db.pool)
+            .await
+            .unwrap();
+        assert_eq!(
+            cids.get::<i64, _>("c"),
+            0,
+            "slug- and id-keyed child rows removed"
+        );
+        assert!(db.get_repo("z6MkReal", "keeper").await.unwrap().is_some());
+    }
+}

crates/gitlawb-node/src/main.rs

@@ -1,3 +1,4 @@
+mod admin;
 mod api;
 mod arweave;
 mod auth;
@@ -50,6 +51,15 @@ async fn main() -> Result<()> {
 
     let mut config = Config::parse();
 
+    // Admin subcommands run instead of the daemon and exit.
+    if let Some(cmd) = config.command.clone() {
+        match cmd {
+            config::Command::PurgeSpam(args) => {
+                return admin::purge_spam(&config, &args).await;
+            }
+        }
+    }
+
     // Merge the embedded seed list of public network nodes into the runtime
     // bootstrap peers. Operators can opt out via GITLAWB_BOOTSTRAP_DISABLE_SEEDS.
     bootstrap::merge_seeds(&mut config);