feat: ingest protocol 2, and keep the record a wipe cannot erase

The module half of the read path. Seven tables, an ingest cursor, four public
routes, and one file whose only job is deciding who may see what.

**The record and the window are different things.** `rust_player_wipe_stats` and
`rust_gather_totals` are permanent and per-wipe, so all-time is those rows SUMmed
rather than a second set of counters that can disagree with them — that is R12's
"per-wipe detail plus all-time rollups" in one table instead of two.
`rust_events` is a bounded 30-day window of raw frames for the killfeed, and
`rust_presence` is a board: replaced wholesale, never appended.

**The feed is a cursor, not a socket, and the header says why.** Core runs Node
20, where a global WebSocket is still behind a flag, so a socket means taking
`ws` — against a release that asserts it has no runtime dependencies (D5). The
deciding argument is the other one though: a socket needs a cursor anyway, for
whatever it missed while the module was restarting, and the catch-up path is the
one that has to be right. A cursor alone is one mechanism exercised every five
seconds rather than two where the second only runs after an outage.

**The cursor advances after the batch, never before.** A crash between the two
re-reads events already counted, which inflates a total; the other order loses
them silently and for ever. One is visible and bounded, the other is invisible
and permanent, so the code fails in the visible direction. A server with no
cursor starts at the sidecar's current END rather than at zero — replaying a
fortnight of deaths into stats for wipes the site never saw is not a catch-up.

**`catalogue.js` is a security boundary, default-deny.** Protocol 2 carries IP
addresses (login attempts, approvals, bans), one player's report about another,
and the grid reference of somebody's base. They are stored, because an operator
chasing ban evasion needs them; they are not served below the admin tier. The
allowlist lives here rather than as a field on the wire, because a boundary
declared by the sender is one a compromised or merely out-of-date game host can
widen — the same reason core's own shard fan-out filters on the serving side. A
kind this build has never heard of is not public, and a test holds the list
against PROTOCOL.md §8.4 so that adding a kind to the protocol without
classifying it fails a build.

`PROTOCOL_VERSION` goes to 2 here in the same change as the emitters, though this
module consumes none of the new frames yet: the sidecar refuses a mismatched
client with a 409, so a module left on 1 would stop being able to read the board
it has been reading all along. A constant that lags the deployment is an outage
with a version number on it.

95 server tests, 20 client tests, every guard green, and `routes.manifest.json`
regenerated against a real core at the pinned ref: 10 routes, all documented,
none of core's moved.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016wDDVXWMDz82WqE1i969r4

server/model/events/events.db.js

