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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011qPmpmVH1xGCiZoz9m9vW3

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

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",

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",

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 }

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 }

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,
+}

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,
+}

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 {

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 }

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 }

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 }