website/server/src/engagement/engine.js

// ── The engagement engine ──────────────────────────────────────────────────
//
// ENGAGEMENT.md Phase 4a. `ctx.events.emit` validated a payload against a
// declaration and stopped (Phase 2); this is what it now hands the validated
// event to. The engine's whole job is to answer, for one event, **who gets told,
// on what, and not too often** - and then to write that down as outbox rows.
// It never delivers: `engagementWorker` drains the outbox, and what actually
// carries a message arrives with the channels' `deliver` in Phases 6 and 7.
//
// **The order of the gates is the design, and each one is here because skipping
// it is a way to mail the wrong people or too many of them:**
//
//   1. enabled rules for this trigger      - nothing is seeded, nothing is on by default
//   2. conditions                          - is this particular firing interesting
//   3. audience -> user ids                - core's tables, or a composed segment
//   4. ceiling re-check (G24)              - re-run at SEND time, not only at save
//   5. per-channel preference              - a user's own opt-in, effective mode
//   6. per-rule hourly ceiling (§7.1 Q3)   - the hard stop that makes rules-as-data safe
//   7. cooldown, per (rule, user, subject) - one statement, so two emits cannot race
//   8. enqueue, deduped                    - a replayed event is one row, not two
//
// Steps 6 and 7 are in that order deliberately. The hourly ceiling is about the
// RULE and is the thing that stops a mail storm; the cooldown is about one
// recipient and one subject. Checking the cheap global bound before consuming a
// per-recipient cooldown slot means a rule that has hit its ceiling does not also
// silently burn everybody's cooldowns on sends that never happen.
//
// **Nothing here throws at its caller.** It is invoked from inside a game-event
// handler by way of `ctx.events.emit`, and a database problem of core's must not
// become a module's control flow (the same posture the emit validator takes).

const rulesDb = require('../model/engagement/engagementRules.db')
const outboxDb = require('../model/engagement/engagementOutbox.db')
const cooldownsDb = require('../model/engagement/engagementCooldowns.db')
const sendsDb = require('../model/engagement/engagementSends.db')
const recipients = require('../model/engagement/engagementRecipients.db')
const conditions = require('./conditions')
const audiences = require('./audiences')
const channels = require('./channels')
const scopedPrefs = require('./scopedPrefs')
const log = require('../utils/logger')('engagement')

const HOUR_MS = 60 * 60 * 1000

// Channels one of whose payloads can REFERENCE another's result, earliest first
// (Phase 7). Only one pair qualifies today: a push tickle's `ref` deep-links to
// the inbox row `inapp` writes, and the outbox is swept `ORDER BY due_at, id`,
// so enqueueing in-app first is what makes that ref resolve on the first pass
// rather than on a retry. Everything not named here keeps the operator's own
// order, which is the order the rules screen shows.
//
// It is an ordering, not a dependency: `pushChannel` treats a missing ref as
// null and the app pulls regardless, so a rule that names only push, or a row
// that gets retried out of sequence, is still correct.
const CHANNEL_ORDER = ['inapp']
const channelRank = (id) => {
  const i = CHANNEL_ORDER.indexOf(id)
  return i === -1 ? CHANNEL_ORDER.length : i
}

/**
 * Which of a rule's channels are actually deliverable right now?
 *
 * A rule stores channel ids as data (`channels JSON`), so it can name one whose
 * module has been removed since. An unregistered channel is dropped rather than
 * failing the rule: the other channels of that rule are still correct, and a
 * dropped one is visible in the log line below.
 */
const liveChannels = (rule) =>
  (rule.channels || [])
    .filter((c) => channels.has(c))
    .sort((a, b) => channelRank(a) - channelRank(b))

/**
 * The EFFECTIVE mode each candidate holds for (id, channel), given the event's
 * scope.
 *
 * Effective, not stored: a row exists only where a user has expressed something,
 * and absence means the channel's `defaultMode` (§3.1). Reading the stored rows
 * and applying the default here keeps that answer in the registry, which is the
 * invariant Phase 3 established.
 *
 * **A scoped preference wins outright where one exists** (Phase 6, decision 4).
 * `team_notification_prefs` stayed where it is and `scopedPrefs` is the adapter;
 * for a Team-scoped event that table is the preference, exactly as it has been
 * since Teams shipped. The argument for replacing rather than intersecting is in
 * scopedPrefs.js's header, and it is short: intersecting would have silenced
 * every existing Team-email subscriber on the deploy that migrated them.
 */
