Review multiple PRs in parallel

README.md

@@ -37,6 +37,8 @@ Hare is a CodeRabbit-style reviewer you host yourself:
    - Posts a walkthrough + inline findings as the bot user
    - Sets the **`Hare`** commit status (`success` or `failure`)
 
+Webhooks return immediately and reviews run on a queue (up to 4 PRs in parallel). Sync reviews every open PR on watched repos, not a cap of 3.
+
 Findings are graded:
 
 | Severity | Meaning | Merge gate |

package.json

@@ -26,8 +26,8 @@
     "preview:stop": "node scripts/preview.mjs stop",
     "typecheck": "tsc --noEmit",
     "check:auth": "node scripts/check-auth-invariant.mjs",
-    "test": "node --test 'scripts/**/*.test.mjs' && node --experimental-strip-types --test src/lib/app-data/app-data.test.ts src/lib/app-data/readiness-schedule.test.ts src/lib/auth/gate-identity.test.ts src/lib/auth/sign-in-gate.test.ts src/lib/hare/migrations.test.ts",
-    "test:ci": "node --experimental-strip-types --test src/lib/app-data/app-data.test.ts src/lib/app-data/readiness-schedule.test.ts src/lib/auth/gate-identity.test.ts src/lib/auth/sign-in-gate.test.ts src/lib/hare/migrations.test.ts",
+    "test": "node --test 'scripts/**/*.test.mjs' && node --experimental-strip-types --test src/lib/app-data/app-data.test.ts src/lib/app-data/readiness-schedule.test.ts src/lib/auth/gate-identity.test.ts src/lib/auth/sign-in-gate.test.ts src/lib/hare/migrations.test.ts src/lib/hare/queue.test.ts",
+    "test:ci": "node --experimental-strip-types --test src/lib/app-data/app-data.test.ts src/lib/app-data/readiness-schedule.test.ts src/lib/auth/gate-identity.test.ts src/lib/auth/sign-in-gate.test.ts src/lib/hare/migrations.test.ts src/lib/hare/queue.test.ts",
     "lint": "eslint .",
     "format": "prettier --write ."
   },

src/lib/hare/engine.ts

@@ -36,6 +36,7 @@ import {
   suggestContextPaths,
 } from "./reviewer";
 import { mergeMigrationFindings, scanMigrationIssues } from "./migrations";
