// ── Announcement pipeline: SQL ───────────────────────────────────────────── // // Two tables since PR 4 (MODULE_SYSTEM.md §1.8): `announce_jobs` is one row per // publish event, `announce_job_legs` one row per delivery leg of that job. The // leg set is registered rather than fixed, so a leg is a stored VALUE now instead // of a group of leg-prefixed columns — which is what lets a module bring its own // leg without altering a core table. // // Every read returns the job with a `legs` array attached, so a caller never has // to remember to fetch the second table. const { query } = require('../../utils/db') const COLS = 'id, post_id, status, created_at, updated_at' const LEG_COLS = 'job_id, leg, status, attempts, last_error, next_attempt_at' async function legsFor(jobIds) { if (jobIds.length === 0) return new Map() const marks = jobIds.map(() => '?').join(', ') const rows = await query( `SELECT ${LEG_COLS} FROM announce_job_legs WHERE job_id IN (${marks}) ORDER BY job_id, leg`, jobIds, ) const byJob = new Map(jobIds.map((id) => [id, []])) for (const row of rows) byJob.get(row.job_id).push(row) return byJob } async function attachLegs(jobs) { const byJob = await legsFor(jobs.map((j) => j.id)) for (const job of jobs) job.legs = byJob.get(job.id) || [] return jobs } // Create a job and its leg rows in one go. `legs` is the registered leg id list — // an empty list is legal and yields a job with nothing to deliver. async function create(postId, legs = []) { const res = await query('INSERT INTO announce_jobs (post_id) VALUES (?)', [postId]) const jobId = Number(res.insertId) if (legs.length > 0) { const values = legs.map(() => '(?, ?)').join(', ') await query( `INSERT INTO announce_job_legs (job_id, leg) VALUES ${values}`, legs.flatMap((leg) => [jobId, leg]), ) } return jobId } // Add any registered legs this job is missing. A job enqueued before a module was // installed has no row for that module's leg, and without this it could never // deliver one — the worker only ever sees rows that exist. async function ensureLegs(jobId, legs = []) { if (legs.length === 0) return const values = legs.map(() => '(?, ?)').join(', ') await query( `INSERT IGNORE INTO announce_job_legs (job_id, leg) VALUES ${values}`, legs.flatMap((leg) => [jobId, leg]), ) } async function findById(id) { const rows = await query(`SELECT ${COLS} FROM announce_jobs WHERE id = ? LIMIT 1`, [id]) if (rows.length === 0) return null return (await attachLegs(rows))[0] } async function findByPostId(postId) { const rows = await query( `SELECT ${COLS} FROM announce_jobs WHERE post_id = ? ORDER BY id DESC LIMIT 1`, [postId], ) if (rows.length === 0) return null return (await attachLegs(rows))[0] } // Jobs with at least one leg that is due now: pending and either never scheduled // (next_attempt_at IS NULL — a fresh enqueue) or past its backoff time. Returns // whole jobs with every leg attached; the worker decides which legs to run, so // this stays one query regardless of how many legs are registered. async function findDue(now = new Date(), limit = 25) { const rows = await query( `SELECT ${COLS} FROM announce_jobs j WHERE EXISTS ( SELECT 1 FROM announce_job_legs l WHERE l.job_id = j.id AND l.status = 'pending' AND (l.next_attempt_at IS NULL OR l.next_attempt_at <= ?)) ORDER BY j.id ASC LIMIT ?`, [now, limit], ) return attachLegs(rows) } // Update one leg's row. `leg` is a bound VALUE, not an interpolated column name — // the reason the old leg allowlist that guarded that interpolation is gone. A // module's leg id could not have passed it anyway. async function updateLeg(jobId, leg, { status, attempts, lastError, nextAttemptAt }) { await query( `UPDATE announce_job_legs SET status = ?, attempts = ?, last_error = ?, next_attempt_at = ? WHERE job_id = ? AND leg = ?`, [status, attempts, lastError ?? null, nextAttemptAt ?? null, jobId, leg], ) } async function setStatus(id, status) { await query('UPDATE announce_jobs SET status = ? WHERE id = ?', [status, id]) } module.exports = { create, ensureLegs, findById, findByPostId, findDue, updateLeg, setStatus }