website/server/src/utils/engagementWorker.js

// ── Engagement outbox worker ───────────────────────────────────────────────
//
// ENGAGEMENT.md §4.2a, Phase 4a. Every ENGAGEMENT_POLL_MS it sweeps
// `engagement_outbox` for rows whose `due_at` has passed, claims each one, hands
// it to its channel, and records the outcome in `engagement_sends`. Same
// setInterval + unref + stop() shape as `announceWorker` and the three Team
// sweepers, wired into server.js start/shutdown beside them.
//
// **Claiming is a compare-and-set, not a lock** (§7.1 Q2, settled by the org lead
// 2026-08-29 over `SELECT ... FOR UPDATE SKIP LOCKED`): an
// `UPDATE ... SET status='sending' WHERE id=? AND status='scheduled'`, and the
// instance the server reports `affectedRows = 1` to owns the row. No transaction
// to hold open, no MariaDB version floor, and it uses a status the ENUM already
// carried for exactly this. What it makes safe is the outbox; the four existing
// workers are still single-instance, so this does not by itself make the
// deployment multi-instance.
//
// **Nothing is delivered in this phase, and that is visible rather than
// pretended.** A channel's `deliver` arrives with email in Phase 6 and the in-app
// inbox in Phase 7; until then `channels.get(id)` has no such function, the row
// finishes as `failed` and the send log says why in as many words. The
// alternatives were both worse: recording 'sent' would be a lie in the one table
// whose whole purpose is answering "did they get it", and leaving the row
// scheduled would mean an IDOC warning queued today arriving three weeks later
// on the deploy that first shipped a mailer.
//
// In practice this path is unreachable on a real deployment for now: core seeds
// no rules and `enabled` defaults to 0, so the outbox stays empty until an
// operator turns a rule on from the screen Phase 4b builds.

const outboxDb = require('../model/engagement/engagementOutbox.db')
const sendsDb = require('../model/engagement/engagementSends.db')
const channels = require('../engagement/channels')
const log = require('./logger')('engagement-worker')

const POLL_MS = Number(process.env.ENGAGEMENT_POLL_MS) || 30_000
// How many rows one sweep will look at. A bound rather than a target: the sweep
// runs again in POLL_MS, and an unbounded batch is how a backlog turns one tick
// into a stall.
const BATCH = Number(process.env.ENGAGEMENT_BATCH) || 100

// A transient failure is retried with a flat backoff, and then given up on.
// Flat rather than exponential because `due_at` is also the grace window's clock
// and a doubling backoff would push a delayed message arbitrarily far past the
// moment it was about.
const MAX_ATTEMPTS = 5
const RETRY_MS = 5 * 60 * 1000

// A row claimed into 'sending' by a process that then died is invisible to every
// other sweeper - `status='scheduled'` will never match it again. This window is
// how long a claim may look alive before it is taken back; it has to be
// comfortably longer than the slowest legitimate send or a slow one gets sent
// twice.
const STALE_MS = 15 * 60 * 1000

/**
 * Deliver one claimed row.
 *
 * @returns {{ outcome: 'sent'|'retry'|'terminal'|'suppressed'|'bounced', detail?: string, transport?: string, addressHash?: string }}
 */
async function deliver(row) {
  const channel = channels.get(row.channel)
  if (!channel) {
    // The channel's module was removed between enqueue and now. Terminal: there
    // is nothing to retry towards, and leaving the row scheduled would make it
    // sweep forever.
    return { outcome: 'terminal', detail: `channel "${row.channel}" is no longer registered` }
  }
  if (typeof channel.deliver !== 'function') {
    return { outcome: 'terminal', detail: `channel "${row.channel}" has no delivery implementation yet` }
  }
  try {
    const result = await channel.deliver(row)
    // `addressHash` rides on every outcome, success or not: a bounce (Phase 9)
    // arrives with an address and has to find the row it belongs to, and the rows
    // worth correlating include the ones that already failed once.
    const hash = (result && result.addressHash) || undefined
    if (result && result.ok) {
      return { outcome: 'sent', transport: result.transport, detail: result.detail, addressHash: hash }
    }
    // Its own outcome rather than a flavour of 'terminal' (Phase 9). Both statuses
    // the ENUMs already carried for it say something a 'failed' row cannot: the
    // outbox row was not attempted, and the send log's `suppressed` is the
    // difference between "we tried and the relay refused" and "we declined to
    // try". An operator reading a screen of failures needs those separated, and
    // so does anybody counting deliverability.
    if (result && result.suppressed) {
      return { outcome: 'suppressed', detail: result.detail || 'suppressed', addressHash: hash }
    }
    // A hard bounce. Terminal like any other refusal, but recorded under its own
    // name: "the relay would not take this" and "this mailbox does not exist"
    // send an operator to two different places.
    if (result && result.bounced) {
      return { outcome: 'bounced', detail: result.detail || 'hard bounce', addressHash: hash }
    }
    if (result && result.retry) {
      return { outcome: 'retry', detail: result.detail || 'transient failure', addressHash: hash }
    }
    return { outcome: 'terminal', detail: (result && result.detail) || 'delivery refused', addressHash: hash }
  } catch (err) {
    // A channel shouldn't throw, but if one does it is a transient failure
    // rather than a crashed tick - announceWorker's posture with its legs.
    log.error('channel deliver threw', { outbox: row.id, channel: row.channel, message: err.message })
    return { outcome: 'retry', detail: err.message }
  }
}