+import { isStaleRunning } from "./queue";
 import type { ChangedFile, ReviewerOutput } from "./types";
 
 export function newSecret(): string {
@@ -261,7 +262,10 @@ export async function reviewPullForUser(input: {
 
   if (!input.force) {
     const existing = await getReviewBySha(pr.id, pr.head_sha);
-    if (existing?.status === "complete" || existing?.status === "running") {
+    if (existing?.status === "complete") {
+      return { ok: true };
+    }
+    if (existing?.status === "running" && !isStaleRunning(existing.createdAt)) {
       return { ok: true };
     }
   }

src/lib/hare/queue.test.ts

@@ -0,0 +1,42 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+import { isStaleRunning, jobKey, mapPool } from "./queue.ts";
+
+describe("jobKey", () => {
+  it("is unique per user and PR", () => {
+    assert.equal(
+      jobKey({ userId: "u1", owner: "acme", repo: "api", number: 18 }),
+      "u1:acme/api#18",
+    );
+    assert.notEqual(
+      jobKey({ userId: "u1", owner: "acme", repo: "api", number: 18 }),
+      jobKey({ userId: "u1", owner: "acme", repo: "api", number: 19 }),
+    );
+  });
+});
+
+describe("isStaleRunning", () => {
+  it("treats missing timestamps as stale", () => {
+    assert.equal(isStaleRunning(null), true);
+    assert.equal(isStaleRunning(""), true);
+  });
+  it("keeps a fresh running review", () => {
+    assert.equal(isStaleRunning(new Date().toISOString()), false);
+  });
+  it("expires an 8+ minute running review", () => {
+    const old = new Date(Date.now() - 9 * 60 * 1000).toISOString();
+    assert.equal(isStaleRunning(old), true);
+  });
+});
+
+describe("mapPool", () => {
+  it("runs every item and preserves order", async () => {
+    const seen: number[] = [];
+    const out = await mapPool([1, 2, 3, 4, 5], 2, async (n) => {
+      seen.push(n);
+      return n * 10;
+    });
+    assert.deepEqual(out, [10, 20, 30, 40, 50]);
+    assert.equal(seen.length, 5);
+  });
+});

src/lib/hare/queue.ts

@@ -0,0 +1,85 @@
+
+export type ReviewJob = {
+  userId: string;
+  owner: string;
+  repo: string;
+  number: number;
+  force?: boolean;
+};
+
+const CONCURRENCY = 4;
+const STALE_MS = 8 * 60 * 1000;
+
+type QueueState = {
+  pending: ReviewJob[];
+  inflight: Set<string>;
+  active: number;
+};
+
+const g = globalThis as typeof globalThis & { __hareReviewQueue?: QueueState };
+
+function state(): QueueState {
+  if (!g.__hareReviewQueue) {
+    g.__hareReviewQueue = { pending: [], inflight: new Set(), active: 0 };
+  }
+  return g.__hareReviewQueue;
+}
+
+export function jobKey(job: ReviewJob): string {
+  return `${job.userId}:${job.owner}/${job.repo}#${job.number}`;
+}
+
+export function isStaleRunning(createdAt: string | null | undefined): boolean {
+  if (!createdAt) return true;
+  const t = Date.parse(createdAt);
+  if (!Number.isFinite(t)) return true;
+  return Date.now() - t > STALE_MS;
+}
+
+export function enqueueReview(job: ReviewJob): boolean {
+  const s = state();
+  const key = jobKey(job);
+  if (s.inflight.has(key) || s.pending.some((p) => jobKey(p) === key)) return false;
+  s.pending.push(job);
+  pump();
+  return true;
+}
+
+export async function mapPool<T, R>(
+  items: T[],
+  limit: number,
+  fn: (item: T) => Promise<R>,
+): Promise<R[]> {
+  if (items.length === 0) return [];
+  const out: R[] = new Array(items.length);
+  let cursor = 0;
+  async function worker() {
+    while (cursor < items.length) {
+      const idx = cursor++;
+      out[idx] = await fn(items[idx]!);
+    }
+  }
+  const n = Math.min(Math.max(1, limit), items.length);
+  await Promise.all(Array.from({ length: n }, () => worker()));
+  return out;
+}
+
+function pump(): void {
+  const s = state();
+  while (s.active < CONCURRENCY && s.pending.length > 0) {
+    const job = s.pending.shift()!;
+    const key = jobKey(job);
+    s.inflight.add(key);
+    s.active += 1;
+    void import("./engine")
+      .then(({ reviewPullForUser }) => reviewPullForUser(job))
+      .catch((err) => {
+        console.error("[hare] queued review failed", key, err);
+      })
+      .finally(() => {
+        s.inflight.delete(key);
+        s.active -= 1;
+        pump();
+      });
+  }
+}

src/lib/hare/server.ts

@@ -23,6 +23,7 @@ import {
   reviewPullForUser,
   seedDemoReview,
 } from "./engine";
+import { mapPool } from "./queue";
 import {
   getAuthenticatedUser,
   getPullDiff,
@@ -31,7 +32,7 @@ import {
   listUserRepos,
 } from "./github";
 
-const MAX_AUTO_REVIEWS = 3;
+const REVIEW_CONCURRENCY = 4;
 
 export const getDashboard = createServerFn({ method: "GET" })
   .middleware([authMiddleware])
@@ -226,15 +227,13 @@ export const syncInbox = createServerFn({ method: "POST" })
       }
     }
 
-    let reviewed = 0;
-    for (const item of pending) {
-      if (reviewed >= MAX_AUTO_REVIEWS) break;
-      const result = await reviewPullForUser({
+    const results = await mapPool(pending, REVIEW_CONCURRENCY, (item) =>
+      reviewPullForUser({
         userId: context.userId,
         ...item,
-      });
-      if (result.ok) reviewed += 1;
-    }
+      }),
+    );
+    const reviewed = results.filter((r) => r.ok).length;
 
     return { ok: true as const, reviewed, pulled, errors };
   });

src/lib/hare/webhook.ts

@@ -1,6 +1,6 @@
 import { createHmac, timingSafeEqual } from "node:crypto";
-import { findWatchedByRepo } from "./db";
-import { runReviewFromWebhook } from "./server";
+import { findWatchedByRepo, upsertPull } from "./db";
+import { enqueueReview } from "./queue";
 
 export function verifyGithubSignature(
   secret: string,
@@ -80,33 +80,37 @@ export async function handleGithubWebhook(
     return { status: 401, body: { ok: false, message: "invalid signature" } };
   }
 
-  let ran = 0;
+  let queued = 0;
   for (const watcher of targets) {
     if (!watcher.autoReview) continue;
-    await runReviewFromWebhook({
+    await upsertPull(watcher.userId, {
+      owner,
+      repo,
+      number: pr.number,
+      title: pr.title,
+      body: pr.body,
+      author: pr.user?.login ?? "unknown",
+      state: pr.state,
+      draft: Boolean(pr.draft),
+      htmlUrl: pr.html_url,
+      headSha: pr.head.sha,
+      baseSha: pr.base.sha,
+      headRef: pr.head.ref,
+      baseRef: pr.base.ref,
+      additions: pr.additions ?? 0,
+      deletions: pr.deletions ?? 0,
+      changedFiles: pr.changed_files ?? 0,
+      isDemo: false,
+      githubUpdatedAt: pr.updated_at ?? new Date().toISOString(),
+    });
+    enqueueReview({
       userId: watcher.userId,
       owner,
       repo,
       number: pr.number,
-      pull: {
-        title: pr.title,
-        body: pr.body,
-        author: pr.user?.login ?? "unknown",
-        state: pr.state,
-        draft: Boolean(pr.draft),
-        htmlUrl: pr.html_url,
-        headSha: pr.head.sha,
-        baseSha: pr.base.sha,
-        headRef: pr.head.ref,
-        baseRef: pr.base.ref,
-        additions: pr.additions ?? 0,
-        deletions: pr.deletions ?? 0,
-        changedFiles: pr.changed_files ?? 0,
-        updatedAt: pr.updated_at ?? new Date().toISOString(),
-      },
     });
-    ran += 1;
+    queued += 1;
   }
 
-  return { status: 200, body: { ok: true, message: `reviewed ${ran}` } };
+  return { status: 200, body: { ok: true, message: `queued ${queued}` } };
 }

src/routes/api/github/action.ts

@@ -1,8 +1,9 @@
 import { createFileRoute } from "@tanstack/react-router";
 import { findWatched } from "@/lib/hare/db";
 import { getPull } from "@/lib/hare/github";
-import { getConnection } from "@/lib/hare/db";
-import { runReviewFromWebhook, userIdForActionSecret } from "@/lib/hare/server";
+import { getConnection, upsertPull } from "@/lib/hare/db";
+import { enqueueReview } from "@/lib/hare/queue";
+import { userIdForActionSecret } from "@/lib/hare/server";
 
 async function post({ request }: { request: Request }) {
   const secret = request.headers.get("x-hare-secret")?.trim();
@@ -34,29 +35,28 @@ async function post({ request }: { request: Request }) {
     return Response.json({ ok: false, message: "GitHub is not connected" }, { status: 400 });
   }
   const pr = await getPull(conn.token, owner, repo, number);
-  const result = await runReviewFromWebhook({
-    userId,
+  await upsertPull(userId, {
     owner,
     repo,
     number,
-    pull: {
-      title: pr.title,
-      body: pr.body,
-      author: pr.user?.login ?? "unknown",
-      state: pr.state,
-      draft: Boolean(pr.draft),
-      htmlUrl: pr.html_url,
-      headSha: pr.head.sha,
-      baseSha: pr.base.sha,
-      headRef: pr.head.ref,
-      baseRef: pr.base.ref,
-      additions: pr.additions ?? 0,
-      deletions: pr.deletions ?? 0,
-      changedFiles: pr.changed_files ?? 0,
-      updatedAt: pr.updated_at,
-    },
+    title: pr.title,
+    body: pr.body,
+    author: pr.user?.login ?? "unknown",
+    state: pr.state,
+    draft: Boolean(pr.draft),
+    htmlUrl: pr.html_url,
+    headSha: pr.head.sha,
+    baseSha: pr.base.sha,
+    headRef: pr.head.ref,
+    baseRef: pr.base.ref,
+    additions: pr.additions ?? 0,
+    deletions: pr.deletions ?? 0,
+    changedFiles: pr.changed_files ?? 0,
+    isDemo: false,
+    githubUpdatedAt: pr.updated_at,
   });
-  return Response.json(result, { status: result.ok ? 200 : 500 });
+  enqueueReview({ userId, owner, repo, number });
+  return Response.json({ ok: true, queued: true, number });
 }
 
 export const Route = createFileRoute("/api/github/action")({