Module-Rust/server/model/events/events.db.js

// ── 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 WEAPONS = 'rust_weapon_kills'
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])],
  )
}

/**
 * The tally's counts for the chat-title conditions (PLAN_REDESIGNS §5.1): the
 * wire field each is, and the column it adds to. All sums.
 */
const TITLE_SUMS = {
  animalKills: 'animal_kills',
  headshots: 'headshots',
  apcKills: 'apc_kills',
  heliKills: 'heli_kills',
  built: 'built',
  repaired: 'repaired',
  healed: 'healed',
  rockets: 'rockets',
  explosives: 'explosives',
  missions: 'missions',
}

/** The two *best* fields and their columns: the longest single kill, never a sum. */
const TITLE_BESTS = { bestPvpM: 'best_pvp_m', bestPveM: 'best_pve_m' }

/** Item categories the plugin sends under `crafted`, and the column each counts in. */
const CRAFTED = { attire: 'crafted_attire', weapon: 'crafted_weapons' }

/**
 * Adds one tally's title counts to the player's row for the wipe. A sum adds;
 * a best is GREATEST of the stored and the sent, because a tally carries its
 * interval's maximum and two maxima added are not a distance. Nothing is
 * written when the tally carried none of them, so an older plugin's frame
 * costs no statement.
 */
async function addTitleStats({ serverId, wipeId, steamId }, frame = {}) {
  if (!serverId || !steamId) return

  const sums = {}
  for (const [field, col] of Object.entries(TITLE_SUMS)) sums[col] = Math.max(0, Math.floor(Number(frame[field]) || 0))

  const crafted = frame.crafted && typeof frame.crafted === 'object' ? frame.crafted : {}
  for (const [category, col] of Object.entries(CRAFTED)) sums[col] = Math.max(0, Math.floor(Number(crafted[category]) || 0))

  const bests = {}
  for (const [field, col] of Object.entries(TITLE_BESTS)) {
    const m = Number(frame[field])
    bests[col] = Number.isFinite(m) && m > 0 ? Math.min(Math.round(m * 10) / 10, 999999.9) : 0
  }

  if (![...Object.values(sums), ...Object.values(bests)].some((v) => v > 0)) return

  const cols = [...Object.keys(sums), ...Object.keys(bests)]

  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
       ${Object.keys(sums).map((c) => `${c} = ${c} + VALUES(${c})`).join(',\n       ')},
       ${Object.keys(bests).map((c) => `${c} = GREATEST(${c}, VALUES(${c}))`).join(',\n       ')},
       last_seen = CURRENT_TIMESTAMP`,
    [serverId, wipeId || '', steamId, ...Object.values(sums), ...Object.values(bests)],
  )
}

/** One weapon's credited kills (§5.3), added as a delta like a gathered resource. */
async function addWeaponKills({ serverId, wipeId, steamId }, weapon, kills) {
  if (!serverId || !steamId || !weapon || !(kills > 0)) return

  await core.query(
    `INSERT INTO ${WEAPONS} (server_id, wipe_id, steam_id, weapon, kills)
     VALUES (?, ?, ?, ?, ?)
     ON DUPLICATE KEY UPDATE kills = kills + VALUES(kills)`,
    [serverId, wipeId || '', steamId, String(weapon).slice(0, 64), kills],
  )
}

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

/**
 * Login attempts in `[from, to]` that no approval answered (D64).
 *
 * An attempt is answered by a `player.approved` for the same Steam id on the
 * same server stamped from `slackMs` before it to `windowMs` after it. The
 * slack is clock grain: both frames come off one game thread, and an approval
 * stamped a millisecond "early" is still the answer.
 *
 * Grouped on (steam id, t) because a cursor replayed after a crash can store the
 * same attempt twice, and one attempt is one denial however often it was
 * written down.
 */
async function unapprovedLogins({ serverId, from, to, windowMs, slackMs }) {
  return core.query(
    `SELECT a.steam_id AS steamId, a.t AS t,
            MAX(JSON_UNQUOTE(JSON_EXTRACT(a.raw, '$.name'))) AS name
       FROM ${EVENTS} a
      WHERE a.server_id = ? AND a.kind = 'player.login.attempt'
        AND a.steam_id IS NOT NULL AND a.t BETWEEN ? AND ?
        AND NOT EXISTS (
          SELECT 1 FROM ${EVENTS} b
           WHERE b.server_id = a.server_id AND b.kind = 'player.approved'
             AND b.steam_id = a.steam_id
             AND b.t BETWEEN a.t - ? AND a.t + ?
        )
      GROUP BY a.steam_id, a.t
      ORDER BY a.t ASC
      LIMIT 200`,
    [serverId, from, to, slackMs, windowMs],
  )
}

module.exports = {
  getCursor,
  setCursor,
  insertEvent,
  touchWipe,
  touchPlayer,
  addStats,
  addGathered,
  addTitleStats,
  addWeaponKills,
  replacePresence,
  pruneEvents,
  recentEvents,
  leaderboard,
  listWipes,
  presenceFor,
  unapprovedLogins,
}