// publisher.js — deliver newly-authored posts to follower inboxes.
//
// WHY THIS EXISTS AT ALL, given the outbox already serves every post: Mastodon does NOT read an
// actor's outbox for timelines, and does NOT backfill on follow — it shows only what was DELIVERED
// to its inbox. So a correct, public, fetchable outbox puts nothing in anyone's timeline. Delivery
// is what makes a post arrive. (The outbox still matters: it is how a server verifies an object it
// was told about, and how anything that DOES read outboxes finds us.)
//
// EVENT-DRIVEN, not polled. DidPeer fires `onHead` on every newly VERIFIED head, which is the
// trigger; a DELIVERED-SET then answers "which records are new". A polling timer here would be the
// anti-pattern the fleet has repeatedly had to rip out.
//
// ★ WHY A SET AND NOT A HIGH-WATERMARK. The obvious design is one watermark rkey plus "everything
// lexicographically greater is new", justified by rkeys being TIDs so that byte order IS time
// order. That is FALSE in practice: wj62's repo contains `e2e-overlay-1784495258`, a hand-written
// rkey from an unrelated test. It sorts above every TID (`e` > `3`) while being the OLDEST record,
// so it became the watermark and every genuinely new post — TID-prefixed, therefore "lower" —
// would have been silently skipped forever. A set is ordering-independent and cannot be defeated
// by whatever rkey scheme a record happens to use.
//
// ⚠ NEW POSTS ONLY. On first run the watermark is initialised to the CURRENT tip and nothing is
// delivered — publishing three weeks of history into every follower's timeline at once is both
// rude and not what any other server does. The cost is that a follower sees nothing until the next
// post; that is the same deal every fediverse account offers.
//
// Deliberately imports NOTHING from records.js: pulling it in would drag atproto-did-peer ->
// atproto-store -> cborg into the test process, which touches TextEncoder at module load and
// needs the Bare globals shim. Staying dependency-free keeps this offline-testable like every
// other lib module.
export class Publisher {
constructor ({ records, followers, deliver, actorId, newId, log = () => {}, state, backfill = 0, buildActorDoc = null, blobResolves = null }) {
// blobResolves: optional async (url) -> boolean. Used ONLY by reproject() to skip a post whose
// media cannot be served (see there). Absent in the normal drain, which must not add a network
// call per post to the hot path.
// buildActorDoc: () => the current AS2 actor. Needed for Update{Person} — see refreshProfile.
this.buildActorDoc = buildActorDoc
this.blobResolves = blobResolves
// backfill: on the FIRST prime only, leave the N most recent posts undelivered so they go out
// once. Deliberately not a mode that changes steady-state behaviour — see primeWatermark.
this.backfill = Math.max(0, Number(backfill) || 0)
this.records = records
this.followers = followers
this.deliver = deliver
this.actorId = actorId
this.newId = newId
this.log = log
this.state = state // { getWatermark(), setWatermark(rkey) } — persisted across restarts
this._running = false
this._again = false
}
// Mark everything that already exists as delivered, WITHOUT delivering it. Called once at boot.
//
// AP_BACKFILL exists because Mastodon never reads an outbox for timelines and never backfills on
// follow, so history that is published and fetchable still appears nowhere. It leaves the N most
// recent posts OUT of the primed set, so the first drain delivers them.
//
// ★ ONE-SHOT BY CONSTRUCTION, and that matters. Backfilling into your OWN instance where you are
// the only follower is harmless; the same code path aimed at a stranger who just followed would
// blast them with your entire history. So it only ever applies to the first prime, and it
// self-disarms — those rkeys land in the delivered-set as they go out, so a restart with the
// variable still set does nothing. It is an operation, not a mode.
async primeWatermark () {
// ⛔ A PRIME THAT RECORDED NOTHING IS NOT A PRIME. `primed` alone was the guard, and a run whose
// delivered-set came out EMPTY still set it — so the flag said "history accounted for" while
// nothing had been. Re-priming when the set is empty is safe in both directions: with no
// backfill it marks the history delivered (which is what the flag claimed anyway), and with a
// backfill it holds the same posts back and sends them, which is what the backfill was for.
const delivered = await this.state.getDelivered()
if (await this.state.isPrimed() && delivered.size) return // a later run must not re-prime and swallow new posts
const { notes } = await this.records.listNotes({ limit: MAX_TRACKED })
// ⛔ AND AN EMPTY LISTING IS NOT AN EMPTY FEED. A held feed's log replicates asynchronously, so
// a prime taken at boot routinely sees ZERO notes on a peer with a full history — and recording
// that reads as "there was nothing to account for". Measured live: ap-bridge primed ashatron at
// boot before the feed arrived, wrote primed=true with an empty set, and the 16 posts already in
// its outbox were left neither delivered nor accounted for. The two states are indistinguishable
// from here and only one of them is safe to record, so this records neither and tries again.
if (!notes.length) {
this.log({ msg: 'publisher: nothing to prime yet — not recording a prime against an empty listing' })
return
}
// listNotes is newest-first, so the first N are the most recent. Ordering is only a heuristic
// for "which history to send" — correctness never depends on it (see the header note).
const hold = this.backfill > 0 ? notes.slice(0, this.backfill) : []
const heldKeys = new Set(hold.map((n) => rkeyIn(this.actorId, n)).filter(Boolean))
const primed = notes
.map((n) => [rkeyIn(this.actorId, n), noteFingerprint(n)])
.filter(([k]) => k && !heldKeys.has(k))
await this.state.markDelivered(primed)
await this.state.setPrimed()
this.log({
msg: hold.length ? 'publisher primed WITH BACKFILL — recent posts will be delivered once' : 'publisher primed — existing posts will NOT be delivered',
existing: notes.length,
backfill: hold.length
})
// ★ Kick the drain. Delivery is otherwise triggered ONLY by onHead, so held-back posts would
// sit undelivered until some unrelated future post advanced the head — which is not "delivered
// once", it is "delivered eventually, if you happen to post again". Backfill has to be
// self-starting to mean anything.
if (hold.length) this.onHead()
}
// ★ Mastodon CACHES actor documents and will not re-fetch one just because it changed. A new
// avatar, banner, display name or bio is invisible to every follower until an Update{Person}
// tells them to look again — which is why "the profile is correct at /ap/actor" and "the profile
// is correct on Mastodon" are different claims.
//
// Fires only when the projected actor actually CHANGED (compared by a stored fingerprint), so a
// restart does not spam followers with no-op Updates.
async refreshProfile ({ force = false } = {}) {
if (!this.buildActorDoc) return
let doc
try { doc = await this.buildActorDoc() } catch (err) { return this.log({ msg: 'profile refresh skipped', error: err.message }) }
const fp = fingerprint(doc)
// ⚠ `force` resends an unchanged profile. Needed because a remote server can hold a STALE copy
// that our own fingerprint says is current — it cached the actor during a window when we served
// it wrongly, and nothing about our state records that. Better than deleting the state file,
// which is persistent and not ours to remove on a hunch.
if (!force && fp === await this.state.getProfileFingerprint()) return
const recipients = await this.followers.list()
// ⛔ RECORD IT ONLY WHEN SOMEBODY WAS TOLD. Storing the fingerprint first meant a run with no
// followers recorded "already sent this" — so when a follower did arrive, the profile was
// permanently un-sendable and the only symptom was silence. Measured: ashatron's publisher
// stored a fingerprint on its first no-op run and every PATCH after it short-circuited.
if (!recipients.length) {
// ⛔ SAY SO. This return was silent, and a silent return here is indistinguishable from the
// method never being called: a forced PATCH /ap/register answered {"ok":true,"forced":true}
// and produced no log line at all, which sent me looking at the wrong layer entirely. The
// actual cause was that the only follower had just unfollowed.
this.log({ msg: 'profile refresh: nobody to tell', actor: this.actorId })
return
}
const update = {
'@context': ['https://www.w3.org/ns/activitystreams', { atproto: 'https://atproto.com/ns#' }],
id: `${this.actorId.replace(/\/actor$/, '')}/activities/${this.newId()}`,
type: 'Update',
actor: this.actorId,
to: ['https://www.w3.org/ns/activitystreams#Public'],
// ⛔ `updated` OR NOTHING HAPPENS. Mastodon silently discards an Update whose object carries
// no updated timestamp — 2xx, no edit, no media re-fetch — so every profile change this
// bridge ever sent was accepted and ignored. The note path sets it (below); this one did not,
// which is why a stale display name never corrected itself.
object: { ...doc, updated: new Date().toISOString() }
}
this.log({ msg: 'profile changed — sending Update', followers: recipients.length })
const landed = await Promise.all(recipients.map((to) =>
this.deliver({ to, activity: update })
.then(() => { this.log({ msg: 'profile Update delivered', to }); return true })
.catch((err) => { this.log({ msg: 'profile Update failed', to, error: err.message }); return false })
))
// ⛔ RECORDED AFTER, AND ONLY IF SOMETHING LANDED. The guard above fixed "nobody to tell"; this
// is the same defect one step later. Every delivery is caught and logged rather than thrown, so
// a run where all of them failed used to store the fingerprint anyway and the profile became
// permanently un-sendable — the same silence, from a different cause. A receiver that was told
// is the only thing that makes this true.
if (landed.some(Boolean)) return this.state.setProfileFingerprint(fp)
this.log({ msg: 'profile Update reached nobody — not recording it as sent', followers: recipients.length })
}
// The onHead handler. Serialised: a head can advance again while we are still delivering, and
// two concurrent passes would double-deliver the same record.
onHead () {
if (this._running) { this._again = true; return }
this._running = true
this._drain()
.catch((err) => this.log({ msg: 'publisher drain failed', error: err.message }))
.finally(() => {
this._running = false
if (this._again) { this._again = false; this.onHead() }
})
}
async _drain () {
const delivered = await this.state.getDelivered()
const { notes } = await this.records.listNotes({ limit: 50 })
// Anything not already delivered is new. No ordering assumption is made about rkeys — see the
// header note on why a high-watermark was wrong here.
// Three cases per post: never sent (Create), sent but the projection has CHANGED since
// (Update — Mastodon's edit mechanism), or sent and unchanged (skip). The fingerprint is what
// distinguishes them, and it means every future projection improvement — images today, facets
// and quotes later — reaches already-delivered posts automatically instead of needing a
// one-off republish each time.
const fresh = []
for (const n of notes) {
const rkey = rkeyIn(this.actorId, n)
if (!rkey) continue
const fp = noteFingerprint(n)
if (!delivered.has(rkey)) fresh.push({ note: n, rkey, fp, verb: 'Create' })
else if (delivered.get(rkey) !== fp) fresh.push({ note: n, rkey, fp, verb: 'Update' })
}
// ⛔ LOG EVERY DRAIN, INCLUDING THE EMPTY ONES. `publisher: outgoing` is only reached when there
// is something to send, so a drain that found nothing was indistinguishable from a drain that
// never ran — and that is exactly the state that cost an evening: primed WITH BACKFILL 16 in the
// log, an outbox serving 16, a follower recorded, and no Create ever sent, with nothing on the
// running node able to say which of the two it was. Same lesson as the mirror's counters.
this.log({ msg: 'publisher: drain', notes: notes.length, delivered: delivered.size, fresh: fresh.length })
const posts = fresh.length ? await this._send(fresh) : { sent: 0 }
const interactions = await this._drainInteractions(delivered)
const deletes = await this._drainDeletes(delivered)
return { ...posts, interactions: interactions.sent, deleted: deletes.sent }
}
// ── deletes ─────────────────────────────────────────────────────────────────────────────────
//
// ⛔ ON A TOMBSTONE ONLY. `_drainInteractions` explains why absence cannot drive an undo: a paged
// listing cannot prove a record is gone. A favela feed CAN — its owner's delete is a tombstone that
// replicates — so a source that reports `listDeleted()` gives the positive evidence, and only a post
// we actually delivered (and have not already retracted) gets a Delete.
//
// ⚠ Recorded per post, and only when somebody was told — the same rule as _send.
async _drainDeletes (delivered) {
if (typeof this.records.listDeleted !== 'function') return { sent: 0 }
let listed
try { listed = await this.records.listDeleted() } catch (err) {
this.log({ msg: 'publisher: deletes unavailable', error: err.message })
return { sent: 0 }
}
const due = (listed.deleted || []).filter((d) => d && d.rkey && delivered.has(d.rkey) && delivered.get(d.rkey) !== RETRACTED)
if (!due.length) return { sent: 0 }
const recipients = await this.followers.list()
if (!recipients.length) {
this.log({ msg: 'publisher: deletes — nobody to tell', held: due.length })
return { sent: 0 }
}
this.log({ msg: 'publisher: outgoing deletes', count: due.length, followers: recipients.length })
for (const { rkey, id } of due) {
const activity = {
'@context': 'https://www.w3.org/ns/activitystreams',
id: `${this.actorId.replace(/\/actor$/, '')}/activities/${this.newId()}`,
type: 'Delete',
actor: this.actorId,
to: ['https://www.w3.org/ns/activitystreams#Public'],
object: { id, type: 'Tombstone' }
}
await Promise.all(recipients.map((to) =>
this.deliver({ to, activity })
.then(() => this.log({ msg: 'published', verb: 'Delete', to, object: id }))
.catch((err) => this.log({ msg: 'publish delivery failed', verb: 'Delete', to, object: id, error: err.message }))
))
await this.state.markDelivered([[rkey, RETRACTED]])
}
return { sent: due.length }
}
// ── likes and reposts ───────────────────────────────────────────────────────────────────────
//
// Interactions are ACTIVITIES, not objects wrapped in a Create, so they do not go through _send:
// there is no Note to wrap and no Update semantics — a like is sent once or not at all.
//
// ⛔ MOST INTERACTIONS ARE NOT SENDABLE, AND THAT IS NOT A FAILURE. `listInteractions` returns
// only the ones whose subject we can name in AP terms (today: our own posts), and counts the rest.
// The count is logged rather than dropped, because "no likes" and "twelve likes of bsky posts we
// cannot address" are different states and only the second one means the feature is limited.
//
// ⚠ Un-liking is NOT wired. projectUndo exists and is tested, but sending it requires PROVING a
// record was deleted, and a paged listing cannot prove absence — a like outside the window looks
// identical to one that was removed. Getting that wrong un-boosts things that are still there, so
// it needs a full-collection read, deliberately, as its own change.
async _drainInteractions (delivered) {
let listed
try {
listed = await this.records.listInteractions({ limit: 50 })
} catch (err) {
this.log({ msg: 'publisher: interactions unavailable', error: err.message })
return { sent: 0 }
}
const fresh = listed.interactions.filter((i) => !delivered.has(i.key))
if (listed.unaddressable) {
this.log({ msg: 'publisher: interactions not addressable in AP terms', count: listed.unaddressable })
}
if (!fresh.length) return { sent: 0 }
const recipients = await this.followers.list()
this.log({ msg: 'publisher: outgoing interactions', count: fresh.length, followers: recipients.length })
for (const { key, activity, kind } of fresh) {
await Promise.all(recipients.map((to) =>
this.deliver({ to, activity })
.then(() => this.log({ msg: 'published', verb: activity.type, to, object: activity.object }))
.catch((err) => this.log({ msg: 'publish delivery failed', verb: activity.type, to, object: activity.object, error: err.message }))
))
// Fingerprint is the activity id: an interaction has no editable projection, so "delivered"
// is the whole of its state.
await this.state.markDelivered([[key, activity.id]])
}
return { sent: fresh.length }
}
// The delivery loop, shared by the head-driven drain and by reproject(). Extracted rather than
// duplicated: two copies of a delivery path is how one of them quietly stops matching the other.
async _send (fresh) {
if (!fresh.length) return { sent: 0 }
// Oldest-first by DECLARED time so a follower's timeline reads in order. createdAt is
// client-declared and not an authoritative clock, so this is presentation only — correctness
// does not depend on it.
fresh.sort((a, b) => String(a.note.published || '').localeCompare(String(b.note.published || '')))
const recipients = await this.followers.list()
const creates = fresh.filter((f) => f.verb === 'Create').length
this.log({ msg: 'publisher: outgoing', create: creates, update: fresh.length - creates, followers: recipients.length })
// ⛔ A DELIVERY THAT DID NOT HAPPEN IS NOT RECORDED AS ONE. This used to mark 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. That is the profile-fingerprint defect of 0.9.71 — "recorded as
// told when nobody was told" — one layer over, in the post path.
//
// ⚠ So posts made while unfollowed go out when a follower arrives. That is a deliberate,
// bounded consequence of only ever recording what happened, and the honest direction to fail in:
// the alternative silently destroys them.
if (!recipients.length) {
this.log({ msg: 'publisher: nobody to send to — holding', held: fresh.length })
return { sent: 0 }
}
for (const { note, rkey, fp, verb } of fresh) {
// ⛔ AN UPDATE WITHOUT `updated` IS SILENTLY IGNORED BY MASTODON. It accepts the POST, returns
// 2xx, and does nothing — no edit, no re-fetch of attachments, edited_at stays null. Verified
// live 2026-08-05: a forced Update for 3mrln2axjlc22 was delivered and accepted, and the
// status still showed 0 media and edited_at: None.
//
// Mastodon has required this since 3.5.0 — an "unofficial requirement" per the community
// write-ups, and exactly the kind of accepted-but-ignored failure this codebase keeps paying
// for (#24, #26: "everything we control reports success").
//
// Set at SEND time, deliberately not part of the projection: if `updated` were in the note the
// fingerprint would change on every drain and every post would look permanently edited.
const object = verb === 'Update' ? { ...note, updated: new Date().toISOString() } : note
const create = {
'@context': ['https://www.w3.org/ns/activitystreams', { atproto: 'https://atproto.com/ns#' }],
id: `${this.actorId.replace(/\/actor$/, '')}/activities/${this.newId()}`,
type: verb,
actor: this.actorId,
published: note.published,
to: note.to,
cc: note.cc,
object
}
// Deliver to every follower independently. One unreachable server must not block the rest,
// and a permanent rejection from one is not a reason to withhold the post from others.
await Promise.all(recipients.map((to) =>
this.deliver({ to, activity: create })
.then(() => this.log({ msg: 'published', verb, to, object: object.id }))
.catch((err) => this.log({ msg: 'publish delivery failed', verb, to, object: object.id, error: err.message }))
))
// Record per-post, so a crash mid-batch re-delivers at most the current post rather than
// replaying the whole batch. Receivers dedup on activity/object id, so a repeat is benign.
await this.state.markDelivered([[rkey, fp]])
}
return { sent: fresh.length }
}
// ── reproject ───────────────────────────────────────────────────────────────────────────────
//
// Re-derive every projection and Update whatever has CHANGED since it was delivered.
//
// WHY IT IS NEEDED even though _drain already handles Update. The drain runs only on onHead(), so
// a projection improvement — images today, facets and quotes later — reaches already-delivered
// posts only when some unrelated FUTURE post advances the head. That is not "delivered", it is
// "delivered eventually, if you happen to post again". #26 recorded this exact trap twice:
// "event-driven quietly meant never happens". An event-driven system needs a defined
// reconciliation it can be ASKED to run.
//
// Measured 2026-08-05: 5 posts hold servable image bytes here and show zero media on prsm.space,
// because they were delivered before attachments projected. Mastodon caches and never refetches;
// Update{Note} is its edit mechanism and therefore the only way to correct them.
//
// rkey — restrict to one post
// force — ignore fingerprints and re-send regardless (receivers dedup on activity id)
// verify — drop any post whose media cannot actually be served (default ON, see below)
async reproject ({ rkey = null, force = false, verify = true, limit = 50 } = {}) {
const delivered = await this.state.getDelivered()
const { notes } = await this.records.listNotes({ limit })
const skipped = []
const fresh = []
for (const n of notes) {
const k = rkeyIn(this.actorId, n)
if (!k) continue
if (rkey && k !== rkey) continue
const fp = noteFingerprint(n)
const known = delivered.has(k)
if (!force && known && delivered.get(k) === fp) continue
// ⛔ NEVER re-send a post whose media 404s. One post (3mpquxgrvu227) has attachments whose
// bytes were lost before the overlay migration, so its projection points at a URL that cannot
// resolve. Re-sending teaches the receiver the same absence a second time and burns the one
// edit that would have fixed it later if the bytes were ever recovered.
if (verify && this.blobResolves && Array.isArray(n.attachment) && n.attachment.length) {
const urls = n.attachment.map((a) => a && a.url).filter(Boolean)
const oks = await Promise.all(urls.map((u) => this.blobResolves(u).catch(() => false)))
if (oks.some((ok) => !ok)) { skipped.push({ rkey: k, reason: 'media not servable' }); continue }
}
fresh.push({ note: n, rkey: k, fp, verb: known ? 'Update' : 'Create' })
}
const updates = fresh.filter((f) => f.verb === 'Update').length
this.log({ msg: 'reproject', candidates: notes.length, sending: fresh.length, updates, creates: fresh.length - updates, skipped: skipped.length, force, verify })
for (const s of skipped) this.log({ msg: 'reproject: SKIPPED', ...s })
const res = await this._send(fresh)
return { sent: res.sent || 0, updates, creates: fresh.length - updates, skipped }
}
}
// Only the fields a follower actually sees — so an unrelated field appearing in the actor doc
// does not trigger a pointless Update to everyone.
// Everything a receiver renders. If any of it changes, the post they hold is stale.
export function noteFingerprint (note) {
return JSON.stringify([note?.content, note?.attachment, note?.inReplyTo, note?.contentMap, note?.published])
}
function fingerprint (doc) {
return JSON.stringify([doc?.name, doc?.summary, doc?.icon?.url, doc?.image?.url, doc?.publicKey?.publicKeyPem])
}
const MAX_TRACKED = 500 // bounded: old rkeys can never reappear as new
// The delivered-map value for a post we RETRACTED. Not a fingerprint any Note can produce (those are
// JSON arrays), so a retracted post is never mistaken for one delivered and unchanged, or re-sent.
const RETRACTED = 'retracted'
/**
* The rkey of one of OUR OWN notes, relative to this publisher's actor.
*
* ⛔ IT USED TO BE A FIXED PATTERN — `/\/ap\/objects\/([^/?#]+)$/` — which is the PRIMARY's id shape.
* A registered user's objects live at `/ap/users/<name>/objects/<rkey>`, so it returned null for
* every one of their notes, and it broke both ends silently and in opposite directions:
* primeWatermark's delivered-set came out empty however many posts existed, and _drain skipped every
* note at `if (!rkey) continue` so `fresh` was permanently 0. Measured live: notes 16, delivered 0,
* fresh 0, with the outbox serving all 16 and a follower recorded. Not one Create was ever sent.
*
* ⛔ AND IT IS DERIVED FROM `actorId`, not loosened to any `/objects/<rkey>`. The rkey is a key into
* a delivered-set; matching another actor's ids would let two users collide on one rkey and mark
* each other's posts as sent. Ownership is the point of the check, not a side effect of it.
*/
const rkeyIn = (actorId, note) => {
const prefix = String(actorId || '').replace(/\/actor$/, '') + '/objects/'
const id = String(note?.id || '')
if (!id.startsWith(prefix)) return null
const rest = id.slice(prefix.length).split(/[/?#]/)[0]
return rest || null
}