From 872cc831bf899c6f4b6ef0ec5d2d9780ac2be2f1 Mon Sep 17 00:00:00 2001
From: hacka0wi <124970567+hacka0wi@users.noreply.github.com>
Date: Fri, 8 May 2026 00:19:28 +0700
Subject: [PATCH] feat(queue): POST /queues/:name/purge to drain stuck PENDING
 jobs (#13)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Operators currently have no way to clear bad messages from a queue
short of dropping the entire queue. After updating a flow's code to
fix a runtime bug, messages enqueued under the old broken code keep
retrying on a 5-minute visibility cycle until max_attempts hits the
DLQ — that's ~15 minutes of latency before fresh enqueues even get
picked up.

Purge deletes only PENDING rows (status='PENDING'). PROCESSING rows
are intentionally left alone so we don't yank a row out from under an
in-flight worker; those age out naturally via visibility timeout if
their handler dies. PG backend only — Redis/Kafka/RabbitMQ users have
their broker UIs for this.

Returns the deleted row count so an operator can confirm.

Refs #13. This is a mitigation, not a root-cause fix: see comment on
the issue for analysis of the reporter's "stale cached code" hypothesis
(loadFlowConfig already SELECTs fresh on every delivery; the more likely
root cause is messages stuck in retry loop).
---
 .../internal/api/queue_handlers.go            | 24 +++++++++++++++++++
 apps/runloop-engine/main.go                   |  1 +
 2 files changed, 25 insertions(+)

diff --git a/apps/runloop-engine/internal/api/queue_handlers.go b/apps/runloop-engine/internal/api/queue_handlers.go
index 27c3cbf..46bccb4 100644
--- a/apps/runloop-engine/internal/api/queue_handlers.go
+++ b/apps/runloop-engine/internal/api/queue_handlers.go
@@ -436,3 +436,27 @@ func (h *Handler) DeleteJob(c *fiber.Ctx) error {
 	}
 	return c.JSON(fiber.Map{"success": true})
 }
+
+// PurgeQueue removes every PENDING job from a queue. PROCESSING jobs are
+// left alone so we don't yank a row out from under an in-flight worker —
+// they'll naturally age out via visibility timeout if their handler dies.
+//
+// Use case: an operator updated a flow's code to fix a bug, but messages
+// enqueued under the old broken code are still occupying the queue and
+// being retried on a 5-minute visibility cycle until max_attempts is hit.
+// Without purge the only options are wait (~15+ minutes per stuck msg) or
+// rebind the queue to a new flow id. PG backend only — Redis/Kafka/Rabbit
+// have their own native purge mechanisms exposed by the broker UI.
+func (h *Handler) PurgeQueue(c *fiber.Ctx) error {
+	name := c.Params("name")
+	tag, err := h.db.Pool.Exec(c.Context(),
+		`DELETE FROM job_queue_items WHERE queue_name=$1 AND status='PENDING'`, name,
+	)
+	if err != nil {
+		return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": err.Error()})
+	}
+	return c.JSON(fiber.Map{
+		"success": true,
+		"deleted": tag.RowsAffected(),
+	})
+}
diff --git a/apps/runloop-engine/main.go b/apps/runloop-engine/main.go
index e9968ad..439f930 100644
--- a/apps/runloop-engine/main.go
+++ b/apps/runloop-engine/main.go
@@ -447,6 +447,7 @@ func setupRoutes(app *fiber.App, handler *api.Handler, webhookHandler *webhook.H
 	apiGroup.Get("/queues/:name/stats", handler.QueueStats)
 	apiGroup.Post("/queues/:name/jobs/:id/retry", handler.RetryJob)
 	apiGroup.Delete("/queues/:name/jobs/:id", handler.DeleteJob)
+	apiGroup.Post("/queues/:name/purge", handler.PurgeQueue)
 
 	// Plugins (SDK-style extension nodes)
 	apiGroup.Get("/plugins", handler.ListPlugins)