Module-Rust/server/ingest.js

// ── Reading a sidecar's feed, and turning it into a record ────────────────
//
// One job: move each server's cursor forward, and apply what it passed.
//
// ── Why a cursor and not a socket ─────────────────────────────────────────
//
// The obvious design is a WebSocket — the sidecar has one, and module-uo takes
// exactly that route for the UO bridge. This module polls a cursor instead, and
// the reason is not laziness about latency.
//
// Core runs on Node 20, where a global `WebSocket` is still behind a flag, so a
// socket means taking `ws` as a runtime dependency — and this module's release
// asserts that it has none (D5: everything it needs arrives on `ctx`, and the
// bundle ships no `node_modules`). That is a cost worth paying for latency, but
// the deciding argument is the other one: **a socket needs a cursor anyway.**
// Whatever a feed misses while a module is restarting has to be caught up from
// somewhere, and the catch-up path is the one that must be right. A socket on
// top of a cursor is two mechanisms where the second is load-bearing; a cursor
// alone is one mechanism that is exercised every few seconds rather than only
// after an outage nobody planned.
//
// What it costs is seconds of latency on a killfeed. What it buys is that the
// path which recovers from a five-hour outage is the same path that ran a moment
// ago.
//
// ── The ordering the whole thing rests on ─────────────────────────────────
//
// **The cursor advances after the batch is written, never before.** A crash
// between the two re-reads events already counted, which inflates a total; a
// crash the other way round loses them silently and for ever. Neither is good and
// they are not equally bad — one is visible and bounded, the other is invisible
// and permanent — so the code is arranged to fail in the visible direction.

const core = require('./core')

const clans = require('./model/clans/clans.model')
const configDb = require('./model/config/config.db')
const configModel = require('./model/config/config.model')
const db = require('./model/events/events.db')
const engagement = require('./engagement/emit')
const eventWorld = require('./eventWorld')
const links = require('./model/links/links.model')
const npcs = require('./model/npcs/npcs.model')
const npcsDb = require('./model/npcs/npcs.db')
const permissionsDb = require('./model/permissions/permissions.db')
const sidecar = require('./sidecarClient')

const log = core.logger('ingest')

/** How many events to ask for at once. */
const BATCH = 200

/**
 * How many batches one tick will drain before letting the loop breathe.
 *
 * A module that has been down for a day has thousands of events waiting, and
 * draining them in one unbounded loop would hold the tick — and a pool
 * connection — for as long as that takes. Bounded, it catches up over several
 * ticks and the site stays responsive while it does.
 */
const MAX_BATCHES_PER_TICK = 10

/**
 * Applies one feed item.
 *
 * Every frame is stored raw, and only some of them move a counter. That split is
 * deliberate: the raw row is what an admin reads and what a later phase can
 * re-derive from, and the counters are what a leaderboard sums. A kind this
 * build has never heard of still lands in `rust_events` — it costs nothing and
 * the alternative is losing the one copy of an event the next version will know
 * how to read.
 */
