website/server/src/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 uoLinkConfigModel = require('../model/uoLinkConfig/uoLinkConfig.model')
const broadcaster = require('./shardBroadcast')
const pushDispatch = require('./pushDispatch')
const defaultLog = require('./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 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
}

// 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)
}

// Apply the state-change side effect for a kind (if any). Returns a promise.
async function applyStateChange(event, deps) {
  const { shardState, uoLinkConfig, log } = deps
  switch (event.kind) {
    case 'server.hello': {
      const incoming = event.bootId || null
      if (incoming && state.bootId && incoming !== state.bootId) {
        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 })
      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,
        builtOn: event.builtOn,
        lastRefreshed: event.lastRefreshed,
      })
      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
    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':
      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
    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,
    uoLinkConfig: deps.uoLinkConfig || uoLinkConfigModel,
    broadcast: deps.broadcast || broadcaster.broadcast,
    pushDispatch: deps.pushDispatch || pushDispatch.fromShardEvent,
    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

  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 }