@@ -0,0 +1,289 @@
+// ── SQL for the read path ─────────────────────────────────────────────────
+//
+// Writes come from one caller (`server/ingest.js`) and reads from the routers.
+// They live together because they are the same tables and the invariants are
+// easier to keep true when the UPDATE and the SELECT are on the same screen.
+//
+// Raw parameterised SQL through `core.query`, no ORM. Placeholders always —
+// except for one place where a list of kinds is expanded into placeholders, and
+// that expansion is checked in `events.model.js` before it ever reaches here.
+
+const core = require('../../core')
+
+const EVENTS = 'rust_events'
+const STATS = 'rust_player_wipe_stats'
+const GATHER = 'rust_gather_totals'
+const PLAYERS = 'rust_players'
+const WIPES = 'rust_wipes'
+const PRESENCE = 'rust_presence'
+const CURSOR = 'rust_ingest_cursor'
+
+// ── The cursor ────────────────────────────────────────────────────────────
+
+async function getCursor(serverId) {
+  const rows = await core.query(
+    `SELECT server_id AS serverId, last_event_id AS lastEventId, events_seen AS eventsSeen
+       FROM ${CURSOR} WHERE server_id = ?`,
+    [serverId],
+  )
+  return rows[0] || null
+}
+
+/**
+ * Moves a server's cursor forward, counting what it passed.
+ *
+ * **Called only after the batch it describes has been written.** The whole
+ * correctness of the ingest is in that ordering: if this ran first, a crash
+ * between the two would skip events for ever, silently, with no way to notice.
+ * Running it last means a crash re-reads events it has already counted at worst
+ * — see `ingest.js` for what makes that survivable.
+ */
+async function setCursor(serverId, lastEventId, seen = 0) {
+  await core.query(
+    `INSERT INTO ${CURSOR} (server_id, last_event_id, events_seen, updated_at)
+     VALUES (?, ?, ?, CURRENT_TIMESTAMP)
+     ON DUPLICATE KEY UPDATE
+       last_event_id = VALUES(last_event_id),
+       events_seen = events_seen + VALUES(events_seen),
+       updated_at = CURRENT_TIMESTAMP`,
+    [serverId, lastEventId, seen],
+  )
+}
+
+// ── Writes ────────────────────────────────────────────────────────────────
+
+async function insertEvent({ serverId, wipeId, kind, t, steamId, raw }) {
+  await core.query(
+    `INSERT INTO ${EVENTS} (server_id, wipe_id, kind, t, steam_id, raw)
+     VALUES (?, ?, ?, ?, ?, ?)`,
+    [serverId, wipeId || null, kind, t, steamId || null, JSON.stringify(raw)],
+  )
+}
+
+/**
+ * Notes that a wipe exists, from any frame that mentions it.
+ *
+ * There is no "a wipe started" call, because the website is not there when one
+ * does — a wipe happens to a game server that was restarted while nobody was
+ * watching. A wipe is therefore created by being mentioned, and `last_seen`
+ * moves every time it is mentioned again.
+ */
+async function touchWipe(serverId, wipeId, saveCreatedAt = null) {
+  if (!wipeId) return
+
+  await core.query(
+    `INSERT INTO ${WIPES} (server_id, wipe_id, save_created_at, first_seen, last_seen)
+     VALUES (?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+     ON DUPLICATE KEY UPDATE
+       last_seen = CURRENT_TIMESTAMP,
+       save_created_at = COALESCE(VALUES(save_created_at), save_created_at)`,
+    [serverId, wipeId, saveCreatedAt],
+  )
+}
+
+/**
+ * Notes that a player exists and what they were last called.
+ *
+ * `name` is COALESCEd rather than overwritten so that a frame which carries no
+ * name — a ban by id, a tally — cannot blank out the name every other frame
+ * supplied.
+ */
+async function touchPlayer(steamId, name = null) {
+  if (!steamId) return
+
+  await core.query(
+    `INSERT INTO ${PLAYERS} (steam_id, name, first_seen, last_seen)
+     VALUES (?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+     ON DUPLICATE KEY UPDATE
+       name = COALESCE(VALUES(name), name),
+       last_seen = CURRENT_TIMESTAMP`,
+    [steamId, name],
+  )
+}
+
+/**
+ * Adds to one player's counters for one wipe.
+ *
+ * Every column is a running total that only rises within a wipe, so this is an
+ * upsert that ADDS rather than sets. `deltas` names only what moved; a `+ 0` on
+ * everything else is what keeps the caller from having to read the row first.
+ */
+async function addStats({ serverId, wipeId, steamId }, deltas = {}) {
+  if (!serverId || !steamId) return
+
+  const cols = ['kills', 'deaths', 'suicides', 'npc_kills', 'structures', 'sessions', 'playtime_sec']
+  const values = {
+    kills: deltas.kills || 0,
+    deaths: deltas.deaths || 0,
+    suicides: deltas.suicides || 0,
+    npc_kills: deltas.npcKills || 0,
+    structures: deltas.structures || 0,
+    sessions: deltas.sessions || 0,
+    playtime_sec: deltas.playtimeSec || 0,
+  }
+
+  await core.query(
+    `INSERT INTO ${STATS} (server_id, wipe_id, steam_id, ${cols.join(', ')}, last_seen)
+     VALUES (?, ?, ?, ${cols.map(() => '?').join(', ')}, CURRENT_TIMESTAMP)
+     ON DUPLICATE KEY UPDATE
+       ${cols.map((c) => `${c} = ${c} + VALUES(${c})`).join(',\n       ')},
+       last_seen = CURRENT_TIMESTAMP`,
+    [serverId, wipeId || '', steamId, ...cols.map((c) => values[c])],
+  )
+}
+
+async function addGathered({ serverId, wipeId, steamId }, resource, amount) {
+  if (!serverId || !steamId || !resource || !(amount > 0)) return
+
+  await core.query(
+    `INSERT INTO ${GATHER} (server_id, wipe_id, steam_id, resource, amount)
+     VALUES (?, ?, ?, ?, ?)
+     ON DUPLICATE KEY UPDATE amount = amount + VALUES(amount)`,
+    [serverId, wipeId || '', steamId, resource, amount],
+  )
+}
+
+/**
+ * Replaces a server's presence rows with exactly what the board said.
+ *
+ * Two statements, delete then insert, because a board is a REPLACEMENT: a player
+ * who left between two boards has to disappear, and an upsert alone would leave
+ * them online for ever. It is not wrapped in a transaction on purpose — the
+ * window between the two is a fraction of a second of a page possibly showing an
+ * empty player list, against holding a lock on a table two routes read.
+ */
+async function replacePresence(serverId, players = []) {
+  await core.query(`DELETE FROM ${PRESENCE} WHERE server_id = ?`, [serverId])
+
+  for (const p of players) {
+    if (!p || !p.steamId) continue
+
+    await core.query(
+      `INSERT INTO ${PRESENCE} (server_id, steam_id, name, sleeping, connected_at, updated_at)
+       VALUES (?, ?, ?, ?, ${p.connectedAt ? 'FROM_UNIXTIME(? / 1000)' : 'NULL'}, CURRENT_TIMESTAMP)
+       ON DUPLICATE KEY UPDATE
+         name = VALUES(name), sleeping = VALUES(sleeping), updated_at = CURRENT_TIMESTAMP`,
+      p.connectedAt
+        ? [serverId, p.steamId, p.name || null, p.sleeping ? 1 : 0, p.connectedAt]
+        : [serverId, p.steamId, p.name || null, p.sleeping ? 1 : 0],
+    )
+  }
+}
+
+/** Deletes raw events older than `days`. Totals are never touched — that is the point of them. */
+async function pruneEvents(days) {
+  if (!(days > 0)) return 0
+
+  const res = await core.query(
+    `DELETE FROM ${EVENTS} WHERE created_at < DATE_SUB(CURRENT_TIMESTAMP, INTERVAL ? DAY)`,
+    [days],
+  )
+  return (res && res.affectedRows) || 0
+}
+
+// ── Reads ─────────────────────────────────────────────────────────────────
+
+/**
+ * Recent events, newest first, restricted to `kinds`.
+ *
+ * **`kinds` is never optional.** A default of "all kinds" is one forgotten
+ * argument away from publishing an IP address, so the caller is made to say it
+ * every time; `events.model.js` builds the list from the catalogue's allowlist
+ * and an empty list answers with no rows rather than with everything.
+ */
+async function recentEvents({ serverId, kinds, wipeId = null, limit = 50 }) {
+  if (!Array.isArray(kinds) || kinds.length === 0) return []
+
+  const holes = kinds.map(() => '?').join(', ')
+  const params = [serverId, ...kinds]
+
+  let sql = `SELECT id, server_id AS serverId, wipe_id AS wipeId, kind, t, steam_id AS steamId, raw
+               FROM ${EVENTS}
+              WHERE server_id = ? AND kind IN (${holes})`
+
+  if (wipeId) {
+    sql += ' AND wipe_id = ?'
+    params.push(wipeId)
+  }
+
+  sql += ' ORDER BY id DESC LIMIT ?'
+  params.push(limit)
+
+  return core.query(sql, params)
+}
+
+/**
+ * The leaderboard for one wipe, or across every wipe when `wipeId` is null.
+ *
+ * All-time is a SUM over the per-wipe rows rather than a separate set of
+ * counters, which is what makes it impossible for the two to disagree — there
+ * is only ever one number, added up differently.
+ */
+async function leaderboard({ serverId, wipeId = null, sort = 'kills', limit = 25 }) {
+  const column = { kills: 'kills', deaths: 'deaths', npcKills: 'npc_kills', playtime: 'playtime_sec' }[sort] || 'kills'
+
+  const params = [serverId]
+  let where = 's.server_id = ?'
+
+  if (wipeId) {
+    where += ' AND s.wipe_id = ?'
+    params.push(wipeId)
+  }
+
+  params.push(limit)
+
+  return core.query(
+    `SELECT s.steam_id AS steamId,
+            p.name AS name,
+            SUM(s.kills) AS kills,
+            SUM(s.deaths) AS deaths,
+            SUM(s.npc_kills) AS npcKills,
+            SUM(s.structures) AS structures,
+            SUM(s.playtime_sec) AS playtimeSec,
+            MAX(s.last_seen) AS lastSeen
+       FROM ${STATS} s
+       LEFT JOIN ${PLAYERS} p ON p.steam_id = s.steam_id
+      WHERE ${where}
+      GROUP BY s.steam_id, p.name
+      ORDER BY SUM(s.${column}) DESC, MAX(s.last_seen) DESC
+      LIMIT ?`,
+    params,
+  )
+}
+
+async function listWipes(serverId) {
+  return core.query(
+    `SELECT wipe_id AS wipeId, save_created_at AS saveCreatedAt,
+            first_seen AS firstSeen, last_seen AS lastSeen
+       FROM ${WIPES}
+      WHERE server_id = ?
+      ORDER BY wipe_id DESC`,
+    [serverId],
+  )
+}
+
+async function presenceFor(serverId) {
+  return core.query(
+    `SELECT steam_id AS steamId, name, sleeping, connected_at AS connectedAt
+       FROM ${PRESENCE}
+      WHERE server_id = ?
+      ORDER BY name ASC`,
+    [serverId],
+  )
+}
+
+module.exports = {
+  getCursor,
+  setCursor,
+  insertEvent,
+  touchWipe,
+  touchPlayer,
+  addStats,
+  addGathered,
+  replacePresence,
+  pruneEvents,
+  recentEvents,
+  leaderboard,
+  listWipes,
+  presenceFor,
+}

