Add task queue with dedup and concurrency control
Publish Image / publish (push) Successful in 54s

- Tasks keyed by {taskType}-{repo}-{issue} for deduplication
- Duplicate requests (pending or running) are dropped
- Configurable concurrency via QUEUE_CONCURRENCY env (default 2)
- Workspace names now {taskType}-{repo}-{issue} for observability
- GET /queue endpoint shows pending/running tasks
- All handlers route through the queue instead of calling Coder directly

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Daniel Samson
2026-04-12 13:30:27 +01:00
co-authored by Claude Opus 4.6
parent 6062d1b518
commit f23fb4e2e0
10 changed files with 168 additions and 107 deletions
+5 -8
View File
@@ -1,14 +1,14 @@
import type { Context } from "hono";
import type { IssueCommentEvent, TaskRequest } from "../types.js";
import type { CoderClient } from "../services/coder.js";
import type { TaskQueue } from "../queue.js";
import type { Config } from "../config.js";
/**
* Issue comment created → re-trigger analyst if still in analysis.
* Issue comment created or edited → re-trigger analyst if still in analysis.
*
* Replaces: issue-comment-reply.json
*/
export function issueComment(config: Config, coder: CoderClient) {
export function issueComment(config: Config, queue: TaskQueue) {
return async (c: Context) => {
const event = await c.req.json<IssueCommentEvent>();
@@ -43,11 +43,8 @@ export function issueComment(config: Config, coder: CoderClient) {
giteaToken: config.giteaDevToken,
};
console.log(`[issue-comment] #${task.issueNumber} → re-trigger analyst`);
const queued = queue.enqueue(task);
const workspace = await coder.createWorkspace(task);
console.log(`[issue-comment] workspace created: ${workspace.name}`);
return c.json({ ok: true, workspace: workspace.name });
return c.json({ ok: true, queued });
};
}
+7 -15
View File
@@ -1,7 +1,6 @@
import type { Context } from "hono";
import type { IssueEvent, TaskRequest } from "../types.js";
import type { CoderClient } from "../services/coder.js";
import type { GiteaClient } from "../services/gitea.js";
import type { TaskQueue } from "../queue.js";
import type { Config } from "../config.js";
const LABEL_TO_TASK: Record<string, string> = {
@@ -19,7 +18,7 @@ const REVIEW_TASKS = new Set(["test"]);
*
* Replaces: issue-stage-transition.json
*/
export function issueLabel(config: Config, coder: CoderClient, gitea: GiteaClient) {
export function issueLabel(config: Config, queue: TaskQueue) {
return async (c: Context) => {
const event = await c.req.json<IssueEvent>();
@@ -55,18 +54,11 @@ export function issueLabel(config: Config, coder: CoderClient, gitea: GiteaClien
giteaToken,
};
console.log(`[issue-label] #${task.issueNumber} → ${taskType}`);
const queued = queue.enqueue(task, {
useReviewAccount: false,
body: `🤖 SDLC stage transition: **${taskType}** — workspace queued.`,
});
const workspace = await coder.createWorkspace(task);
console.log(`[issue-label] workspace created: ${workspace.name}`);
await gitea.commentOnIssue(
org,
repo,
event.issue.number,
`🤖 SDLC stage transition: **${taskType}** — workspace created.`,
);
return c.json({ ok: true, workspace: workspace.name });
return c.json({ ok: true, queued });
};
}
+7 -15
View File
@@ -1,7 +1,6 @@
import type { Context } from "hono";
import type { IssueEvent, TaskRequest } from "../types.js";
import type { CoderClient } from "../services/coder.js";
import type { GiteaClient } from "../services/gitea.js";
import type { TaskQueue } from "../queue.js";
import type { Config } from "../config.js";
/**
@@ -9,7 +8,7 @@ import type { Config } from "../config.js";
*
* Replaces: gitea-issue-triage.json
*/
export function issueTriage(config: Config, coder: CoderClient, gitea: GiteaClient) {
export function issueTriage(config: Config, queue: TaskQueue) {
return async (c: Context) => {
const event = await c.req.json<IssueEvent>();
@@ -32,18 +31,11 @@ export function issueTriage(config: Config, coder: CoderClient, gitea: GiteaClie
giteaToken: config.giteaDevToken,
};
console.log(`[issue-triage] #${task.issueNumber} → ${taskType}`);
const queued = queue.enqueue(task, {
useReviewAccount: false,
body: `🤖 Issue received. Starting **${taskType}** stage.`,
});
const workspace = await coder.createWorkspace(task);
console.log(`[issue-triage] workspace created: ${workspace.name}`);
await gitea.commentOnIssue(
org,
repo,
event.issue.number,
`🤖 Issue received. Starting **${taskType}** stage.`,
);
return c.json({ ok: true, workspace: workspace.name });
return c.json({ ok: true, queued });
};
}
+4 -18
View File
@@ -1,4 +1,4 @@
import type { CoderClient } from "../services/coder.js";
import type { TaskQueue } from "../queue.js";
import type { Config } from "../config.js";
import type { TaskRequest } from "../types.js";
@@ -9,15 +9,11 @@ export interface MaintenanceRepo {
}
/**
* Weekly maintenance cron → create a maintenance workspace.
* Weekly maintenance cron → enqueue a maintenance workspace per repo.
*
* Replaces: scheduled-maintenance.json
*/
export function runMaintenance(
config: Config,
coder: CoderClient,
repos: MaintenanceRepo[],
) {
export function runMaintenance(config: Config, queue: TaskQueue, repos: MaintenanceRepo[]) {
return async () => {
for (const entry of repos) {
const task: TaskRequest = {
@@ -29,17 +25,7 @@ export function runMaintenance(
giteaToken: config.giteaDevToken,
};
try {
const workspace = await coder.createWorkspace(task);
console.log(
`[maintenance] ${entry.org}/${entry.repo} → workspace: ${workspace.name}`,
);
} catch (err) {
console.error(
`[maintenance] failed for ${entry.org}/${entry.repo}:`,
err,
);
}
queue.enqueue(task);
}
};
}
+4 -7
View File
@@ -1,6 +1,6 @@
import type { Context } from "hono";
import type { PullRequestEvent, TaskRequest } from "../types.js";
import type { CoderClient } from "../services/coder.js";
import type { TaskQueue } from "../queue.js";
import type { GiteaClient } from "../services/gitea.js";
import type { Config } from "../config.js";
@@ -10,7 +10,7 @@ import type { Config } from "../config.js";
*
* Replaces: pr-review-trigger.json
*/
export function prReview(config: Config, coder: CoderClient, gitea: GiteaClient) {
export function prReview(config: Config, queue: TaskQueue, gitea: GiteaClient) {
return async (c: Context) => {
const event = await c.req.json<PullRequestEvent>();
@@ -40,11 +40,8 @@ export function prReview(config: Config, coder: CoderClient, gitea: GiteaClient)
giteaToken: config.giteaReviewToken,
};
console.log(`[pr-review] PR #${prNumber} → review-code`);
const queued = queue.enqueue(task);
const workspace = await coder.createWorkspace(task);
console.log(`[pr-review] workspace created: ${workspace.name}`);
return c.json({ ok: true, workspace: workspace.name });
return c.json({ ok: true, queued });
};
}
+7 -15
View File
@@ -1,7 +1,6 @@
import type { Context } from "hono";
import type { PullRequestReviewEvent, TaskRequest } from "../types.js";
import type { CoderClient } from "../services/coder.js";
import type { GiteaClient } from "../services/gitea.js";
import type { TaskQueue } from "../queue.js";
import type { Config } from "../config.js";
/**
@@ -9,7 +8,7 @@ import type { Config } from "../config.js";
*
* Replaces: pr-review-rework.json
*/
export function prRework(config: Config, coder: CoderClient, gitea: GiteaClient) {
export function prRework(config: Config, queue: TaskQueue) {
return async (c: Context) => {
const event = await c.req.json<PullRequestReviewEvent>();
@@ -43,18 +42,11 @@ export function prRework(config: Config, coder: CoderClient, gitea: GiteaClient)
giteaToken,
};
console.log(`[pr-rework] PR #${prNumber} ${reviewState} → ${taskType}`);
const queued = queue.enqueue(task, {
useReviewAccount: false,
body: `🤖 Review outcome: **${taskType}** — workspace queued.`,
});
const workspace = await coder.createWorkspace(task);
console.log(`[pr-rework] workspace created: ${workspace.name}`);
await gitea.commentOnIssue(
org,
repo,
prNumber,
`🤖 Review outcome: starting **${taskType}** stage.`,
);
return c.json({ ok: true, workspace: workspace.name });
return c.json({ ok: true, queued });
};
}
+7 -15
View File
@@ -1,7 +1,6 @@
import type { Context } from "hono";
import type { IssueEvent, TaskRequest } from "../types.js";
import type { CoderClient } from "../services/coder.js";
import type { GiteaClient } from "../services/gitea.js";
import type { TaskQueue } from "../queue.js";
import type { Config } from "../config.js";
/**
@@ -10,7 +9,7 @@ import type { Config } from "../config.js";
*
* Replaces: release-process.json
*/
export function release(config: Config, coder: CoderClient, gitea: GiteaClient) {
export function release(config: Config, queue: TaskQueue) {
return async (c: Context) => {
const event = await c.req.json<IssueEvent>();
@@ -52,18 +51,11 @@ export function release(config: Config, coder: CoderClient, gitea: GiteaClient)
giteaToken: config.giteaDevToken,
};
console.log(`[release] #${task.issueNumber} → release`);
const queued = queue.enqueue(task, {
useReviewAccount: false,
body: `🤖 Release process started.`,
});
const workspace = await coder.createWorkspace(task);
console.log(`[release] workspace created: ${workspace.name}`);
await gitea.commentOnIssue(
org,
repo,
event.issue.number,
`🤖 Release process started.`,
);
return c.json({ ok: true, workspace: workspace.name });
return c.json({ ok: true, queued });
};
}