fix(admin): env filter and honest count on DLQ replay (alt to #76)

docs/openapi/chimely.yaml

@@ -130,7 +130,11 @@ paths:
       tags:
       - admin
       summary: Replay dead letter
-      description: Requeue one failed job for another attempt.
+      description: |-
+        Requeue one failed job for another attempt.
+
+        Without an environment filter the replay reaches across environments and
+        `replayed` reports every row moved.
       operationId: adminReplayDeadLetter
       parameters:
       - name: job_id
@@ -139,6 +143,12 @@ paths:
         required: true
         schema:
           type: string
+      - name: environment
+        in: query
+        description: Environment slug. Pins the replay to that environment.
+        required: false
+        schema:
+          type: string
       responses:
         '200':
           description: Replayed.
@@ -167,7 +177,7 @@ paths:
                   code: forbidden
                   message: your role does not permit this action
         '404':
-          description: No such parked job.
+          description: No such parked job, or no such environment.
           content:
             application/json:
               schema:

packages/client/src/generated/api.d.ts

@@ -50,6 +50,9 @@ export interface paths {
         /**
          * Replay dead letter
          * @description Requeue one failed job for another attempt.
+         *
+         *     Without an environment filter the replay reaches across environments and
+         *     `replayed` reports every row moved.
          */
         post: operations["adminReplayDeadLetter"];
         delete?: never;
@@ -1292,7 +1295,10 @@ export interface operations {
     };
     adminReplayDeadLetter: {
         parameters: {
-            query?: never;
+            query?: {
+                /** @description Environment slug. Pins the replay to that environment. */
+                environment?: string;
+            };
             header?: never;
             path: {
                 /** @description Job TypeID (job_…). */
@@ -1345,7 +1351,7 @@ export interface operations {
                     "application/json": components["schemas"]["Error"];
                 };
             };
-            /** @description No such parked job. */
+            /** @description No such parked job, or no such environment. */
             404: {
                 headers: {
                     [name: string]: unknown;

server/src/api/admin.rs

@@ -1215,6 +1215,13 @@ pub struct AdminReplayResult {
     pub replayed: i64,
 }
 
+#[derive(Deserialize, IntoParams)]
+#[into_params(parameter_in = Query)]
+pub struct AdminReplayFilter {
+    /// Environment slug. Pins the replay to that environment.
+    pub environment: Option<String>,
+}
+
 #[utoipa::path(
     get,
     path = "/admin/api/dlq",
@@ -1265,8 +1272,14 @@ pub async fn list_dlq(
     tag = "admin",
     operation_id = "adminReplayDeadLetter",
     summary = "Replay dead letter",
-    description = "Requeue one failed job for another attempt.",
-    params(("job_id" = String, Path, description = "Job TypeID (job_…).")),
+    description = r#"Requeue one failed job for another attempt.
+
+Without an environment filter the replay reaches across environments and
+`replayed` reports every row moved."#,
+    params(
+        ("job_id" = String, Path, description = "Job TypeID (job_…)."),
+        AdminReplayFilter
+    ),
     responses(
         (status = 200, description = "Replayed.", body = AdminReplayResult),
         (
@@ -1283,7 +1296,7 @@ pub async fn list_dlq(
         ),
         (
             status = 404,
-            description = "No such parked job.",
+            description = "No such parked job, or no such environment.",
             body = crate::api::contract::Error,
             example = json!({"error": {"code": "not_found", "message": "no such parked job"}}),
         ),
@@ -1294,18 +1307,30 @@ pub async fn replay_dead_letter(
     auth: AdminAuth,
     State(state): State<AppState>,
     Path(job_id): Path<String>,
+    ApiQuery(filter): ApiQuery<AdminReplayFilter>,
 ) -> Result<Json<AdminReplayResult>, ApiError> {
     auth.require(Capability::DlqReplay)?;
     let id = ids::parse_typeid(ids::JOB, &job_id)
         .ok_or_else(|| ApiError::not_found("no such parked job"))?;
-    // Cross-environment admin path: job ids are globally unique UUIDv7s.
-    let replayed = dlq::replay(&state.pool, id, None)
+    let environment = match &filter.environment {
+        Some(slug) => Some(
+            dlq::environment_by_slug(&state.pool, slug)
+                .await
+                .map_err(ApiError::from)?
+                .ok_or_else(|| ApiError::not_found("no such environment"))?,
+        ),
+        None => None,
+    };
+    // The same job id can be parked in more than one environment under the
+    // (environment_id, id) key. An unfiltered replay moves every copy, so the
+    // response carries the real count.
+    let replayed = dlq::replay(&state.pool, id, environment)
         .await
-        .map_err(ApiError::from)?;
-    if !replayed {
+        .map_err(ApiError::from)? as i64;
+    if replayed == 0 {
         return Err(ApiError::not_found("no such parked job"));
     }
-    Ok(Json(AdminReplayResult { replayed: 1 }))
+    Ok(Json(AdminReplayResult { replayed }))
 }
 
 #[utoipa::path(

server/src/dlq.rs

@@ -41,11 +41,11 @@ pub async fn list(pool: &PgPool) -> anyhow::Result<Vec<DeadLetter>> {
 }
 
 /// Replay one parked job by id, optionally pinned to one environment.
-/// environment_id is part of every key. An unscoped id match would reach
-/// across environments. Returns false if no such dead letter.
-pub async fn replay(pool: &PgPool, id: Uuid, environment: Option<Uuid>) -> anyhow::Result<bool> {
-    let moved = replay_where(pool, Some(id), environment).await?;
-    Ok(moved > 0)
+/// environment_id is part of every key. An unscoped id match reaches across
+/// environments, so the moved-row count is returned. Zero means no such
+/// dead letter.
+pub async fn replay(pool: &PgPool, id: Uuid, environment: Option<Uuid>) -> anyhow::Result<u64> {
+    replay_where(pool, Some(id), environment).await
 }
 
 /// Replay every parked job, optionally only one environment's.
@@ -83,7 +83,8 @@ async fn replay_where(
     Ok(moved)
 }
 
-/// Resolve an environment slug for the CLI's `--env` flag.
+/// Resolve an environment slug for the CLI's `--env` flag and the admin
+/// replay filter.
 pub async fn environment_by_slug(pool: &PgPool, slug: &str) -> anyhow::Result<Option<Uuid>> {
     Ok(
         sqlx::query_scalar!("SELECT id FROM environments WHERE slug = $1", slug)

server/src/main.rs

@@ -212,7 +212,7 @@ async fn dlq_command(args: Vec<String>) -> anyhow::Result<()> {
                 let id = ids::parse_typeid(ids::JOB, id)
                     .or_else(|| id.parse().ok())
                     .context("expected a job_… TypeID or a raw UUID")?;
-                if dlq::replay(&pool, id, environment).await? {
+                if dlq::replay(&pool, id, environment).await? > 0 {
                     println!("replayed {}", ids::typeid(ids::JOB, id));
                 } else {
                     eprintln!("no such dead letter");

server/tests/redteam_dlq_env_scope.rs

@@ -7,6 +7,7 @@
 
 mod support;
 
+use serde_json::json;
 use uuid::Uuid;
 
 async fn park_dead_letter_with_id(pool: &sqlx::PgPool, env: Uuid, id: Uuid) {
@@ -73,7 +74,7 @@ async fn replay_by_id_replays_only_the_scoped_environment() {
     let replayed = chimely::dlq::replay(&app.pool, shared, Some(app.env.id))
         .await
         .expect("replay env A's copy");
-    assert!(replayed, "env A's copy was replayed");
+    assert_eq!(replayed, 1, "only env A's copy was replayed");
 
     assert_eq!(
         dead_letters_in(&app.pool, app.env.id).await,
@@ -86,3 +87,87 @@ async fn replay_by_id_replays_only_the_scoped_environment() {
         "env B's same-id dead letter must survive an env-A-scoped replay"
     );
 }
+
+#[tokio::test]
+async fn admin_replay_honors_the_environment_query_param() {
+    let app = support::spawn().await;
+    let env_b = app.create_environment(true).await;
+
+    let shared = chimely::ids::new_uuid();
+    park_dead_letter_with_id(&app.pool, app.env.id, shared).await;
+    park_dead_letter_with_id(&app.pool, env_b.id, shared).await;
+
+    let job = chimely::ids::typeid(chimely::ids::JOB, shared);
+    let res = app
+        .admin_post(
+            &format!("/admin/api/dlq/{job}/replay?environment={}", app.env.slug),
+            json!({}),
+        )
+        .send()
+        .await
+        .expect("replay request");
+    assert_eq!(res.status(), 200);
+    let body: serde_json::Value = res.json().await.expect("replay body");
+    assert_eq!(body["replayed"], json!(1));
+
+    assert_eq!(
+        dead_letters_in(&app.pool, app.env.id).await,
+        0,
+        "env A's copy moved back to jobs"
+    );
+    assert_eq!(
+        dead_letters_in(&app.pool, env_b.id).await,
+        1,
+        "env B's same-id dead letter must survive an env-scoped HTTP replay"
+    );
+}
+
+#[tokio::test]
+async fn admin_replay_reports_the_actual_moved_count_when_unscoped() {
+    let app = support::spawn().await;
+    let env_b = app.create_environment(true).await;
+
+    // Without an environment filter the admin path keeps its documented
+    // cross-environment reach. The response must report every row it moved.
+    let shared = chimely::ids::new_uuid();
+    park_dead_letter_with_id(&app.pool, app.env.id, shared).await;
+    park_dead_letter_with_id(&app.pool, env_b.id, shared).await;
+
+    let job = chimely::ids::typeid(chimely::ids::JOB, shared);
+    let res = app
+        .admin_post(&format!("/admin/api/dlq/{job}/replay"), json!({}))
+        .send()
+        .await
+        .expect("replay request");
+    assert_eq!(res.status(), 200);
+    let body: serde_json::Value = res.json().await.expect("replay body");
+    assert_eq!(body["replayed"], json!(2), "both moved rows are counted");
+
+    assert_eq!(dead_letters_in(&app.pool, app.env.id).await, 0);
+    assert_eq!(dead_letters_in(&app.pool, env_b.id).await, 0);
+}
+
+#[tokio::test]
+async fn admin_replay_404s_for_an_unknown_environment_slug() {
+    let app = support::spawn().await;
+
+    let id = chimely::ids::new_uuid();
+    park_dead_letter_with_id(&app.pool, app.env.id, id).await;
+
+    let job = chimely::ids::typeid(chimely::ids::JOB, id);
+    let res = app
+        .admin_post(
+            &format!("/admin/api/dlq/{job}/replay?environment=no-such-env"),
+            json!({}),
+        )
+        .send()
+        .await
+        .expect("replay request");
+    assert_eq!(res.status(), 404);
+
+    assert_eq!(
+        dead_letters_in(&app.pool, app.env.id).await,
+        1,
+        "a rejected replay must not move the parked job"
+    );
+}