import test from 'brittle'
import { Publisher, noteFingerprint } from '../lib/publisher.js'
import { MemoryState } from '../lib/state.js'
import { MemoryFollowerStore } from '../lib/followers.js'
const ACTOR = 'https://ap.7rnx.net/ap/actor'
const note = (rkey, published) => ({
id: `https://ap.7rnx.net/ap/objects/${rkey}`,
type: 'Note', content: `<p>${rkey}</p>`, published,
to: ['https://www.w3.org/ns/activitystreams#Public'],
cc: ['https://ap.7rnx.net/ap/followers']
})
// listNotes is newest-first, and rkeys are TIDs so lexicographic order IS chronological.
function fakeRecords (notes) {
return { async listNotes ({ limit = 50 } = {}) { return { notes: notes.slice(0, limit), cursor: null } } }
}
function make (notes, { seen = [], primed = false, followers = ['https://prsm.space/users/fleeky'] } = {}) {
const delivered = []
const p = new Publisher({
records: fakeRecords(notes),
followers: new MemoryFollowerStore(followers),
deliver: async (d) => { delivered.push(d) },
actorId: ACTOR,
newId: (() => { let n = 0; return () => `id${n++}` })(),
log: () => {},
state: new MemoryState(seen, primed)
})
return { p, delivered }
}
const settle = async (p) => { p.onHead(); await new Promise((r) => setTimeout(r, 20)) }
// ★ Publishing three weeks of history into every follower's timeline at once is rude and is not
// what any other server does.
test('★ priming does NOT deliver existing posts', async (t) => {
const { p, delivered } = make([note('3b', 'x'), note('3a', 'y')])
await p.primeWatermark()
t.ok((await p.state.getDelivered()).has('3b'), 'existing posts are marked delivered')
t.ok((await p.state.getDelivered()).has('3a'))
await settle(p)
t.is(delivered.length, 0)
})
test('★ a new post is delivered to every follower', async (t) => {
const { p, delivered } = make([note('3c', 'z'), note('3b', 'x')], {
seen: [['3b', noteFingerprint(note('3b', 'x'))]], primed: true, followers: ['https://a.net/u/1', 'https://b.net/u/2']
})
await settle(p)
t.is(delivered.length, 2, 'one delivery per follower')
t.is(delivered[0].activity.type, 'Create')
t.is(delivered[0].activity.actor, ACTOR)
t.is(delivered[0].activity.object.id, 'https://ap.7rnx.net/ap/objects/3c')
t.alike(delivered[0].activity.to, delivered[0].activity.object.to, 'wrapper and object agree on audience')
t.ok((await p.state.getDelivered()).has('3c'))
})
test('nothing new → nothing delivered, and it is idempotent', async (t) => {
const { p, delivered } = make([note('3b', 'x')], { seen: [['3b', noteFingerprint(note('3b', 'x'))]], primed: true })
await settle(p)
await settle(p)
t.is(delivered.length, 0)
})
test('multiple new posts deliver oldest-first so a timeline reads in order', async (t) => {
const { p, delivered } = make([note('3d', 'z'), note('3c', 'y'), note('3b', 'x')], { seen: [['3b', noteFingerprint(note('3b', 'x'))]], primed: true })
await settle(p)
t.is(delivered.length, 2)
t.is(delivered[0].activity.object.id.endsWith('/3c'), true)
t.is(delivered[1].activity.object.id.endsWith('/3d'), true)
})
// One unreachable server must not withhold the post from everyone else.
test('a failing follower does not block the others, and the watermark still advances', async (t) => {
const { p } = make([note('3c', 'z')], { seen: [['3b', noteFingerprint(note('3b', 'x'))]], primed: true })
const seen = []
p.deliver = async ({ to }) => { if (to.includes('bad')) throw new Error('502'); seen.push(to) }
p.followers = new MemoryFollowerStore(['https://bad.net/u/1', 'https://good.net/u/2'])
await settle(p)
t.alike(seen, ['https://good.net/u/2'])
t.ok((await p.state.getDelivered()).has('3c'))
})
// A head can advance while a drain is still running; two concurrent passes would double-deliver.
test('overlapping onHead calls are serialised', async (t) => {
const { p, delivered } = make([note('3c', 'z')], { seen: [['3b', noteFingerprint(note('3b', 'x'))]], primed: true })
p.onHead(); p.onHead(); p.onHead()
await new Promise((r) => setTimeout(r, 40))
t.is(delivered.length, 1)
})
// ★ REGRESSION. wj62's repo contains `e2e-overlay-1784495258`, a hand-written rkey that sorts
// ABOVE every TID (`e` > `3`) while being the OLDEST record. A high-watermark design made it the
// watermark and would have silently skipped every genuinely new post forever. Ordering must not
// matter.
test('★ a non-TID rkey that sorts above every TID does not block new posts', async (t) => {
const weird = note('e2e-overlay-1784495258', '2026-07-19T00:00:00Z')
const older = note('3mrxloimxoc22', '2026-07-31T18:00:00Z')
const fresh = note('3mrzzzzzzzz22', '2026-08-01T02:00:00Z')
const { p, delivered } = make([weird, older], { seen: [], primed: false })
await p.primeWatermark()
await settle(p)
t.is(delivered.length, 0, 'nothing existing is delivered')
p.records = { async listNotes () { return { notes: [weird, older, fresh], cursor: null } } }
await settle(p)
t.is(delivered.length, 1, 'the new post IS delivered despite sorting below the odd rkey')
t.ok(delivered[0].activity.object.id.endsWith('/3mrzzzzzzzz22'))
})
// AP_BACKFILL — an operation, not a mode.
test('★ backfill leaves the N most recent posts undelivered so they go out once', async (t) => {
const notes = [note('3e', '5'), note('3d', '4'), note('3c', '3'), note('3b', '2'), note('3a', '1')]
const { p, delivered } = make(notes, { seen: [], primed: false })
p.backfill = 2
await p.primeWatermark()
await settle(p)
t.is(delivered.length, 2, 'the 2 most recent are delivered')
t.is(delivered[0].activity.object.id.endsWith('/3d'), true, 'oldest of the backfill first')
t.is(delivered[1].activity.object.id.endsWith('/3e'), true)
t.ok((await p.state.getDelivered()).has('3a'), 'older history stays undelivered')
})
// The dangerous version of this feature is one that re-fires. A stranger following later must not
// receive the whole history because the variable was left set.
test('★ backfill self-disarms — a restart with it still set delivers nothing', async (t) => {
const notes = [note('3c', '3'), note('3b', '2'), note('3a', '1')]
const { p, delivered } = make(notes, { seen: [], primed: false })
p.backfill = 1
await p.primeWatermark()
await settle(p)
t.is(delivered.length, 1)
// simulate a restart against the SAME state, variable still set
const { p: p2, delivered: d2 } = make(notes, { seen: [...(await p.state.getDelivered())], primed: true })
p2.backfill = 1
await p2.primeWatermark()
await settle(p2)
t.is(d2.length, 0, 'nothing re-delivered')
})
test('backfill=0 behaves exactly as before', async (t) => {
const { p, delivered } = make([note('3b', '2'), note('3a', '1')], { seen: [], primed: false })
await p.primeWatermark()
await settle(p)
t.is(delivered.length, 0)
})
// Delivery is otherwise triggered only by onHead, so a backfill that does not start itself would
// sit undelivered until an unrelated future post advanced the head.
test('★ backfill is self-starting — no onHead required', async (t) => {
const { p, delivered } = make([note('3c', '3'), note('3b', '2'), note('3a', '1')], { seen: [], primed: false })
p.backfill = 2
await p.primeWatermark()
await new Promise((r) => setTimeout(r, 30)) // no settle() — deliberately never call onHead
t.is(delivered.length, 2, 'delivered without any head advance')
})
// ★ Mastodon caches actor documents and will not re-fetch one just because it changed. "The
// profile is correct at /ap/actor" and "the profile is correct on Mastodon" are different claims.
const ACTOR_DOC = { id: ACTOR, type: 'Person', name: 'tabertron', icon: { url: 'https://h/ap/blobs/a' } }
test('★ a changed profile sends Update{Person} to followers', async (t) => {
const { p, delivered } = make([], { seen: [], primed: true })
p.buildActorDoc = async () => ACTOR_DOC
await p.refreshProfile()
t.is(delivered.length, 1)
t.is(delivered[0].activity.type, 'Update')
t.is(delivered[0].activity.object.name, 'tabertron')
t.is(delivered[0].activity.actor, ACTOR)
})
// A restart must not spam every follower with a no-op Update.
test('an unchanged profile sends nothing', async (t) => {
const { p, delivered } = make([], { seen: [], primed: true })
p.buildActorDoc = async () => ACTOR_DOC
await p.refreshProfile()
await p.refreshProfile()
t.is(delivered.length, 1)
p.buildActorDoc = async () => ({ ...ACTOR_DOC, name: 'renamed' })
await p.refreshProfile()
t.is(delivered.length, 2, 'a real change does fire')
})
test('profile refresh survives a failing actor build', async (t) => {
const { p, delivered } = make([], { seen: [], primed: true })
p.buildActorDoc = async () => { throw new Error('peer down') }
await p.refreshProfile()
t.is(delivered.length, 0, 'no throw, no delivery')
})
// ★ Mastodon models an edit as Update{Note}, so an improved projection can reach posts that were
// already delivered. Without this, every projection improvement (images today, facets and quotes
// later) would only ever apply to FUTURE posts.
test('★ a changed projection sends Update, not a duplicate Create', async (t) => {
const before = note('3c', 'z')
const { p, delivered } = make([before], { seen: [['3c', noteFingerprint(before)]], primed: true })
await settle(p)
t.is(delivered.length, 0, 'unchanged → nothing')
// the projection gains an image attachment
const after = { ...before, attachment: [{ type: 'Document', url: 'https://h/ap/blobs/x' }] }
p.records = { async listNotes () { return { notes: [after], cursor: null } } }
await settle(p)
t.is(delivered.length, 1)
t.is(delivered[0].activity.type, 'Update')
t.ok(delivered[0].activity.object.attachment)
await settle(p)
t.is(delivered.length, 1, 'and it settles — no repeat once recorded')
})
// The v1 state stored bare rkeys with no fingerprint. Those must be brought up to the current
// projection ONCE, not treated as never-sent (which would duplicate) nor as unchanged (which would
// strand them).
test('★ legacy state with no fingerprints sends one Update per post, then settles', async (t) => {
const n = note('3c', 'z')
const { p, delivered } = make([n], { seen: ['3c'], primed: true }) // bare rkey = v1 shape
await settle(p)
t.is(delivered.length, 1)
t.is(delivered[0].activity.type, 'Update', 'migrated, not re-Created')
await settle(p)
t.is(delivered.length, 1, 'settles after one pass')
})
// ── interactions ────────────────────────────────────────────────────────────────────────────────
//
// The wiring, not the projection. as2-interactions.test.js proves a Like is shaped correctly; these
// prove the publisher actually SENDS it, marks it, and does not send it twice — the part that was
// green-but-unexercised the moment the delivery path was added.
const like = (rkey) => ({
id: `https://ap.7rnx.net/ap/activities/like-${rkey}`,
type: 'Like', actor: ACTOR, object: 'https://ap.7rnx.net/ap/objects/3post'
})
const announce = (rkey) => ({
id: `https://ap.7rnx.net/ap/activities/announce-${rkey}`,
type: 'Announce', actor: ACTOR, object: 'https://ap.7rnx.net/ap/objects/3post',
to: ['https://www.w3.org/ns/activitystreams#Public'], cc: ['https://ap.7rnx.net/ap/followers']
})
function makeWithInteractions (interactions, { unaddressable = 0, seen = [], logs = null } = {}) {
const delivered = []
const p = new Publisher({
records: {
async listNotes () { return { notes: [], cursor: null } },
async listInteractions () { return { interactions, unaddressable } }
},
followers: new MemoryFollowerStore(['https://a.net/u/1', 'https://b.net/u/2']),
deliver: async (d) => { delivered.push(d) },
actorId: ACTOR,
newId: (() => { let n = 0; return () => `id${n++}` })(),
log: logs ? (l) => logs.push(l) : () => {},
state: new MemoryState(seen, true)
})
return { p, delivered }
}
test('★ a like is delivered to every follower', async (t) => {
const { p, delivered } = makeWithInteractions([{ kind: 'like', rkey: '3l', key: 'like:3l', activity: like('3l') }])
await settle(p)
t.is(delivered.length, 2, 'both followers')
t.is(delivered[0].activity.type, 'Like')
t.is(delivered[0].activity.object, 'https://ap.7rnx.net/ap/objects/3post')
})
test('★ a boost keeps its public addressing all the way to the wire', async (t) => {
// A boost delivered without to:Public is accepted, returns 2xx, and appears in nobody's timeline —
// the same silent success as an Update with no `updated`. So the addressing is asserted at the
// point it actually leaves, not only where it is built.
const { p, delivered } = makeWithInteractions([{ kind: 'announce', rkey: '3r', key: 'announce:3r', activity: announce('3r') }])
await settle(p)
t.alike(delivered[0].activity.to, ['https://www.w3.org/ns/activitystreams#Public'])
t.alike(delivered[0].activity.cc, ['https://ap.7rnx.net/ap/followers'])
})
test('★ an already-delivered interaction is NOT re-sent', async (t) => {
const { p, delivered } = makeWithInteractions(
[{ kind: 'like', rkey: '3l', key: 'like:3l', activity: like('3l') }],
{ seen: [['like:3l', like('3l').id]] }
)
await settle(p)
t.is(delivered.length, 0)
})
test('★ like and post rkeys do not collide in the delivered map', async (t) => {
// TIDs are unique per COLLECTION, not per repo, so a like and a post really can share an rkey. An
// unnamespaced key would make delivering one silently suppress the other.
const { p, delivered } = makeWithInteractions(
[{ kind: 'like', rkey: '3b', key: 'like:3b', activity: like('3b') }],
{ seen: [['3b', 'some-post-fingerprint']] }
)
await settle(p)
t.is(delivered.length, 2, 'the like is still sent despite a post with the same rkey being delivered')
})
test('★ interactions we cannot address are COUNTED, not silently dropped', async (t) => {
// "no likes" and "twelve likes of bsky posts we cannot name" are different states, and only the
// second means the feature is limited rather than unused.
const logs = []
const { p, delivered } = makeWithInteractions([], { unaddressable: 12, logs })
await settle(p)
t.is(delivered.length, 0)
const line = logs.find((l) => /not addressable/.test(l.msg || ''))
t.ok(line, 'it says so')
t.is(line.count, 12)
})
test('a records source with no listInteractions does not break the drain', async (t) => {
// ap-bridge fronts an injected reader; an older or fake one without the method must degrade, not
// take posting down with it.
const delivered = []
const p = new Publisher({
records: { async listNotes () { return { notes: [note('3z', 'a')], cursor: null } } },
followers: new MemoryFollowerStore(['https://a.net/u/1']),
deliver: async (d) => { delivered.push(d) },
actorId: ACTOR,
newId: () => 'id0',
log: () => {},
state: new MemoryState([], true)
})
await settle(p)
t.is(delivered.length, 1, 'the post still went out')
})
// ⛔⛔ A PRIME AGAINST AN EMPTY LISTING IS NOT A PRIME.
//
// A held feed's log replicates asynchronously, so a prime taken at boot routinely sees ZERO notes
// on a peer with a full history. Recording that reads as "there was nothing to account for", and
// the flag then suppresses every later prime.
//
// Measured live 2026-09-08: ap-bridge primed ashatron at boot before its feed had arrived and wrote
// primed=true with an empty delivered-set. The 16 posts already served from its outbox were left
// neither delivered nor accounted for — and with AP_BACKFILL=20 configured, the backfill that was
// supposed to send them had already been consumed by a listing of nothing.
test('⛔⛔ priming against an empty listing records nothing and retries', async (t) => {
const state = new MemoryState()
let notes = []
const sent = []
const p = new Publisher({
records: { async listNotes () { return { notes, cursor: null } }, async listInteractions () { return { interactions: [] } } },
followers: { async list () { return ['https://elsewhere.net/users/bob'] } },
deliver: async (d) => { sent.push(d) },
actorId: ACTOR,
newId: () => 'n',
state,
log: () => {}
})
await p.primeWatermark()
t.absent(await state.isPrimed(), 'not recorded — "no notes" and "not replicated yet" are the same view from here')
// The log arrives.
notes = [note('a'), note('b')]
await p.primeWatermark()
t.ok(await state.isPrimed(), 'now it primes')
t.is((await state.getDelivered()).size, 2, 'against the history that actually exists')
})
test('★ a real prime is not repeated', async (t) => {
const state = new MemoryState()
const p = new Publisher({
records: { async listNotes () { return { notes: [note('a')], cursor: null } }, async listInteractions () { return { interactions: [] } } },
followers: { async list () { return [] } },
deliver: async () => {},
actorId: ACTOR,
newId: () => 'n',
state,
log: () => {}
})
await p.primeWatermark()
const first = [...(await state.getDelivered()).keys()]
await p.primeWatermark()
t.alike([...(await state.getDelivered()).keys()], first, 'a second call changes nothing')
})
// ⛔⛔ THE ONE THAT MEANT A REGISTERED USER COULD NEVER PUBLISH.
//
// rkeyOf matched `/\/ap\/objects\/([^/?#]+)$/`. The primary's object ids are
// https://host/ap/objects/<rkey> and a registered user's are
// https://host/ap/users/<name>/objects/<rkey> — so it returned null for every one of their notes.
//
// Both ends broke silently and in opposite directions. In primeWatermark the delivered-set came out
// empty however many posts existed; in _drain every note hit `if (!rkey) continue`, so `fresh` was
// always 0 and not one Create was ever sent. Measured live: notes 16, delivered 0, fresh 0 — with
// the outbox serving all 16 and a follower recorded.
//
// This is the same family as blobUrl, objectIdFor and keyId({host}): a helper written when there was
// one actor, silently returning the wrong thing for the second one rather than failing.
const perUserNote = (rkey, published) => ({
id: `https://ap.7rnx.net/ap/users/ashatron/objects/${rkey}`,
type: 'Note', content: `<p>${rkey}</p>`, published,
to: ['https://www.w3.org/ns/activitystreams#Public'],
cc: ['https://ap.7rnx.net/ap/users/ashatron/followers']
})
test('⛔⛔ a registered user\'s notes are addressable — Creates actually go out', async (t) => {
const state = new MemoryState()
const sent = []
const p = new Publisher({
records: {
async listNotes () { return { notes: [perUserNote('01k2b', '2'), perUserNote('01k2a', '1')], cursor: null } },
async listInteractions () { return { interactions: [] } }
},
followers: { async list () { return ['https://prsm.space/users/fleeky'] } },
deliver: async (d) => { sent.push(d) },
actorId: 'https://ap.7rnx.net/ap/users/ashatron/actor',
newId: () => 'n',
state,
log: () => {}
})
await p.primeWatermark()
t.is((await state.getDelivered()).size, 2, 'prime accounted for their history — an unreadable rkey made this 0 forever')
// Nothing to send after a prime with no backfill; a NEW post must go out.
p.records.listNotes = async () => ({ notes: [perUserNote('01k2c', '3'), perUserNote('01k2b', '2'), perUserNote('01k2a', '1')], cursor: null })
await p._drain()
t.is(sent.length, 1, 'exactly the new one')
t.is(sent[0].activity.type, 'Create')
t.is(sent[0].activity.object.id, 'https://ap.7rnx.net/ap/users/ashatron/objects/01k2c')
})
test('⛔ an id belonging to a DIFFERENT actor is not claimed as ours', async (t) => {
const state = new MemoryState()
const p = new Publisher({
records: {
// The primary's id, seen by ashatron's publisher. Matching it loosely on `/objects/<rkey>`
// would let two actors collide on one rkey and mark each other's posts delivered.
async listNotes () { return { notes: [{ id: 'https://ap.7rnx.net/ap/objects/01k2x', type: 'Note', published: '1' }], cursor: null } },
async listInteractions () { return { interactions: [] } }
},
followers: { async list () { return [] } },
deliver: async () => {},
actorId: 'https://ap.7rnx.net/ap/users/ashatron/actor',
newId: () => 'n',
state,
log: () => {}
})
await p.primeWatermark()
t.is((await state.getDelivered()).size, 0, 'not ours to account for')
})
// ⛔⛔ A DELIVERY THAT DID NOT HAPPEN MUST NOT BE RECORDED AS ONE.
//
// _send marked every post delivered whether or not there was anybody to send it to, so a post made
// while the account had no followers was burned: permanently in the delivered-set, never sent, and
// invisible the moment somebody did follow. Exactly the profile-fingerprint defect fixed in 0.9.71
// — "recorded as told when nobody was told" — one layer over, in the post path.
//
// Found while setting up a live test that this would have silently sabotaged: ashatron had 0
// followers at the time, so the marker post being sent to prove delivery works would have been
// consumed by the very act of proving it.
test('⛔⛔ with no followers, a post is NOT marked delivered', async (t) => {
const state = new MemoryState()
const sent = []
let followers = []
const p = new Publisher({
records: {
async listNotes () { return { notes: [note('01k2a', '1')], cursor: null } },
async listInteractions () { return { interactions: [] } }
},
followers: { async list () { return followers } },
deliver: async (d) => { sent.push(d) },
actorId: ACTOR,
newId: () => 'n',
state,
log: () => {}
})
await p._drain()
t.is(sent.length, 0, 'nothing sent — there was nobody')
t.is((await state.getDelivered()).size, 0, 'and nothing recorded as sent')
// Somebody follows. The post must still be deliverable.
followers = ['https://prsm.space/users/fleeky']
await p._drain()
t.is(sent.length, 1, 'it goes out now')
t.is((await state.getDelivered()).size, 1, 'and is recorded only once it actually went')
})
// ── deletes ──────────────────────────────────────────────────────────────────────────────────────
//
// ⛔ A Delete goes out ONLY for a post the source reports POSITIVELY deleted (a favela tombstone) and
// that we actually delivered. The header of _drainInteractions says why absence is not enough: a
// paged listing cannot prove a record is gone, and retracting something still there is worse than
// leaving something deleted.
function makeDel (deleted, { seen = [], followers = ['https://prsm.space/users/fleeky'] } = {}) {
const sent = []
const p = new Publisher({
records: { ...fakeRecords([]), async listDeleted () { return { deleted } } },
followers: new MemoryFollowerStore(followers),
deliver: async (d) => { sent.push(d) },
actorId: ACTOR,
newId: (() => { let n = 0; return () => `id${n++}` })(),
log: () => {},
state: new MemoryState(seen, true)
})
return { p, sent }
}
const OBJ = (rkey) => `https://ap.7rnx.net/ap/objects/${rkey}`
test('★★★ a delivered post that was deleted is retracted with Delete{Tombstone}, once', async (t) => {
const { p, sent } = makeDel([{ rkey: '3d', id: OBJ('3d') }], { seen: [['3d', 'fp']], followers: ['https://a.net/u/1', 'https://b.net/u/2'] })
await settle(p)
t.is(sent.length, 2, 'one Delete per follower')
t.is(sent[0].activity.type, 'Delete')
t.is(sent[0].activity.actor, ACTOR)
t.alike(sent[0].activity.object, { id: OBJ('3d'), type: 'Tombstone' })
await settle(p)
t.is(sent.length, 2, 'and never again')
})
test('★★ a deleted post that was never delivered sends nothing', async (t) => {
const { p, sent } = makeDel([{ rkey: '3d', id: OBJ('3d') }])
await settle(p)
t.is(sent.length, 0)
})
test('⛔ with nobody to tell, the Delete is not recorded as sent', async (t) => {
const { p, sent } = makeDel([{ rkey: '3d', id: OBJ('3d') }], { seen: [['3d', 'fp']], followers: [] })
await settle(p)
t.is(sent.length, 0)
t.is((await p.state.getDelivered()).get('3d'), 'fp', 'still recorded as the delivered post, not as retracted')
})
test('a source with no listDeleted publishes as before', async (t) => {
const { p, delivered } = make([note('3c', 'z')], { primed: true })
await settle(p)
t.is(delivered.length, 1)
t.is(delivered[0].activity.type, 'Create')
})