async function apply(serverId, item, server = null) {
  const frame = (item && item.frame) || {}
  const kind = item.kind || frame.kind
  const wipeId = frame.wipeId || null

  // A wipe exists because something mentioned it. There is no "a wipe started"
  // call and there must not be one: the website is not there when a wipe happens.
  await db.touchWipe(serverId, wipeId, frame.saveCreatedAt || null)

  await db.insertEvent({
    serverId,
    wipeId,
    kind,
    t: Number(frame.t) || item.t || Date.now(),
    steamId: frame.steamId || null,
    raw: frame,
  })

  // What core's engagement engine is told (PLAN.md §25). BEFORE the frame is
  // applied, because applying a disband deletes the roster the notification is
  // for. Never throws, and does not hold the cursor on core: `onEvent` resolves
  // who a frame is about and hands it over, and delivery is core's own time.
  if (server) await engagement.onEvent(server, item)

  const at = { serverId, wipeId, steamId: frame.steamId }

  switch (kind) {
    case 'player.connected':
      await db.touchPlayer(frame.steamId, frame.name || null)
      break

    case 'player.disconnected': {
      await db.touchPlayer(frame.steamId, frame.name || null)

      // `sessionSec` is ABSENT when the plugin never saw the connect — a player
      // already on the server when it loaded. Absent is not zero: adding a zero
      // would be recording a session of no length, which is a different claim
      // from recording no session, and it is the one that quietly under-reports
      // playtime for ever.
      const seconds = Number(frame.sessionSec)
      await db.addStats(at, {
        sessions: Number.isFinite(seconds) ? 1 : 0,
        playtimeSec: Number.isFinite(seconds) && seconds > 0 ? seconds : 0,
      })
      break
    }

    case 'player.death': {
      await db.touchPlayer(frame.steamId, frame.name || null)

      // A suicide is a death AND a suicide, not one instead of the other: the
      // deaths column is "how many times did this player die", and a leaderboard
      // that silently omitted self-inflicted ones would disagree with the
      // killfeed sitting next to it on the same page.
      await db.addStats(at, { deaths: 1, suicides: frame.attackerType === 'self' ? 1 : 0 })

      // Only a real player's kill counts. `npc` and `environment` have no
      // attacker to credit, and `self` must not credit the victim with a kill —
      // which is the one line here that would look right in review and produce a
      // leaderboard topped by whoever died the most.
      if (frame.attackerType === 'player' && frame.attackerId) {
        await db.touchPlayer(frame.attackerId, frame.attackerName || null)
        await db.addStats({ ...at, steamId: frame.attackerId }, { kills: 1 })
      }
      break
    }

    case 'player.tally': {
      // Protocol 14 (§11.2.2): the player's Economics balance, a VALUE written
      // over rather than added — unlike everything below it.
      const hasBalance = frame.balance !== undefined && frame.balance !== null
      if (hasBalance) await db.setBalance(at, frame.balance)

      // A balance with no name is an offline player's: a payment or a staff
      // adjustment reaches somebody who is not there, and the plugin names only
      // a connected player. It says nothing about play, so it must not move
      // anybody's last-seen — which every writer below does.
      if (hasBalance && !frame.name) break

      await db.touchPlayer(frame.steamId, frame.name || null)
      await db.addStats(at, {
        npcKills: Number(frame.npcKills) || 0,
        structures: Number(frame.structures) || 0,
      })

      // A tally is a DELTA since the last flush, which is what makes adding it
      // correct. If it ever becomes a running total this loop doubles every
      // number in it, slowly, and looks right the whole time.
      const gathered = frame.gathered || {}
      for (const [resource, amount] of Object.entries(gathered)) {
        await db.addGathered(at, resource, Number(amount) || 0)
      }

      // Protocol 13, the chat-title conditions (PLAN_REDESIGNS §5.1). Deltas
      // too, except the two distances, which are this interval's best.
      await db.addTitleStats(at, frame)

      const weapons = frame.weaponKills && typeof frame.weaponKills === 'object' ? frame.weaponKills : {}
      for (const [weapon, kills] of Object.entries(weapons)) {
        await db.addWeaponKills(at, weapon, Math.floor(Number(kills) || 0))
      }

      // RunicNPC's NPCs, by profile name (D247). Credited to the site profile
      // that name was last pushed as on this server, so "this profile only"
      // can be counted; 0 for a name the site never pushed.
      const profiles = frame.npcProfileKills && typeof frame.npcProfileKills === 'object' ? frame.npcProfileKills : {}
      for (const [profile, kills] of Object.entries(profiles)) {
        const n = Math.floor(Number(kills) || 0)
        if (n > 0) await npcsDb.addKills(at, profile, await npcs.siteProfileFor(serverId, profile), n)
      }
      break
    }

    case 'player.chat':
    case 'player.respawned':
      await db.touchPlayer(frame.steamId, frame.name || null)
      break

    // ── Protocol 3: the one frame that changes something other than a counter ──
    //
    // `/unlink` in game severs the site's link, and it is the only way out of a
    // link on the wrong account: the site REFUSES to move a Steam id another
    // website account already holds (D23), so without this a player who linked
    // while signed in as the wrong account would need staff.
    //
    // It arrives here rather than through a route because the plugin has nothing
    // to delete — the site is the author of record and the game holds no link —
    // so `/unlink` is the game reporting what the player asked for, applied off
    // the feed like every other frame.
    //
    // **The authority is the Steam account itself.** Whoever is connected to the
    // game as it is who it is, which is a stronger proof of ownership than the
    // site can obtain any other way, so this is not scoped by website user.
    case 'account.unlinked':
      await db.touchPlayer(frame.steamId, frame.name || null)
      await links.unlinkFromGame(frame.steamId)
      break

    // Stored and counted as a sighting, nothing more. The code is deliberately
    // NOT on this frame — it travels through the player — so there is nothing
    // here to redeem and no pending state for the site to hold. It exists so an
    // operator can see linking being used at all.
    case 'account.link.requested':
      await db.touchPlayer(frame.steamId, frame.name || null)
      break

    // ── Protocol 4: somebody changed the permission store, and it was not us ──
    //
    // The plugin raises this only for writes it did not make itself — its own
    // sync suppresses the hooks while it applies (PROTOCOL.md §10.4). What
    // arrives here is therefore a hand edit, a console command, or another
    // plugin granting something.
    //
    // **It is a reason to reconcile, not the reconciliation.** This frame cannot
    // say whether the change is foreign: only the desired set can, and that
    // comparison happens in the sync. So the server is marked dirty and the next
    // tick produces the authoritative answer — which means a hook that stops
    // firing on a framework upgrade costs latency and nothing else. The audit
    // interval finds the same drift within fifteen minutes either way.
    case 'perm.drift':
      await permissionsDb.markDirty(serverId)
      break

    // ── Protocol 13: a plugin loaded or unloaded (F8, D184) ─────────────────
    //
    // A grant for a plugin that was not loaded stays `unresolved` until that
    // plugin comes back, and the first walk watched one land thirteen minutes
    // late, on the fifteen-minute audit. The frame carries the permissions the
    // plugin added or removed, and only a non-empty list is a reason to sync —
    // a plugin that registers nothing cannot have changed what a grant resolves
    // to. `perm.drift`'s posture: a reason, and the next tick (30 s) is the answer.
    case 'plugin.loaded':
    case 'plugin.unloaded':
      if (Array.isArray(frame.permissions) && frame.permissions.length > 0) {
        await permissionsDb.markDirty(serverId)
      }
      break

    // ── Protocol 13: a zone reached its deadline (F13, F14, D170, D183) ──────
    //
    // Core records the run's resource row as `expired`. Not awaited: it is
    // core's bookkeeping, fire-and-forget by contract.
    case 'world.expired':
      eventWorld.expired(serverId, frame)
      break

    // ── Protocol 13: how a configuration save's reload ended (F9, D179) ─────
    //
    // The save was answered before the reload finished and recorded as
    // `reloading`; this is the rest of it. Only a row still `reloading` moves,
    // so a replayed frame changes nothing, and one that arrives after the page
    // gave up on it (`lost`) still lands — the audit row ends up true either way.
    case 'config.outcome': {
      const outcome = configModel.outcomeOf(frame)
      if (outcome) await configDb.settleWrite(serverId, outcome)
      break
    }

    // ── Protocol 14: Economics wiped every balance (§11.2.2) ────────────────
    //
    // `ecowipe` puts everyone back at the start in the wipe in play, so none of
    // its balances is known any more. A new save's wipe (`newSave`) opens a new
    // map whose balances arrive as players connect; the last wipe's stay as the
    // record of how it ended.
    case 'economics.wiped':
      if (!frame.newSave) await db.clearBalances(serverId, wipeId)
      break

    // ── Protocol 6: first-party clans ──────────────────────────────────────
    //
    // Each one is told to core as it happens (`ctx.teams.publish`) and written
    // to the clan's Team feed as a members-only line (D49). Neither is the
    // record: the `clans` board the plugin re-sends a few seconds later is what
    // the store is rebuilt from, so an event this module never saw costs a
    // feed line and nothing else.
    //
    // No `touchPlayer` here, on purpose: it moves `last_seen`, and a kick is
    // done TO somebody who may be offline. `model/clans` notes names without it.
    case 'clan.created':
    case 'clan.disbanded':
    case 'clan.member.added':
    case 'clan.member.left':
    case 'clan.member.kicked':
      await clans.applyEvent(serverId, frame)
      break

    default:
      // Stored, not counted. Moderation frames, the server lifecycle, and
      // anything a newer protocol sends that this build does not understand.
      break
  }
}

