Module-uo/server/utils/shardIngest.js

// ── Shard event ingest dispatcher ──────────────────────────────────────────
//
// The single entry point for every event that arrives on the uo-link WebSocket
// feed (and for backfilled /history events on reconnect). It routes by kind:
//   • state-changing kinds update shard_online / shard_economy / shard_houses,
//   • notable kinds are appended to the append-only shard_events log,
//   • every kind is fanned out to the SSE broadcaster (which decides public vs
//     admin visibility).
// High-frequency kinds (char.vitals, economy.supply) are deliberately NOT logged
// to shard_events — they only update state — keeping the event log lean.
//
// Dependencies are injected (defaulting to the real models) so the routing can
// be unit-tested with mocked writes.

const shardEventsModel = require('../model/shardEvents/shardEvents.model')
const shardStateModel = require('../model/shardState/shardState.model')
const shardLinksModel = require('../model/shardLinks/shardLinks.model')
const shardMarketModel = require('../model/shardMarket/shardMarket.model')
const uoLinkConfigModel = require('../model/uoLinkConfig/uoLinkConfig.model')
const { settings: settingsModel, events: coreEvents } = 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
// (char.vitals, economy.supply, mob.login/logout, account.login.attempt,
// gold.change, vendor.buy/sell) are excluded on purpose. house.decay is handled
// specially — logged only on the transition INTO IDOC.
const LOGGED_KINDS = new Set([
  'vendor.sale',
  'player.death',
  'player.murdered',
  'mob.killed',
  'quest.complete',
  'skill.gain',
  'fame.change',
  'karma.change',
  'audit.set',
  'audit.command',
  'admin.audit',
  'cheat.fastwalk',
  'link.request',
  'server.hello',
  'server.shutdown',
  'server.crashed',
  // Protocol 2.0: a real-time guild join (the board itself is state, not logged).
  'guild.join',
  // Protocol 4: the departure counterpart to guild.join, and logged for the same
  // reason — it is what a "so-and-so left" feed reads. `guild.roster` deliberately
  // stays out: it is board state like guild.update, and it is the one fat frame on
  // the wire (~69 bytes per member), so logging it would bloat shard_events with
  // a full membership snapshot on every membership change.
  'guild.leave',
  // Protocol 2.0 provisioning audit (admin channel only — not in PUBLIC_KINDS).
  'account.audit',
  'account.unlinked',
])

// Tracks the current shard boot id so a restart (changed bootId on server.hello)
// can be detected and stale online state dropped. Module-level so it survives
// across events within a process; reset() is exposed for tests.
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?
function shouldLog(event) {
  if (event.kind === 'house.decay') return String(event.to).toUpperCase() === 'IDOC'
  return LOGGED_KINDS.has(event.kind)
}

// ServUO's stock Server.cfg name. An operator who never set one publishes this
// verbatim, so it carries no more information than a blank — matched
// case-insensitively and trim-tolerantly, but ONLY as an exact whole value: a
// shard genuinely called "My Shard Reborn" keeps its name.
const STOCK_SHARD_NAME = 'my shard'

/**
 * The name to publish for the shard: its own, or this instance's when it has
 * effectively not given one.
 *
 * Deliberately not a general "blank means brand" rule applied across the wire —
 * it is scoped to this one field, where the two names denote the same thing.
 */
async function resolveShardName(shard, deps) {
  const given = String(shard ?? '').trim()
  if (given !== '' && given.toLowerCase() !== STOCK_SHARD_NAME) return given
  try {
    return (await deps.settings.getInstanceName()) || given
  } catch {
    // A ruleset that publishes the stock name is still better than one that
    // fails to store because the settings read hiccuped.
    return given
  }
}

