feat: job maintenance API (remove / drain / clean / obliterate)

README.md

@@ -219,6 +219,7 @@ In-repo: [`docs/engine.md`](docs/engine.md) (engine internals), [`docs/history.m
 | Priorities | Future | ✓ | ✓ | — |
 | Rate limiter | Future | ✓ | ✓ | — |
 | Pause / Resume | ✓ | ✓ | ✓ | — |
+| Job maintenance (`remove` / `drain` / `clean` / `obliterate`) | ✓ | ✓ | ✓ | — |
 | Parent / child dependencies | Future | ✓ | — | — |
 | Web UI | Future | ✓ | ✓ | — |
 | Optimized for | Throughput | Jobs | Jobs | Messages |

chasquimq-cli/src/main.rs

@@ -1,6 +1,7 @@
 mod dlq;
 mod events;
 mod inspect;
+mod maintenance;
 mod pause;
 mod repeatable;
 mod watch;
@@ -87,6 +88,44 @@ enum Command {
         /// Queue name (without the `{chasqui:...}` hash-tag wrapping).
         queue: String,
     },
+    /// Age- and state-filtered bulk delete. Removes up to `--limit` jobs
+    /// in `--state` that are older than `--grace-ms` milliseconds and
+    /// prints the removed job ids.
+    Clean {
+        /// Queue name (without the `{chasqui:...}` hash-tag wrapping).
+        queue: String,
+        /// Job state to clean: completed | failed | delayed | waiting.
+        #[arg(long, default_value = "completed", value_name = "STATE")]
+        state: String,
+        /// Only remove jobs older than this many milliseconds.
+        #[arg(long, default_value_t = 0, value_name = "MS")]
+        grace_ms: u64,
+        /// Cap on how many jobs are removed in this call.
+        #[arg(long, default_value_t = 1000)]
+        limit: u32,
+        /// Consumer group used for stream acks. Matches the engine
+        /// default `ConsumerConfig::group`.
+        #[arg(long, default_value = DEFAULT_GROUP, value_name = "NAME")]
+        group: String,
+        /// Skip the interactive confirmation prompt.
+        #[arg(long)]
+        yes: bool,
+    },
+    /// Tear an entire queue down — delete every Redis key backing it:
+    /// the stream and its consumer groups, the DLQ, delayed jobs, all
+    /// per-job side-indexes and results, repeatable specs, the pause
+    /// flag, and the events stream. Destructive and not reversible.
+    Obliterate {
+        /// Queue name (without the `{chasqui:...}` hash-tag wrapping).
+        queue: String,
+        /// Consumer group accepted for symmetry; obliterate drops every
+        /// group regardless.
+        #[arg(long, default_value = DEFAULT_GROUP, value_name = "NAME")]
+        group: String,
+        /// Skip the interactive confirmation prompt.
+        #[arg(long)]
+        yes: bool,
+    },
 }
 
 #[derive(Subcommand, Debug)]
@@ -147,5 +186,16 @@ async fn main() -> anyhow::Result<()> {
         Command::Events { queue, from } => events::run(redis_url, &queue, &from).await,
         Command::Pause { queue } => pause::pause(redis_url, &queue).await,
         Command::Resume { queue } => pause::resume(redis_url, &queue).await,
+        Command::Clean {
+            queue,
+            state,
+            grace_ms,
+            limit,
+            group,
+            yes,
+        } => maintenance::clean(redis_url, &queue, &group, &state, grace_ms, limit, yes).await,
+        Command::Obliterate { queue, group, yes } => {
+            maintenance::obliterate(redis_url, &queue, &group, yes).await
+        }
     }
 }

chasquimq-cli/src/maintenance.rs

@@ -0,0 +1,94 @@
+//! `chasqui clean` and `chasqui obliterate` — operator-facing wrappers
+//! over the engine's job maintenance API.
+
+use anyhow::{Context, Result};
+use chasquimq::{JobState, Producer, ProducerConfig};
+use std::io::{self, Write};
+
+async fn connect(redis_url: &str, queue: &str) -> Result<Producer<rmpv::Value>> {
+    let cfg = ProducerConfig {
+        queue_name: queue.to_string(),
+        ..Default::default()
+    };
+    Producer::<rmpv::Value>::connect(redis_url, cfg)
+        .await
+        .map_err(|e| anyhow::anyhow!("redis connect failed: {e}"))
+}
+
+fn confirm(prompt: &str) -> Result<bool> {
+    eprint!("{prompt} [y/N]: ");
+    io::stderr().flush().ok();
+    let mut buf = String::new();
+    io::stdin()
+        .read_line(&mut buf)
+        .context("failed to read confirmation from stdin")?;
+    Ok(matches!(
+        buf.trim().to_ascii_lowercase().as_str(),
+        "y" | "yes"
+    ))
+}
+
+/// `chasqui clean <queue> --state <s> --grace-ms <ms> --limit <n>` —
+/// age- and state-filtered bulk delete. Prints the removed job ids.
+pub async fn clean(
+    redis_url: &str,
+    queue: &str,
+    group: &str,
+    state: &str,
+    grace_ms: u64,
+    limit: u32,
+    yes: bool,
+) -> Result<()> {
+    let parsed = JobState::parse(state).ok_or_else(|| {
+        anyhow::anyhow!(
+            "unknown state '{state}'; expected one of completed | failed | delayed | waiting"
+        )
+    })?;
+    if !yes
+        && !confirm(&format!(
+            "Clean up to {limit} '{state}' jobs older than {grace_ms}ms on queue {queue}?"
+        ))?
+    {
+        anyhow::bail!("aborted");
+    }
+
+    let producer = connect(redis_url, queue).await?;
+    let removed = producer
+        .clean(group, grace_ms, limit as usize, parsed)
+        .await
+        .context("clean failed")?;
+
+    println!(
+        "cleaned {} '{}' job(s) from queue {}",
+        removed.len(),
+        state,
+        queue
+    );
+    for id in &removed {
+        println!("  {id}");
+    }
+    Ok(())
+}
+
+/// `chasqui obliterate <queue>` — delete the entire `{chasqui:<queue>}`
+/// keyspace. Destructive; gated behind an interactive confirm unless
+/// `--yes` is passed.
+pub async fn obliterate(redis_url: &str, queue: &str, group: &str, yes: bool) -> Result<()> {
+    if !yes
+        && !confirm(&format!(
+            "Obliterate the ENTIRE queue {queue}? This deletes every job, the DLQ, \
+             delayed jobs, repeatable specs, and results. This cannot be undone."
+        ))?
+    {
+        anyhow::bail!("aborted");
+    }
+
+    let producer = connect(redis_url, queue).await?;
+    let removed = producer
+        .obliterate(group)
+        .await
+        .context("obliterate failed")?;
+
+    println!("obliterated queue {queue}: removed {removed} Redis key(s)");
+    Ok(())
+}

chasquimq-node/README.md

@@ -68,7 +68,7 @@ main()
 
 | Surface | What it does |
 |---|---|
-| `Queue<DataType, ResultType, NameType>` | Producer + queue inspection. `add` / `addBulk` / `addUnique` / `getJob` / `getJobs` / `getJobState` / `getJobCounts` / `getWaitingCount` / `getActiveCount` / `getDelayedCount` / `getCompletedCount` / `getFailedCount` / `count` / `getJobResult` / `peekDlq` / `replayDlq` / `cancelDelayed` / `getRepeatableJobs` / `removeRepeatableByKey` / `pause` / `resume` / `isPaused`. `[Symbol.asyncDispose]`. |
+| `Queue<DataType, ResultType, NameType>` | Producer + queue inspection. `add` / `addBulk` / `addUnique` / `getJob` / `getJobs` / `getJobState` / `getJobCounts` / `getWaitingCount` / `getActiveCount` / `getDelayedCount` / `getCompletedCount` / `getFailedCount` / `count` / `getJobResult` / `peekDlq` / `replayDlq` / `cancelDelayed` / `getRepeatableJobs` / `removeRepeatableByKey` / `pause` / `resume` / `isPaused` / `remove` / `removeReport` / `drain` / `clean` / `obliterate`. `[Symbol.asyncDispose]`. |
 | `Worker<DataType, ResultType, NameType>` | Consumer pool. tokio-side dispatch, opt-in result storage (`storeResults: true`), `EventEmitter` events (`completed` / `failed` / `error`), `pause` / `resume` / `isPaused`. `[Symbol.asyncDispose]`. |
 | `Job<DataType, ResultType, NameType>` | Read-only handle. `id`, `name`, `data`, `attemptsMade`, `waitForResult({ timeoutMs, intervalMs, signal })`. |
 | `QueueEvents` | Cross-process pub/sub via the events stream. Subscribe to `completed` / `failed` / `dlq` / `retry-scheduled` / `delayed` / `drained`. `[Symbol.asyncDispose]`. |
@@ -220,6 +220,39 @@ Counts are cheap (~5 Redis round trips, one per state column). `completed` runs
 const queue = new Queue("emails", { connection, consumerGroup: "primary" })
 ```
 
+### Maintenance
+
+Tear individual jobs — or a whole queue — down. All four methods are off the hot path; every scan is bounded so a single call never blocks Redis.
+
+```ts
+const queue = new Queue("emails", { connection })
+
+// Remove one job from everywhere it lives — delayed stage, a waiting or
+// in-flight stream entry, the DLQ, and the stored result. Idempotent.
+const surfaces = await queue.remove("job-123")
+// => number of surfaces the job was removed from (0 if not found)
+
+const report = await queue.removeReport("job-123")
+// => { delayed, stream, dlq, result } booleans
+
+// Clear every waiting job. In-flight jobs keep running. Delayed jobs go
+// too unless you pass `false`.
+const drained = await queue.drain() // => count removed
+await queue.drain(false) // keep scheduled future jobs
+
+// Age- + state-filtered bulk delete. Removes up to `limit` jobs in the
+// given state older than `grace` ms; returns the removed job ids.
+const ids = await queue.clean(60_000, 1000, "completed")
+//          clean(graceMs,  limit, "completed" | "failed" | "delayed" | "waiting")
+
+// Nuke the entire queue keyspace — stream, DLQ, delayed, results,
+// repeatable specs, the pause flag, the events stream. Returns the
+// Redis key count removed.
+await queue.obliterate()
+```
+
+`remove` is idempotent — a job id that exists on no surface resolves without error (count `0`). `clean` ignores `grace` for `"completed"` (a stored result has no creation timestamp; rely on result TTL for time-based expiry). `clean` with `"active"` is a no-op — use `remove` for the deliberate per-job case. `obliterate` cannot be undone.
+
 ## Power-user surface
 
 The native engine handles ship from the same top-level package: