// ── 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, }