server/model/events/events.model.js

@@ -0,0 +1,162 @@
+// ── The read path's logic ─────────────────────────────────────────────────
+//
+// Everything that decides WHAT a caller gets, separated from the SQL that
+// fetches it, so this file can be tested with no database and `events.db.js` has
+// no branching to test.
+//
+// The decision that matters here is not a business rule, it is a boundary: what
+// a signed-out visitor may see. Protocol 2 carries IP addresses and player
+// reports, and the only thing standing between them and a public page is
+// `catalogue.js`'s allowlist and the fact that **every read on this file takes an
+// explicit viewer**. There is no default, because a default is what a caller
+// gets when they forget — and the safe value is never the one that is easier to
+// type.
+
+const catalogue = require('../../catalogue')
+const db = require('./events.db')
+
+/** Hard ceiling on a page, whatever a caller asks for. */
+const MAX_LIMIT = 200
+
+function boundedLimit(requested, fallback = 50) {
+  const n = Number(requested)
+  if (!Number.isFinite(n) || n <= 0) return fallback
+  return Math.min(Math.trunc(n), MAX_LIMIT)
+}
+
+/**
+ * Parses a `kind` query parameter into a list.
+ *
+ * Accepts `?kind=player.death` and `?kind=player.death,player.chat`, and answers
+ * `null` for anything empty — which means "whatever this viewer may see" rather
+ * than "nothing", and is then narrowed by the catalogue.
+ */
+function parseKinds(raw) {
+  if (!raw) return null
+
+  const list = String(raw)
+    .split(',')
+    .map((k) => k.trim())
+    .filter(Boolean)
+
+  return list.length > 0 ? list : null
+}
+
+/**
+ * Recent events for one server, already narrowed to what this viewer may see.
+ *
+ * **`admin` is a parameter, not a default.** A route that forgets it gets the
+ * public list, which is the direction it is safe to be wrong in. And a kind the
+ * caller asked for that they may not see is dropped silently rather than
+ * refused: naming it in an error would confirm the kind exists, which is a small
+ * thing to leak and a free one to avoid.
+ */
+async function recent({ serverId, admin = false, kind = null, wipeId = null, limit }) {
+  const kinds = catalogue.kindsFor({ admin, requested: parseKinds(kind) })
+
+  // Every requested kind was refused. Answering with an empty list is right —
+  // the events they asked for are, as far as they are concerned, not there.
+  if (kinds.length === 0) return []
+
+  const rows = await db.recentEvents({
+    serverId,
+    kinds,
+    wipeId,
+    limit: boundedLimit(limit),
+  })
+
+  return rows.map(shape)
+}
+
+/**
+ * One stored row as an API object.
+ *
+ * `raw` comes back from the database as text and is parsed here rather than in
+ * the db layer, because a row whose JSON will not parse is a reporting problem
+ * and not a query problem: it answers with the envelope it does know and an
+ * empty body, instead of failing a whole page over one bad row.
+ */
+function shape(row) {
+  let frame = {}
+
+  try {
+    frame = typeof row.raw === 'string' ? JSON.parse(row.raw) : row.raw || {}
+  } catch {
+    frame = {}
+  }
+
+  return {
+    id: Number(row.id),
+    kind: row.kind,
+    t: Number(row.t),
+    wipeId: row.wipeId || null,
+    steamId: row.steamId || null,
+    frame,
+  }
+}
+
+/**
+ * The leaderboard for a server, per wipe or all-time.
+ *
+ * All-time is the same rows summed differently rather than a second set of
+ * counters, so the two can never disagree — which is the whole reason R12's
+ * "per-wipe detail plus all-time rollups" is one table and not two.
+ */
+async function leaderboard({ serverId, wipeId = null, sort = 'kills', limit }) {
+  const rows = await db.leaderboard({
+    serverId,
+    wipeId,
+    sort,
+    limit: boundedLimit(limit, 25),
+  })
+
+  return rows.map((r) => ({
+    steamId: r.steamId,
+    name: r.name || null,
+    kills: Number(r.kills) || 0,
+    deaths: Number(r.deaths) || 0,
+    npcKills: Number(r.npcKills) || 0,
+    structures: Number(r.structures) || 0,
+    playtimeSec: Number(r.playtimeSec) || 0,
+    lastSeen: r.lastSeen || null,
+  }))
+}
+
+/**
+ * Every wipe this server has had, newest first.
+ *
+ * The list is what makes the per-wipe view navigable, and it is also the proof
+ * R12 asks for: a wipe that ended is still here, with its stats still attached.
+ */
+async function wipes(serverId) {
+  const rows = await db.listWipes(serverId)
+
+  return rows.map((r) => ({
+    wipeId: r.wipeId,
+    saveCreatedAt: r.saveCreatedAt || null,
+    firstSeen: r.firstSeen,
+    lastSeen: r.lastSeen,
+  }))
+}
+
+/**
+ * Who is on the server right now.
+ *
+ * Read from the presence board rather than counted from connect and disconnect
+ * events: the board is re-sent on every bridge connect and every minute, so it
+ * is right even after this module has missed something. Counting transitions
+ * instead would drift, and drift in exactly the direction people notice —
+ * players who never left.
+ */
+async function online(serverId) {
+  const rows = await db.presenceFor(serverId)
+
+  return rows.map((r) => ({
+    steamId: r.steamId,
+    name: r.name || null,
+    sleeping: Boolean(r.sleeping),
+    connectedAt: r.connectedAt || null,
+  }))
+}
+
+module.exports = { recent, leaderboard, wipes, online, parseKinds, boundedLimit, MAX_LIMIT }