// Apply the state-change side effect for a kind (if any). Returns a promise.
async function applyStateChange(event, deps) {
  const { shardState, uoLinkConfig, eventsReconcile, log } = deps
  switch (event.kind) {
    case 'server.hello': {
      const incoming = event.bootId || null
      const restarted = Boolean(incoming && state.bootId && incoming !== state.bootId)
      if (restarted) {
        log.warn('shard restarted (bootId changed) — clearing online roster', {
          from: state.bootId,
          to: incoming,
        })
        await shardState.clearOnline()
      }
      if (incoming) state.bootId = incoming
      await uoLinkConfig.recordStatus({ pluginConnected: true, bootId: incoming, lastEventAt: event.t })
      if (restarted) {
        // EVENTS.md F: core has no concept of the game being up, so the module
        // says when a ledger of live shard resources has become a claim about a
        // world that no longer exists. This is that moment, and a changed
        // `bootId` is the only thing that distinguishes it from a sidecar
        // reconnect — which changes nothing in the game and must not orphan a row.
        //
        // **After `recordStatus`, and that ordering is load-bearing.** Every
        // action's `reconcile()` decides what is still in force by comparing its
        // stamp against the CURRENT boot id, which it reads back out of this
        // row. Asking first would have every resource compared against the boot
        // that has just ended, and every one of them would look live.
        //
        // Fire-and-forget by the contract: core logs what it orphaned, and there
        // is nothing an ingest handler could correctly do with the answer.
        eventsReconcile()
      }
      return
    }
    case 'server.shutdown':
    case 'server.crashed':
      // Shard is going away — nobody is online anymore.
      await shardState.clearOnline()
      await uoLinkConfig.recordStatus({ pluginConnected: false })
      return
    case 'mob.login': {
      const who = event.who || {}
      await shardState.upsertOnline({
        serial: who.serial,
        name: who.name,
        acct: who.acct,
        webId: event.webId,
        map: event.map,
        x: event.x,
        y: event.y,
        z: event.z,
      })
      return
    }
    case 'mob.logout': {
      const who = event.who || {}
      if (who.serial) await shardState.setOffline(who.serial)
      return
    }
    case 'char.vitals':
      await shardState.upsertOnline({
        serial: event.serial,
        hits: event.hits,
        hitsMax: event.hitsMax,
        mana: event.mana,
        manaMax: event.manaMax,
        stam: event.stam,
        stamMax: event.stamMax,
        str: event.str,
        dex: event.dex,
        int: event.int,
        map: event.map,
        x: event.x,
        y: event.y,
      })
      return
    case 'economy.supply':
      await shardState.addEconomySample({ accounts: event.accounts, gold: event.gold, t: event.t })
      return
    case 'house.decay':
      await shardState.upsertHouse({
        serial: event.serial,
        stage: event.to,
        map: event.map,
        x: event.x,
        y: event.y,
        z: event.z,
        region: event.region,
        name: event.name,
        ownerSerial: event.ownerSerial,
        ownerAcct: event.ownerAcct,
        // Protocol 5. `ownerName` used to arrive only on house.update, so a house
        // that had decayed but never been swept into the registry named an account
        // and no character. It rides house.decay now, which is the frame the IDOC
        // page is actually built from.
        ownerName: event.ownerName,
        builtOn: event.builtOn,
        lastRefreshed: event.lastRefreshed,
        schedule: event.schedule,
      })
      return
    case 'champ.update':
      await shardState.upsertChamp(event)
      return
    case 'champ.remove':
      await shardState.removeChamp(event.serial)
      return
    case 'page.new':
    case 'page.updated':
      await shardState.upsertPage(event)
      return
    case 'page.closed':
      await shardState.removePage(event.pageId)
      return
    // ── Protocol 2.0 boards ──────────────────────────────────────────────
    case 'guild.update':
      await shardState.upsertGuild(event)
      return
    case 'guild.remove':
      await shardState.removeGuild(event.id)
      return
    // Protocol 4: membership. A roster arrives in one frame for any realistic
    // guild and in several for one over the shard's cap — upsertGuildRoster
    // handles both. guild.leave is advisory; the next roster would converge
    // anyway, but applying it shows the departure at once.
    case 'guild.roster':
      await shardState.upsertGuildRoster(event)
      return
    case 'guild.leave':
      await shardState.removeGuildMember(event)
      return
    case 'city.update':
      // Upserts the board AND captures term history (idempotent).
      await shardState.upsertGovernor(event)
      return
    case 'presence.online':
      await shardState.setPresence(event)
      return
    case 'house.update':
      await shardState.upsertHouseRegistry(event)
      return
    case 'house.remove':
      await shardState.removeHouse(event.serial)
      return
    // ── Protocol 3.0 ─────────────────────────────────────────────────────
    // The shard re-emits its whole ruleset on every sidecar connect, so this is
    // an overwrite, not an append — and deliberately NOT in LOGGED_KINDS: it
    // would put a duplicate row in the event log on every reconnect, and
    // server.hello already marks each of those.
    case 'world.ruleset':
      // A shard whose operator never edited Server.cfg publishes ServUO's stock
      // "My Shard". That is the shard saying *unnamed*, not a name, so the site
      // answers with its own — the rules page reading "My Shard" under a header
      // reading UOMysticmoon is the shard failing to introduce itself.
      //
      // Normalized HERE rather than on read because the ruleset is also live: the
      // same `event` object is handed to the SSE broadcast a few lines below, and
      // a read-time fix would be undone by the next reconnect's frame.
      event.shard = await resolveShardName(event.shard, deps)
      await shardState.setRuleset(event)
      return
    // Board state, like guild.update — the newest frame for a system replaces the
    // previous one, so it is NOT in LOGGED_KINDS. Logging would append a row every
    // time anyone's score moved the top ten, which is a board, not an event.
    case 'points.board':
      await shardState.upsertPointsBoard(event)
      return
    // Player-vendor market index. Each frame is authoritative for one shop, so
    // the model replaces that vendor's whole listing set rather than merging.
    //
    // NOT in LOGGED_KINDS, and this is the strongest case of the three v3 kinds:
    // one frame carries up to 250 listings, the sweep re-emits a shop on any
    // price change, and appending each of those to the event log would make
    // shard_events mostly a price history nobody reads. The market IS the state.
    case 'vendor.listing':
      await deps.shardMarket.upsertVendor(event)
      return
    case 'vendor.listing.remove':
      await deps.shardMarket.removeVendor(event.serial)
      return
    case 'account.unlinked':
      // A player ran [unlink in game (or a site-side unlink echoed back) — drop
      // our local link mirror so attribution stops immediately.
      if (event.account) await deps.shardLinks.removeByAccount(event.account)
      return
    // guild.join / account.audit → logged; region.enter → broadcast-only.
    default:
      // No state side effect (e.g. vendor.sale, audit.*, cheat.*) — logging and
      // broadcasting still happen in ingest().
  }
}

