website/server/src/utils/uoLinkSocket.js

// ── 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 shardState = require('../model/shardState/shardState.model')
const newsGump = require('./newsGump')
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
}

// One guarded board snapshot: fetch, verify `data[key]` is an array, hand it to
// `apply`, and (when given) log `label` with the row count. Isolated so a
// failed/absent board never aborts the rest of backfill — and so backfill()
// stays a flat sequence rather than nine repetitions of the same guard.
async function snapshot(fetchFn, key, apply, label) {
  const res = await fetchFn()
  if (!res.ok || !res.data || !Array.isArray(res.data[key])) return
  await apply(res.data[key])
  if (label) log.info(label, { count: res.data[key].length })
}

// Replay events through the dispatcher oldest-first (history/economy arrive
// newest-first) so latest-wins state settles correctly.
async function ingestReversed(events) {
  for (const ev of [...events].reverse()) await shardIngest.ingest(ev, { fromBackfill: true })
}
async function ingestEach(events) {
  for (const ev of events) await shardIngest.ingest(ev, { fromBackfill: true })
}

// ── Market backfill ────────────────────────────────────────────────────────
//
// The market is the only board that does not fit in one response, so /market is
// paged and this walks it. Two bounds, both deliberate:
//
//   • MARKET_SNAPSHOT_MAX caps the walk. A pathological world (or a sidecar whose
//     store was never pruned) must not be able to hang startup — backfill runs
//     before the site is serving the live feed, so an unbounded loop here is
//     downtime, not slowness.
//   • The loop stops on a SHORT page as well as on `total`, because a concurrent
//     sweep can shrink the index underneath the walk and paging to a stale total
//     would spin.
//
// Vendors are upserted, never reconciled-by-replacement. A vendor absent from the
// snapshot is absent because the sidecar dropped it on vendor.listing.remove —
// which our own ingest already processed — so clearing the table first would only
// create a window where the market page is empty.
const MARKET_SNAPSHOT_MAX = 5000
const MARKET_PAGE = 200

async function backfillMarket() {
  let offset = 0
  let seen = 0

  for (;;) {
    const res = await uoLinkClient.getMarket({ limit: MARKET_PAGE, offset })
    if (!res.ok || !res.data || !Array.isArray(res.data.vendors)) return

    const page = res.data.vendors
    if (page.length === 0) break

    await ingestEach(page)
    seen += page.length
    offset += page.length

    if (page.length < MARKET_PAGE) break
    if (seen >= MARKET_SNAPSHOT_MAX) {
      log.warn('market snapshot truncated at the safety cap', {
        cap: MARKET_SNAPSHOT_MAX,
        total: res.data.total,
      })
      break
    }
    if (Number.isFinite(res.data.total) && offset >= res.data.total) break
  }

  if (seen > 0) log.info('snapshotted player-vendor market from /market', { count: seen })
}

// 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 {
    await snapshot(() => uoLinkClient.getHistory({ limit: BACKFILL_LIMIT }), 'events', ingestReversed, 'backfilled events from /history')
    await snapshot(() => uoLinkClient.getEconomy(200), 'series', ingestReversed)

    // Champ board + help-page queue have no replay stream — snapshot the
    // authoritative current state directly (the sidecar guide's advice for both),
    // reconciling our tables to it so a stale row from before a disconnect can't
    // linger. Live champ.*/page.* deltas keep them fresh thereafter.
    await snapshot(() => uoLinkClient.getChamps(), 'spawns', (s) => shardState.replaceChamps(s), 'snapshotted champ board from /champs')
    await snapshot(() => uoLinkClient.getPages(), 'pages', (p) => shardState.replacePages(p), 'snapshotted help-page queue from /pages')

    // ── Protocol 2.0 boards ──────────────────────────────────────────────
    // Same as champs/pages: snapshot the authoritative current state and
    // reconcile our tables to it. Each call is independently guarded so a
    // failed/absent board (e.g. no City Loyalty → empty /governors) never wipes
    // another. Governors are NOT cleared before upsert (cities are fixed and the
    // term-capture is idempotent, so a reconnect can't spawn spurious terms).
    await snapshot(() => uoLinkClient.getGuilds(), 'guilds', (g) => shardState.replaceGuilds(g), 'snapshotted guild board from /guilds')
    await snapshot(() => uoLinkClient.getGovernors(), 'cities', (c) => shardState.replaceGovernors(c), 'snapshotted governor board from /governors')
    await snapshot(() => uoLinkClient.getHouses(), 'houses', ingestEach, 'snapshotted house registry from /houses')

    // ── Protocol 3.0 ─────────────────────────────────────────────────────
    // The ruleset is object-shaped, not a board, so it can't go through
    // snapshot() (which asserts an array under `key`). The shard also re-emits
    // world.ruleset on its own connect — this covers the other order, where the
    // sidecar was already up and holding the ruleset when WE reconnected.
    //
    // Routed through the dispatcher rather than straight to shardState, exactly as
    // ingestEach does for the array-shaped boards: the two orders must produce the
    // same stored frame, and calling setRuleset directly here made this a second
    // write path that silently skipped the shard-name normalization the live frame
    // gets. One writer, one set of rules.
    const ruleset = await uoLinkClient.getRuleset()
    if (ruleset.ok && ruleset.data && ruleset.data.ruleset) {
      await shardIngest.ingest(ruleset.data.ruleset, { fromBackfill: true })
      log.info('snapshotted shard ruleset from /ruleset', { rev: ruleset.data.ruleset.rev })
    }

    // Points boards ARE array-shaped, so they go through snapshot() — but with
    // ingestEach rather than a replace*: there is no points.remove and the shard's
    // system set is fixed, so upserting is the whole reconciliation. A system the
    // operator has since excluded keeps its last-known board rather than vanishing,
    // which is the right answer for a month-scale standing.
    await snapshot(() => uoLinkClient.getPoints(), 'boards', ingestEach, 'snapshotted points boards from /points')

    await backfillMarket()

    const presence = await uoLinkClient.getPresence()
    if (presence.ok && presence.data && typeof presence.data.count === 'number') {
      await shardState.setPresence(presence.data)
      log.info('snapshotted online population from /online', { count: presence.data.count })
    }

    // Re-assert our published news into the in-game Town Cryer News gump. The
    // website is the source of truth; this reconciles the gump on every
    // (re)connect (and recovers any article whose original live push failed).
    // Silent (announce:false) so a reconnect never re-proclaims old news.
    await newsGump.reassertAll()
  } 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 || 3
  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', handleOpen)
  ws.on('message', handleMessage)
  ws.on('close', handleClose)
  ws.on('error', (err) => {
    log.warn('uo-link WS error', { message: err.message })
    // 'close' fires after 'error'; reconnect is scheduled there.
  })
}

// WS lifecycle handlers, split out of connect() so it stays a flat setup path.
async function handleOpen() {
  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()
}

// A ws.hello frame: mark it seen and, on a protocol mismatch, record the error
// and close (we won't run against an incompatible sidecar).
async function handleHello(event) {
  helloSeen = true
  if (!event.protocol || event.protocol === state.protocol) return
  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 */
  }
}

async function handleMessage(raw) {
  let event
  try {
    event = JSON.parse(raw.toString())
  } catch {
    log.warn('dropping non-JSON WS frame')
    return
  }

  if (event.kind === 'ws.hello') return handleHello(event)
  if (event.kind === 'pong') return // sidecar heartbeat — ignore

  state.lastEventAt = Number.isFinite(event.t) ? event.t : Date.now()
  await shardIngest.ingest(event)
}

async function handleClose() {
  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()
}

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