server/model/servers/servers.db.js

@@ -79,7 +79,8 @@ async function listState() {
   return core.query(
     `SELECT server_id AS serverId, reachable, online, players, max_players AS maxPlayers,
             hostname, level, seed, world_size AS worldSize, boot_id AS bootId,
-            save_created_at AS saveCreatedAt, protocol, updated_at AS updatedAt
+            save_created_at AS saveCreatedAt, wipe_id AS wipeId, protocol,
+            updated_at AS updatedAt
        FROM ${STATE}`,
   )
 }
@@ -99,13 +100,14 @@ async function putState(state) {
   await core.query(
     `INSERT INTO ${STATE}
        (server_id, reachable, online, players, max_players, hostname, level, seed,
-        world_size, boot_id, save_created_at, protocol, raw, updated_at)
-     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
+        world_size, boot_id, save_created_at, wipe_id, protocol, raw, updated_at)
+     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
      ON DUPLICATE KEY UPDATE
        reachable = VALUES(reachable), online = VALUES(online), players = VALUES(players),
        max_players = VALUES(max_players), hostname = VALUES(hostname), level = VALUES(level),
        seed = VALUES(seed), world_size = VALUES(world_size), boot_id = VALUES(boot_id),
-       save_created_at = VALUES(save_created_at), protocol = VALUES(protocol),
+       save_created_at = VALUES(save_created_at), wipe_id = VALUES(wipe_id),
+       protocol = VALUES(protocol),
        raw = VALUES(raw), updated_at = CURRENT_TIMESTAMP`,
     [
       state.serverId,
@@ -119,6 +121,7 @@ async function putState(state) {
       state.worldSize === undefined ? null : state.worldSize,
       state.bootId || null,
       state.saveCreatedAt || null,
+      state.wipeId || null,
       state.protocol === undefined ? null : state.protocol,
       state.raw ? JSON.stringify(state.raw) : null,
     ],