// ── Announcement dispatcher worker ────────────────────────────────────────── // // A lightweight, in-process table poller (no Redis/BullMQ in the stack). Every // ANNOUNCE_POLL_MS it sweeps announce_jobs for legs that are due — freshly // enqueued or past their backoff — and dispatches each one: // • town crier → uoLinkClient.postTownCrier (sidecar → in-game) // • discord → botInternalClient.announce (bot → #news channel) // Both clients never throw (they return { ok, status, error }); the model turns // each result into done / retry / terminal and owns the backoff + rollup. One // leg failing never touches the other. Same setInterval + unref + stop() shape // as middleware/botScore's sweeper, wired into server.js start/shutdown. const announceJobs = require('../model/announceJobs/announceJobs.model') const announceJobsDb = require('../model/announceJobs/announceJobs.db') const logic = require('../model/announceJobs/announceJobs.logic') const posts = require('../model/posts/posts.model') const uoLinkClient = require('./uoLinkClient') const botInternalClient = require('./botInternalClient') const log = require('./logger')('announce-worker') const POLL_MS = Number(process.env.ANNOUNCE_POLL_MS) || 15_000 const TOWNCRIER_DURATION_SEC = Number(process.env.TOWNCRIER_DURATION_SEC) || 3600 function baseUrl() { return (process.env.APP_BASE_URL || 'http://localhost:5173').replace(/\/+$/, '') } // ── Leg dispatchers ───────────────────────────────────────────────────────── // Return the raw client result ({ ok, status, data, error }); classification is // the model/logic's job. async function dispatchTownCrier(post) { const lines = logic.buildTownCrierText(post, { baseUrl: baseUrl() }) // Stable id: re-posting `post-` REPLACES the prior town-crier entry rather // than stacking a duplicate, so a retry after a partial failure is safe. return uoLinkClient.postTownCrier({ id: `post-${post.id}`, lines, durationSec: TOWNCRIER_DURATION_SEC, }) } async function dispatchDiscord(post) { const base = baseUrl() // Stored image paths are relative ("/uploads/x.png"); Discord embeds need an // absolute URL. const imageUrl = post.image_url ? new URL(post.image_url, base).toString() : null return botInternalClient.announce({ title: post.title, excerpt: post.excerpt, url: `${base}/site/news`, imageUrl, }) } // Process a single due leg of a job: fetch the post, dispatch, classify, record. async function processLeg(job, leg) { const post = await posts.getById(job.post_id) if (!post) { // Post was deleted between enqueue and dispatch (the CASCADE usually reaps // the job first, but guard anyway). Nothing to announce — fail the leg. await announceJobs.recordOutcome(job, leg, { outcome: 'terminal', error: 'post no longer exists' }) return } let result let classification try { if (leg === 'towncrier') { result = await dispatchTownCrier(post) classification = logic.classifyTownCrier(result) } else { result = await dispatchDiscord(post) classification = logic.classifyDiscord(result) } } catch (err) { // Clients shouldn't throw, but if one does, treat it as a transient failure // rather than crashing the tick. log.error('dispatch threw', { jobId: job.id, leg, message: err.message }) classification = { outcome: 'retry', error: err.message } } await announceJobs.recordOutcome(job, leg, classification) } // One sweep: find due jobs and process each due leg. A job may have both legs due // (a fresh enqueue) — process the ones that are actually pending. `job` is a // snapshot from the SELECT; recordOutcome re-reads for the rollup, so processing // the two legs sequentially off the same snapshot is fine (each leg only writes // its own columns). async function tick(now = new Date()) { let jobs try { jobs = await announceJobsDb.findDue(now) } catch (err) { log.error('failed to load due jobs', { message: err.message }) return } if (!jobs || jobs.length === 0) return for (const job of jobs) { if (isLegDue(job, 'towncrier', now)) await processLeg(job, 'towncrier') if (isLegDue(job, 'discord', now)) await processLeg(job, 'discord') } } function isLegDue(job, leg, now) { if (job[`${leg}_status`] !== 'pending') return false const next = job[`${leg}_next_attempt_at`] return next == null || new Date(next) <= now } let timer = null function start() { if (timer) return timer timer = setInterval(() => { tick().catch((err) => log.error('announce tick failed', { message: err.message })) }, POLL_MS) if (timer.unref) timer.unref() // don't keep the event loop alive (tests, shutdown) log.info('announcement dispatcher started', { pollMs: POLL_MS }) return timer } function stop() { if (timer) { clearInterval(timer) timer = null } } module.exports = { start, stop, tick, processLeg, dispatchTownCrier, dispatchDiscord }