// ── Announcement pipeline: orchestration ──────────────────────────────────── // // Sits between the DB rows and the worker: creates jobs on publish, records each // leg's outcome, keeps the parent `status` rollup in sync, stamps the post's // announced_at when every leg lands, and resets a leg for the admin retry button. // The pure decisions (backoff, rollup) live in .logic.js; which legs exist at all // is modules/registries.js's answer, not this file's (MODULE_SYSTEM.md §1.8). const db = require('./announceJobs.db') const logic = require('./announceJobs.logic') const registries = require('../../modules/registries') const posts = require('../posts/posts.model') const log = require('../../utils/logger')('announce') // Enqueue an announcement for a freshly-published news post: one job row, one leg // row per registered leg (all pending, due immediately), plus a back-pointer on // the post so the admin panel can find it. Returns the new job id. async function enqueue(postId) { const jobId = await db.create(postId, registries.announceLegIds()) await posts.linkAnnounceJob(postId, jobId) log.info('announce job enqueued', { jobId, postId }) return jobId } // Should publishing this post fire the pipeline? Only on a real transition INTO // "published news" — a false→true publish while in news, or a category change // into news while already published — and never twice (guarded by the post's // existing announce_job_id). Editing an already-announced post does not re-fire. function shouldEnqueue(post, { wasPublished, wasNews }) { if (!post || post.category !== 'news' || !post.published) return false if (post.announce_job_id) return false const wasNewsPublished = Boolean(wasPublished) && Boolean(wasNews) return !wasNewsPublished } // Convenience used by the post controller: enqueue iff shouldEnqueue. Never // throws — a pipeline hiccup must not break saving/publishing a post. async function enqueueIfNeeded(post, transition) { try { if (!shouldEnqueue(post, transition)) return null return await enqueue(post.id) } catch (err) { log.error('enqueueIfNeeded failed', { postId: post && post.id, message: err.message }) return null } } // Record a leg's dispatch outcome and refresh the rollup. `outcome` is one of a // leg's classify() results: 'done' | 'retry' | 'terminal'. For 'retry' we bump the // attempt count and schedule the next run (or fail the leg once the cap is hit). // Returns the updated job row. async function recordOutcome(job, leg, { outcome, error }) { const row = (job.legs || []).find((l) => l.leg === leg) const attempts = Number(row && row.attempts) || 0 if (outcome === 'done') { await db.updateLeg(job.id, leg, { status: 'done', attempts, lastError: null, nextAttemptAt: null }) } else if (outcome === 'terminal') { await db.updateLeg(job.id, leg, { status: 'failed', attempts: attempts + 1, lastError: error, nextAttemptAt: null }) log.warn('announce leg failed (terminal)', { jobId: job.id, leg, error }) } else { const nextAttempts = attempts + 1 const delay = logic.scheduleAfter(nextAttempts) if (delay === null) { await db.updateLeg(job.id, leg, { status: 'failed', attempts: nextAttempts, lastError: error, nextAttemptAt: null }) log.warn('announce leg failed (retries exhausted)', { jobId: job.id, leg, attempts: nextAttempts, error }) } else { const nextAttemptAt = new Date(Date.now() + delay) await db.updateLeg(job.id, leg, { status: 'pending', attempts: nextAttempts, lastError: error, nextAttemptAt }) log.info('announce leg retry scheduled', { jobId: job.id, leg, attempts: nextAttempts, nextAttemptAt }) } } return refreshStatus(job.id) } // Recompute and persist the parent status from the legs; stamp the post's // announced_at the moment every leg has delivered. async function refreshStatus(jobId) { const job = await db.findById(jobId) if (!job) return null const status = logic.rollupStatus(job.legs.map((l) => l.status)) if (status !== job.status) await db.setStatus(jobId, status) job.status = status if (status === 'done') { try { await posts.markAnnounced(job.post_id) } catch (err) { log.warn('markAnnounced failed', { jobId, postId: job.post_id, message: err.message }) } } return job } // Admin retry button: reset one leg to pending, clear its error/backoff, and let // the worker pick it up on the next tick. Resets the attempt count so a retry // after a config fix gets a full budget again. async function resetLeg(postId, leg) { if (!registries.announceLeg(leg)) throw new Error(`unknown announce leg: ${leg}`) const job = await db.findByPostId(postId) if (!job) return null // A job enqueued before this leg was registered has no row for it; create it so // the retry button works on an existing post after a module is installed. await db.ensureLegs(job.id, [leg]) await db.updateLeg(job.id, leg, { status: 'pending', attempts: 0, lastError: null, nextAttemptAt: null }) log.info('announce leg reset for retry', { jobId: job.id, postId, leg }) // Labelled, because this is the response body the admin panel re-renders from. return withLabels(await refreshStatus(job.id)) } // Decorate a job's legs with the label their registration carries, so the admin // panel renders a module's leg with a real name and no client change // (MODULE_SYSTEM.md §1.8). An unregistered leg — a stale row from a module that // was since removed — keeps its id as the label rather than disappearing. function withLabels(job) { if (!job) return job job.legs = (job.legs || []).map((l) => { const registered = registries.announceLeg(l.leg) return { ...l, label: registered ? registered.label : l.leg } }) return job } async function getByPostId(postId) { return withLabels(await db.findByPostId(postId)) } module.exports = { enqueue, shouldEnqueue, enqueueIfNeeded, recordOutcome, refreshStatus, resetLeg, getByPostId, withLabels, }