Track running Granthi review queue jobs

src/lib/db.js

@@ -35,6 +35,7 @@ const DDL = [
     state TEXT NOT NULL DEFAULT 'queued',
     outcome TEXT,
     created_ts TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+    started_ts TEXT,
     finished_ts TEXT
   )`,
   `CREATE INDEX IF NOT EXISTS idx_queue_jobs_state ON queue_jobs(state)`
@@ -44,9 +45,17 @@ export function openDb(dbPath) {
   mkdirSync(dirname(dbPath), { recursive: true });
   const db = new DatabaseSync(dbPath);
   for (const stmt of DDL) db.prepare(stmt).run();
+  ensureColumn(db, 'queue_jobs', 'started_ts', 'TEXT');
   return db;
 }
 
+function ensureColumn(db, table, column, type) {
+  const columns = db.prepare(`PRAGMA table_info(${table})`).all();
+  if (!columns.some((c) => c.name === column)) {
+    db.prepare(`ALTER TABLE ${table} ADD COLUMN ${column} ${type}`).run();
+  }
+}
+
 export function insertReview(db, r) {
   const stmt = db.prepare(`
     INSERT INTO reviews (repo, sha, pr, event, pusher, authors, trailers, axes,
@@ -76,6 +85,13 @@ export function enqueueJob(db, job) {
   return res.lastInsertRowid;
 }
 
+export function startJob(db, id) {
+  db.prepare(
+    `UPDATE queue_jobs SET state = 'running',
+       started_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?`
+  ).run(id);
+}
+
 export function finishJob(db, id, outcome = null) {
   db.prepare(
     `UPDATE queue_jobs SET state = 'done', outcome = ?,

src/server.js

@@ -1,6 +1,6 @@
 import { createServer } from 'node:http';
 import { loadConfig } from './config.js';
-import { openDb, listReviews, enqueueJob, finishJob, recoverPendingJobs } from './lib/db.js';
+import { openDb, listReviews, enqueueJob, startJob, finishJob, recoverPendingJobs } from './lib/db.js';
 import { GiteaClient } from './lib/gitea.js';
 import { ReviewQueue } from './lib/queue.js';
 import { verifySignature, jobFromWebhook } from './lib/webhook.js';
@@ -34,7 +34,10 @@ const queue = new ReviewQueue();
 // whatever the outcome (runReview itself posts a final or fail-open status).
 function dispatchJob(queueRowId, job) {
   return queue.enqueue(`${job.owner}/${job.repo}`, (withGlobal) =>
-    runReview({ cfg, gitea, adminGitea, db, job, withGlobal, log })
+    {
+      startJob(db, queueRowId);
+      return runReview({ cfg, gitea, adminGitea, db, job, withGlobal, log });
+    }
   ).then(
     (r) => finishJob(db, queueRowId, r?.ok ? (r.verdict || 'ok') : `error:${String(r?.error || '').slice(0, 200)}`),
     (e) => {

test/persistqueue.test.js

@@ -1,6 +1,6 @@
 import { test } from 'node:test';
 import assert from 'node:assert/strict';
-import { openDb, enqueueJob, finishJob, recoverPendingJobs } from '../src/lib/db.js';
+import { openDb, enqueueJob, startJob, finishJob, recoverPendingJobs } from '../src/lib/db.js';
 
 const JOB = (over = {}) => ({
   event: 'push', owner: 'nirpa', repo: 'demo', sha: 'a'.repeat(40),
@@ -32,6 +32,19 @@ test('finished jobs are not recovered again', () => {
   assert.equal(recoverPendingJobs(db).length, 0);
 });
 
+test('running jobs are visible and recovered on startup', () => {
+  const db = openDb(':memory:');
+  const id = enqueueJob(db, JOB({ sha: '4'.repeat(40) }));
+  startJob(db, id);
+
+  const row = db.prepare(`SELECT state, started_ts FROM queue_jobs WHERE id = ?`).get(id);
+  assert.equal(row.state, 'running');
+  assert.match(row.started_ts, /^\d{4}-\d{2}-\d{2}T/);
+
+  const pending = recoverPendingJobs(db);
+  assert.deepEqual(pending.map(p => p.id), [id]);
+});
+
 test('unparseable persisted job is marked done, not re-run forever', () => {
   const db = openDb(':memory:');
   db.prepare(`INSERT INTO queue_jobs (repo, sha, job) VALUES ('a/b', 'c', '{broken')`).run();