// inbound.js — DESIGN §7.6's inbound arm, assembled. One controller the daemon starts and stops.
//
// This is only composition: SpacedustClient (live edge) + Constellation (historical floor) + LinkDrain
// (tier routing) + retention hooks (fetch/verify/retain/evict). Every part is tested on its own; what
// this adds is the wiring, the roster binding, and the reconnect gap-fill.
//
// DEFAULT OFF. The daemon constructs this only when explicitly configured. An always-on subscription to
// a third-party service is not something a node should inherit by upgrading.
//
// THE ROSTER IS THE FILTER. `wantedSubjectDids` = the DIDs we serve, so spacedust sends only links that
// touch our own content — §7.6's "one hop from our records", enforced server-side rather than by
// discarding a firehose locally. Caps are 10k DIDs / 1k sources, far above a community.
//
// CONTENT-BEARING SOURCES ONLY. We deliberately do NOT subscribe to likes/reposts/follows. The drain
// would skip them anyway (Tier C), so subscribing would spend bandwidth to deliver events we discard.
// Counts come from Constellation on demand — that is the whole point of the tiering.
//
// RECONNECT MEANS BACKFILL. /subscribe has no cursor and replays nothing, so every reconnect leaves a
// gap. On each (re)connect we walk Constellation for OUR OWN RECORD at-uris (not our DIDs — see the
// asymmetry note on backfill()) and push those pointers into the same drain. Pointer shapes are
// identical, so they dedup against whatever the live stream already handled.
import { SpacedustClient, SOURCES, CONTENT_BEARING } from './spacedust.js'
import { LinkDrain } from './link-drain.js'
import { createConstellationClient } from './constellation.js'
import { createFetchRecord, createVerifyRecord, createIdentityResolver } from './record-fetch.js'
import { createRetentionHooks } from './retention.js'
// Only the sources that carry text. Derived from CONTENT_BEARING so the two can never drift.
export const DEFAULT_SOURCES = Object.values(SOURCES).filter((s) => CONTENT_BEARING.has(s))
export function createInboundLens ({
roster = [], // DIDs we serve — the subject filter
openStore, // async (did) -> OverlayRepoStore (the daemon's getRecordStore)
appview = null, // the Appview, for the index fold
index, // retention index (bee-backed in production)
doFetch, // bare-fetch
createSocket, // from spacedust-socket.js
listSubjectUris = null, // async (rosterDids) -> [at-uri] OUR OWN records to ask Constellation about.
// Required for backfill: Constellation matches subjects EXACTLY, so DIDs
// match nothing. See the note on backfill() below.
plc = null, // optional PlcDirectory override (tests)
endpoint = undefined,
constellationEndpoint = undefined,
budgetBytes = null,
sources = DEFAULT_SOURCES,
backfillPages = 2,
backfillRetryMs = 5000, // cold start: the appview is EMPTY until the P2P ingest lands (see start())
backfillRetries = 6,
onEvent = null
} = {}) {
if (typeof openStore !== 'function') throw new Error('createInboundLens requires openStore')
if (typeof doFetch !== 'function') throw new Error('createInboundLens requires doFetch')
if (typeof createSocket !== 'function') throw new Error('createInboundLens requires createSocket')
if (!index) throw new Error('createInboundLens requires an index')
const emit = (ev, detail) => { if (onEvent) onEvent(ev, detail) }
const resolveIdentity = createIdentityResolver({ fetchImpl: doFetch, plc })
const retention = createRetentionHooks({ openStore, appview, index, budgetBytes, onEvent })
const constellation = createConstellationClient({
doFetch,
...(constellationEndpoint ? { endpoint: constellationEndpoint } : {})
})
let subjects = roster.slice()
const client = new SpacedustClient(
{ subjectDids: subjects, sources },
{ createSocket, ...(endpoint ? { endpoint } : {}) }
)
const drain = new LinkDrain(client, {
fetchRecord: createFetchRecord({ doFetch, resolveIdentity }),
verifyRecord: createVerifyRecord({ onFailure: (reason, p) => emit('verify-failed', { reason, pointer: p }) }),
retainRecord: retention.retainRecord,
evictRecord: retention.evictRecord,
knownRev: retention.knownRev
})
let backfills = 0
let running = false
let lastUris = 0 // how many of OUR at-uris the last sweep had to ask about
let retryTimer = null
let retriesLeft = backfillRetries
// Push Constellation's historical pointers into the drain's queue. Uses the client's own enqueue path
// so backfilled and live pointers share one bounded queue and one dedup surface.
//
// ⛔ SUBJECTS HERE ARE at-uris, NOT DIDs — the two services are NOT symmetric, measured live 2026-07-30.
// Spacedust filters on `subject_did`, which its source documents as matching "both bare-DIDs and DIDs
// extracted from at-uris", so the LIVE filter is correctly keyed by roster DID. Constellation's
// `subject` is an EXACT target with no such extraction: a reply's reply.parent.uri is always
// `at://did/collection/rkey` and never a bare DID, so querying subject=<our DID> against a
// content-bearing source matches NOTHING, always. That returned 0 and looked exactly like a quiet
// roster — the silent-zero failure this design keeps warning about.
//
// So backfill needs OUR OWN RECORD at-uris, which only the caller can enumerate (they come from the
// appview). With no provider we do NOT fall back to DID queries — that is the bug, not a degraded
// mode — we skip and say so.
async function backfill () {
if (subjects.length === 0) return 0
if (typeof listSubjectUris !== 'function') {
emit('backfill-skipped', { reason: 'no listSubjectUris provider — DID subjects match nothing in Constellation' })
return 0
}
lastUris = 0
let uris = []
try {
uris = (await listSubjectUris(subjects)) || []
} catch (err) {
emit('backfill-skipped', { reason: 'listSubjectUris failed', error: (err && err.message) || String(err) })
return 0
}
lastUris = uris.length
if (uris.length === 0) {
emit('backfill', { offered: 0, duplicates: 0, uris: 0, note: 'no own records to ask about yet' })
return 0
}
const { offered, duplicates } = await constellation.backfillInto(
(pointer) => client.enqueue(pointer),
{ subjects: uris, sources, maxPages: backfillPages, onTruncate: (t) => emit('backfill-truncated', t) }
)
backfills++
// `duplicates` is the cross-source collapse (one reply links via parent AND root). Reported because
// a silently-deduped sweep and an empty sweep look identical from the outside.
emit('backfill', { offered, duplicates, uris: uris.length })
return offered
}
// COLD START IS A RACE, and losing it is silent. `inbound.start()` runs as soon as the daemon has an
// Appview, but the P2P ingest that POPULATES that appview finishes later — observed live: the sweep
// logged `uris: 0` and only then did `ingest: backfill complete` arrive. A one-shot backfill therefore
// asks about nothing at boot and never asks again, which reads exactly like "nobody has replied to us".
//
// Retrying rather than hooking the ingest lifecycle keeps this decoupled and also covers the honest
// second case: a roster DID that genuinely has no posts yet. Bounded, so a permanently-empty roster
// costs a handful of no-op calls, not a poll forever.
async function backfillWithRetry () {
const offered = await backfill()
if (running && lastUris === 0 && retriesLeft > 0) {
retriesLeft--
emit('backfill-retry', { retriesLeft, delayMs: backfillRetryMs, reason: 'no own records indexed yet' })
retryTimer = setTimeout(() => {
retryTimer = null
backfillWithRetry().catch(() => {}) // fail soft — a retry must never take the daemon down
}, backfillRetryMs)
}
return offered
}
return {
get stats () {
return {
running,
subjects: subjects.length,
sources: sources.length,
backfills,
backfillUris: lastUris,
backfillRetriesLeft: retriesLeft,
socket: client.stats,
drain: drain.stats,
retention: retention.stats,
pending: client.pending
}
},
async start () {
if (running) return this
running = true
client.start()
drain.start()
// Cold start is a gap by definition — everything that happened while we were down. Retries
// because the appview is not populated yet at this point (see backfillWithRetry).
await backfillWithRetry()
emit('started', { subjects: subjects.length, sources })
return this
},
// Roster changed: update the server-side filter WITHOUT reconnecting, then backfill the newcomers.
async updateRoster (dids) {
const next = (dids || []).filter(Boolean)
const added = next.filter((d) => !subjects.includes(d))
subjects = next
client.updateFilter({ subjectDids: subjects })
// A new member's HISTORY needs their own record at-uris, for the same reason cold start does —
// `added` is a list of DIDs and would match nothing. Re-running the provider is the honest way to
// pick them up; it returns the union for the current roster.
if (added.length > 0) {
retriesLeft = backfillRetries // a new member's posts may not be indexed yet either
const offered = await backfillWithRetry()
emit('roster-backfill', { added: added.length, offered })
}
emit('roster-updated', { subjects: subjects.length })
return subjects.length
},
backfill,
async stop () {
running = false
if (retryTimer !== null) { clearTimeout(retryTimer); retryTimer = null }
client.close()
await drain.stop()
emit('stopped', {})
}
}
}