Module-uo/server/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 events to browsers over Server-Sent Events
// (plain HTTP — works through any reverse proxy).
//
// Since Protocol 3.0 the split is no longer "one public channel with a static
// allowlist plus one admin channel". Each subscriber carries the audience rung
// it resolved to at subscribe time, and every frame is
//
//   1. mapped kind → feature (an UNMAPPED kind reaches nobody below admin —
//      fail closed; see utils/shardVisibility.js rule 2),
//   2. gated on that feature being enabled, streamed, and within the viewer's
//      rung, and
//   3. passed through field projection, so `acct` / `webId` and any field an
//      admin has re-gated are stripped per viewer.
//
// **This is the security boundary.** It used to be the PUBLIC_KINDS set in this
// file; it is now the kind map plus the visibility config. PUBLIC_KINDS still
// exists and is still exported, but it is now DERIVED from the kind map (see
// shardVisibility.js) so the two can no longer drift.
//
// shardIngest calls broadcast(event) for each ingested event; the public/admin
// SSE route handlers call subscribe(req, res, channel).

const visibility = require('./shardVisibility')
const log = require('../core').logger('shard-broadcast')

// Re-exported for back-compat: shardEvents `/feed` filtering and
// config/notificationStreams.js both ask "is this kind public-safe?".
const { PUBLIC_KINDS } = visibility

// Open streams. Each entry is { res, level }. The admin bucket is kept separate
// because it is unconditional and must not depend on a config read.
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.
//
// The viewer's rung is resolved ONCE, here, and frozen for the life of the
// connection — a long-lived stream must not silently gain privilege because the
// caller's session changed underneath it. (Config changes, by contrast, DO take
// effect live: the config is read per broadcast, cached ~5s.)
async function subscribe(req, res, channel) {
  const bucket = clients[channel]
  if (!bucket) {
    res.status(400).end()
    return
  }

  let level = 'admin'
  if (channel === 'public') {
    try {
      level = await visibility.viewerLevel(req)
    } catch (err) {
      // Fail closed: an unresolvable viewer is anonymous, not privileged.
      log.warn('viewerLevel failed on subscribe; treating as anonymous', { message: err.message })
      level = 'anonymous'
    }
  }

  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')

  const client = { res, level, ping: null }
  bucket.add(client)

  client.ping = setInterval(() => {
    try {
      res.write(': ping\n\n')
    } catch {
      /* write after close — cleanup below handles it */
    }
  }, KEEPALIVE_MS)

  const cleanup = () => drop(bucket, client)
  req.on('close', cleanup)
  res.on('error', cleanup)
}

// The ONLY way a client leaves a bucket. Clearing the keepalive here (rather
// than only in the close handler) matters: a client dropped because its write
// threw never fires `req.close`, so its interval would otherwise keep firing on
// a dead socket for the life of the process.
function drop(bucket, client) {
  clearInterval(client.ping)
  bucket.delete(client)
}

function writeTo(bucket, client, payload) {
  try {
    client.res.write(payload)
  } catch (err) {
    log.warn('sse write failed; dropping client', { message: err.message })
    drop(bucket, client)
  }
}

// Fan an ingested event out. The admin channel gets it verbatim, always. Public
// subscribers are filtered and projected per their own rung — so two viewers on
// the same channel can legitimately receive different versions of one frame, or
// one of them nothing at all.
async function broadcast(event) {
  if (!event || !event.kind) return

  if (clients.admin.size) {
    const frame = `data: ${JSON.stringify(event)}\n\n`
    for (const client of [...clients.admin]) writeTo(clients.admin, client, frame)
  }

  if (!clients.public.size) return

  let config
  try {
    config = await visibility.getConfig()
  } catch (err) {
    // Fail closed: without a config we cannot prove a frame is safe to send.
    log.error('visibility config unavailable; withholding public frame', err)
    return
  }

  // Most frames land on one rung set, so cache the serialised payload per level
  // instead of re-projecting and re-stringifying for every subscriber.
  const byLevel = new Map()
  for (const client of [...clients.public]) {
    let frame = byLevel.get(client.level)
    if (frame === undefined) {
      frame = visibility.kindVisibleTo(event.kind, client.level, config)
        ? `data: ${JSON.stringify(visibility.projectFeature(visibility.KIND_FEATURE.get(event.kind), event, client.level, config))}\n\n`
        : null
      byLevel.set(client.level, frame)
    }
    if (frame) writeTo(clients.public, client, frame)
  }
}

// Close every open stream (graceful shutdown). Clears each keepalive timer too —
// without that the intervals keep the event loop alive after the streams are
// gone, and the process won't exit.
function closeAll() {
  for (const bucket of Object.values(clients)) {
    for (const client of [...bucket]) {
      drop(bucket, client)
      try {
        client.res.end()
      } catch {
        /* ignore */
      }
    }
  }
}

function stats() {
  return { publicClients: clients.public.size, adminClients: clients.admin.size }
}

module.exports = { subscribe, broadcast, closeAll, stats, PUBLIC_KINDS }