feat(engagement): declare 24 shard triggers and 3 audiences (Phase 11a)

module-uo's half of ENGAGEMENT.md Phase 11: every trigger DECLARATION, the
wire-kind mapping that fires them, and the three registered audiences. No rule
and no template is seeded here -- that is 11b -- so nothing this adds sends
anybody anything until an operator writes a rule.

server/config/shardTriggers.js declares the 24, grouped by the audience kind
each family exercises, and every variable carries the `example` the template
editor previews and test-sends with. Ceilings: 10 `owner`, 2 `members`, 7
`authenticated`, 2 `staff`, 3 `admin` (the value core adds in the same window).
`uo.cheat.detected` at `staff` is the declaration the lattice exists for.

server/utils/shardEngagement.js maps the wire to those ids, hung off
shardIngest.ingest beside the SSE broadcast and the push tickle, and reads like
shardPush.js on purpose -- owner resolution is why neither can be a pure mapper.
Three things live here because a rule cannot express them:

  * Transitions. champ.update and city.update are full-state upserts, so without
    a per-process tracker a sidecar reconnect reads as twenty spawns starting.
    A FIRST sighting is never a transition.
  * Thresholds. conditions.js compares a declared variable against a LITERAL, so
    "within 24 hours of dismissal" is not expressible; and vendor.listing is a
    sweep frame re-emitted on any price change, so per-frame would flood. The
    crossing is tracked here and `hoursRemaining` is declared so an operator can
    still narrow with `is at most`.
  * The members audience. "The members of THIS guild" differs every firing, so
    it travels on the envelope as recipientUserIds (Phase 6 decision 2).

**The fan-out runs BEFORE the state write, and that ordering is load-bearing.**
account.unlinked drops the shard_account_links row that names the one person who
needs to be told; house.remove drops the house whose stored ownerAcct is the only
place a collapsed house's owner appears; guild.leave/remove need the roster and
board mirrors to name who left. Resolving afterwards finds nobody, every time.

