// inbound lens — the composition: roster binding, cold-start backfill, reconnect gap-fill, stats.
// Run: node test/inbound.test.mjs
//
// Every part is tested on its own elsewhere; this covers only what composition adds. The assertion that
// matters most is that a backfilled pointer and a live pointer land in the SAME queue — if they ever
// diverge, the drain quietly grows two paths with two sets of semantics.
import * as dagCbor from '@ipld/dag-cbor'
import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
import { createRepo } from 'atproto-repo/lib/repo.js'
import { keyFromPrivate } from 'atproto-identity/lib/atproto-identity.js'
import { createInboundLens, DEFAULT_SOURCES } from '../lib/inbound.js'
import { createMemoryIndex } from '../lib/retention.js'
import { SOURCES, CONTENT_BEARING } from '../lib/spacedust.js'
let pass = 0
const ok = (c, m) => { if (!c) { console.error('FAIL:', m); process.exit(1) } console.log(' ok -', m); pass++ }
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
const AUTHOR = keyFromPrivate(Buffer.alloc(32, 3))
const AUTHOR_DID = 'did:plc:author0000000000000000'
const OURS = 'did:plc:wj62000000000000000000000'
const PDS = 'https://pds.author.test'
const repo = await createRepo({
did: AUTHOR_DID,
rev: '3lccc',
privateKey: AUTHOR.privateKey,
records: [{
collection: 'app.bsky.feed.post',
rkey: 'reply1',
record: { $type: 'app.bsky.feed.post', text: 'a reply to you', createdAt: '2026-07-30T00:00:00Z' }
}]
})
function fakeSocket () {
const h = {}
const s = {
written: [],
on (ev, cb) { (h[ev] = h[ev] || []).push(cb); return s },
emit (ev, a) { for (const cb of h[ev] || []) cb(a) },
write (d) { s.written.push(d) },
end () {}
}
return s
}
// doFetch serves: Constellation pages, the PLC doc, and the CAR.
function makeFetch ({ backlinkRows = [] } = {}) {
const calls = { constellation: 0, plc: 0, car: 0 }
return {
calls,
async doFetch (url) {
if (url.includes('blue.microcosm.links.getBacklinks')) {
calls.constellation++
return { ok: true, json: async () => ({ total: backlinkRows.length, records: backlinkRows }) }
}
if (url.includes('plc.directory') || url.includes('/did:plc:')) {
calls.plc++
return {
ok: true,
json: async () => ({
did: AUTHOR_DID,
verificationMethods: { atproto: AUTHOR.didKey },
services: { atproto_pds: { endpoint: PDS } }
})
}
}
if (url.includes('com.atproto.sync.getRecord')) {
calls.car++
const b = repo.carBytes
return { ok: true, arrayBuffer: async () => b.buffer.slice(b.byteOffset, b.byteOffset + b.byteLength) }
}
return { ok: false }
}
}
}
function fakeStore () {
const files = new Map()
return {
puts: 0,
async put (c, r, v) { this.puts++; files.set(`${c}/${r}`, v) },
// Models OverlayRepoStore.getRaw — the no-lossy-overwrite guard reads it before writing.
async getRaw (c, r) { const v = files.get(`${c}/${r}`); return v ? { bytes: dagCbor.encode(v), record: v } : null },
async del (c, r) { return files.delete(`${c}/${r}`) },
get count () { return files.size }
}
}
// ── sources: content-bearing only ────────────────────────────────────────────────────────────────
ok(DEFAULT_SOURCES.every((s) => CONTENT_BEARING.has(s)),
'the default subscription covers ONLY content-bearing sources')
ok(!DEFAULT_SOURCES.includes(SOURCES.LIKE),
'likes are NOT subscribed — the drain would skip them anyway, so subscribing would spend bandwidth to deliver discards')
ok(DEFAULT_SOURCES.includes(SOURCES.REPLY_PARENT) && DEFAULT_SOURCES.includes(SOURCES.QUOTE),
'replies and quotes are')
// ── construction guards ──────────────────────────────────────────────────────────────────────────
{
const base = { openStore: async () => fakeStore(), doFetch: async () => ({ ok: false }), createSocket: () => fakeSocket(), index: createMemoryIndex() }
for (const missing of ['openStore', 'doFetch', 'createSocket', 'index']) {
let threw = false
const o = { ...base }; delete o[missing]
try { createInboundLens(o) } catch { threw = true }
ok(threw, `${missing} is required — the daemon must not half-construct this`)
}
}
// ── cold start: backfill runs, and its pointers reach the drain ──────────────────────────────────
{
const f = makeFetch({ backlinkRows: [{ did: AUTHOR_DID, collection: 'app.bsky.feed.post', rkey: 'reply1' }] })
const store = fakeStore()
const events = []
const lens = createInboundLens({
roster: [OURS],
openStore: async () => store,
appview: { async applyCommit () { return 1 } },
index: createMemoryIndex(),
doFetch: f.doFetch,
createSocket: () => fakeSocket(),
listSubjectUris: async () => [`at://${OURS}/app.bsky.feed.post/mypost`],
plc: { getData: async () => ({ did: AUTHOR_DID, verificationMethods: { atproto: AUTHOR.didKey }, services: { atproto_pds: { endpoint: PDS } } }) },
onEvent: (e, d) => events.push([e, d])
})
await lens.start()
ok(f.calls.constellation > 0, 'a cold start BACKFILLS — everything that happened while we were down is a gap by definition')
ok(events.some(([e]) => e === 'backfill'), 'and reports it')
await sleep(600) // drain idles 250ms between empty polls — must outlast one full cycle
await lens.stop()
ok(store.puts === 1, 'a backfilled pointer is fetched, VERIFIED against the real signature, and retained')
ok(lens.stats.drain.retained === 1, 'and the drain accounts for it')
ok(lens.stats.retention.retained === 1, 'as does retention')
}
// ── one queue: backfilled and live pointers share the bound and the drain ────────────────────────
{
const f = makeFetch({ backlinkRows: Array.from({ length: 4 }, (_, i) => ({ did: AUTHOR_DID, collection: 'app.bsky.feed.post', rkey: `rk${i}` })) })
const lens = createInboundLens({
roster: [OURS],
openStore: async () => fakeStore(),
index: createMemoryIndex(),
doFetch: f.doFetch,
createSocket: () => fakeSocket(),
listSubjectUris: async () => [`at://${OURS}/app.bsky.feed.post/mypost`],
plc: { getData: async () => null }
})
await lens.backfill()
ok(lens.stats.pending > 0, 'backfilled pointers land in the SAME bounded queue the socket feeds — one drain, one dedup surface')
await lens.stop()
}
// ── roster change updates the filter in place and backfills newcomers ────────────────────────────
{
let sock
const f = makeFetch({ backlinkRows: [] })
const events = []
const lens = createInboundLens({
roster: [OURS],
openStore: async () => fakeStore(),
index: createMemoryIndex(),
doFetch: f.doFetch,
createSocket: () => (sock = fakeSocket()),
listSubjectUris: async () => [`at://${OURS}/app.bsky.feed.post/mypost`],
plc: { getData: async () => null },
onEvent: (e, d) => events.push([e, d])
})
await lens.start()
const before = f.calls.constellation
const n = await lens.updateRoster([OURS, 'did:plc:newmember00000000000000'])
ok(n === 2, 'the roster grows')
ok(sock.written.length > 0, 'the filter is updated over the EXISTING socket — no reconnect, so a mutating roster stays live')
const sent = JSON.parse(sock.written[sock.written.length - 1])
ok(sent.payload.wantedSubjectDids.length === 2, 'and carries the complete new subject set')
ok(f.calls.constellation > before, 'the NEW member is backfilled — otherwise their history would be invisible until something linked to them again')
ok(events.some(([e]) => e === 'roster-backfill'), 'and that is reported')
await lens.updateRoster([OURS])
ok(lens.stats.subjects === 1, 'shrinking the roster works too')
await lens.stop()
}
// ── the DID-vs-at-uri asymmetry (the bug this fix exists for) ────────────────────────────────────
{
let queriedSubjects = []
const f = {
async doFetch (url) {
if (url.includes('getBacklinks')) {
queriedSubjects.push(new URL(url).searchParams.get('subject'))
return { ok: true, json: async () => ({ records: [] }) }
}
return { ok: false }
}
}
const lens = createInboundLens({
roster: [OURS],
openStore: async () => fakeStore(),
index: createMemoryIndex(),
doFetch: f.doFetch,
createSocket: () => fakeSocket(),
listSubjectUris: async () => [`at://${OURS}/app.bsky.feed.post/mypost`],
plc: { getData: async () => null }
})
await lens.backfill()
ok(queriedSubjects.length > 0, 'backfill queries Constellation')
ok(queriedSubjects.every((s) => s.startsWith('at://')),
'and every subject is an at-uri, NEVER a bare DID — Constellation matches subjects EXACTLY, so a DID subject returns zero forever and looks like a quiet roster')
ok(!queriedSubjects.includes(OURS), 'our DID is specifically not used as a subject')
await lens.stop()
}
{
// Without the provider we must SKIP, not fall back to DID queries — the fallback IS the bug.
let queried = 0
const events = []
const lens = createInboundLens({
roster: [OURS],
openStore: async () => fakeStore(),
index: createMemoryIndex(),
doFetch: async (url) => { if (url.includes('getBacklinks')) queried++; return { ok: false } },
createSocket: () => fakeSocket(),
plc: { getData: async () => null },
onEvent: (e, d) => events.push([e, d])
})
ok((await lens.backfill()) === 0 && queried === 0,
'with no listSubjectUris provider, backfill issues NO queries rather than useless DID ones')
ok(events.some(([e]) => e === 'backfill-skipped'), 'and says so loudly — a silent skip is the failure mode being fixed')
await lens.stop()
}
// ── cold start is a RACE: the appview is empty until P2P ingest lands ────────────────────────────
{
// listSubjectUris returns nothing at first (empty appview), then starts returning a uri — exactly the
// live ordering: `backfill uris:0` fired BEFORE `ingest: backfill complete`.
let calls = 0
let queried = 0
const events = []
const lens = createInboundLens({
roster: [OURS],
openStore: async () => fakeStore(),
index: createMemoryIndex(),
doFetch: async (url) => {
if (url.includes('getBacklinks')) { queried++; return { ok: true, json: async () => ({ records: [] }) } }
return { ok: false }
},
createSocket: () => fakeSocket(),
listSubjectUris: async () => (++calls < 3 ? [] : [`at://${OURS}/app.bsky.feed.post/mypost`]),
plc: { getData: async () => null },
backfillRetryMs: 5,
backfillRetries: 6,
onEvent: (e, d) => events.push([e, d])
})
await lens.start()
ok(queried === 0, 'the first sweep asks Constellation NOTHING — there are no own records indexed yet')
ok(events.some(([e]) => e === 'backfill-retry'), 'so it schedules a retry instead of giving up silently')
await sleep(80)
ok(queried > 0, 'once the appview has our posts, a retry actually sweeps — the cold-start race no longer loses forever')
ok(lens.stats.backfillUris > 0, 'and stats show it found subjects to ask about')
const retriesAfter = lens.stats.backfillRetriesLeft
await sleep(40)
ok(lens.stats.backfillRetriesLeft === retriesAfter,
'retrying STOPS once a sweep has subjects — it is a cold-start fix, not a permanent poll')
await lens.stop()
}
{
// The retry budget is bounded: a permanently empty roster must not poll forever.
let calls = 0
const lens = createInboundLens({
roster: [OURS],
openStore: async () => fakeStore(),
index: createMemoryIndex(),
doFetch: async () => ({ ok: false }),
createSocket: () => fakeSocket(),
listSubjectUris: async () => { calls++; return [] },
plc: { getData: async () => null },
backfillRetryMs: 2,
backfillRetries: 3
})
await lens.start()
await sleep(60)
ok(calls === 4, `bounded: 1 initial + 3 retries, then it stops (saw ${calls}) — a genuinely empty roster costs a handful of no-ops, not a forever poll`)
ok(lens.stats.backfillRetriesLeft === 0, 'and the budget is visibly exhausted rather than silently')
await lens.stop()
}
{
// A pending retry must not outlive stop().
let calls = 0
const lens = createInboundLens({
roster: [OURS],
openStore: async () => fakeStore(),
index: createMemoryIndex(),
doFetch: async () => ({ ok: false }),
createSocket: () => fakeSocket(),
listSubjectUris: async () => { calls++; return [] },
plc: { getData: async () => null },
backfillRetryMs: 15,
backfillRetries: 10
})
await lens.start()
await lens.stop()
const after = calls
await sleep(60)
ok(calls === after, 'stop() cancels the pending retry — a stopped lens must not keep calling back into a torn-down daemon')
}
// ── lifecycle ────────────────────────────────────────────────────────────────────────────────────
{
const f = makeFetch()
const empty = createInboundLens({
roster: [],
openStore: async () => fakeStore(),
index: createMemoryIndex(),
doFetch: f.doFetch,
createSocket: () => fakeSocket(),
plc: { getData: async () => null }
})
ok(empty.stats.running === false, 'not running before start')
await empty.start()
ok(empty.stats.running === true, 'running after start')
ok(f.calls.constellation === 0, 'an EMPTY roster backfills nothing — no subjects means no queries, not a full-network sweep')
ok(empty.stats.backfills === 0, 'and no sweep is counted, because none happened')
await empty.stop()
ok(empty.stats.running === false, 'stopped')
// Idempotency needs a real subject, or backfill short-circuits before the counter.
const lens = createInboundLens({
roster: [OURS],
openStore: async () => fakeStore(),
index: createMemoryIndex(),
doFetch: f.doFetch,
createSocket: () => fakeSocket(),
listSubjectUris: async () => [`at://${OURS}/app.bsky.feed.post/mypost`],
plc: { getData: async () => null }
})
await lens.start()
await lens.start()
ok(lens.stats.backfills === 1, 'start is idempotent — a double start does not double-backfill a live subscription')
await lens.stop()
ok(lens.stats.running === false, 'stopped')
}
// ── stats surface ────────────────────────────────────────────────────────────────────────────────
{
const lens = createInboundLens({
roster: [OURS], openStore: async () => fakeStore(), index: createMemoryIndex(),
doFetch: async () => ({ ok: false }), createSocket: () => fakeSocket(), plc: { getData: async () => null }
})
const s = lens.stats
ok('socket' in s && 'drain' in s && 'retention' in s,
'stats expose all three layers — drops and refusals are invisible unless the daemon can report them')
ok(s.sources === DEFAULT_SOURCES.length, 'and how many sources are subscribed')
}
console.log(`\nINBOUND OK: ${pass} assertions`)