A webhook-driven code reviewer. When a pull request opens or updates, TurtleCode runs a local LLM over the diff and posts a review comment back to the PR.
Architecturally it is a standard queue + worker service: the webhook endpoint is a
producer, a Redis-backed queue buffers work, and workers consume jobs. The only unusual
part is what the worker does — it calls a (deliberately small, local) LLM — so the design
work is (1) the normal reliability of a queue/worker app and (2) guardrails that keep a
weak model honest. Nothing here is a demo; it is small scale, built to production standard.
See ARCHITECTURE.md for the full design.
GitHub ──webhook──▶ Web (producer) Worker (consumer)
verify HMAC reserve job
filter PR events skip if already done / superseded
normalize + enqueue fetch diff ─▶ chunk ─▶ LLM review (guarded)
return 202 upsert one comment on the PR
│ ▲
└────▶ Redis + BullMQ ────┘
(persistent, retries, DLQ)
The web tier returns 202 immediately — it never waits on the model, so a slow LLM can't
cause webhook timeouts. Everything slow or failure-prone happens in the worker, backed by a
durable queue with retries and a dead-letter queue.
- Ollama with a code model:
ollama pull codellama ollama serve
- Redis (backs the queue):
docker run -p 6379:6379 redis:7
- Dependencies:
npm init -y npm install express bullmq ioredis @octokit/rest zod dotenv
- Environment (
.env):GITHUB_WEBHOOK_SECRET=... # HMAC secret set on the GitHub webhook GITHUB_TOKEN=... # token with PR read + comment write REDIS_URL=redis://localhost:6379 OLLAMA_BASE_URL=http://localhost:11434 LLM_MODEL=codellama WORKER_CONCURRENCY=2 # global LLM concurrency = (#worker procs) * this
Run both roles in one process for local use, or separately to scale:
ROLE=all node index.js # web + worker
ROLE=web node index.js # producer only
ROLE=worker WORKER_CONCURRENCY=4 node index.js # add as many as the model can serverequire('dotenv').config();
const express = require('express');
const crypto = require('crypto');
const { Queue, Worker } = require('bullmq');
const IORedis = require('ioredis');
const { Octokit } = require('@octokit/rest');
const { z } = require('zod');
// --- 1. CONFIG ---
const cfg = {
role: process.env.ROLE || 'all', // 'web' | 'worker' | 'all'
port: Number(process.env.PORT || 3000),
redisUrl: process.env.REDIS_URL || 'redis://localhost:6379',
ollamaUrl: process.env.OLLAMA_BASE_URL || 'http://localhost:11434',
model: process.env.LLM_MODEL || 'codellama',
webhookSecret: process.env.GITHUB_WEBHOOK_SECRET || '',
githubToken: process.env.GITHUB_TOKEN || '',
concurrency: Number(process.env.WORKER_CONCURRENCY || 2), // global LLM cap = workers * this
llmTimeoutMs: Number(process.env.LLM_TIMEOUT_MS || 30000),
maxAttempts: Number(process.env.MAX_ATTEMPTS || 5),
};
const QUEUE_NAME = 'pr-reviews';
const DLQ_NAME = 'pr-reviews-dead';
// BullMQ requires a Redis connection with maxRetriesPerRequest: null
const connection = new IORedis(cfg.redisUrl, { maxRetriesPerRequest: null });
// --- 2. STRUCTURED LOGGING (keyed by deliveryId so one PR is traceable end to end) ---
const log = (level, deliveryId, msg, extra = {}) =>
console.log(JSON.stringify({ ts: new Date().toISOString(), level, deliveryId, msg, ...extra }));
// --- 3. QUEUE (Redis + BullMQ): persistent, dedup, backoff, dead-letter ---
const reviewQueue = new Queue(QUEUE_NAME, { connection });
const deadQueue = new Queue(DLQ_NAME, { connection });
const processedKey = (repo, pr, sha) => `processed:${repo}:${pr}:${sha}`;
// --- 4. WEB / PRODUCER: verify → filter → normalize → enqueue → 202 ---
function startWeb() {
const app = express();
// Keep the raw bytes so the HMAC is verified over exactly what GitHub signed
app.use(express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }));
app.post('/webhook', async (req, res) => {
const deliveryId = req.get('X-GitHub-Delivery') || '';
const event = req.get('X-GitHub-Event');
if (!verifySignature(req)) {
log('warn', deliveryId, 'invalid signature');
return res.status(401).json({ error: 'invalid signature' });
}
// Only actionable PR events; acknowledge and ignore everything else
const p = req.body;
const actionable = event === 'pull_request' &&
['opened', 'synchronize', 'reopened'].includes(p.action);
if (!actionable) return res.status(204).end();
const job = {
deliveryId,
repoOwner: p.repository.owner.login,
repoName: p.repository.name,
prNumber: p.pull_request.number,
headSha: p.pull_request.head.sha,
};
// jobId = deliveryId dedups redeliveries; attempts + backoff drive retries; keep fails for DLQ
await reviewQueue.add('review', job, {
jobId: deliveryId,
attempts: cfg.maxAttempts,
backoff: { type: 'exponential', delay: 2000 },
removeOnComplete: 1000,
removeOnFail: false,
});
log('info', deliveryId, 'queued', { repo: job.repoName, pr: job.prNumber });
res.status(202).json({ status: 'queued', deliveryId }); // return without waiting on the model
});
app.get('/healthz', (_req, res) => res.json({ ok: true }));
return app.listen(cfg.port, () => log('info', '-', `web listening on :${cfg.port}`));
}
function verifySignature(req) {
if (!cfg.webhookSecret) return false;
const sig = req.get('X-Hub-Signature-256') || '';
const expected = 'sha256=' + crypto
.createHmac('sha256', cfg.webhookSecret)
.update(req.rawBody)
.digest('hex');
const a = Buffer.from(sig);
const b = Buffer.from(expected);
return a.length === b.length && crypto.timingSafeEqual(a, b); // constant-time compare
}
// --- 5. WORKER / CONSUMER ---
function startWorker() {
const worker = new Worker(QUEUE_NAME, processReview, {
connection,
concurrency: cfg.concurrency, // global LLM concurrency = (#worker procs) * concurrency
});
worker.on('failed', async (job, err) => {
log('error', job?.data?.deliveryId, 'job failed', { attempt: job?.attemptsMade, err: err.message });
// Out of attempts → dead-letter it. Never dropped silently.
if (job && job.attemptsMade >= (job.opts.attempts || 1)) {
await deadQueue.add('dead', { ...job.data, error: err.message }, { removeOnComplete: false });
log('error', job.data.deliveryId, 'moved to DLQ');
}
});
log('info', '-', `worker up (concurrency ${cfg.concurrency})`);
return worker;
}
const octokit = () => new Octokit({ auth: cfg.githubToken });
async function processReview(job) {
const { deliveryId, repoOwner, repoName, prNumber, headSha } = job.data;
const gh = octokit();
// Idempotency: already reviewed this exact commit?
if (await connection.exists(processedKey(repoName, prNumber, headSha))) {
log('info', deliveryId, 'already processed; skipping');
return;
}
// Supersession: is this still the PR head? If a newer commit landed, drop the stale job.
const pr = await gh.pulls.get({ owner: repoOwner, repo: repoName, pull_number: prNumber });
if (pr.data.head.sha !== headSha) {
log('info', deliveryId, 'stale commit superseded; dropping', { was: headSha, now: pr.data.head.sha });
return;
}
// Fetch the real diff, split into small hunks, review each under guardrails
const diff = await gh.pulls.get({
owner: repoOwner, repo: repoName, pull_number: prNumber,
mediaType: { format: 'diff' },
});
const chunks = chunkDiff(diff.data);
const findings = [];
for (const chunk of chunks) {
const finding = await reviewChunk(chunk, deliveryId);
if (finding) findings.push(finding);
}
// One comment per PR, upserted by head SHA → retries/redeliveries converge, never duplicate
await upsertReviewComment(gh, { repoOwner, repoName, prNumber, headSha, findings });
await connection.set(processedKey(repoName, prNumber, headSha), '1', 'EX', 60 * 60 * 24 * 7);
log('info', deliveryId, 'review posted', { findings: findings.length });
}
// --- 6. DIFF CHUNKING ---
// A small model reviews better with small, scoped prompts. Split per file, then per hunk.
// Upgrade path: tree-sitter for language-aware, function-level units.
function chunkDiff(rawDiff) {
const files = rawDiff.split(/^diff --git .*$/m).filter(s => s.trim());
return files.flatMap(file => {
const nameMatch = file.match(/\+\+\+ b\/(.+)/);
const name = nameMatch ? nameMatch[1].trim() : 'unknown';
const hunks = file.split(/^(?=@@ )/m).filter(h => h.startsWith('@@'));
return hunks.map(h => ({ file: name, diff: h.slice(0, 4000) })); // cap prompt size
});
}
// --- 7. THE REVIEW STEP (small model → strict guardrails) ---
const FindingSchema = z.object({
severity: z.enum(['info', 'warning', 'critical']),
line: z.number().int().nonnegative(),
comment: z.string().min(1).max(500),
confidence: z.number().min(0).max(1),
});
async function reviewChunk(chunk, deliveryId) {
const prompt = buildPrompt(chunk);
// Malformed/invalid model output is retryable, handled HERE so one bad hunk degrades to
// "no finding" instead of failing (and re-running) the whole job.
for (let attempt = 1; attempt <= 3; attempt++) {
try {
const raw = await withTimeout(signal => callModel(prompt, signal), cfg.llmTimeoutMs);
const finding = FindingSchema.parse(JSON.parse(raw));
// Guardrail: the cited line must actually be in this hunk (anti-hallucination)
if (!chunkHasLine(chunk, finding.line)) throw new Error(`cited line ${finding.line} not in diff`);
// Guardrail: drop low-confidence noise
if (finding.confidence < 0.6) { log('info', deliveryId, 'suppressed low-confidence finding'); return null; }
return { ...finding, file: chunk.file };
} catch (err) {
log('warn', deliveryId, `chunk review attempt ${attempt} failed`, { err: err.message });
if (attempt === 3) return null; // give up on this hunk; keep the job alive
}
}
}
// Call Ollama's HTTP API directly. format:'json' constrains output to a JSON string.
async function callModel(prompt, signal) {
const res = await fetch(`${cfg.ollamaUrl}/api/generate`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ model: cfg.model, prompt, stream: false, format: 'json', options: { temperature: 0.1 } }),
signal,
});
if (!res.ok) throw new Error(`ollama ${res.status}`);
return (await res.json()).response;
}
function buildPrompt(chunk) {
return `You are a senior code reviewer. Review ONLY the diff hunk below.
Respond with a SINGLE JSON object and nothing else:
{"severity":"info|warning|critical","line":<number>,"comment":"<max 50 words>","confidence":<0..1>}
File: ${chunk.file}
Diff hunk:
---
${chunk.diff}
---`;
}
function chunkHasLine(chunk, line) {
// New-file line range lives in the hunk header: @@ -a,b +c,d @@
const m = chunk.diff.match(/@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@/);
if (!m) return true; // can't verify → don't over-reject
const start = Number(m[1]);
const len = Number(m[2] || 1);
return line >= start && line <= start + len;
}
// --- 8. TIMEOUT (real cancellation via AbortController, not just abandonment) ---
function withTimeout(fn, ms) {
const ac = new AbortController();
const timer = setTimeout(() => ac.abort(new Error('LLM timeout')), ms);
return Promise.resolve(fn(ac.signal)).finally(() => clearTimeout(timer));
}
// --- 9. OUTPUT (idempotent): one bot comment per PR, upserted by head SHA ---
async function upsertReviewComment(gh, { repoOwner, repoName, prNumber, headSha, findings }) {
const marker = `<!-- turtlecode:${headSha} -->`;
const body = `${marker}\n${renderReview(findings)}`;
const existing = await gh.issues.listComments({ owner: repoOwner, repo: repoName, issue_number: prNumber });
const mine = existing.data.find(c => c.body?.startsWith('<!-- turtlecode:'));
if (mine) {
await gh.issues.updateComment({ owner: repoOwner, repo: repoName, comment_id: mine.id, body });
} else {
await gh.issues.createComment({ owner: repoOwner, repo: repoName, issue_number: prNumber, body });
}
}
function renderReview(findings) {
if (!findings.length) return 'TurtleCode: no issues found.';
const rank = s => ({ info: 0, warning: 1, critical: 2 }[s] ?? 0);
const lines = findings
.sort((a, b) => rank(b.severity) - rank(a.severity))
.map(f => `- **${f.severity}** \`${f.file}:${f.line}\` — ${f.comment}`);
return `**TurtleCode review**\n\n${lines.join('\n')}`;
}
// --- 10. BOOTSTRAP + GRACEFUL SHUTDOWN ---
const servers = [];
let worker;
if (cfg.role === 'web' || cfg.role === 'all') servers.push(startWeb());
if (cfg.role === 'worker' || cfg.role === 'all') worker = startWorker();
log('info', '-', `TurtleCode started (role=${cfg.role})`);
async function shutdown(sig) {
log('info', '-', `shutting down (${sig})`);
await Promise.allSettled([
...servers.map(s => new Promise(r => s.close(r))),
worker?.close(),
reviewQueue.close(),
deadQueue.close(),
]);
await connection.quit();
process.exit(0);
}
process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));This is a plain queue/worker app; the notes below are just where the reliability lives.
- Decoupled producer/consumer. The webhook enqueues and returns
202. LLM latency (seconds) never touches the HTTP path, so GitHub never sees a slow webhook. - Durable, at-least-once queue. BullMQ on Redis persists jobs, reserves→acks so a crashed worker's job returns, and dedups redeliveries by
jobId = deliveryId. - No silent loss. Failures retry with exponential backoff; exhausted jobs go to a dead-letter queue for inspection/replay.
- Idempotency. A processed-key skips repeats, and the review is a single comment upserted by head SHA — retries and redeliveries converge to one comment instead of piling up.
- Bounded latency. Every model call is time-boxed and genuinely cancelled via
AbortController. - Supersession. If a newer commit landed, the stale job is dropped rather than posting a review for an old diff.
- Guardrails for a small model. Output is JSON-constrained and validated against a schema; a hallucinated line reference is rejected; low-confidence findings are suppressed; bad output is treated as a retryable error, never posted.
- Scaling. Web and worker are stateless roles. Global LLM concurrency is capped by
(#workers) × WORKER_CONCURRENCY, because the model server — not CPU — is the bottleneck.
- tree-sitter chunking — language-aware, function-level units instead of raw hunks, for large diffs.
- Structural context — if reviews need cross-file facts (callers, types), fetch them with the language server / grep / code search. That is exact and cheap.
- No vector DB. Semantic search returns plausible-but-wrong neighbors for code; the context a reviewer needs is structural and exact. A vector store is only worth it later, for fuzzy recall over a large history of past reviews — not at this scale.
- GitHub App auth — installation tokens instead of a PAT (least privilege, higher rate limits).
- Global rate limiter — a Redis token-bucket in front of the model for a hard fleet-wide cap independent of worker count.
- Observability — metrics (queue depth, retries, DLQ size, model p50/p95, cost per job) and alerts on DLQ growth / model unavailability.