Four rows of 8.6 deliberately do not ship, each with its reason recorded in
docs (docs#194): uo.market.item_listed (a saved search, no per-user query store),
uo.guild.joined (core's team.member.joined already fires for it -- a UO guild IS
a Team and this module is the provider), uo.link.requested (no addressable
recipient by construction, ~5-minute TTL), and uo.points.rank_changed's personal
half (top[] names a serial, links are keyed by account).

coreApi -> ^1.8.0: the module now calls registerEventTriggers and declares
`ceiling: 'admin'`, so a 1.7.0 core would refuse the ceiling and a 1.6.0 one
would not have the method at all.

39 new tests; 509/509 pass. check:imports, check:bundle and check:swagger clean.

Co-Authored-By: Claude <noreply@anthropic.com>

server/utils/shardEngagement.js

@@ -0,0 +1,781 @@
+// ── Shard event → engagement trigger ───────────────────────────────────────
+//
+// ENGAGEMENT.md Phase 11. The third fan-out off `shardIngest.ingest`, beside the
+// SSE broadcast and the push tickle, and the one that produces a PER-PERSON
+// notification subject to a rule, a preference and a suppression. It is the twin
+// of `shardPush.js` and reads deliberately like it:
+//
+//   • `shardStreams.mapShardEvent` turns a frame into push targets;
+//     `mapShardEvent` here turns a frame into engagement events.
+//   • Owner resolution is why neither can be a pure mapper: an owner-keyed target
+//     names a GAME ACCOUNT, and turning that into a website user needs
+//     `shardLinks`. An unlinked account is simply nobody to notify.
+//
+// **Nothing here decides who is told.** It says what happened and (for an
+// owner- or members-shaped event) who it is ABOUT; the engine applies the rules,
+// the ceiling, the preferences and the suppression list. That split is the
+// module boundary: a module cannot send mail (§1.2) and this is not the back door.
+//
+// **Never throws, never blocks ingest.** `ingest()` calls this fire-and-forget
+// exactly as it calls the broadcast and the push dispatch, and every mapper below
+// is wrapped so one bad frame cannot stop the feed. This is the same reason the
+// C# side's `Emit()` enqueues and returns rather than touching the socket from the
+// Core thread.
+//
+// ── Three things that are NOT a plain field mapping ────────────────────────
+//
+// Most of §8.6's rows are "read four fields off the frame and emit". Three are
+// not, and each is here rather than in a rule because a rule cannot express it:
+//
+//   1. **Transitions.** `champ.update` and `city.update` are full-state UPSERTS
+//      re-emitted on any change, not discrete "started"/"elected" events. Without
+//      a per-process transition tracker, a reconnect snapshot is read as twenty
+//      champion spawns starting at once. `shardStreams.js` already solved this for
+//      push and this file uses the same shape — and the same rule that a FIRST
+//      sighting is never a transition.
+//   2. **Thresholds.** `uo.vendor.expiring` and `uo.economy.milestone` fire when a
+//      value CROSSES a line. `conditions.js` compares a declared variable against
+//      a literal and has no relative-time or previous-value operator, so
+//      "within 24 hours of dismissal" and "gold passed a billion" are not
+//      expressible as conditions — and `vendor.listing` is a sweep frame
+//      re-emitted on every price change, so emitting per frame would flood. The
+//      crossing is tracked here; the operator still narrows with
+//      `hoursRemaining is at most N`.
+//   3. **Audience resolution for `members`.** A guild event is about the members
+//      of THAT guild, which is a different answer for every firing and therefore
+//      cannot be a saved segment (whose params are constants). The access-checked
+//      set travels on the envelope as `recipientUserIds` — Phase 6's decision 2,
+//      and the mechanism the Team fan-out was built on.
+
+const shardLinks = require('../model/shardLinks/shardLinks.model')
+const shardState = require('../model/shardState/shardState.model')
+const { TRIGGER_IDS } = require('../config/shardTriggers')
+const core = require('../core')
+
+const log = core.logger('shard-engagement')
+
+// ── Thresholds ─────────────────────────────────────────────────────────────
+
+// When a vendor becomes "expiring". Hours rather than pay periods, because a pay
+// period is a real day under the new vendor system and a UO day (~2 real hours)
+// under the old one — the exact factor-of-twelve trap `docs/link/v5.md` records,
+// and the reason the wire carries `dismissalAt` as an instant.
+//
+// 48 hours is one full real day of warning even on a shard whose owner logs in
+// daily, and it is the OUTER edge: the mapper fires once on the way in, and the
+// operator narrows further with `hoursRemaining is at most 24` if they want less.
+const VENDOR_WARN_HOURS = 48
+
+// Gold-supply reporting lines, ascending. Crossing one in either direction is one
+// `uo.economy.milestone`. They are the module's rather than the operator's for
+// now: an admin-configurable ladder is a settings surface, and this phase's job
+// is the trigger. An operator who wants a different line writes a rule condition
+// on `value`.
+const GOLD_THRESHOLDS = [
+  100_000_000, 250_000_000, 500_000_000, 1_000_000_000,
+  2_500_000_000, 5_000_000_000, 10_000_000_000,
+]
+
+// The same, for account count.
+const ACCOUNT_THRESHOLDS = [100, 250, 500, 1000, 2500, 5000, 10_000]
+
+// Which line a value sits above, as an index. -1 means "below the first".
+const bandOf = (value, thresholds) => {
+  let band = -1
+  for (let i = 0; i < thresholds.length; i += 1) if (value >= thresholds[i]) band = i
+  return band
+}
+
+// ── The transition tracker ─────────────────────────────────────────────────
+
+/**
+ * Per-process state for the upsert kinds and the threshold kinds.
+ *
+ * Injectable so a test gets a fresh one; a module-level default backs the live
+ * dispatcher. It is deliberately NOT persisted: its whole job is to say "has
+ * this process seen a previous value", and a value restored from a database
+ * would make the first frame after a restart a transition against state the
+ * shard may have left behind hours ago.
+ */
+function createTracker() {
+  return {
+    champActive: new Map(), // spawn serial → boolean
+    champBossUp: new Map(), // spawn serial → boolean
+    cityGovernor: new Map(), // city → governor serial or null
+    cityPhase: new Map(), // city → electionPhase
+    vendorWarned: new Map(), // vendor serial → boolean (already inside the window)
+    pointsLeader: new Map(), // points system → leader serial
+    economyBand: new Map(), // metric → band index
+    serverUp: null, // boolean or null (never seen)
+  }
+}
+const defaultTracker = createTracker()
+
+/** Reset the module-level tracker. For tests and for `shardIngest.reset()`. */
+function reset() {
+  const fresh = createTracker()
+  for (const key of Object.keys(fresh)) defaultTracker[key] = fresh[key]
+}
+
+// ── Small shared shapes ────────────────────────────────────────────────────
+
+// "Felucca 1480, 1600" — one string rather than four variables, because a
+// template that has to assemble coordinates is a template every author gets
+// slightly differently. Returns undefined when there is nothing to format, so it
+// drops out of an optional variable rather than rendering "undefined , ".
+function place(ev) {
+  const map = ev.map || (ev.location && ev.location.map)
+  const x = ev.x ?? (ev.location && ev.location.x)
+  const y = ev.y ?? (ev.location && ev.location.y)
+  if (!map && x == null) return undefined
+  const coords = x == null || y == null ? '' : ` ${x}, ${y}`
+  const region = ev.region || (ev.location && ev.location.region)
+  const suffix = region ? ` (${region})` : ''
+  return `${map || ''}${coords}${suffix}`.trim() || undefined
+}
+
+// An actor object's display name, whichever of the shard's shapes it arrives in.
+const actorName = (actor) => (actor && typeof actor === 'object' ? actor.name : undefined) || undefined
+const actorAcct = (actor) => (actor && typeof actor === 'object' ? actor.acct : undefined) || undefined
+
+// Drop the undefined values before they reach `emit`. A declared OPTIONAL
+// variable that arrives as `undefined` is dropped by `validatePayload` anyway,
+// but building the object without them keeps the emit log's `variables` list
+// honest about what the frame actually carried.
+const defined = (obj) => Object.fromEntries(Object.entries(obj).filter(([, v]) => v !== undefined))
+
+// ── The mappers ────────────────────────────────────────────────────────────
+//
+// Each returns an array of `{ triggerId, data, ownerAccount?, guildId?, subject?,
+// dedupeKey? }`. Resolution — account → user id, guild → member ids — happens in
+// `dispatch` below, because it needs the database and these must not.
+//
+// `ownerAccount` is the same field name `shardStreams.js` uses for the same idea,
+// so the two mappers can be read side by side.
+
+const decayName = (ev) => ev.name || undefined
+
+const MAPPERS = {
+  // ── Owned asset at risk ────────────────────────────────────────────────
+  'house.decay': (ev, tracker, out) => {
+    const to = String(ev.to || '').toUpperCase()
+    const serial = ev.serial == null ? null : String(ev.serial)
+    if (!serial) return
+    // COLLAPSED is its own trigger; the late stages are the warning. `LikeNew`
+    // and the early stages are not news — a house being refreshed is the normal
+    // case and mailing it would make the warning worthless.
+    if (to === 'COLLAPSED') {
+      out.push({
+        triggerId: 'uo.house.collapsed',
+        ownerAccount: ev.ownerAcct,
+        data: defined({
+          houseSerial: serial,
+          houseName: decayName(ev),
+          region: ev.region || undefined,
+          location: place(ev),
+        }),
+      })
+      return
+    }
+    if (!['FAIRLY', 'GREATLY', 'IDOC'].includes(to)) return
+    const schedule = ev.schedule && typeof ev.schedule === 'object' ? ev.schedule : {}
+    out.push({
+      triggerId: 'uo.house.idoc_warning',
+      ownerAccount: ev.ownerAcct,
+      data: defined({
+        houseSerial: serial,
+        houseName: decayName(ev),
+        stage: ev.to,
+        previousStage: ev.from || undefined,
+        region: ev.region || undefined,
+        location: place(ev),
+        // **Both optional, and both genuinely absent much of the time.** A v4
+        // overlay sends no `schedule` at all; a dynamic-decay shard omits
+        // `estimatedCollapse` at every stage before IDOC because ServUO draws
+        // each stage's duration at random when the stage is entered. Passing
+        // `undefined` through is the honest thing — `docs/link/v5.md` is explicit
+        // that absence means "not knowable", never "not yet read", and computing
+        // a fallback here would republish exactly the guess the shard refused to.
+        nextStage: schedule.nextStage || undefined,
+        estimatedCollapse: schedule.estimatedCollapse || undefined,
+        lastRefreshed: ev.lastRefreshed || undefined,
+      }),
+    })
+  },
+
+  // `house.remove` carries ONLY a serial — the house is gone, so the frame has
+  // nothing else to say. The owner comes from this module's own registry mirror,
+  // which is a database read and therefore happens in `dispatch`.
+  'house.remove': (ev, tracker, out) => {
+    if (ev.serial == null) return
+    out.push({
+      triggerId: 'uo.house.collapsed',
+      houseSerial: String(ev.serial),
+      data: { houseSerial: String(ev.serial) },
+    })
+  },
+
+  'vendor.listing': (ev, tracker, out) => {
+    const serial = ev.serial == null ? null : String(ev.serial)
+    if (!serial || !ev.ownerAcct) return
+    const fees = ev.fees && typeof ev.fees === 'object' ? ev.fees : null
+    // A pre-v5 overlay sends no `fees`; a commission vendor sends `{exempt:true}`
+    // and is NEVER dismissed for them. Both mean "nothing to warn about", and
+    // conflating exempt with a distant date is how a vendor that cannot expire
+    // ends up in an expiry warning (`docs/link/v5.md`).
+    if (!fees || fees.exempt === true || !fees.dismissalAt) {
+      tracker.vendorWarned.delete(serial)
+      return
+    }
+    const at = new Date(fees.dismissalAt)
+    if (Number.isNaN(at.getTime())) return
+    const hours = Math.floor((at.getTime() - Date.now()) / 3_600_000)
+    const inWindow = hours <= VENDOR_WARN_HOURS
+    const wasWarned = tracker.vendorWarned.get(serial) === true
+    tracker.vendorWarned.set(serial, inWindow)
+    // **Only the CROSSING.** The sweep re-emits a shop on any price change, so
+    // without this a vendor inside the window mails its owner every time somebody
+    // reprices a longsword. Leaving the window (a deposit) clears the flag above,
+    // so the next approach warns again — which is the behaviour an owner wants.
+    if (!inWindow || wasWarned) return
+    out.push({
+      triggerId: 'uo.vendor.expiring',
+      ownerAccount: ev.ownerAcct,
+      data: defined({
+        vendorSerial: serial,
+        shopName: ev.shopName || undefined,
+        dismissalAt: fees.dismissalAt,
+        // Never negative: a vendor already past its dismissal tick is being
+        // destroyed, and "-3 hours remaining" in a mail is worse than "0".
+        hoursRemaining: Math.max(0, hours),
+        periodsRemaining: Number.isFinite(fees.periodsRemaining) ? fees.periodsRemaining : undefined,
+        funds: Number.isFinite(fees.funds) ? fees.funds : undefined,
+        chargePerPeriod: Number.isFinite(fees.chargePerPeriod) ? fees.chargePerPeriod : undefined,
+        location: place(ev),
+      }),
+    })
+  },
+
+  'vendor.listing.remove': (ev, tracker) => {
+    if (ev.serial != null) tracker.vendorWarned.delete(String(ev.serial))
+  },
+
+  // ── Passive income ─────────────────────────────────────────────────────
+  'vendor.sale': (ev, tracker, out) => {
+    if (!ev.ownerAcct) return
+    out.push({
+      triggerId: 'uo.vendor.sale',
+      ownerAccount: ev.ownerAcct,
+      data: defined({
+        vendorSerial: String(ev.vendorSerial ?? ''),
+        itemName: ev.itemType || 'an item',
+        amount: Number.isFinite(ev.amount) ? ev.amount : undefined,
+        price: Number.isFinite(ev.price) ? ev.price : 0,
+        commission: Number.isFinite(ev.commission) ? ev.commission : undefined,
+      }),
+    })
+  },
+
+  // ── Personal security ──────────────────────────────────────────────────
+  //
+  // **`account.login.result` and NOT `account.login.attempt`.** The attempt fires
+  // from `EventSink.AccountLogin`, which runs before the auth decision — the
+  // emitter's own comment says so — and `AccountLoginEventArgs` constructs with
+  // `Accepted = true`, so a rule on it would have mailed a security alert every
+  // time the player logged in successfully. That inversion is why protocol 5 adds
+  // this kind and why the trigger is named `login_failed` rather than `attempt`.
+  'account.login.result': (ev, tracker, out) => {
+    if (!ev.acct) return
+    if (ev.accepted !== false) return
+    out.push({
+      triggerId: 'uo.account.login_failed',
+      ownerAccount: ev.acct,
+      data: defined({
+        account: String(ev.acct),
+        reason: ev.reason || undefined,
+        ip: ev.ip || undefined,
+      }),
+    })
+  },
+
+  // **Resolved BEFORE ingest drops the link mirror**, which is the whole reason
+  // this file is called from `ingest()` ahead of the state write rather than
+  // after it. `applyStateChange` removes the `shard_account_links` row for this
+  // account, so an owner lookup that ran afterwards would find nobody and the one
+  // person who needs to know their account was unlinked would never be told.
+  'account.unlinked': (ev, tracker, out) => {
+    if (!ev.account) return
+    out.push({
+      triggerId: 'uo.account.unlinked',
+      ownerAccount: ev.account,
+      data: defined({
+        account: String(ev.account),
+        characterName: ev.char || undefined,
+      }),
+    })
+  },
+
+  // ── Personal milestone ─────────────────────────────────────────────────
+  'skill.gain': (ev, tracker, out) => {
+    // The cap, and only the cap. `skill.gain` fires on every tenth of a point;
+    // `base >= cap` is the milestone and everything else is noise.
+    if (!Number.isFinite(ev.base) || !Number.isFinite(ev.cap) || ev.base < ev.cap) return
+    const acct = actorAcct(ev.who)
+    if (!acct) return
+    out.push({
+      triggerId: 'uo.skill.capped',
+      ownerAccount: acct,
+      data: defined({
+        characterName: actorName(ev.who) || 'your character',
+        skill: String(ev.skill || 'a skill'),
+        cap: ev.cap,
+      }),
+    })
+  },
+
+  'quest.complete': (ev, tracker, out) => {
+    const acct = actorAcct(ev.who)
+    if (!acct) return
+    out.push({
+      triggerId: 'uo.quest.complete',
+      ownerAccount: acct,
+      data: defined({
+        characterName: actorName(ev.who) || 'your character',
+        quest: String(ev.quest || 'a quest'),
+      }),
+    })
+  },
+
+  'player.death': (ev, tracker, out) => {
+    const acct = actorAcct(ev.who)
+    if (!acct) return
+    out.push({
+      triggerId: 'uo.character.death',
+      ownerAccount: acct,
+      data: defined({
+        characterName: actorName(ev.who) || 'your character',
+        killerName: actorName(ev.killer),
+      }),
+    })
+  },
+
+  'player.murdered': (ev, tracker, out) => {
+    const acct = actorAcct(ev.victim)
+    if (!acct) return
+    out.push({
+      triggerId: 'uo.character.murdered',
+      ownerAccount: acct,
+      data: defined({
+        characterName: actorName(ev.victim) || 'your character',
+        murdererName: actorName(ev.murderer),
+      }),
+    })
+  },
+
+  // ── Social / civic ─────────────────────────────────────────────────────
+  //
+  // `uo.guild.joined` is NOT here: core's `team.member.joined` already fires for
+  // it on every roster reconcile, because a UO guild is a Team and this module is
+  // the Team provider. See ENGAGEMENT.md §8.6 for the carve-out.
+  'guild.leave': (ev, tracker, out) => {
+    if (ev.id == null) return
+    out.push({
+      triggerId: 'uo.guild.left',
+      guildId: ev.id,
+      // `who` is a bare SERIAL string here, not an actor object — the mobile has
+      // already left, so there is nothing for the shard to attribute. The name is
+      // looked up from the roster mirror in `dispatch`.
+      memberSerial: ev.who == null ? null : String(ev.who),
+      data: defined({ guildName: ev.name || `guild ${ev.id}` }),
+    })
+  },
+
+  'guild.remove': (ev, tracker, out) => {
+    if (ev.id == null) return
+    out.push({
+      triggerId: 'uo.guild.disbanded',
+      guildId: ev.id,
+      // The frame carries ONLY the id, so the name comes from the board mirror in
+      // `dispatch` — and it has to be read there before `applyStateChange` drops
+      // the row, the same ordering `account.unlinked` depends on.
+      data: {},
+    })
+  },
+
+  'city.update': (ev, tracker, out) => {
+    const { city } = ev
+    if (!city) return
+
+    // A new governor. Never on FIRST sight (`prev === undefined`), so a reconnect
+    // snapshot is not read as eight simultaneous elections.
+    const gov = ev.governor && ev.governor.serial != null ? String(ev.governor.serial) : null
+    const prevGov = tracker.cityGovernor.get(city)
+    tracker.cityGovernor.set(city, gov)
+    if (prevGov !== undefined && gov && gov !== prevGov) {
+      out.push({
+        triggerId: 'uo.governor.elected',
+        data: defined({
+          city: String(city),
+          governorName: actorName(ev.governor) || 'a new governor',
+          previousGovernorName: undefined,
+        }),
+      })
+    }
+
+    // An election opening. `autoPickAt` is REQUIRED on the trigger, so a phase
+    // change that arrives without one is not emitted at all rather than emitted
+    // as a deadline-less call to action — which is what a "vote now" mail with
+    // nothing to act by would be.
+    const phase = ev.electionPhase || 'none'
+    const prevPhase = tracker.cityPhase.get(city)
+    tracker.cityPhase.set(city, phase)
+    if (
+      prevPhase !== undefined
+      && phase !== prevPhase
+      && (phase === 'nominate' || phase === 'vote')
+      && ev.autoPickAt
+    ) {
+      out.push({
+        triggerId: 'uo.election.opened',
+        data: defined({
+          city: String(city),
+          phase,
+          autoPickAt: ev.autoPickAt,
+          candidates: Number.isFinite(ev.candidates) ? ev.candidates : undefined,
+        }),
+      })
+    }
+  },
+
+  // ── Come online now ────────────────────────────────────────────────────
+  'champ.update': (ev, tracker, out) => {
+    const serial = ev.serial == null ? null : String(ev.serial)
+    if (!serial) return
+    const isActive = ev.active === true
+    const wasActive = tracker.champActive.get(serial)
+    tracker.champActive.set(serial, isActive)
+
+    const base = defined({
+      spawnSerial: serial,
+      spawnName: ev.name || ev.type || 'a champion spawn',
+      category: ev.category || undefined,
+      location: place(ev),
+    })
+
+    if (wasActive !== undefined && isActive && wasActive !== true) {
+      out.push({ triggerId: 'uo.champ.started', data: base })
+    }
+
+    const bossUp = ev.bossUp === true
+    const wasBossUp = tracker.champBossUp.get(serial)
+    tracker.champBossUp.set(serial, bossUp)
+    if (wasBossUp !== undefined && bossUp && wasBossUp !== true) {
+      out.push({
+        triggerId: 'uo.champ.boss_up',
+        data: defined({ ...base, bossName: ev.boss || undefined }),
+      })
+    }
+  },
+
+  'champ.remove': (ev, tracker) => {
+    if (ev.serial == null) return
+    tracker.champActive.delete(String(ev.serial))
+    tracker.champBossUp.delete(String(ev.serial))
+  },
+
+  // **`server.hello` fires on every sidecar reconnect, not only on a shard
+  // restart** — which is exactly the flapping this trigger must not amplify. The
+  // tracker's `serverUp` is the guard: a hello while we already believe the shard
+  // is up is a reconnect and emits nothing. The seeded rule's hard cooldown is the
+  // second line of defence, for a shard genuinely bouncing.
+  'server.hello': (ev, tracker, out) => {
+    const wasUp = tracker.serverUp
+    tracker.serverUp = true
+    if (wasUp === true) return
+    out.push({
+      triggerId: 'uo.server.up',
+      data: defined({ shardName: ev.shard || undefined }),
+    })
+  },
+
+  'server.shutdown': (ev, tracker, out) => {
+    if (tracker.serverUp === false) return
+    tracker.serverUp = false
+    out.push({ triggerId: 'uo.server.down', data: { clean: true } })
+  },
+
+  'server.crashed': (ev, tracker, out) => {
+    if (tracker.serverUp === false) return
+    tracker.serverUp = false
+    out.push({ triggerId: 'uo.server.down', data: { clean: false } })
+  },
+
+  // ── Leaderboard ────────────────────────────────────────────────────────
+  //
+  // `subscribers` only. `top[]` names a mobile SERIAL and `shard_account_links`
+  // is keyed by game ACCOUNT, so the "you were pushed out" half of §8.6's row is
+  // carved out rather than resolved for whoever happens to be online.
+  'points.board': (ev, tracker, out) => {
+    const system = ev.system
+    const top = Array.isArray(ev.top) ? ev.top : []
+    if (!system || !top.length) return
+    const leader = top.find((e) => e && e.rank === 1) || top[0]
+    if (!leader || leader.serial == null) return
+    const serial = String(leader.serial)
+    const prev = tracker.pointsLeader.get(system)
+    tracker.pointsLeader.set(system, serial)
+    if (prev === undefined || prev === serial) return
+    out.push({
+      triggerId: 'uo.points.rank_changed',
+      data: defined({
+        system: String(system),
+        systemName: ev.nameString || undefined,
+        leaderName: leader.name || 'a new leader',
+        points: Number.isFinite(leader.points) ? leader.points : undefined,
+      }),
+    })
+  },
+
+  // ── Staff-facing ───────────────────────────────────────────────────────
+  'page.new': (ev, tracker, out) => {
+    out.push({
+      triggerId: 'uo.page.new',
+      data: defined({
+        pageType: String(ev.type || 'Other'),
+        senderName: actorName(ev.sender),
+        message: ev.message || undefined,
+        location: place(ev),
+      }),
+    })
+  },
+
+  'cheat.fastwalk': (ev, tracker, out) => {
+    out.push({
+      triggerId: 'uo.cheat.detected',
+      data: defined({
+        characterName: actorName(ev.who) || 'an unnamed character',
+        account: actorAcct(ev.who),
+        ip: ev.ip || undefined,
+        detector: 'fastwalk',
+      }),
+    })
+  },
+
+  // ── Operator-facing ────────────────────────────────────────────────────
+  'audit.set': (ev, tracker, out) => {
+    out.push({
+      triggerId: 'uo.audit.staff_action',
+      data: defined({
+        staffName: actorName(ev.staff) || (typeof ev.staff === 'string' ? ev.staff : undefined),
+        action: 'set',
+        detail: ev.prop ? `${ev.prop}: ${ev.old ?? '?'} → ${ev.new ?? '?'}` : undefined,
+        target: ev.target || undefined,
+        origin: 'in-game',
+      }),
+    })
+  },
+
+  'audit.command': (ev, tracker, out) => {
+    out.push({
+      triggerId: 'uo.audit.staff_action',
+      data: defined({
+        staffName: actorName(ev.staff) || (typeof ev.staff === 'string' ? ev.staff : undefined),
+        action: 'command',
+        detail: ev.command ? `${ev.command} ${ev.args || ''}`.trim() : undefined,
+        origin: 'in-game',
+      }),
+    })
+  },
+
+  'admin.audit': (ev, tracker, out) => {
+    out.push({
+      triggerId: 'uo.audit.staff_action',
+      data: defined({
+        staffName: typeof ev.actor === 'string' ? ev.actor : actorName(ev.actor),
+        action: String(ev.action || 'action'),
+        detail: ev.reason || undefined,
+        target: ev.target || undefined,
+        origin: ev.origin || undefined,
+      }),
+    })
+  },
+
+  'economy.supply': (ev, tracker, out) => {
+    for (const [metric, value, thresholds] of [
+      ['gold', ev.gold, GOLD_THRESHOLDS],
+      ['accounts', ev.accounts, ACCOUNT_THRESHOLDS],
+    ]) {
+      if (!Number.isFinite(value)) continue
+      const band = bandOf(value, thresholds)
+      const prev = tracker.economyBand.get(metric)
+      tracker.economyBand.set(metric, band)
+      // First sighting establishes the band and reports nothing. Otherwise a
+      // sidecar reconnect on a mature shard announces "gold passed a billion"
+      // about a line it crossed months ago.
+      if (prev === undefined || prev === band) continue
+      // The line that was crossed is the HIGHER of the two bands under a rise and
+      // the one just left under a fall, so both directions name the line the
+      // reader is thinking about.
+      const crossed = band > prev ? thresholds[band] : thresholds[prev]
+      out.push({
+        triggerId: 'uo.economy.milestone',
+        data: defined({
+          metric,
+          value: Math.round(value),
+          threshold: crossed,
+          direction: band > prev ? 'up' : 'down',
+        }),
+      })
+    }
+  },
+
+  'world.save.after': (ev, tracker, out) => {
+    out.push({
+      triggerId: 'uo.world.saved',
+      data: defined({
+        items: Number.isFinite(ev.items) ? ev.items : undefined,
+        mobiles: Number.isFinite(ev.mobiles) ? ev.mobiles : undefined,
+      }),
+    })
+  },
+}
+
+/**
+ * Map one shard event to zero or more engagement events. Pure given `tracker`.
+ *
+ * Exported so the mapping can be tested without a database, exactly as
+ * `shardStreams.mapShardEvent` is.
+ */
+function mapShardEvent(event, tracker = defaultTracker) {
+  if (!event || typeof event.kind !== 'string') return []
+  const mapper = MAPPERS[event.kind]
+  if (!mapper) return []
+  const out = []
+  mapper(event, tracker, out)
+  // **Defence in depth, and the exact counterpart of `shardStreams.js`'s
+  // public-allowlist filter.** A target naming an id this module does not declare
+  // cannot be delivered — `emit` would refuse it anyway, throwing in dev and
+  // logging in prod — so catching it here turns a typo into one warning with the
+  // id in it rather than an exception on the ingest path.
+  return out.filter((t) => {
+    if (TRIGGER_IDS.has(t.triggerId)) return true
+    log.warn('mapper produced an undeclared trigger id', { kind: event.kind, triggerId: t.triggerId })
+    return false
+  })
+}
+
+// ── Resolution and dispatch ────────────────────────────────────────────────
+
+/**
+ * Turn one mapped target into the envelope `ctx.events.emit` takes, or null when
+ * there is nobody to tell.
+ *
+ * This is the half that reaches the database, and it is why the mapping above is
+ * separate: an owner-keyed target names a GAME ACCOUNT and a members-keyed one
+ * names a GUILD, and neither is a website user until something asks.
+ */
+async function resolveTarget(target, deps) {
+  const { links, state } = deps
+  const data = { ...target.data }
+
+  // `owner` — one account, one user. An unlinked account is nobody to notify,
+  // which is a normal outcome and not an error: most game accounts on most shards
+  // have never been linked.
+  if (target.ownerAccount) {
+    const link = await links.getByAccount(target.ownerAccount)
+    if (!link || link.user_id == null) return null
+    return { data, ownerUserId: Number(link.user_id) }
+  }
+
+  // `uo.house.collapsed` off `house.remove`, whose frame carries only a serial.
+  // The owner comes from this module's registry mirror — and this read has to
+  // happen before `applyStateChange` drops the row, which is why ingest calls the
+  // engagement fan-out ahead of the state write.
+  if (target.houseSerial) {
+    const houses = await state.listHouses()
+    const house = houses.find((h) => String(h.serial) === target.houseSerial)
+    if (!house || !house.ownerAcct) return null
+    const link = await links.getByAccount(house.ownerAcct)
+    if (!link || link.user_id == null) return null
+    if (house.name) data.houseName = house.name
+    if (house.region) data.region = house.region
+    return { data, ownerUserId: Number(link.user_id) }
+  }
+
+  // `members` — the guild's roster, resolved to website users through
+  // `shard_account_links` rather than through the roster's mirrored `web_id`.
+  // The mirror is a copy of what the wire said; the links table is the answer.
+  if (target.guildId != null) {
+    const accounts = await state.listGuildMemberAccounts(target.guildId)
+    const userIds = await links.userIdsForAccounts(accounts)
+    if (!userIds.length) return null
+
+    // Fill in the two names the frames do not carry, from the board mirror.
+    if (!data.guildName || !data.abbreviation) {
+      const guilds = await state.listGuilds()
+      const guild = guilds.find((g) => String(g.id) === String(target.guildId))
+      if (guild) {
+        if (!data.guildName) data.guildName = guild.name || `guild ${target.guildId}`
+        if (guild.abbr && data.abbreviation === undefined) data.abbreviation = guild.abbr
+      }
+    }
+    if (!data.guildName) data.guildName = `guild ${target.guildId}`
+
+    // Who left, from the roster mirror — the departing member's row is still
+    // there, because `guild.leave`'s state write has not run yet.
+    if (target.memberSerial) {
+      const members = await state.listGuildMembers(target.guildId)
+      const gone = members.find((m) => String(m.serial) === target.memberSerial)
+      if (gone && gone.name) data.memberName = gone.name
+    }
+    return { data, recipientUserIds: userIds }
+  }
+
+  // Everything else — `subscribers`, `staff`, `admin` — has no per-event
+  // audience to resolve. The rule's audience is the whole answer.
+  return { data }
+}
+
+/**
+ * Fan one shard event out to the engagement engine.
+ *
+ * Never throws. Called fire-and-forget from `shardIngest.ingest`, beside the SSE
+ * broadcast and the push dispatch, and held to the same promise all three make:
+ * a slow or failing notification path must never delay or fail ingest.
+ */
+async function fromShardEvent(event, deps = {}) {
+  const d = {
+    links: deps.shardLinks || shardLinks,
+    state: deps.shardState || shardState,
+    emit: deps.emit || core.events.emit,
+    tracker: deps.tracker || defaultTracker,
+  }
+
+  for (const target of mapShardEvent(event, d.tracker)) {
+    try {
+      const resolved = await resolveTarget(target, d)
+      // Nobody to tell. Not an error and deliberately not logged at warn: an
+      // unlinked house owner is the common case on every shard.
+      if (!resolved) continue
+      d.emit(target.triggerId, {
+        data: resolved.data,
+        ...(resolved.ownerUserId ? { ownerUserId: resolved.ownerUserId } : {}),
+        ...(resolved.recipientUserIds ? { recipientUserIds: resolved.recipientUserIds } : {}),
+        ...(target.dedupeKey ? { dedupeKey: target.dedupeKey } : {}),
+        occurredAt: Number.isFinite(event.t) ? new Date(event.t) : undefined,
+      })
+    } catch (err) {
+      log.warn('engagement target failed', { triggerId: target.triggerId, message: err.message })
+    }
+  }
+}
+
+module.exports = {
+  fromShardEvent,
+  mapShardEvent,
+  createTracker,
+  reset,
+  VENDOR_WARN_HOURS,
+  GOLD_THRESHOLDS,
+  ACCOUNT_THRESHOLDS,
+}

server/utils/shardIngest.js

@@ -20,6 +20,7 @@ const uoLinkConfigModel = require('../model/uoLinkConfig/uoLinkConfig.model')
 const { settings: settingsModel } = require('../core')
 const broadcaster = require('./shardBroadcast')
 const shardPush = require('./shardPush')
+const shardEngagement = require('./shardEngagement')
 const defaultLog = require('../core').logger('shard-ingest')
 
 // Notable kinds appended to the shard_events log. High-frequency/session kinds
@@ -62,6 +63,11 @@ const LOGGED_KINDS = new Set([
 const state = { bootId: null }
 function reset() {
   state.bootId = null
+  // The engagement mapper's transition/threshold tracker is per-process state of
+  // exactly the same kind as `bootId`, so it is reset by the same call. A test
+  // that reset one and not the other would see a champion spawn that started in
+  // the previous test.
+  shardEngagement.reset()
 }
 
 // Should this event be written to the append-only log?
@@ -285,6 +291,7 @@ function resolveDeps(deps) {
     settings: deps.settings || settingsModel,
     broadcast: deps.broadcast || broadcaster.broadcast,
     pushDispatch: deps.pushDispatch || shardPush.fromShardEvent,
+    engagement: deps.engagement || shardEngagement.fromShardEvent,
     log: deps.log || defaultLog,
   }
 }
@@ -300,6 +307,34 @@ async function ingest(event, deps = {}) {
   let stored = false
   let logged = false
 
+  // **The engagement fan-out runs BEFORE the state write, and that ordering is
+  // load-bearing rather than incidental** (ENGAGEMENT.md Phase 11). Three of the
+  // mappings read a row that `applyStateChange` is about to delete or replace:
+  //
+  //   • `account.unlinked` drops the `shard_account_links` row — the row that
+  //     turns the account into the one person who needs to be told it was
+  //     unlinked. Resolving afterwards finds nobody, every time.
+  //   • `house.remove` drops the house, whose stored `ownerAcct` is the only place
+  //     the owner of a collapsed house is named (the frame carries a serial alone).
+  //   • `guild.leave` / `guild.remove` need the roster and the board mirror to
+  //     name who left and which guild it was.
+  //
+  // Awaited, unlike the broadcast and the push tickle below, and this is the one
+  // place this file waits on a notification path. It has to: the whole point is
+  // that the read happens first, and a fire-and-forget promise would race the
+  // DELETE it is trying to precede. `fromShardEvent` never throws and never opens
+  // a socket — it resolves ids and hands the engine an envelope, which does its
+  // own work off the caller's stack (`emit` is deliberately not awaited inside).
+  // Backfilled frames are excluded for the same reason the broadcast is: a
+  // reconnect replay must not re-notify anyone about events from hours ago.
+  if (!deps.fromBackfill) {
+    try {
+      await d.engagement(event)
+    } catch (err) {
+      d.log.warn('engagement fan-out failed', { kind: event.kind, message: err.message })
+    }
+  }
+
   try {
     await applyStateChange(event, d)
   } catch (err) {