async function effectiveModes(userIds, streamId, channel, scopeKey) {
  const modes = new Map()
  if (!userIds.length) return modes
  const scoped = await scopedPrefs.resolve(userIds, channel, scopeKey)
  const stored = await recipients.storedModes(userIds, streamId, channel)
  const fallback = channels.defaultMode(channel)
  for (const id of userIds) modes.set(id, scoped.get(id) ?? stored.get(id) ?? fallback)
  return modes
}

/**
 * The candidates who should get an OUTBOX ROW for this channel.
 *
 * `off` is excluded for the obvious reason. **`digest` is excluded too, and that
 * corrects what Phase 4a said here** — its comment read "a 'digest' preference is
 * kept, not dropped… what changes in Phase 6 is who drains it", and what changed
 * in Phase 6 is that nothing drains it. §4.2b keeps `teamDigestWorker`'s
 * compute-at-send-time design, so a digest is re-derived from the source tables
 * when it goes out, not assembled from snapshots taken hours earlier. An outbox
 * row for a digest recipient would be a second copy of the content with none of
 * the three properties that design exists for — most importantly, it would mail
 * a user who lost access between the post and the send.
 */
async function subscribedTo(userIds, streamId, channel, scopeKey = null) {
  const modes = await effectiveModes(userIds, streamId, channel, scopeKey)
  return userIds.filter((id) => modes.get(id) === 'instant')
}

/**
 * Run one rule against one event. Returns a small summary, for the log line and
 * for tests; it is not read by the caller for control flow.
 */
async function applyRule(rule, event, now) {
  const summary = {
    ruleId: rule.id,
    enqueued: 0,
    deduped: 0,
    cooled: 0,
    capped: 0,
    // Phase 9: people a CHANNEL refused to enqueue for, by reason. Counted
    // separately from `cooled` and `capped` because those are the engine holding
    // a message back and this is a channel saying it cannot carry one at all.
    ineligible: {},
    skipped: null,
  }

  if (!conditions.evaluate(rule.conditions, event.data)) {
    summary.skipped = 'conditions'
    return summary
  }

  const resolved = await audiences.resolveForRule(rule, event)
  if (resolved.dormant) {
    summary.skipped = resolved.reason || 'dormant'
    return summary
  }
  if (!resolved.userIds.length) {
    summary.skipped = resolved.reason || 'empty audience'
    return summary
  }

  // G24, re-run at send time. A rule saved when its trigger permitted a wider
  // audience must not keep reaching it after a module upgrade narrowed the
  // declaration - and that was the only way this could fail, since the save path
  // ran the same check.
  //
  // **Phase 10 gave it a second way, and it is the one that fires in practice:**
  // the event may carry a narrowing ceiling of its own. A rehearsal emits
  // `event.run.started` with `ceiling: 'staff'`, and every rule an operator wrote
  // for the real thing is then refused here rather than mailing subscribers about
  // an event that is not happening. Nothing about the rule changed; the occasion
  // did. See `audiences.permitted`.
  if (!audiences.permitted(event.triggerId, resolved.ceiling, event.ceiling)) {
    log.warn('rule audience exceeds its trigger ceiling - refusing', {
      rule: rule.id,
      trigger: event.triggerId,
      audience: resolved.ceiling,
      emitted: event.ceiling || null,
    })
    summary.skipped = 'ceiling'
    return summary
  }

  const live = liveChannels(rule)
  if (!live.length) {
    summary.skipped = 'no registered channel'
    return summary
  }

  // The per-rule hourly ceiling (§7.1 Q3). Counted once for the whole event
  // rather than per channel: an operator setting "100 an hour" means a hundred
  // messages, not a hundred per channel per event.
  const sentThisHour = await sendsDb.countSentSince(rule.id, new Date(now.getTime() - HOUR_MS))
  let budget = Math.max(0, rule.max_sends_per_hour - sentThisHour)
  if (budget === 0) {
    log.warn('rule is at its hourly send ceiling', {
      rule: rule.id,
      trigger: event.triggerId,
      ceiling: rule.max_sends_per_hour,
    })
    summary.skipped = 'hourly ceiling'
    return summary
  }

  const subjectKey = (event.subject ?? '').toString().slice(0, 190)
  const dueAt = new Date(now.getTime() + Math.max(0, rule.delay_seconds) * 1000)

  for (const channel of live) {
    // Phase 9, and it runs BEFORE the preference filter rather than after. Both
    // orders reach the same recipients; this one costs one query against the
    // narrower set only when the channel declares an `eligible` at all, and it
    // means `summary.ineligible` counts people the CHANNEL cannot reach rather
    // than people who happened to also be opted in. Channels that declare none —
    // push and in-app — pass straight through.
    const gated = await channels.eligibleFor(channel, resolved.userIds)
    for (const [why, n] of Object.entries(gated.excluded)) {
      summary.ineligible[why] = (summary.ineligible[why] || 0) + n
    }
    const eligible = await subscribedTo(gated.userIds, event.triggerId, channel, event.scopeKey)
    for (const userId of eligible) {
      if (budget <= 0) {
        summary.capped += 1
        continue
      }
      // Guarded on the interval, so two concurrent emits cannot both pass a
      // read-then-write check (§4.1).
      //
      // **Keyed on the CHANNEL as well**, which is what makes this loop correct
      // rather than what makes it work. Without the channel, the first channel of
      // a rule claims the cooldown and every later one is refused as cooling —
      // and `inapp` is ranked first above, so a rule naming email + in-app would
      // deliver the in-app item and silently never the mail. Found on Phase 11b's
      // live rig; a cooldown is per delivery, not per occasion.
      const allowed = await cooldownsDb.claim(
        rule.id, userId, subjectKey, channel, rule.cooldown_seconds, now,
      )
      if (!allowed) {
        summary.cooled += 1
        continue
      }
      const id = await outboxDb.enqueue({
        rule_id: rule.id,
        trigger_id: event.triggerId,
        user_id: userId,
        channel,
        subject_key: subjectKey,
        // The scope a PREFERENCE and an UNSUBSCRIBE are keyed on, which is not
        // `subject_key`: for the Team triggers the subject is `teamName` (what a
        // cooldown counts) and the scope is `team:12` (what survives a rename).
        scope_key: event.scopeKey ?? null,
        payload: event.data,
        // Scoped per (rule, user, channel) by the unique index, so one event
        // fanned out to fifty people is fifty rows carrying the same key.
        dedupe_key: event.dedupeKey,
        due_at: dueAt,
      })
      if (id === null) summary.deduped += 1
      else {
        summary.enqueued += 1
        budget -= 1
      }
    }
  }

  return summary
}

