feat(server): port the UO models, utils and schema fragment

The data half of the extraction: 8 model directories, 13 utils, the shard
stream catalog and the 27-table schema fragment with its purge.

server/core.js is what makes the port a one-line import change per file rather
than a signature change per function. Ported code requires its dependencies at
file scope -- `const { query } = require('../../core')` -- which runs before
register() has been called and before any ctx exists. So every member is a
stable function that resolves ctx when CALLED, and nothing may be destructured
off ctx at init either, because core is free to hand over a getter.

Two helpers are vendored rather than taken from ctx, and the line between them
is the point. utils/excerpt.js is core's deriveExcerpt -- nine lines of pure
text handling. Core's sanitiser next to it was NOT copied: a second copy of a
security control diverges silently the moment either is fixed. announceLinks.js
vendors legError and articleUrl the same way, but baseUrl could not be: core's
reads APP_BASE_URL, and §2.7 forbids a module reading core's environment, so it
comes off ctx.site.baseUrl.

The schema fragment is core's 27 shard_*/uo_link_* statements, verbs CREATE,
ALTER and UPDATE only, every CREATE TABLE guarded. Two of its tables carry a
foreign key INTO users, which is allowed and is why the replay order matters --
core's schema is in place before this runs. The reverse never occurs and must
not: it would make core unable to boot without a module installed.

One real port bug caught by the integration run, not by tests: the atlas art
map resolved `../../../db/data`, which pointed at core's tree when this file
lived there and points outside server/ now. A path that happens to resolve is
exactly what survives a green suite, because the absent-file branch returns {}
and looks like the normal case.

Co-Authored-By: Claude <noreply@anthropic.com>

server/utils/shardBroadcast.js

@@ -0,0 +1,166 @@
+// ── 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 }