kgod
ed172bf369
基于 OpenCodeReview 的 webhook 服务：监听 Gitea 的 push 与 Pull Request 事件， 调用 OCR 审查 diff，把结果发布回 Gitea，并按阻断阈值决定是否自动合并。 主要能力： - Push / PR 事件触发，支持分支 glob 过滤与 PR-only / push-only 范围 - PR 内联评论（按 diff 行号定位）、汇总评论、Issue 生命周期、提交状态 - 可配置阻断阈值（严重级别 / 类别 / 任意意见） - 无阻断问题时自动合并，审查覆盖不完整时拒绝合并 - 内置 Web 后台：仓库配置、任务日志、失败重跑、连通性自检 - SQLite 持久化，worker 重启回收卡死任务，失败自动重试 实现为独立服务而非 Gitea Action：本机 act_runner 指向的实例不可达， 且后台配置与任务历史需要独立进程承载。
79 lines
2.2 KiB
JavaScript
79 lines
2.2 KiB
JavaScript
/** Sequential job worker with retry and stale-job recovery. */
|
|
import { claimNextJob, updateJob } from "./db.js";
|
|
import { SkipJob } from "./review.js";
|
|
|
|
const STALE_MS = 2 * 60 * 60 * 1000;
|
|
|
|
export class JobQueue {
|
|
constructor({ db, engine, logger = console, pollMs = 3000, maxAttempts = 2 }) {
|
|
this.db = db;
|
|
this.engine = engine;
|
|
this.logger = logger;
|
|
this.pollMs = pollMs;
|
|
this.maxAttempts = maxAttempts;
|
|
this.running = false;
|
|
this.busy = false;
|
|
this.timer = null;
|
|
this.currentJobId = null;
|
|
}
|
|
|
|
start() {
|
|
if (this.running) return;
|
|
this.running = true;
|
|
this.recoverStale();
|
|
this.tick();
|
|
}
|
|
|
|
stop() {
|
|
this.running = false;
|
|
if (this.timer) clearTimeout(this.timer);
|
|
}
|
|
|
|
recoverStale() {
|
|
const cutoff = new Date(Date.now() - STALE_MS).toISOString().replace("T", " ").slice(0, 19);
|
|
const stale = this.db.prepare(
|
|
"SELECT id FROM jobs WHERE status = 'running' AND started_at IS NOT NULL AND started_at < ?",
|
|
).all(cutoff);
|
|
for (const row of stale) {
|
|
updateJob(this.db, row.id, {
|
|
status: "queued", phase: null,
|
|
error: "requeued after worker restart or timeout",
|
|
});
|
|
this.logger.warn?.(`requeued stale job #${row.id}`);
|
|
}
|
|
}
|
|
|
|
async tick() {
|
|
if (!this.running) return;
|
|
if (this.busy) {
|
|
this.timer = setTimeout(() => this.tick(), this.pollMs);
|
|
return;
|
|
}
|
|
const job = claimNextJob(this.db);
|
|
if (!job) {
|
|
this.timer = setTimeout(() => this.tick(), this.pollMs);
|
|
return;
|
|
}
|
|
this.busy = true;
|
|
this.currentJobId = job.id;
|
|
try {
|
|
await this.engine.execute(job);
|
|
} catch (err) {
|
|
if (err instanceof SkipJob) {
|
|
await this.engine.failJob(job, err);
|
|
} else if (job.attempts < this.maxAttempts) {
|
|
this.logger.warn?.(`job #${job.id} failed (attempt ${job.attempts}): ${err.message}; retrying`);
|
|
updateJob(this.db, job.id, {
|
|
status: "queued", phase: null,
|
|
error: `${err.name || "Error"}: ${err.message}`,
|
|
});
|
|
} else {
|
|
await this.engine.failJob(job, err);
|
|
}
|
|
} finally {
|
|
this.busy = false;
|
|
this.currentJobId = null;
|
|
}
|
|
this.timer = setTimeout(() => this.tick(), this.pollMs);
|
|
}
|
|
} |