// spacedust.js — live inbound link subscription. The transport for DESIGN §7.6's inbound arm.
//
// WHAT IT IS. Spacedust (microcosm.blue) watches the global firehose, extracts every LINK it sees, and
// re-emits them over a filtered WebSocket. We subscribe with our roster's DIDs as the subjects, so we
// receive exactly the interactions that touch our own content — replies, quotes, likes, follows — and
// nothing else. This is NOT the firehose: the filtering happens server-side, and the caps (10k subject
// DIDs, 1k sources) are far above what a community needs.
//
// WHAT IT IS NOT. It does not fetch, verify, or store anything. Spacedust deliberately ships only a
// LOCATOR — `{did, collection, rkey}` reconstructed from `source_record`, plus the subject it points at.
// Their own source says so ("include the record too? ... no, not for now"), and that is the right call:
// an unsigned record from a third-party index is not something we retain. This module emits pointers;
// §7.6's pipeline (resolveMiniDoc -> getRecord -> verify -> OverlayRepoStore) turns them into data.
//
// THE 21-SECOND BUFFER IS LOAD-BEARING — LEAVE IT ON. By default spacedust holds every link for 21s
// before emitting, so interactions that are undone almost immediately never reach us (~<1% of links are
// deleted within 21s). That is precisely what Tier B wants: we should never spend a fetch+verify+write
// on a reply that was deleted two seconds later. `instant: true` bypasses it and is for UIs that want
// noise, not for a retention pipeline.
//
// BACKPRESSURE IS THE REAL HAZARD, AND IT IS ARCHITECTURAL. Spacedust DROPS a subscriber that lags
// (`RecvError::Lagged` -> cancel). If we did fetch/verify/store work inside the socket handler we would
// block the Bare event loop and be disconnected — the same failure class as pgv3's `reconcileAll` and
// pgh-sidecar's `spawnSync`. So the handler does one thing: parse and enqueue. All work happens in a
// drain loop. If the queue fills we drop the OLDEST pointer and count it, because a dropped pointer is
// recoverable (Constellation can backfill the same link by subject+source) while a lag-disconnect loses
// the entire stream until reconnect. Trading a recoverable loss for an unrecoverable one is the whole
// point of the bound.
//
// PURE + INJECTED. `createSocket` and `now` are parameters, so the URL building, event parsing, filter
// framing and queue behaviour are all unit-testable with no network and no clock.
//
// NO RUNTIME-SPECIFIC IMPORTS HERE — deliberately. `bare-ws` pulls native addons (bare-dns) that cannot
// load under Node, so importing it at the top of this file would make the whole module Bare-only and the
// "pure + injected" claim above a fiction. The real socket factory lives in `spacedust-socket.js`;
// `createSocket` is therefore REQUIRED, and tests inject a fake.
export const DEFAULT_ENDPOINT = 'wss://spacedust.microcosm.blue/subscribe'
// They instrument by user-agent and ask consumers to identify themselves.
export const USER_AGENT = 'favela-web (+https://github.com/favela)'
// Link sources are `<collection NSID>:<record property path>` with NO leading dot — spacedust strips it
// (`link.path.strip_prefix('.')`). These are the inbound interactions Tier B/C care about.
//
// VERIFIED against live traffic 2026-07-31 via Constellation's link enumeration on a post with 426
// quotes: `.embed.record.uri` 376 + `.embed.record.record.uri` 50 = 426 exactly. BOTH quote paths are
// real — plain quote vs recordWithMedia — so subscribing to only the first silently missed ~12% of
// quotes. Reply paths confirmed the same way (.reply.parent.uri / .reply.root.uri).
export const SOURCES = {
REPLY_PARENT: 'app.bsky.feed.post:reply.parent.uri',
REPLY_ROOT: 'app.bsky.feed.post:reply.root.uri',
QUOTE: 'app.bsky.feed.post:embed.record.uri',
QUOTE_MEDIA: 'app.bsky.feed.post:embed.record.record.uri', // recordWithMedia — a DIFFERENT path
LIKE: 'app.bsky.feed.like:subject.uri',
REPOST: 'app.bsky.feed.repost:subject.uri',
FOLLOW: 'app.bsky.graph.follow:subject'
}
// Content-bearing sources (Tier B — worth a fetch+verify+retain) vs count-bearing (Tier C — reference
// only, answered on demand by Constellation's getBacklinksCount). §7.6's pressure valve lives here: a
// post with 50k likes must not become 50k retained records.
export const CONTENT_BEARING = new Set([
SOURCES.REPLY_PARENT, SOURCES.REPLY_ROOT, SOURCES.QUOTE, SOURCES.QUOTE_MEDIA
])
const DEFAULTS = {
endpoint: DEFAULT_ENDPOINT,
instant: false, // see the 21-second note above — do not flip this for a retention pipeline
maxQueue: 1000,
reconnectMinMs: 1000,
reconnectMaxMs: 60_000
}
// Build the subscribe URL. Params are camelCase and REPEAT rather than comma-separating:
// `wantedSubjectDids=a&wantedSubjectDids=b`. Server-side caps are 50k subjects / 100 prefixes /
// 10k subject DIDs / 1k sources; we do not enforce them here (the server rejects, loudly).
export function buildSubscribeUrl (filter = {}, { endpoint = DEFAULT_ENDPOINT, instant = false } = {}) {
const url = new URL(endpoint)
const add = (name, values) => {
for (const v of values || []) url.searchParams.append(name, v)
}
add('wantedSubjectDids', filter.subjectDids)
add('wantedSubjects', filter.subjects)
add('wantedSubjectPrefixes', filter.subjectPrefixes)
add('wantedSources', filter.sources)
if (instant) url.searchParams.set('instant', 'true')
return url.toString()
}
// The runtime filter-update frame. Sent as a TEXT frame over the SAME socket — no reconnect, which is
// what lets a mutating roster stay live. Note the shape asymmetry: the envelope is snake_case-tagged
// (serde `tag="type", content="payload"`) while the payload is camelCase. It REPLACES the whole query
// (`*self = opts`), so always send the complete filter, never a delta.
export function optionsUpdateFrame (filter = {}) {
return JSON.stringify({
type: 'options_update',
payload: {
wantedSubjectDids: filter.subjectDids || [],
wantedSubjects: filter.subjects || [],
wantedSubjectPrefixes: filter.subjectPrefixes || [],
wantedSources: filter.sources || []
}
})
}
// Split an at-uri into a record locator. Returns null on anything that is not `at://<did>/<nsid>/<rkey>`
// — a plain URL subject, or a bare DID, is legitimate and simply is not a record.
export function parseAtUri (uri) {
if (typeof uri !== 'string' || !uri.startsWith('at://')) return null
const parts = uri.slice(5).split('/')
if (parts.length !== 3) return null
const [did, collection, rkey] = parts
if (!did || !collection || !rkey) return null
return { did, collection, rkey }
}
// Parse one link event into a normalized POINTER. Returns null for anything unexpected rather than
// throwing — a malformed frame must never take down the subscription.
//
// Wire shape (spacedust/src/lib.rs ClientEvent/ClientLinkEvent):
// { kind:"link", origin:"live"|"replay"|"backfill",
// link:{ operation:"create"|"delete", source, source_record, source_rev, subject } }
export function parseLinkEvent (raw) {
let body
try {
body = typeof raw === 'string' ? JSON.parse(raw) : JSON.parse(raw.toString())
} catch {
return null // see the framing note in the class below
}
if (!body || body.kind !== 'link' || !body.link) return null
const l = body.link
if (typeof l.source !== 'string' || typeof l.subject !== 'string') return null
if (l.operation !== 'create' && l.operation !== 'delete') return null
const source = parseAtUri(l.source_record)
if (source === null) return null // the linking record must be addressable, or we cannot fetch it
return {
operation: l.operation,
source: l.source, // the `<nsid>:<path>` spec, e.g. app.bsky.feed.like:subject.uri
subject: l.subject, // what was linked TO (one of ours)
// The linking record — this is the thing to fetch+verify+retain if content-bearing.
did: source.did,
collection: source.collection,
rkey: source.rkey,
// Repo revision of the commit that produced this link. The backfilling guide's skip test: if this
// rev is <= the rev we already hold for this DID, we already have this state.
rev: typeof l.source_rev === 'string' ? l.source_rev : null,
contentBearing: CONTENT_BEARING.has(l.source),
origin: body.origin || null
}
}
// Exponential backoff with full jitter. Jitter matters: if the service restarts, every subscriber
// reconnecting on the same schedule is a thundering herd (DESIGN §7.5 makes the same point about ghost
// sweeps).
export function backoffDelay (attempt, { min, max }, random = Math.random) {
const ceiling = Math.min(max, min * 2 ** Math.max(0, attempt - 1))
return Math.floor(random() * ceiling)
}
// Does this filter narrow the stream BY SUBJECT? `sources` does not count: it defaults to every
// source, so a sources-only filter looks configured and narrows nothing. Without a subject axis the
// server sends the whole network (see buildSubscribeUrl — an empty array appends no param at all).
export function hasSubjectAxis (filter = {}) {
return (filter.subjectDids || []).length > 0 ||
(filter.subjects || []).length > 0 ||
(filter.subjectPrefixes || []).length > 0
}
export class SpacedustClient {
// FAIL CLOSED on an empty roster (hyper-social #47). 0.3.58 made the roster empty-by-default so an
// unconfigured node would not silently index someone else's timeline — but this path turned the same
// condition into a subscription to EVERYONE'S. Measured on ashatron: 98 retained posts from one
// unrelated DID, zero from the roster. Unconfigured has to mean inert, not indiscriminate.
//
// `allowUnfiltered: true` preserves the URL builder's documented contract for a caller that really
// does want the firehose — it just has to say so, rather than arrive there via an empty array.
constructor (filter = {}, opts = {}) {
this.opts = { ...DEFAULTS, ...opts }
this._allowUnfiltered = !!opts.allowUnfiltered
this.filter = {
subjectDids: filter.subjectDids || [],
subjects: filter.subjects || [],
subjectPrefixes: filter.subjectPrefixes || [],
sources: filter.sources || Object.values(SOURCES)
}
// REQUIRED — see the no-runtime-imports note above. Production wiring passes the factory from
// `spacedust-socket.js`; tests pass a fake.
if (typeof opts.createSocket !== 'function') {
throw new Error('SpacedustClient requires opts.createSocket (see lib/spacedust-socket.js)')
}
this._createSocket = opts.createSocket
this._now = opts.now || (() => Date.now())
this._socket = null
this._queue = []
this._closed = false
this._attempt = 0
this._reconnectTimer = null
// Observability — §7.6 budgets are meaningless if drops are silent.
this.stats = { received: 0, parsed: 0, dropped: 0, unparseable: 0, connects: 0, disconnects: 0, refusedUnfiltered: 0 }
}
// Pointers waiting to be drained. The caller owns the drain loop, so it can do fetch/verify/retain at
// whatever rate it can sustain WITHOUT ever blocking the socket.
get pending () {
return this._queue.length
}
start () {
if (this._closed) throw new Error('client is closed')
this._connect()
return this
}
_connect () {
// Refuse rather than widen. Guarded here, not in start(), so a RECONNECT cannot quietly restore an
// unfiltered subscription after the roster was emptied at runtime.
if (!hasSubjectAxis(this.filter) && !this._allowUnfiltered) {
this.stats.refusedUnfiltered++
console.error(JSON.stringify({
msg: 'spacedust: refusing to subscribe with no subject filter',
reason: 'an empty roster would subscribe to the ENTIRE network, not to nothing',
fix: 'set FAVELA_ROSTER (or FAVELA_LIST), or pass allowUnfiltered:true to mean it'
}))
return
}
const url = buildSubscribeUrl(this.filter, this.opts)
let socket
try {
socket = this._createSocket(url)
} catch {
return this._scheduleReconnect() // fail soft — a bad connect is a retry, not a crash
}
this._socket = socket
this.stats.connects++
// bare-ws auto-replies PONG with the echoed payload (lib/socket.js:163-165), which is exactly what
// spacedust's 30s keepalive checks (it verifies the 8 random bytes come back). Nothing to do here.
socket.on('data', (chunk) => this._onMessage(chunk))
socket.on('error', () => this._onDisconnect())
socket.on('close', () => this._onDisconnect())
socket.on('end', () => this._onDisconnect())
}
// THE HOT PATH. Parse and enqueue. Nothing else. Any await here risks a lag-disconnect.
//
// Framing note: bare-ws is a non-objectMode Duplex that pushes one Buffer per WebSocket message
// (lib/socket.js:195), and flowing-mode 'data' emits per push — so one chunk is one message. We do not
// rely on that silently: an unparseable chunk is counted and dropped rather than throwing, and a lost
// pointer is recoverable via Constellation backfill by (subject, source).
_onMessage (chunk) {
this.stats.received++
const pointer = parseLinkEvent(chunk)
if (pointer === null) {
this.stats.unparseable++
return
}
this.stats.parsed++
this.enqueue(pointer)
}
// Enqueue an already-parsed pointer. PUBLIC because the live socket is not the only source: the
// Constellation backfill pushes historical pointers through this same queue, so both share one bound,
// one drain and one dedup surface rather than growing a second path.
//
// Bounded: drop the OLDEST. Losing one recoverable pointer beats being disconnected and losing the
// stream — Constellation can re-offer a dropped pointer, nothing can re-offer a missed subscription.
enqueue (pointer) {
if (!pointer) return false
if (this._queue.length >= this.opts.maxQueue) {
this._queue.shift()
this.stats.dropped++
}
this._queue.push({ ...pointer, receivedAt: this._now() })
return true
}
// Drain up to `max` pointers. Returns them; the caller runs the §7.6 pipeline.
drain (max = Infinity) {
if (max >= this._queue.length) {
const all = this._queue
this._queue = []
return all
}
return this._queue.splice(0, max)
}
// Change the filter WITHOUT reconnecting. Send the complete filter — the server replaces wholesale.
// Falls back to a reconnect if the socket is not writable, so a roster change is never silently lost.
updateFilter (filter = {}) {
// The server REPLACES the whole query, so an empty update is the same unfiltered subscription
// reached by a different door. Refuse WITHOUT mutating this.filter: the previous (narrower) query
// stays in force, which is stale but bounded — never wider than what was already asked for.
const next = { ...this.filter, ...filter }
if (!hasSubjectAxis(next) && !this._allowUnfiltered) {
this.stats.refusedUnfiltered++
console.error(JSON.stringify({
msg: 'spacedust: refusing a filter update that would remove every subject',
reason: 'the server replaces the whole query — an empty one matches the ENTIRE network'
}))
return false
}
this.filter = next
if (this._socket === null) return false
try {
this._socket.write(optionsUpdateFrame(this.filter))
return true
} catch {
this._onDisconnect()
return false
}
}
_onDisconnect () {
if (this._socket === null) return // already handled; close/end/error can all fire
this._socket = null
this.stats.disconnects++
if (!this._closed) this._scheduleReconnect()
}
_scheduleReconnect () {
if (this._closed || this._reconnectTimer !== null) return
this._attempt++
const delay = backoffDelay(this._attempt, {
min: this.opts.reconnectMinMs,
max: this.opts.reconnectMaxMs
})
this._reconnectTimer = setTimeout(() => {
this._reconnectTimer = null
if (!this._closed) this._connect()
}, delay)
}
// NOTE for the caller: a reconnect leaves a GAP. Spacedust replays nothing — there is no cursor in the
// subscribe contract. Gap-filling is Constellation's job (getBacklinks by subject+source, paginated),
// which is why §7.6 pairs the live subscription with the historical index rather than treating either
// as sufficient alone.
close () {
this._closed = true
if (this._reconnectTimer !== null) {
clearTimeout(this._reconnectTimer)
this._reconnectTimer = null
}
const socket = this._socket
this._socket = null
if (socket !== null) {
try {
socket.end()
} catch {
// already gone
}
}
}
}