diff --git a/src/handlers/issue-comment.ts b/src/handlers/issue-comment.ts index 1f7bbb2..b9afa64 100644 --- a/src/handlers/issue-comment.ts +++ b/src/handlers/issue-comment.ts @@ -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(); @@ -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 }); }; } diff --git a/src/handlers/issue-label.ts b/src/handlers/issue-label.ts index a0aa2e0..00ab17d 100644 --- a/src/handlers/issue-label.ts +++ b/src/handlers/issue-label.ts @@ -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 = { @@ -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(); @@ -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 }); }; } diff --git a/src/handlers/issue-triage.ts b/src/handlers/issue-triage.ts index 24f2ce2..330886d 100644 --- a/src/handlers/issue-triage.ts +++ b/src/handlers/issue-triage.ts @@ -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(); @@ -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 }); }; } diff --git a/src/handlers/maintenance.ts b/src/handlers/maintenance.ts index 9036c0a..0f0227b 100644 --- a/src/handlers/maintenance.ts +++ b/src/handlers/maintenance.ts @@ -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); } }; } diff --git a/src/handlers/pr-review.ts b/src/handlers/pr-review.ts index 8fc86b1..78f3f82 100644 --- a/src/handlers/pr-review.ts +++ b/src/handlers/pr-review.ts @@ -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(); @@ -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 }); }; } diff --git a/src/handlers/pr-rework.ts b/src/handlers/pr-rework.ts index c9e47a9..c5fca88 100644 --- a/src/handlers/pr-rework.ts +++ b/src/handlers/pr-rework.ts @@ -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(); @@ -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 }); }; } diff --git a/src/handlers/release.ts b/src/handlers/release.ts index 5579596..cec6906 100644 --- a/src/handlers/release.ts +++ b/src/handlers/release.ts @@ -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(); @@ -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 }); }; } diff --git a/src/index.ts b/src/index.ts index 1caac32..6310191 100644 --- a/src/index.ts +++ b/src/index.ts @@ -6,6 +6,7 @@ import cron from "node-cron"; import { loadConfig } from "./config.js"; import { CoderClient } from "./services/coder.js"; import { GiteaClient } from "./services/gitea.js"; +import { TaskQueue } from "./queue.js"; import { issueTriage } from "./handlers/issue-triage.js"; import { issueLabel } from "./handlers/issue-label.js"; @@ -23,26 +24,30 @@ const config = loadConfig(); const coderClient = new CoderClient(config); const giteaClient = new GiteaClient(config); +const concurrency = parseInt(process.env.QUEUE_CONCURRENCY || "2", 10); +const queue = new TaskQueue(coderClient, giteaClient, concurrency); + const app = new Hono(); app.use("*", logger()); // --------------------------------------------------------------------------- -// Health check +// Health check + queue status // --------------------------------------------------------------------------- app.get("/health", (c) => c.json({ status: "ok" })); +app.get("/queue", (c) => c.json(queue.status())); // --------------------------------------------------------------------------- // Webhook endpoints — same paths as the n8n webhooks so Gitea config // can stay unchanged (just point to new host). // --------------------------------------------------------------------------- -app.post("/webhook/gitea-issue-triage", issueTriage(config, coderClient, giteaClient)); -app.post("/webhook/gitea-issue-label", issueLabel(config, coderClient, giteaClient)); -app.post("/webhook/gitea-issue-comment", issueComment(config, coderClient)); -app.post("/webhook/gitea-pr-review", prReview(config, coderClient, giteaClient)); -app.post("/webhook/gitea-pr-review-rework", prRework(config, coderClient, giteaClient)); -app.post("/webhook/gitea-release", release(config, coderClient, giteaClient)); +app.post("/webhook/gitea-issue-triage", issueTriage(config, queue)); +app.post("/webhook/gitea-issue-label", issueLabel(config, queue)); +app.post("/webhook/gitea-issue-comment", issueComment(config, queue)); +app.post("/webhook/gitea-pr-review", prReview(config, queue, giteaClient)); +app.post("/webhook/gitea-pr-review-rework", prRework(config, queue)); +app.post("/webhook/gitea-release", release(config, queue)); // --------------------------------------------------------------------------- // Scheduled maintenance cron (Monday 9:00 AM) @@ -50,7 +55,7 @@ app.post("/webhook/gitea-release", release(config, coderClient, giteaClient)); const maintenanceRepos = parseMaintenanceRepos(); if (maintenanceRepos.length > 0) { - const maintenanceFn = runMaintenance(config, coderClient, maintenanceRepos); + const maintenanceFn = runMaintenance(config, queue, maintenanceRepos); cron.schedule("0 9 * * 1", () => { console.log("[cron] running weekly maintenance"); @@ -69,7 +74,7 @@ if (maintenanceRepos.length > 0) { // --------------------------------------------------------------------------- serve({ fetch: app.fetch, port: config.port }, (info) => { - console.log(`SDLC Orchestrator listening on :${info.port}`); + console.log(`SDLC Orchestrator listening on :${info.port} (concurrency: ${concurrency})`); }); // --------------------------------------------------------------------------- diff --git a/src/queue.ts b/src/queue.ts new file mode 100644 index 0000000..b9f1d20 --- /dev/null +++ b/src/queue.ts @@ -0,0 +1,111 @@ +import type { TaskRequest } from "./types.js"; +import type { CoderClient } from "./services/coder.js"; +import type { GiteaClient } from "./services/gitea.js"; + +interface QueuedTask { + task: TaskRequest; + comment?: { useReviewAccount: boolean; body: string }; +} + +/** + * Task queue with deduplication and concurrency control. + * + * - Tasks are keyed by `{taskType}-{repo}-{issue}` for dedup + * - If a task with the same key is already pending or running, new requests are dropped + * - Concurrency is configurable (defaults to 2) + */ +export class TaskQueue { + private pending = new Map(); + private running = new Set(); + private concurrency: number; + private coder: CoderClient; + private gitea: GiteaClient; + + constructor(coder: CoderClient, gitea: GiteaClient, concurrency = 2) { + this.coder = coder; + this.gitea = gitea; + this.concurrency = concurrency; + } + + /** Build a dedup key for a task. */ + static key(task: TaskRequest): string { + return `${task.taskType}-${task.giteaRepo}-${task.issueNumber}`; + } + + /** Build the workspace name (matches the dedup key). */ + static workspaceName(task: TaskRequest): string { + return `${task.taskType}-${task.giteaRepo}-${task.issueNumber}`; + } + + /** + * Enqueue a task. Returns true if queued, false if deduplicated (dropped). + */ + enqueue( + task: TaskRequest, + comment?: { useReviewAccount: boolean; body: string }, + ): boolean { + const key = TaskQueue.key(task); + + if (this.running.has(key)) { + console.log(`[queue] dropped (running): ${key}`); + return false; + } + + if (this.pending.has(key)) { + console.log(`[queue] dropped (pending): ${key}`); + return false; + } + + this.pending.set(key, { task, comment }); + console.log( + `[queue] enqueued: ${key} (pending: ${this.pending.size}, running: ${this.running.size}/${this.concurrency})`, + ); + + this.drain(); + return true; + } + + /** Process queued tasks up to the concurrency limit. */ + private drain(): void { + while (this.running.size < this.concurrency && this.pending.size > 0) { + const [key, entry] = this.pending.entries().next().value!; + this.pending.delete(key); + this.running.add(key); + this.process(key, entry); + } + } + + private async process(key: string, entry: QueuedTask): Promise { + const { task, comment } = entry; + + try { + console.log(`[queue] processing: ${key}`); + const workspace = await this.coder.createWorkspace(task); + console.log(`[queue] workspace created: ${workspace.name}`); + + if (comment) { + await this.gitea.commentOnIssue( + task.giteaOrg, + task.giteaRepo, + task.issueNumber, + comment.body, + comment.useReviewAccount, + ); + } + } catch (err) { + console.error(`[queue] failed: ${key}`, err); + } finally { + this.running.delete(key); + this.drain(); + } + } + + /** Current queue status for health/debug endpoints. */ + status(): { pending: string[]; running: string[]; concurrency: number } { + return { + pending: [...this.pending.keys()], + running: [...this.running.keys()], + concurrency: this.concurrency, + }; + } +} diff --git a/src/services/coder.ts b/src/services/coder.ts index 470386d..0a89e1d 100644 --- a/src/services/coder.ts +++ b/src/services/coder.ts @@ -1,5 +1,6 @@ import type { Config } from "../config.js"; import type { TaskRequest } from "../types.js"; +import { TaskQueue } from "../queue.js"; export class CoderClient { private baseUrl: string; @@ -15,11 +16,7 @@ export class CoderClient { } async createWorkspace(task: TaskRequest): Promise<{ id: string; name: string }> { - const timestamp = new Date() - .toISOString() - .replace(/[-:T]/g, "") - .slice(8, 14); // HHmmss - const name = `${task.taskType}-${task.issueNumber}-${timestamp}`; + const name = TaskQueue.workspaceName(task); const body = { name,