website/server/src/utils/announceWorker.js

// ── 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-<id>` 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 }