News post → town crier + Discord announcement pipeline
Replace the fire-and-forget Discord-only announce on publish with a
retry-safe, two-leg pipeline. When a post transitions into published-news
(false→true publish while in news, or category→news while published), an
announce_jobs row is enqueued with two INDEPENDENT delivery legs:
• town crier — sidecar POST /towncrier via uoLinkClient (stable id
`post-<id>` so a retry replaces rather than duplicates)
• discord — bot POST /internal/announce via botInternalClient
(single source of truth for the #news channel stays in the bot)
An in-process poller (utils/announceWorker) sweeps the table every
ANNOUNCE_POLL_MS and dispatches each due leg with its own exponential
backoff (30s→2h, 6 attempts). A leg is retried on transient failures
(503/504/network) and failed fast on data/config errors (400 over-cap,
401/409). Publishing never blocks on the sidecar or Discord — enqueue is
local DB only. Parent `status` is a done/partial/failed rollup of the two
legs; posts.announced_at is stamped once both deliver.
Admin visibility: GET /admin/posts/:id/announce + a per-leg Retry
(POST .../announce/retry) surfaced in the PostEditor for news posts.
Pure decisions (text build/caps, classification, backoff, rollup) live in
announceJobs.logic and are unit-tested (server/test/announceJobs.test.js,
10 tests). The old manual /admin/uo-link/towncrier form is untouched.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0114TpmrNW4wNXsHq5CR72jQ
This commit is contained in:
116
server/src/model/announceJobs/announceJobs.model.js
Normal file
116
server/src/model/announceJobs/announceJobs.model.js
Normal file
@@ -0,0 +1,116 @@
|
||||
// ── 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 both legs land, and resets a leg for the admin retry button.
|
||||
// The pure decisions (backoff, rollup, classification) live in .logic.js.
|
||||
|
||||
const db = require('./announceJobs.db')
|
||||
const logic = require('./announceJobs.logic')
|
||||
const posts = require('../posts/posts.model')
|
||||
const log = require('../../utils/logger')('announce')
|
||||
|
||||
// Enqueue an announcement for a freshly-published news post: one job row (both
|
||||
// legs 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)
|
||||
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
|
||||
// logic.classify*'s 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 attempts = Number(job[`${leg}_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 two legs; stamp the post's
|
||||
// announced_at the moment both legs have delivered.
|
||||
async function refreshStatus(jobId) {
|
||||
const job = await db.findById(jobId)
|
||||
if (!job) return null
|
||||
const status = logic.rollupStatus(job.towncrier_status, job.discord_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 (!db.LEGS.includes(leg)) throw new Error(`unknown announce leg: ${leg}`)
|
||||
const job = await db.findByPostId(postId)
|
||||
if (!job) return null
|
||||
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 })
|
||||
return refreshStatus(job.id)
|
||||
}
|
||||
|
||||
async function getByPostId(postId) {
|
||||
return db.findByPostId(postId)
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
enqueue,
|
||||
shouldEnqueue,
|
||||
enqueueIfNeeded,
|
||||
recordOutcome,
|
||||
refreshStatus,
|
||||
resetLeg,
|
||||
getByPostId,
|
||||
}
|
||||
Reference in New Issue
Block a user