/**
 * Brings one server's cursor up to date.
 *
 * Returns the number of events applied, for the log and for the tests.
 */
async function ingestServer(server) {
  const cursor = await db.getCursor(server.id)

  // A server this module has never ingested starts at the sidecar's CURRENT end,
  // not at zero. A module installed today against a sidecar that has been running
  // for a month should read what happens next — replaying a fortnight of deaths
  // into stats for wipes it never saw is not a catch-up, it is inventing a
  // history it was not present for. `/feed` with no `since` asks exactly that
  // question, which is why the sidecar answers it that way.
  if (!cursor) {
    const tail = await sidecar.feedTail(server)

    if (!tail.ok || !tail.data) {
      // Unreachable. Write nothing: a cursor of 0 written now would replay the
      // whole retained history the moment the sidecar came back.
      return 0
    }

    await db.setCursor(server.id, Number(tail.data.lastId) || 0, 0)
    log.info('cursor started at the feed tail', { server: server.id, at: tail.data.lastId })
    return 0
  }

  let since = Number(cursor.lastEventId) || 0
  let applied = 0

  for (let batch = 0; batch < MAX_BATCHES_PER_TICK; batch += 1) {
    const res = await sidecar.feed(server, since, BATCH)

    if (!res.ok || !res.data) return applied

    const items = Array.isArray(res.data.items) ? res.data.items : []

    for (const item of items) {
      try {
        await apply(server.id, item, server)
        applied += 1
      } catch (err) {
        // One malformed event must not wedge a server's cursor for ever. It is
        // logged with its id so it can be found, and the cursor moves past it:
        // the alternative is an ingest that stops at a single bad row and then
        // silently stops being a feed at all.
        log.warn('could not apply an event', {
          server: server.id,
          id: item && item.id,
          kind: item && item.kind,
          error: err.message,
        })
      }
    }

    const lastId = Number(res.data.lastId)

    if (Number.isFinite(lastId) && lastId > since) {
      // AFTER the batch. See the header.
      await db.setCursor(server.id, lastId, items.length)
      since = lastId
    }

    if (!res.data.more) break
  }

  if (applied > 0) {
    log.info('ingested', { server: server.id, events: applied, cursor: since })
    // A leader can only change when something was applied. Never throws.
    await engagement.checkLeader(server)
  }

  return applied
}

