From 9d9f5aac28ed1b263c5427646884967eb0d89d07 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 02:08:56 -0500 Subject: [PATCH] Add uo-link WS ingest, storage tables and SSE broadcaster (phase 1) The site now ingests the sidecar's live WebSocket feed and persists it to its own MariaDB, and re-broadcasts curated events to browsers over SSE. - schema: shard_events (append-only notable-kind log, sha1 dedupe_key + INSERT IGNORE for idempotent reconnect backfill), shard_online (current players, upsert/refresh/remove), shard_economy (gold-supply series), shard_houses (per-house decay stage + derived is_idoc). - model/shardEvents + model/shardState: the .db.js/.model.js split; writes take camelCase event data, reads are shaped; online upsert uses COALESCE so a partial char.vitals refresh never blanks login fields. - utils/shardIngest: single dispatcher routing each kind to state writes and/or the event log, then the broadcaster. High-frequency kinds (char.vitals, economy.supply) update state only. A changed server.hello bootId clears the stale online roster. Deps are injected for unit testing. - utils/uoLinkSocket: the server's first outbound WS client (ws dep). Verifies the ws.hello protocol, backfills via /history + /economy on every (re)connect (dedupe handles overlap), reconnects with capped backoff, and mirrors connection state into uo_link_config. Self-guards: only connects when the integration is enabled with a token. - utils/shardBroadcast: SSE fan-out with public (safe kinds only) and admin (all) channels, keepalive pings, per-client cleanup. - server.js: start the ingest socket on boot (no-op until configured) and stop it + close SSE streams on graceful shutdown. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_011qPmpmVH1xGCiZoz9m9vW3 --- server/db/schema.sql | 79 +++++++ server/package-lock.json | 24 ++- server/package.json | 3 +- .../src/model/shardEvents/shardEvents.db.js | 31 +++ .../model/shardEvents/shardEvents.model.js | 52 +++++ server/src/model/shardState/shardState.db.js | 83 ++++++++ .../src/model/shardState/shardState.model.js | 141 +++++++++++++ server/src/server.js | 14 ++ server/src/utils/shardBroadcast.js | 116 +++++++++++ server/src/utils/shardIngest.js | 187 +++++++++++++++++ server/src/utils/uoLinkSocket.js | 195 ++++++++++++++++++ 11 files changed, 923 insertions(+), 2 deletions(-) create mode 100644 server/src/model/shardEvents/shardEvents.db.js create mode 100644 server/src/model/shardEvents/shardEvents.model.js create mode 100644 server/src/model/shardState/shardState.db.js create mode 100644 server/src/model/shardState/shardState.model.js create mode 100644 server/src/utils/shardBroadcast.js create mode 100644 server/src/utils/shardIngest.js create mode 100644 server/src/utils/uoLinkSocket.js diff --git a/server/db/schema.sql b/server/db/schema.sql index 8563126..992f43b 100644 --- a/server/db/schema.sql +++ b/server/db/schema.sql @@ -289,6 +289,85 @@ CREATE TABLE IF NOT EXISTS uo_link_config ( CONSTRAINT chk_uo_link_config_singleton CHECK (id = 1) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +-- Append-only log of notable shard events ingested from the uo-link WebSocket +-- feed. The site OWNS this data (it does not query the sidecar's SQLite): the WS +-- client writes here, and the public/admin read endpoints + live feeds read from +-- here. Only "notable" kinds are logged (sales, deaths, murders, mob.killed, +-- IDOC transitions, quests, skill.gain, fame/karma, audit.*, cheat.*, link.*, +-- server.*). High-frequency kinds (char.vitals, economy.supply) are NOT logged +-- here — they update shard_online / shard_economy instead, keeping the log lean. +-- dedupe_key = sha1(kind + t + stable-json(payload)); with the UNIQUE index it +-- makes INSERT IGNORE idempotent so WS-reconnect backfill never double-inserts. +CREATE TABLE IF NOT EXISTS shard_events ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + kind VARCHAR(48) NOT NULL, + t BIGINT NOT NULL, -- event time, epoch ms (from the sidecar) + boot_id VARCHAR(64) NULL, -- shard boot id at ingest (server.hello.bootId) + payload JSON NOT NULL, -- the full event object + dedupe_key CHAR(40) NOT NULL UNIQUE, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + INDEX idx_shard_events_kind_t (kind, t), + INDEX idx_shard_events_t (t) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +-- Current online players. Upserted on mob.login, refreshed on char.vitals, and +-- removed on mob.logout. Cleared wholesale when the shard restarts (a new +-- server.hello.bootId). web_id is the linked website user id (present when the +-- account is linked), so the roster can be correlated to site accounts. +CREATE TABLE IF NOT EXISTS shard_online ( + serial VARCHAR(20) NOT NULL PRIMARY KEY, -- mobile serial (opaque hex key) + name VARCHAR(120) NULL, + acct VARCHAR(120) NULL, + web_id INT NULL, + map VARCHAR(40) NULL, + x INT NULL, + y INT NULL, + z INT NULL, + hits INT NULL, + hits_max INT NULL, + mana INT NULL, + mana_max INT NULL, + stam INT NULL, + stam_max INT NULL, + str INT NULL, + dex INT NULL, + `int` INT NULL, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + INDEX idx_shard_online_acct (acct) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +-- Total-gold-supply time series (from the periodic economy.supply event). Kept +-- append-only so the public status page can render a supply-over-time sparkline. +CREATE TABLE IF NOT EXISTS shard_economy ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + accounts INT NULL, -- number of accounts included in the total + gold BIGINT NULL, -- total gold supply across all accounts + t BIGINT NOT NULL, -- sample time, epoch ms + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + INDEX idx_shard_economy_t (t) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +-- Current decay stage per house, upserted on house.decay. is_idoc is a derived +-- flag (stage == 'IDOC') so the public "houses in danger" list is a cheap +-- indexed lookup rather than a scan. +CREATE TABLE IF NOT EXISTS shard_houses ( + serial VARCHAR(20) NOT NULL PRIMARY KEY, + stage VARCHAR(24) NULL, -- Somewhat | Fairly | Greatly | IDOC | Collapsed | ... + map VARCHAR(40) NULL, + x INT NULL, + y INT NULL, + z INT NULL, + region VARCHAR(120) NULL, + name VARCHAR(160) NULL, + owner_serial VARCHAR(20) NULL, + owner_acct VARCHAR(120) NULL, + built_on DATETIME NULL, + last_refreshed DATETIME NULL, + is_idoc TINYINT(1) NOT NULL DEFAULT 0, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + INDEX idx_shard_houses_idoc (is_idoc) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + -- Discord bot moderation core (Phase 2). These tables are owned by the bot -- process (its own DB pool, bot/src/db.js) — the main server never reads or -- writes them. They live in the same physical database as everything else diff --git a/server/package-lock.json b/server/package-lock.json index 8deee00..92df4a2 100644 --- a/server/package-lock.json +++ b/server/package-lock.json @@ -26,7 +26,8 @@ "qrcode": "^1.5.4", "sanitize-html": "^2.17.5", "speakeasy": "^2.0.0", - "swagger-ui-express": "^5.0.1" + "swagger-ui-express": "^5.0.1", + "ws": "^8.21.0" }, "devDependencies": { "nodemon": "^3.1.4", @@ -2386,6 +2387,27 @@ "dev": true, "license": "ISC" }, + "node_modules/ws": { + "version": "8.21.0", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.0.tgz", + "integrity": "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + }, "node_modules/y18n": { "version": "4.0.3", "resolved": "https://registry.npmjs.org/y18n/-/y18n-4.0.3.tgz", diff --git a/server/package.json b/server/package.json index de1b81f..8c5a859 100644 --- a/server/package.json +++ b/server/package.json @@ -36,7 +36,8 @@ "qrcode": "^1.5.4", "sanitize-html": "^2.17.5", "speakeasy": "^2.0.0", - "swagger-ui-express": "^5.0.1" + "swagger-ui-express": "^5.0.1", + "ws": "^8.21.0" }, "devDependencies": { "nodemon": "^3.1.4", diff --git a/server/src/model/shardEvents/shardEvents.db.js b/server/src/model/shardEvents/shardEvents.db.js new file mode 100644 index 0000000..792672a --- /dev/null +++ b/server/src/model/shardEvents/shardEvents.db.js @@ -0,0 +1,31 @@ +const { query } = require('../../utils/db') + +// INSERT IGNORE on the UNIQUE dedupe_key — a re-ingested event (WS-reconnect +// backfill overlap) is silently skipped rather than duplicated. Returns true if +// a new row was actually inserted. +async function insertIgnore({ kind, t, bootId, payload, dedupeKey }) { + const res = await query( + `INSERT IGNORE INTO shard_events (kind, t, boot_id, payload, dedupe_key) + VALUES (?, ?, ?, ?, ?)`, + [kind, t, bootId || null, JSON.stringify(payload), dedupeKey], + ) + return res.affectedRows > 0 +} + +// Recent events, newest first. Optional kind filter; limit is clamped by the model. +async function list({ kind, limit }) { + if (kind) { + return query( + `SELECT id, kind, t, boot_id, payload, created_at + FROM shard_events WHERE kind = ? ORDER BY t DESC LIMIT ?`, + [kind, limit], + ) + } + return query( + `SELECT id, kind, t, boot_id, payload, created_at + FROM shard_events ORDER BY t DESC LIMIT ?`, + [limit], + ) +} + +module.exports = { insertIgnore, list } diff --git a/server/src/model/shardEvents/shardEvents.model.js b/server/src/model/shardEvents/shardEvents.model.js new file mode 100644 index 0000000..27818a0 --- /dev/null +++ b/server/src/model/shardEvents/shardEvents.model.js @@ -0,0 +1,52 @@ +// Append-only shard event log. The WS ingest dispatcher calls append() for the +// notable kinds; the public/admin read endpoints call list(). The DB layer only +// sees an already-computed dedupe_key so INSERT IGNORE is idempotent across +// WS-reconnect backfill. + +const crypto = require('crypto') +const db = require('./shardEvents.db') + +const MAX_LIMIT = 1000 +const DEFAULT_LIMIT = 100 + +// Stable stringify — keys sorted — so the dedupe hash is independent of the +// property order the sidecar happened to serialize with. +function stableStringify(value) { + if (value === null || typeof value !== 'object') return JSON.stringify(value) + if (Array.isArray(value)) return `[${value.map(stableStringify).join(',')}]` + const keys = Object.keys(value).sort() + return `{${keys.map((k) => `${JSON.stringify(k)}:${stableStringify(value[k])}`).join(',')}}` +} + +// dedupe_key = sha1(kind + t + stable-json(payload)). Two identical events (same +// kind, same timestamp, same body) collapse to one row. +function dedupeKey(kind, t, payload) { + return crypto.createHash('sha1').update(`${kind}|${t}|${stableStringify(payload)}`).digest('hex') +} + +// Append one event. Returns true if a new row was inserted (false = deduped). +async function append({ kind, t, bootId, payload }) { + return db.insertIgnore({ kind, t, bootId, payload, dedupeKey: dedupeKey(kind, t, payload) }) +} + +function normalizeLimit(limit) { + const n = Number(limit) + if (!Number.isFinite(n) || n <= 0) return DEFAULT_LIMIT + return Math.min(Math.floor(n), MAX_LIMIT) +} + +// Recent events, newest first. Each row's JSON payload is parsed back to an object. +async function list({ kind, limit } = {}) { + const rows = await db.list({ kind, limit: normalizeLimit(limit) }) + return rows.map((row) => ({ + id: row.id, + kind: row.kind, + t: row.t, + bootId: row.boot_id || null, + // mariadb returns JSON columns as strings on some versions; parse defensively. + payload: typeof row.payload === 'string' ? JSON.parse(row.payload) : row.payload, + createdAt: row.created_at, + })) +} + +module.exports = { append, list, dedupeKey } diff --git a/server/src/model/shardState/shardState.db.js b/server/src/model/shardState/shardState.db.js new file mode 100644 index 0000000..0e35276 --- /dev/null +++ b/server/src/model/shardState/shardState.db.js @@ -0,0 +1,83 @@ +const { query } = require('../../utils/db') + +// ── Online players ───────────────────────────────────────────────────────── +const ONLINE_COLS = + 'serial, name, acct, web_id, map, x, y, z, hits, hits_max, mana, mana_max, stam, stam_max, str, dex, `int`, updated_at' + +// Upsert one online player. `fields` already prepared by the model (only the +// columns it wants to write); serial is required and is the primary key. +async function upsertOnline(serial, fields) { + const cols = Object.keys(fields) + const allCols = ['serial', ...cols] + const insertCols = allCols.map((c) => `\`${c}\``).join(', ') + const placeholders = allCols.map(() => '?').join(', ') + // Never overwrite an existing column with NULL on refresh (a char.vitals frame + // that omits acct/name shouldn't blank what mob.login set) — COALESCE keeps the + // prior value when the incoming one is NULL. + const updates = cols.map((c) => `\`${c}\` = COALESCE(VALUES(\`${c}\`), \`${c}\`)`).join(', ') + await query( + `INSERT INTO shard_online (${insertCols}) VALUES (${placeholders}) + ON DUPLICATE KEY UPDATE ${updates}`, + [serial, ...cols.map((c) => fields[c])], + ) +} + +const removeOnline = (serial) => query('DELETE FROM shard_online WHERE serial = ?', [serial]) +const clearOnline = () => query('DELETE FROM shard_online') + +async function countOnline() { + const rows = await query('SELECT COUNT(*) AS n FROM shard_online') + return rows[0] ? Number(rows[0].n) : 0 +} + +const listOnline = () => + query(`SELECT ${ONLINE_COLS} FROM shard_online ORDER BY name ASC`) + +// ── Economy supply series ──────────────────────────────────────────────── +const insertEconomy = ({ accounts, gold, t }) => + query('INSERT INTO shard_economy (accounts, gold, t) VALUES (?, ?, ?)', [ + accounts ?? null, + gold ?? null, + t, + ]) + +const listEconomy = (limit) => + query('SELECT accounts, gold, t FROM shard_economy ORDER BY t DESC LIMIT ?', [limit]) + +async function latestEconomy() { + const rows = await query('SELECT accounts, gold, t FROM shard_economy ORDER BY t DESC LIMIT 1') + return rows[0] || null +} + +// ── Houses / IDOC ──────────────────────────────────────────────────────── +const HOUSE_COLS = + 'serial, stage, map, x, y, z, region, name, owner_serial, owner_acct, built_on, last_refreshed, is_idoc, updated_at' + +async function upsertHouse(serial, fields) { + const cols = Object.keys(fields) + const allCols = ['serial', ...cols] + const insertCols = allCols.map((c) => `\`${c}\``).join(', ') + const placeholders = allCols.map(() => '?').join(', ') + const updates = cols.map((c) => `\`${c}\` = VALUES(\`${c}\`)`).join(', ') + await query( + `INSERT INTO shard_houses (${insertCols}) VALUES (${placeholders}) + ON DUPLICATE KEY UPDATE ${updates}`, + [serial, ...cols.map((c) => fields[c])], + ) +} + +const listIdocHouses = () => + query(`SELECT ${HOUSE_COLS} FROM shard_houses WHERE is_idoc = 1 ORDER BY updated_at DESC`) + +module.exports = { + upsertOnline, + removeOnline, + clearOnline, + countOnline, + listOnline, + insertEconomy, + listEconomy, + latestEconomy, + upsertHouse, + listIdocHouses, +} diff --git a/server/src/model/shardState/shardState.model.js b/server/src/model/shardState/shardState.model.js new file mode 100644 index 0000000..bee0b96 --- /dev/null +++ b/server/src/model/shardState/shardState.model.js @@ -0,0 +1,141 @@ +// Live shard state derived from the WS feed: who is online, the gold-supply +// series, and per-house decay stage. The ingest dispatcher calls the write +// methods; the public read endpoints call the list/count methods. Writes take +// camelCase semantic objects and map to the snake_case columns; only the keys +// present are written (so a char.vitals refresh doesn't clobber login fields). + +const db = require('./shardState.db') + +const MAX_ECONOMY = 1000 + +// Map a camelCase online descriptor to DB columns, dropping undefined keys so a +// partial refresh only touches the fields it carries. +function onlineFields(data) { + const map = { + name: data.name, + acct: data.acct, + web_id: data.webId, + map: data.map, + x: data.x, + y: data.y, + z: data.z, + hits: data.hits, + hits_max: data.hitsMax, + mana: data.mana, + mana_max: data.manaMax, + stam: data.stam, + stam_max: data.stamMax, + str: data.str, + dex: data.dex, + int: data.int, + } + const fields = {} + for (const [k, v] of Object.entries(map)) if (v !== undefined) fields[k] = v + return fields +} + +// Upsert an online player (mob.login) or refresh their vitals (char.vitals). +async function upsertOnline(data) { + if (!data || !data.serial) return + await db.upsertOnline(data.serial, onlineFields(data)) +} + +const setOffline = (serial) => db.removeOnline(serial) +const clearOnline = () => db.clearOnline() +const onlineCount = () => db.countOnline() + +async function listOnline() { + const rows = await db.listOnline() + return rows.map((r) => ({ + serial: r.serial, + name: r.name, + acct: r.acct, + webId: r.web_id, + map: r.map, + x: r.x, + y: r.y, + z: r.z, + hits: r.hits, + hitsMax: r.hits_max, + mana: r.mana, + manaMax: r.mana_max, + stam: r.stam, + stamMax: r.stam_max, + str: r.str, + dex: r.dex, + int: r.int, + updatedAt: r.updated_at, + })) +} + +// Append a gold-supply sample (economy.supply). +async function addEconomySample({ accounts, gold, t }) { + await db.insertEconomy({ accounts, gold, t }) +} + +async function listEconomy(limit = 100) { + const n = Math.min(Math.max(Number(limit) || 100, 1), MAX_ECONOMY) + const rows = await db.listEconomy(n) + // Return oldest → newest for charting. + return rows + .map((r) => ({ accounts: r.accounts, gold: r.gold == null ? null : Number(r.gold), t: r.t })) + .reverse() +} + +async function latestEconomy() { + const r = await db.latestEconomy() + return r ? { accounts: r.accounts, gold: r.gold == null ? null : Number(r.gold), t: r.t } : null +} + +// Upsert a house's decay stage (house.decay). is_idoc is derived from the stage. +async function upsertHouse(data) { + if (!data || !data.serial) return + const fields = { + stage: data.stage ?? null, + map: data.map ?? null, + x: data.x ?? null, + y: data.y ?? null, + z: data.z ?? null, + region: data.region ?? null, + name: data.name ?? null, + owner_serial: data.ownerSerial ?? null, + owner_acct: data.ownerAcct ?? null, + built_on: data.builtOn ? new Date(data.builtOn) : null, + last_refreshed: data.lastRefreshed ? new Date(data.lastRefreshed) : null, + is_idoc: String(data.stage).toUpperCase() === 'IDOC' ? 1 : 0, + } + await db.upsertHouse(data.serial, fields) +} + +async function listIdoc() { + const rows = await db.listIdocHouses() + return rows.map((r) => ({ + serial: r.serial, + stage: r.stage, + map: r.map, + x: r.x, + y: r.y, + z: r.z, + region: r.region, + name: r.name, + ownerSerial: r.owner_serial, + ownerAcct: r.owner_acct, + builtOn: r.built_on, + lastRefreshed: r.last_refreshed, + isIdoc: Boolean(r.is_idoc), + updatedAt: r.updated_at, + })) +} + +module.exports = { + upsertOnline, + setOffline, + clearOnline, + onlineCount, + listOnline, + addEconomySample, + listEconomy, + latestEconomy, + upsertHouse, + listIdoc, +} diff --git a/server/src/server.js b/server/src/server.js index 8ad8e81..12cd45c 100644 --- a/server/src/server.js +++ b/server/src/server.js @@ -4,6 +4,8 @@ const http = require('http') const app = require('./app') const internalApp = require('./internalApp') const botScore = require('./middleware/botScore') +const uoLinkSocket = require('./utils/uoLinkSocket') +const shardBroadcast = require('./utils/shardBroadcast') const { ensureSchema, close } = require('./utils/db') const { seedDefaults, createInitialAdminFromEnv } = require('../db/seed') const settings = require('./model/settings/settings.model') @@ -77,6 +79,16 @@ async function start() { log.info(`internal API listening on http://${HOST}:${INTERNAL_PORT} (server<->bot only — do NOT proxy)`) }) + // Start the uo-link WebSocket ingest client. Self-guards: it only actually + // connects when the admin has enabled the integration and saved a token, so + // this is a no-op on shards that haven't configured the sidecar. Never let a + // sidecar problem block server startup. + try { + await uoLinkSocket.start() + } catch (err) { + log.warn('uo-link socket failed to start (continuing)', { error: err.message }) + } + setupShutdown(server, internalServer) } @@ -87,6 +99,8 @@ function setupShutdown(server, internalServer) { closing = true log.warn(`${signal} received — shutting down gracefully`) botScore.stopSweeper() // stop the bot-store cleanup interval + uoLinkSocket.stop() // close the uo-link WS ingest client + shardBroadcast.closeAll() // end any open shard live-feed SSE streams server.close(() => log.info('http server closed')) if (internalServer) internalServer.close(() => log.info('internal http server closed')) try { diff --git a/server/src/utils/shardBroadcast.js b/server/src/utils/shardBroadcast.js new file mode 100644 index 0000000..13f740c --- /dev/null +++ b/server/src/utils/shardBroadcast.js @@ -0,0 +1,116 @@ +// ── Shard live-feed SSE broadcaster ──────────────────────────────────────── +// +// The browser can't talk to the sidecar's WebSocket directly (the token must +// never reach it, and the WS may be on another host). Instead the server ingests +// the WS feed and re-broadcasts curated events to browsers over Server-Sent +// Events (plain HTTP — works through any reverse proxy). +// +// Two channels: +// • public — safe kinds only (sales, deaths, IDOC, logins, economy). No IPs, +// no account-login attempts, no staff audit / cheat events. +// • admin — everything, including the sensitive kinds above. +// +// shardIngest calls broadcast(event) for each ingested event; the public/admin +// SSE route handlers call subscribe(req, res, channel). + +const log = require('./logger')('shard-broadcast') + +// Kinds safe to expose to unauthenticated browsers. +const PUBLIC_KINDS = new Set([ + 'vendor.sale', + 'player.death', + 'player.murdered', + 'mob.killed', + 'house.decay', + 'quest.complete', + 'skill.gain', + 'fame.change', + 'karma.change', + 'mob.login', + 'mob.logout', + 'economy.supply', + 'server.hello', + 'server.shutdown', + 'server.crashed', +]) + +// Open response streams per channel. +const clients = { public: new Set(), admin: new Set() } + +const KEEPALIVE_MS = 25000 + +// Register an SSE stream on a channel. Sets the SSE headers, sends an initial +// comment, keeps the connection warm with periodic pings, and cleans up on close. +function subscribe(req, res, channel) { + const bucket = clients[channel] + if (!bucket) { + res.status(400).end() + return + } + + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', // disable proxy buffering so events flush immediately + }) + res.write('retry: 5000\n\n') // tell EventSource to reconnect after 5s if dropped + res.write(': connected\n\n') + + bucket.add(res) + + const ping = setInterval(() => { + try { + res.write(': ping\n\n') + } catch { + /* write after close — cleanup below handles it */ + } + }, KEEPALIVE_MS) + + const cleanup = () => { + clearInterval(ping) + bucket.delete(res) + } + req.on('close', cleanup) + res.on('error', cleanup) +} + +function writeTo(bucket, payload) { + for (const res of bucket) { + try { + res.write(payload) + } catch (err) { + log.warn('sse write failed; dropping client', { message: err.message }) + bucket.delete(res) + } + } +} + +// Fan an ingested event out to the admin channel (always) and the public +// channel (safe kinds only). A no-op when nobody is subscribed. +function broadcast(event) { + if (!event || !event.kind) return + const frame = `data: ${JSON.stringify(event)}\n\n` + if (clients.admin.size) writeTo(clients.admin, frame) + if (clients.public.size && PUBLIC_KINDS.has(event.kind)) writeTo(clients.public, frame) +} + +// Close every open stream (graceful shutdown). +function closeAll() { + for (const channel of Object.values(clients)) { + for (const res of channel) { + try { + res.end() + } catch { + /* ignore */ + } + } + channel.clear() + } +} + +function stats() { + return { publicClients: clients.public.size, adminClients: clients.admin.size } +} + +module.exports = { subscribe, broadcast, closeAll, stats, PUBLIC_KINDS } diff --git a/server/src/utils/shardIngest.js b/server/src/utils/shardIngest.js new file mode 100644 index 0000000..8b82c4a --- /dev/null +++ b/server/src/utils/shardIngest.js @@ -0,0 +1,187 @@ +// ── Shard event ingest dispatcher ────────────────────────────────────────── +// +// The single entry point for every event that arrives on the uo-link WebSocket +// feed (and for backfilled /history events on reconnect). It routes by kind: +// • state-changing kinds update shard_online / shard_economy / shard_houses, +// • notable kinds are appended to the append-only shard_events log, +// • every kind is fanned out to the SSE broadcaster (which decides public vs +// admin visibility). +// High-frequency kinds (char.vitals, economy.supply) are deliberately NOT logged +// to shard_events — they only update state — keeping the event log lean. +// +// Dependencies are injected (defaulting to the real models) so the routing can +// be unit-tested with mocked writes. + +const shardEventsModel = require('../model/shardEvents/shardEvents.model') +const shardStateModel = require('../model/shardState/shardState.model') +const uoLinkConfigModel = require('../model/uoLinkConfig/uoLinkConfig.model') +const broadcaster = require('./shardBroadcast') +const defaultLog = require('./logger')('shard-ingest') + +// Notable kinds appended to the shard_events log. High-frequency/session kinds +// (char.vitals, economy.supply, mob.login/logout, account.login.attempt, +// gold.change, vendor.buy/sell) are excluded on purpose. house.decay is handled +// specially — logged only on the transition INTO IDOC. +const LOGGED_KINDS = new Set([ + 'vendor.sale', + 'player.death', + 'player.murdered', + 'mob.killed', + 'quest.complete', + 'skill.gain', + 'fame.change', + 'karma.change', + 'audit.set', + 'audit.command', + 'cheat.fastwalk', + 'link.request', + 'server.hello', + 'server.shutdown', + 'server.crashed', +]) + +// Tracks the current shard boot id so a restart (changed bootId on server.hello) +// can be detected and stale online state dropped. Module-level so it survives +// across events within a process; reset() is exposed for tests. +const state = { bootId: null } +function reset() { + state.bootId = null +} + +// Should this event be written to the append-only log? +function shouldLog(event) { + if (event.kind === 'house.decay') return String(event.to).toUpperCase() === 'IDOC' + return LOGGED_KINDS.has(event.kind) +} + +// Apply the state-change side effect for a kind (if any). Returns a promise. +async function applyStateChange(event, deps) { + const { shardState, uoLinkConfig, log } = deps + switch (event.kind) { + case 'server.hello': { + const incoming = event.bootId || null + if (incoming && state.bootId && incoming !== state.bootId) { + log.warn('shard restarted (bootId changed) — clearing online roster', { + from: state.bootId, + to: incoming, + }) + await shardState.clearOnline() + } + if (incoming) state.bootId = incoming + await uoLinkConfig.recordStatus({ pluginConnected: true, bootId: incoming, lastEventAt: event.t }) + return + } + case 'server.shutdown': + case 'server.crashed': + // Shard is going away — nobody is online anymore. + await shardState.clearOnline() + await uoLinkConfig.recordStatus({ pluginConnected: false }) + return + case 'mob.login': { + const who = event.who || {} + await shardState.upsertOnline({ + serial: who.serial, + name: who.name, + acct: who.acct, + webId: event.webId, + map: event.map, + x: event.x, + y: event.y, + z: event.z, + }) + return + } + case 'mob.logout': { + const who = event.who || {} + if (who.serial) await shardState.setOffline(who.serial) + return + } + case 'char.vitals': + await shardState.upsertOnline({ + serial: event.serial, + hits: event.hits, + hitsMax: event.hitsMax, + mana: event.mana, + manaMax: event.manaMax, + stam: event.stam, + stamMax: event.stamMax, + str: event.str, + dex: event.dex, + int: event.int, + map: event.map, + x: event.x, + y: event.y, + }) + return + case 'economy.supply': + await shardState.addEconomySample({ accounts: event.accounts, gold: event.gold, t: event.t }) + return + case 'house.decay': + await shardState.upsertHouse({ + serial: event.serial, + stage: event.to, + map: event.map, + x: event.x, + y: event.y, + z: event.z, + region: event.region, + name: event.name, + ownerSerial: event.ownerSerial, + ownerAcct: event.ownerAcct, + builtOn: event.builtOn, + lastRefreshed: event.lastRefreshed, + }) + return + default: + // No state side effect (e.g. vendor.sale, audit.*, cheat.*) — logging and + // broadcasting still happen in ingest(). + } +} + +// Ingest one event. Returns { logged, stored } for tests/stats. `fromBackfill` +// suppresses the SSE broadcast (a reconnect replay shouldn't re-animate the +// live ticker). Never throws — a bad single event must not kill the feed. +async function ingest(event, deps = {}) { + const d = { + shardEvents: deps.shardEvents || shardEventsModel, + shardState: deps.shardState || shardStateModel, + uoLinkConfig: deps.uoLinkConfig || uoLinkConfigModel, + broadcast: deps.broadcast || broadcaster.broadcast, + log: deps.log || defaultLog, + } + + if (!event || typeof event.kind !== 'string') return { logged: false, stored: false } + // ws.hello / pong are transport frames, not game events. + if (event.kind === 'ws.hello' || event.kind === 'pong') return { logged: false, stored: false } + + const t = Number.isFinite(event.t) ? event.t : Date.now() + let stored = false + let logged = false + + try { + await applyStateChange(event, d) + } catch (err) { + d.log.warn('state-change write failed', { kind: event.kind, message: err.message }) + } + + if (shouldLog(event)) { + logged = true + try { + stored = await d.shardEvents.append({ kind: event.kind, t, bootId: state.bootId, payload: event }) + } catch (err) { + d.log.warn('event log write failed', { kind: event.kind, message: err.message }) + } + } + + if (!deps.fromBackfill) { + try { + d.broadcast(event) + } catch (err) { + d.log.warn('broadcast failed', { kind: event.kind, message: err.message }) + } + } + + return { logged, stored } +} + +module.exports = { ingest, shouldLog, reset, LOGGED_KINDS, state } diff --git a/server/src/utils/uoLinkSocket.js b/server/src/utils/uoLinkSocket.js new file mode 100644 index 0000000..e82370e --- /dev/null +++ b/server/src/utils/uoLinkSocket.js @@ -0,0 +1,195 @@ +// ── uo-link WebSocket ingest client ──────────────────────────────────────── +// +// Long-lived client that connects to the sidecar's push-only WS feed and pumps +// every frame through the ingest dispatcher. This is the server's first +// outbound WebSocket. Lifecycle: +// • start() — connect if the config is enabled and has a token; verify the +// ws.hello protocol; backfill missed events via /history on every +// (re)connect (INSERT IGNORE dedupes the overlap); reconnect with +// capped backoff. +// • stop() — close the socket and stop reconnecting (graceful shutdown). +// Connection state is mirrored into uo_link_config (plugin_connected / status / +// last_event_at) so the admin panel and public status endpoint have live data. + +const WebSocket = require('ws') + +const uoLinkConfig = require('../model/uoLinkConfig/uoLinkConfig.model') +const uoLinkClient = require('./uoLinkClient') +const shardIngest = require('./shardIngest') +const log = require('./logger')('uo-link-socket') + +const BACKOFF_MIN_MS = 1000 +const BACKOFF_MAX_MS = 30000 +const BACKFILL_LIMIT = 500 + +let ws = null +let reconnectTimer = null +let backoff = BACKOFF_MIN_MS +let running = false // set by start()/stop(); guards auto-reconnect +let helloSeen = false + +const state = { + connected: false, + lastEventAt: null, + lastConnectedAt: null, + reconnects: 0, + protocol: null, +} + +function buildUrl(wsUrl, token) { + const sep = wsUrl.includes('?') ? '&' : '?' + return token ? `${wsUrl}${sep}token=${encodeURIComponent(token)}` : wsUrl +} + +// Pull recent events from the sidecar's own store and replay them through the +// dispatcher (fromBackfill = no SSE re-broadcast). dedupe_key + INSERT IGNORE +// make this idempotent, so overlap with what we already stored is harmless. +async function backfill() { + try { + const hist = await uoLinkClient.getHistory({ limit: BACKFILL_LIMIT }) + if (hist.ok && hist.data && Array.isArray(hist.data.events)) { + // History is newest-first; replay oldest-first so latest-wins state (e.g. + // house.decay stage) settles correctly. + const events = [...hist.data.events].reverse() + for (const ev of events) await shardIngest.ingest(ev, { fromBackfill: true }) + log.info('backfilled events from /history', { count: events.length }) + } + const eco = await uoLinkClient.getEconomy(200) + if (eco.ok && eco.data && Array.isArray(eco.data.series)) { + const series = [...eco.data.series].reverse() + for (const ev of series) await shardIngest.ingest(ev, { fromBackfill: true }) + } + } catch (err) { + log.warn('backfill failed (continuing on live feed)', { message: err.message }) + } +} + +function scheduleReconnect() { + if (!running) return + clearTimeout(reconnectTimer) + reconnectTimer = setTimeout(connect, backoff) + log.info(`reconnecting in ${backoff}ms`) + backoff = Math.min(backoff * 2, BACKOFF_MAX_MS) +} + +async function connect() { + if (!running) return + let config + try { + config = await uoLinkConfig.getWithToken() + } catch (err) { + log.error('could not read uo-link config', err) + scheduleReconnect() + return + } + if (!config || !config.enabled || !config.wsUrl || !config.token) { + log.info('uo-link WS not started (disabled or missing url/token)') + running = false + return + } + + state.protocol = config.protocol || 1 + helloSeen = false + const url = buildUrl(config.wsUrl, config.token) + + try { + ws = new WebSocket(url) + } catch (err) { + log.error('failed to open WS', err) + scheduleReconnect() + return + } + + ws.on('open', async () => { + log.info('uo-link WS connected') + state.connected = true + state.lastConnectedAt = Date.now() + backoff = BACKOFF_MIN_MS + await uoLinkConfig.recordStatus({ status: 'connected', statusDetail: null, pluginConnected: true }).catch(() => {}) + await backfill() + }) + + ws.on('message', async (raw) => { + let event + try { + event = JSON.parse(raw.toString()) + } catch { + log.warn('dropping non-JSON WS frame') + return + } + + if (event.kind === 'ws.hello') { + helloSeen = true + if (event.protocol && event.protocol !== state.protocol) { + log.error('uo-link protocol mismatch on ws.hello — closing', { + expected: state.protocol, + got: event.protocol, + }) + await uoLinkConfig + .recordStatus({ status: 'error', statusDetail: `protocol mismatch: expected ${state.protocol}, got ${event.protocol}` }) + .catch(() => {}) + running = false + try { + ws.close() + } catch { + /* ignore */ + } + } + return + } + if (event.kind === 'pong') return // sidecar heartbeat — ignore + + state.lastEventAt = Number.isFinite(event.t) ? event.t : Date.now() + await shardIngest.ingest(event) + }) + + ws.on('close', async () => { + state.connected = false + if (running) state.reconnects += 1 + log.warn('uo-link WS closed') + await uoLinkConfig + .recordStatus({ status: running ? 'reconnecting' : 'disconnected', pluginConnected: false }) + .catch(() => {}) + ws = null + scheduleReconnect() + }) + + ws.on('error', (err) => { + log.warn('uo-link WS error', { message: err.message }) + // 'close' fires after 'error'; reconnect is scheduled there. + }) +} + +// Begin (or restart) the WS client. Idempotent — a running client is stopped +// first so a config save can re-point it at a new URL/token. +async function start() { + stop() + running = true + backoff = BACKOFF_MIN_MS + await connect() +} + +// Stop the client and cancel any pending reconnect. Called on shutdown and +// before a restart. +function stop() { + running = false + clearTimeout(reconnectTimer) + reconnectTimer = null + if (ws) { + try { + ws.removeAllListeners() + ws.close() + } catch { + /* ignore */ + } + ws = null + } + state.connected = false +} + +// Ingestion stats for the admin panel. +function getState() { + return { ...state, running } +} + +module.exports = { start, stop, getState }