/**
 * Cancel pending rows that this event resolves (§4.2a).
 *
 * This is the actual point of `delay_seconds`: without cancellation a delay is
 * just a late mail. A house repaired back to LikeNew fires a trigger that some
 * rule names in its `cancel_on`, and every still-scheduled row for that
 * (rule, subject) stops.
 *
 * When the resolving event names an owner, only that user's rows are cancelled;
 * when it does not, every user queued about that subject is - which is the
 * house-repaired case, where the event is about the house and not about any one
 * of the people who were going to be told.
 */
async function applyCancellations(event, summary) {
  const rules = await rulesDb.enabledCancelledBy(event.triggerId)
  if (!rules.length) return
  const subjectKey = (event.subject ?? '').toString().slice(0, 190)
  for (const rule of rules) {
    const n = await outboxDb.cancel(rule.id, subjectKey, event.ownerUserId || null)
    if (n) {
      summary.cancelled += n
      log.info('cancelled scheduled sends', {
        rule: rule.id,
        by: event.triggerId,
        subject: subjectKey,
        rows: n,
      })
    }
  }
}

/**
 * Dispatch one validated event. Called by `engagementEmit.emit` after the payload
 * has been checked against the declaration.
 *
 * @param {object} event the envelope `engagementEmit` built
 * @returns {Promise<{ rules: number, enqueued: number, cancelled: number }>}
 */
async function dispatch(event, now = new Date()) {
  const summary = { rules: 0, enqueued: 0, deduped: 0, cooled: 0, capped: 0, cancelled: 0 }
  try {
    const rules = await rulesDb.enabledForTrigger(event.triggerId)
    summary.rules = rules.length

    for (const rule of rules) {
      const result = await applyRule(rule, event, now)
      summary.enqueued += result.enqueued
      summary.deduped += result.deduped
      summary.cooled += result.cooled
      summary.capped += result.capped
    }

    await applyCancellations(event, summary)

    // Keys and counts, never values - the same rule the emit log line follows.
    // A payload carries player names, house locations and forum excerpts, and a
    // log that reproduces them is a second copy of exactly the content
    // `engagement_sends` is careful to keep out of the database.
    if (summary.rules || summary.cancelled) {
      log.info('event dispatched', { trigger: event.triggerId, ...summary })
    }
  } catch (err) {
    // A database problem of core's must not become the module's control flow at
    // three in the morning. The emit already succeeded as a contract; what failed
    // is delivery, and it is logged as core's failure.
    log.error('dispatch failed', { trigger: event.triggerId, message: err.message })
  }
  return summary
}

module.exports = {
  dispatch,
  applyRule,
  applyCancellations,
  subscribedTo,
  effectiveModes,
  liveChannels,
  CHANNEL_ORDER,
  HOUR_MS,
}