/**
 * Applies the boards: what is true right now, rather than what happened.
 *
 * `players.online` replaces the presence rows wholesale, because that is what a
 * board is. Storing it as history is the mistake the wire's `type` field exists
 * to prevent, and it would be a poor return for the sidecar's trouble to make it
 * here after it went out of its way not to make it there.
 */
async function applyBoards(serverId, boards) {
  const presence = boards && boards['players.online']

  if (presence && Array.isArray(presence.players)) {
    await db.replacePresence(serverId, presence.players)
  }

  // Clans only once the game has spoken at all. A sidecar that has never heard
  // from its plugin holds no boards, and recording "no clan board" then would
  // blame the plugin's protocol for a game server that is simply not up. Left
  // alone, the stored board ages past fresh on its own, which is the true answer.
  if (boards && boards['server.hello']) {
    // Fenced: clans are the one board here that core's Teams depend on, and a
    // failure applying them must cost the clans rather than the presence board
    // above or the server state the caller writes next.
    try {
      await clans.applyBoard(serverId, boards.clans)
      await clans.reofferActivity(serverId)
    } catch (err) {
      log.warn('could not apply the clan board', { server: serverId, error: err.message })
    }
  }
}

module.exports = { apply, applyBoards, ingestServer, BATCH, MAX_BATCHES_PER_TICK }