// Ingest one event. Returns { logged, stored } for tests/stats. `fromBackfill`
// suppresses the SSE broadcast (a reconnect replay shouldn't re-animate the
// live ticker). Never throws — a bad single event must not kill the feed.
// Resolve the injectable dependencies to their live defaults (tests override a
// subset). Split out so ingest() isn't penalised for the fan of `|| default`s.
function resolveDeps(deps) {
  return {
    shardEvents: deps.shardEvents || shardEventsModel,
    shardState: deps.shardState || shardStateModel,
    shardLinks: deps.shardLinks || shardLinksModel,
    shardMarket: deps.shardMarket || shardMarketModel,
    uoLinkConfig: deps.uoLinkConfig || uoLinkConfigModel,
    settings: deps.settings || settingsModel,
    broadcast: deps.broadcast || broadcaster.broadcast,
    pushDispatch: deps.pushDispatch || shardPush.fromShardEvent,
    engagement: deps.engagement || shardEngagement.fromShardEvent,
    // MODULE_API 1.10.0 (EVENTS.md F, Phase 8). Injectable for the same reason
    // every member above is: a test that asserted a shard restart triggers a
    // reconcile must be able to see the call without a live event engine behind
    // it.
    eventsReconcile: deps.eventsReconcile || (() => coreEvents.reconcile()),
    log: deps.log || defaultLog,
  }
}

async function ingest(event, deps = {}) {
  const d = resolveDeps(deps)

  if (!event || typeof event.kind !== 'string') return { logged: false, stored: false }
  // ws.hello / pong are transport frames, not game events.
  if (event.kind === 'ws.hello' || event.kind === 'pong') return { logged: false, stored: false }

  const t = Number.isFinite(event.t) ? event.t : Date.now()
  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) {
    d.log.warn('state-change write failed', { kind: event.kind, message: err.message })
  }

  if (shouldLog(event)) {
    logged = true
    try {
      stored = await d.shardEvents.append({ kind: event.kind, t, bootId: state.bootId, payload: event })
    } catch (err) {
      d.log.warn('event log write failed', { kind: event.kind, message: err.message })
    }
  }

  if (!deps.fromBackfill) {
    // Broadcast is async since v3 (it reads the visibility config to decide what
    // each subscriber may see). Fire-and-forget, like the push fan-out below: a
    // slow config read must never delay or fail ingest.
    Promise.resolve(d.broadcast(event)).catch((err) =>
      d.log.warn('broadcast failed', { kind: event.kind, message: err.message }),
    )
    // Opt-in push fan-out, off the same event source as the SSE broadcast.
    // Fire-and-forget (a slow/dead ntfy relay must never delay or fail ingest);
    // fromShardEvent is self-guarding, but .catch() covers any lookup rejection.
    Promise.resolve(d.pushDispatch(event, { shardLinks: d.shardLinks })).catch((err) =>
      d.log.warn('push dispatch failed', { kind: event.kind, message: err.message }),
    )
  }

  return { logged, stored }
}

module.exports = { ingest, shouldLog, reset, LOGGED_KINDS, state }