website/server/src/utils/shardBroadcast.js

// ── 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. Note: vendor.sale is
// deliberately NOT here — sales are owner-private (a linked player sees only
// their own, via /player/shard/sales).
const PUBLIC_KINDS = new Set([
  '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 }