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 db = require('./model/events/events.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) {
  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,
  })

  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': {
      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)
      }
      break
    }

    case 'player.chat':
    case 'player.respawned':
      await db.touchPlayer(frame.steamId, frame.name || null)
      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)
        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 })

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

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