/**
 * Claim, deliver, record. One row, start to finish.
 *
 * `deliverFn` is injectable so a test can drive the retry/give-up path without a
 * channel that fails on demand - the alternative is registering a fake channel,
 * which would make the registry, not this function, the thing under test.
 */
async function processRow(row, now = new Date(), deliverFn = deliver) {
  if (!(await outboxDb.claim(row.id))) return null // another sweeper got there first

  const result = await deliverFn(row)

  if (result.outcome === 'retry' && row.attempts + 1 < MAX_ATTEMPTS) {
    await outboxDb.reschedule(row.id, new Date(now.getTime() + RETRY_MS), result.detail)
    return 'retry'
  }

  // **The two tables diverge here, deliberately.** `engagement_outbox.status` is
  // the ROW's lifecycle and its ENUM has no 'bounced' - from the queue's point of
  // view a bounced message is a row that finished unsuccessfully, which is
  // 'failed'. `engagement_sends.status` is what happened to the MESSAGE, and
  // there 'bounced' is the whole point: it is the difference between "look at
  // your relay" and "this person's mailbox is gone".
  let status = 'failed'
  if (result.outcome === 'sent') status = 'sent'
  else if (result.outcome === 'suppressed') status = 'suppressed'
  else if (result.outcome === 'bounced') status = 'bounced'
  const outboxStatus = status === 'bounced' ? 'failed' : status
  await outboxDb.finish(row.id, outboxStatus, status === 'sent' ? null : result.detail)
  // The send log is written for every terminal outcome, not only success. G15's
  // question is "did user X get the mail?", and "no, and here is why" is an
  // answer that table has to be able to give.
  await sendsDb.record({
    outbox_id: row.id,
    rule_id: row.rule_id,
    trigger_id: row.trigger_id,
    user_id: row.user_id,
    channel: row.channel,
    transport: result.transport ?? null,
    address_hash: result.addressHash ?? null,
    status,
    detail: result.detail ?? null,
  })
  return status
}

async function tick(now = new Date()) {
  try {
    await outboxDb.reclaimStale(new Date(now.getTime() - STALE_MS))
  } catch (err) {
    log.error('failed to reclaim stale rows', { message: err.message })
  }

  let due
  try {
    due = await outboxDb.findDue(now, BATCH)
  } catch (err) {
    log.error('failed to load due rows', { message: err.message })
    return
  }
  if (!due || !due.length) return

  const counts = { sent: 0, failed: 0, suppressed: 0, bounced: 0, retry: 0, taken: 0 }
  for (const row of due) {
    try {
      const outcome = await processRow(row, now)
      if (outcome === null) counts.taken += 1
      else counts[outcome] += 1
    } catch (err) {
      log.error('row failed', { outbox: row.id, message: err.message })
    }
  }
  log.info('outbox swept', counts)
}

let timer = null

function start() {
  if (timer) return timer
  timer = setInterval(() => {
    tick().catch((err) => log.error('engagement tick failed', { message: err.message }))
  }, POLL_MS)
  if (timer.unref) timer.unref() // don't keep the event loop alive (tests, shutdown)
  log.info('engagement outbox worker started', { pollMs: POLL_MS, batch: BATCH })
  return timer
}

function stop() {
  if (timer) {
    clearInterval(timer)
    timer = null
  }
}

module.exports = { start, stop, tick, processRow, deliver, POLL_MS, MAX_ATTEMPTS, RETRY_MS, STALE_MS }