// ap-bridge — the ActivityPub membrane, as a module. See DESIGN.md.
//
// M1 SCOPE (deliberate deviation, DESIGN.md §11): this process holds the actor's private key.
// The target architecture keeps keys on an attested virtual-ap-device and signs over RPC, so the
// bridge stays keyless — that is M2. Proving federation and proving keylessness are two separate
// experiments; running them together means a failure has two possible causes.
//
// ⛔ WHY A MODULE. Until 0.10.15 this was index.js's main(): it bound the public and admin ports, the
// admin recovery socket, a FeedHolder (Corestore + swarm) and an admin-identity Corestore, and shut
// down by closing ONLY the public server and calling process.exit — process exit is what released
// everything else. The favela gateway (hyper-social #80) runs this beside the atproto membrane in one
// process, so stopping it has to free every port, socket and store lock it took. startApBridge() does
// exactly what main() did and returns:
// - `steps`: the ordered stop sequence as [name, fn] pairs, for the caller's bounded shutdown;
// - `handlers`: the control verbs (lib/control.js bridgeHandlers), for whichever socket serves them;
// - the callbacks startControl takes (status, onInvite, onNotifications, onReproject);
// - `port`, `adminPort`, `bridgeKey`.
// Nothing here exits the process or installs a signal handler. index.js is the standalone wrapper.
import './bare-globals.js' // FIRST — deps touch TextEncoder at module load
import fs from 'bare-fs'
import process from 'bare-process'
import { join } from 'bare-path'
import crypto from 'bare-crypto'
import http from 'bare-http1'
import { createServer } from './server.js'
import { actorDocFor } from './router.js'
import { FollowerStore } from './followers.js'
import { InboundStore } from './inbound-store.js'
import { ObservationStore } from './observations.js'
import { RemoteKeyResolver } from './remote-key.js'
import { createDeliverer } from './deliver.js'
import { LocalSigner } from './signer.js'
import { DeliveryQueue, createQueuedDeliverer } from './queue.js'
import { toNotifications } from './inbound-notifications.js'
import { Users } from './users.js'
import { PerUser } from './per-user.js'
import { Registrations } from './registrations.js'
import { FeedHolder } from './feed-holder.js'
import { UserKeys } from './user-keys.js'
import { keyId, actorId, buildActor } from './actor.js'
import { RecordSource } from './records.js'
import { createBlobFetcher } from './blobs.js'
import { httpClient } from './https.js'
import { Publisher } from './publisher.js'
import { JsonState } from './state.js'
import { bridgeHandlers } from './control.js'
import { DidPeer } from 'atproto-did-peer'
import { FavelaSource } from './favela-source.js'
import { resolveDidVerifyKey } from './plc.js'
import Hyperswarm from 'hyperswarm'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
const defaultLog = (o) => console.log(JSON.stringify({ t: new Date().toISOString(), ...o }))
/**
* @param {object} config assertUsable(loadConfig())'s shape
* @param {object} [opts]
* @param {Function} [opts.log]
* @param {Array} [opts.bootstrap] DHT bootstrap; tests pass [] to stay off the public DHT
* @param {string} [opts.version] for the `status` verb
* @param {Map} [opts.adminPages] path -> async (req) => ({ title, body }), served on the admin surface
*/
export async function startApBridge (config, { log = defaultLog, bootstrap = null, version = null, adminPages = null } = {}) {
const privateKey = fs.readFileSync(config.privateKeyPath, 'utf8')
const publicKeyPem = fs.readFileSync(config.publicKeyPath, 'utf8')
const dataDir = config.dataDir || join(process.env.HOME || '.', '.config', 'ap-bridge')
const followers = new FollowerStore(join(dataDir, 'followers.json'))
// What the fediverse sends us — replies, mentions, likes, boosts. Until now these were verified,
// acked with 202 and discarded, so ap-bridge could be replied to and never knew.
// ⛔ THE CAP IS THE OPERATOR'S. It was fixed at 500 here, and for the fediverse an eviction is not a
// cache miss — there is no backlink index to re-ask, so it can be the only copy. See config.js.
const inbound = new InboundStore({ file: join(dataDir, 'inbound.json'), cap: config.inboundCap })
// ⛔ THE DURABLE HALF OF THE SAME ARRIVAL. `inbound` is a capped timeline for THIS bridge's pages;
// an observation is what a registered PEER collects and keeps. Both are written from one inbox POST
// and they have different lifetimes, which is why they are two stores rather than one with a flag —
// see hyper-social/docs/OBSERVED-INTERACTIONS.md §9.
const observations = new ObservationStore({ path: join(dataDir, 'observations.jsonl') })
const resolver = new RemoteKeyResolver()
// Everything this start opens, so stop() can close it. Filled in as each piece comes up.
const open = { didPeer: null, didSwarm: null, didStore: null, reg: null, adminServer: null, adminControl: null, browsers: null, browsersStore: null }
// ⛔ CUSTODY IS NAMED HERE, not implied by passing a key around. This is the primary actor and it
// is co-located with its key on this box, so LocalSigner is honest — signer.js: "a RemoteSigner to
// a peer ON THE SAME BOX buys nothing". Moving custody to the virtual-ap-device (DESIGN.md §4) is
// now a change to this one expression; nothing in deliver.js or http-signature.js moves.
//
// It also parses the PEM once instead of on every delivery, which is what signRequest did when
// handed a PEM string (rsa.js:184 re-parses whenever the key is a string).
const sendNow = createDeliverer({
signer: new LocalSigner({ keyId: keyId(config), privateKeyPem: privateKey }),
log
})
// ⛔ EVERY OUTBOUND ACTIVITY GOES THROUGH THE QUEUE, and is durable before the network is touched.
// Both callers benefit: Publisher's Creates, and router.js:287's `Accept` for an inbound Follow —
// which logged "will retry" while having no retry that survived a restart. A follow that arrives
// while this box is down is the main thing genuinely owed to somebody, and it is now owed durably.
// ⛔ THE QUEUE CARRIES A KEY ID, AND THIS TURNS IT BACK INTO A SIGNER. A signer is a live object
// and cannot survive a restart in a JSON file, so the durable entry records WHO signs and the
// signer is resolved per attempt. Without this every queued activity fell back to the process
// signer: a registered user's posts went out signed with the primary's key against the primary's
// key id, and Mastodon dropped every one of them after answering 202.
//
// ⚠ Assigned below, once PerUser exists. A null resolver means a queued entry with a signerId is
// HELD rather than sent under the wrong identity — the safe direction.
let signerByKeyId = null
// Kept, so metrics() can read pending and dead entries; the deliverer exposes only counts.
const deliveryQueue = new DeliveryQueue(join(dataDir, 'queue.json'))
const outbound = createQueuedDeliverer({
queue: deliveryQueue,
deliver: sendNow,
resolveSigner: (keyId) => (signerByKeyId ? signerByKeyId(keyId) : null),
log
})
const deliver = outbound.deliver
// ⛔ Registration is OFF unless the operator turned it on — `config.registration` is 'open' or
// 'invite'. Absent means no registry at all and /ap/register 404s, so a bridge never starts
// handing out names because nobody said not to.
const registrations = config.registration
? new Registrations({ path: join(dataDir, 'registrations.json'), policy: config.registration })
: null
if (registrations) log({ msg: 'registration', policy: registrations.policy })
// ⛔ Holds registered peers' feeds over the SWARM. A registered peer is on somebody else's
// machine, so the bridge holds their feed read-only by key — the same thing any follower does —
// rather than reaching for a unix socket that only exists when the peer is co-located.
const holder = registrations ? await new FeedHolder({ storage: join(dataDir, 'held'), bootstrap, log }).ready() : null
// ⛔ WHAT THE FEDIVERSE DOES TO A REGISTERED USER goes into an observation feed this bridge owns for them, which
// their nodes hold (hyper-social #82, I1). On FeedHolder's corestore, overlay and SWARM — one store, one swarm,
// no new published key. Non-fatal: a bridge that federates is worth more than one that refuses to boot.
let obsFeeds = null
if (holder) {
try {
const { ObservationFeeds } = await import('./observation-feeds.js')
obsFeeds = await new ObservationFeeds({ corestore: holder.corestore, storage: join(dataDir, 'held'), overlay: holder.overlay, swarm: holder.swarm, log }).ready()
log({ msg: 'observation feeds ready', observerRoot: obsFeeds.root })
} catch (err) {
log({ msg: 'observation feeds unavailable', error: (err && err.message) || String(err) })
}
}
// ⛔ ITS OWN KEYPAIR PER USER. Mastodon indexes actors by publicKey.id, and sharing the primary's
// key made a registered user resolve AS the primary — a duplicate, not a new account. Distinct
// ids fix resolution; distinct key material is what makes the separation real, since a signature
// is what proves who acted.
const userKeys = registrations ? new UserKeys({ dir: join(dataDir, 'user-keys') }) : null
// ⛔ The control half of admission. registrations.js holds the registry and says in its own header
// that proving control "belongs at the route where the challenge lives" — this is that challenge.
// Without it an identity root, which is public, was accepted as proof of holding it.
const { Admission } = await import('./admission.js')
const admission = registrations ? new Admission() : null
let seq = 0
const ctx = {
registrations,
admission,
config,
publicKeyPem,
followers,
inbound,
observations,
resolvePublicKey: (kid) => resolver.resolve(kid),
deliver,
// Activity ids must be unique and stable enough that a receiver can dedup. Timestamp+counter
// is adequate for one actor; a real id scheme belongs with the durable queue.
newId: () => `${Date.now().toString(36)}-${(seq++).toString(36)}`,
log
}
// The router's observation-feed sink (router.js observe/unobserve). Keyed by the REGISTRATION generation, so a
// released and re-claimed name writes to a new feed, never its predecessor's.
// favela feed key → its registration, for OTHER roles in this process (the gateway's bsky ingest, #82 I2b): what
// reaches a registered identity from another network goes into the same observation feed. Filled at admission.
const feedRegs = new Map()
if (obsFeeds && registrations) {
const genOf = async (user) => { const reg = await registrations.get(user.username); return reg ? reg.generation : null }
ctx.observe = {
note: async (user, obs) => { const g = await genOf(user); if (g !== null) await obsFeeds.note(user.username, g, obs) },
remove: async (user, activity) => { const g = await genOf(user); if (g !== null) await obsFeeds.remove(user.username, g, activity) }
}
ctx.observationFeeds = obsFeeds
}
// ★ THE KEYLESS READ SIDE. Front the DID exactly as atproto-bridge and favela do: replicate the
// peer's cores, read records, project to AS2 on demand. No key involved — atproto-did-peer
// refuses to front a repo whose signed commit doesn't verify against the DID's published key,
// which is what makes reading keylessly safe rather than merely convenient.
//
// Failure here is NON-FATAL: without a record source the outbox is empty but discovery, the
// inbox and Follow/Accept all still work. A peer that is briefly unreachable must not stop the
// actor existing.
// In this function's scope: the control surface's `reproject` (below) must reach it, and it was
// once declared inside the try-block that builds it — in scope for onHead, out of scope for
// everything after. node --check does not catch that; only calling reproject would have.
let publisher = null
/**
* ⛔ THE FAVELA SOURCE, when one is configured — records read from the peer's own markdown rather
* than from the atproto projection of them.
*
* `favela-markdown/spec/design.md`: both networks derive from the canonical record, neither from
* the other. Deriving AS2 from `record.text` gives the fediverse bsky's plain text, so a heading
* and a list arrive as paragraphs of literal bullet characters. This path renders the markdown.
*
* ⛔ AND IT NEEDS NO DID. No atproto identifier, no PDS, no MST, no swarm topic — just the peer's
* socket. That is what lets this bridge front an identity holding no atproto credential at all
* (identity.md §4.1), which the branch below cannot do.
*
* ⚠ EXCLUSIVE, and checked rather than assumed. Two sources would mean two sets of object ids for
* one author, and whichever ran second would look like a stranger reposting the first.
*/
if (config.favelaSocket) {
if (config.did) {
throw new Error('ap-bridge: configure favelaSocket OR did, not both — two sources for one ' +
'actor produce two object ids per post, and the second reads as somebody else reposting')
}
ctx.records = new FavelaSource({
sock: config.favelaSocket,
host: config.host,
actorId: actorId(config),
followersId: `https://${config.host}/ap/followers`,
roomType: config.roomType || 'public'
})
log({ msg: 'favela source', sock: config.favelaSocket })
} else if (config.did) {
// GOTCHA (favela hit this): use atproto-store's openCorestore, NOT `new Corestore` from a
// separately-installed copy — RepoStore's `instanceof Corestore` fails across duplicated
// module copies and surfaces as "path... undefined".
const corestore = openCorestore(join(dataDir, 'store'))
const swarm = new Hyperswarm(bootstrap ? { bootstrap } : {})
open.didStore = corestore
open.didSwarm = swarm
try {
const verifyKey = await resolveDidVerifyKey(config.did)
// onHead fires on every newly VERIFIED head — the event-driven trigger for delivery. A
// polling timer here would be the anti-pattern this fleet keeps having to remove.
const peer = new DidPeer(corestore, {
did: config.did,
verifyKey,
// ⛔ A HEAD ADVANCE CARRIES PROFILE CHANGES, NOT JUST POSTS. onHead drained notes and
// nothing else, and refreshProfile was called exactly once at boot — so a changed display
// name, bio or avatar reached Mastodon only if the process happened to restart afterwards.
// Measured 2026-09-09: our actor served a bio prsm had never seen, last Update 21h earlier.
// refreshProfile is fingerprint-guarded, so this is a no-op when nothing changed.
onHead: () => {
if (!publisher) return
publisher.onHead()
publisher.refreshProfile().catch((err) => log({ msg: 'profile refresh failed', error: err.message }))
},
onError: (err) => log({ msg: 'peer error', error: err.message })
})
open.didPeer = peer
// One replicate per connection; the peer attaches to learn its cores (the keet model).
swarm.on('connection', (conn) => peer.attach(conn))
// ★ JOIN THE DID'S TOPIC. Attaching connections is useless without this — nothing ever
// connects, and the only symptom is an outbox that is permanently empty with no error.
// {client,server} BOTH: symmetric join is load-bearing for HOLE-PUNCHING, not just
// discovery — a client:false member never punches outbound so its NAT mapping never opens.
swarm.join(discoveryTopic(config.did), { client: true, server: true })
await swarm.flush()
await peer.ready({ timeout: 20000 })
ctx.records = new RecordSource({
peer,
did: config.did,
host: config.host,
actorId: actorId(config),
followersId: `https://${config.host}/ap/followers`
})
publisher = new Publisher({
records: ctx.records,
followers,
deliver,
actorId: actorId(config),
newId: ctx.newId,
log,
state: new JsonState(join(dataDir, 'publisher.json')),
backfill: config.backfill,
// Used ONLY by reproject(): can we actually serve this attachment? A /ap/blobs/<cid> URL is
// proxied from the PDS, so the honest check is whether the PDS has the bytes. One post's
// blobs were lost before the overlay migration; re-sending it would teach the receiver the
// same 404 again and burn the one edit that would fix it if the bytes ever came back.
blobResolves: async (url) => {
const m = /\/ap\/blobs\/([^/?#]+)$/.exec(String(url || ''))
if (!m) return true // not one of ours — not ours to judge
try { return !!(await ctx.fetchBlob(m[1])) } catch { return false }
},
buildActorDoc: async () => buildActor({
host: config.host,
username: config.username,
displayName: config.displayName,
summary: config.summary,
publicKeyPem,
alsoKnownAs: config.alsoKnownAs || [],
profile: await ctx.records.getProfile()
})
})
// Prime WITHOUT delivering: existing posts stay undelivered, only new ones go out.
await publisher.primeWatermark()
await publisher.refreshProfile()
// ★ Reconcile once at boot. The drain is otherwise triggered ONLY by a head advance, so a
// deploy that IMPROVES the projection (images, and later facets/quotes) would sit inert
// until the next unrelated post — the same trap the backfill hit. Cheap and self-limiting:
// one listNotes plus a fingerprint compare, and nothing is sent if nothing changed.
publisher.onHead()
// The primary holds posts for the same reason and needs the same kick when somebody follows.
ctx.onFollowed = () => { if (publisher) publisher.onHead() }
ctx.fetchBlob = createBlobFetcher({ did: config.did, pdsUrl: config.pdsUrl, doFetch: httpClient })
log({ msg: 'fronting DID (keyless read)', did: config.did })
} catch (err) {
log({ msg: 'could not front DID — outbox will be empty (discovery/inbox unaffected)', did: config.did, error: err.message })
}
}
// ⛔ MULTI-USER, AND STRICTLY ADDITIVE. `PerUser` and `Users` have existed and tested since 0.9.51
// with nothing constructing them (hyper-social #64 §8, #65 §7). This is that construction.
//
// The point is a second actor whose records come from FAVELA MARKDOWN rather than from the
// atproto projection — `PerUser.records()` builds a FavelaSource from `user.favelaSocket`, which
// needs no DID at all.
//
// ⛔ THE INCUMBENT DOES NOT MOVE. `/ap/actor` and every already-federated
// `/ap/objects/<rkey>` must stay byte-identical: prsm.space follows this actor and holds 20 of
// its objects, and an actor that changes its id is a stranger to every server that already knows
// it. The primary keeps the FLAT paths and deliberately inherits the process-wide instances; new
// users live under `/ap/users/<name>/`. A test pins that a single-actor registry produces
// identical ids to having no registry at all.
//
// ⚠ AND SWITCHING THE INCUMBENT'S SOURCE IS NOT AN OPTION, which is why this is additive rather
// than a config flip. Both sources mint `objectIdFor({host, rkey})`, but the DID path's rkey is
// an atproto TID and FavelaSource's is `row.name` — the favela record's own name. Same content,
// different ids. Repointing the live actor would orphan all 20 federated objects and read as a
// deletion followed by a repost.
//
// No `users` in config = no registry = today's behaviour exactly, by the router's own fallback.
// ⛔ A REGISTRY IS NEEDED WHENEVER REGISTRATION IS ON, not only when users are listed in config.
// This bridge is env-configured with a single AP_USERNAME and no users array — so gating the
// registry on `config.users` would have left registration enabled and nothing ever admitted: a
// name claimed, a 200 returned, and no actor.
//
// With no list, the registry starts as the configured primary alone. `Users` with one entry
// produces ids identical to having no registry at all, which is pinned by a test — so this stays
// invisible to the incumbent.
const userEntries = Array.isArray(config.users) && config.users.length
? config.users
: (registrations ? [{ username: config.username, primary: true, favelaSocket: config.favelaSocket || null }] : null)
// Hoisted out of the block below so metrics() can read them; null when there is no user registry.
let users = null
let perUser = null
if (userEntries) {
users = new Users(userEntries, config.host)
perUser = new PerUser({
users,
dataDir,
// ⛔ The overlay's Files, so a held user's avatar resolves from THEIR bytes rather than being
// looked for on the primary's PDS, where it is not and never was.
files: holder ? holder.files : null,
readFile: async (p) => fs.readFileSync(p),
log,
// ⛔ AFTER ctx.records is built. The primary INHERITS these; constructing the
// registry earlier handed it a null source, which is the cross-wiring PerUser exists
// to prevent — pointed the wrong way.
// ⛔ THESE KEY NAMES ARE THE CONTRACT. PerUser reads shared.records / shared.followerStore /
// shared.fetchBlob / shared.signer. Passing `followers` instead of `followerStore` — which is
// what shipped — made `_inherits` never match, so the incumbent and every registered user
// read the SAME follower list and each per-user publisher found no recipients. Nothing threw.
shared: { records: ctx.records, followerStore: followers, fetchBlob: ctx.fetchBlob, signer: null }
})
ctx.users = users
Object.assign(ctx, perUser.resolvers())
// Resolve a queued entry's key id back to that user's signer. Scanning the registry is right
// here: the key id IS the actor id plus '#main-key', so this cannot match somebody else by
// accident, and there are a handful of users rather than thousands.
signerByKeyId = (keyId) => {
for (const u of users.all()) {
const s = perUser.signer(u)
if (s && s.keyId === keyId) return s
}
return null
}
/**
* A registration becomes a served actor: hold the feed, add the user.
*
* ⛔ ADDITIVE ONLY. `Users.add` never makes a registrant primary, so the flat paths — and every
* one of the incumbent's already-federated object ids — stay exactly where they are.
*/
const admit = async ({ username, root, feedKey }) => {
const peer = feedKey && holder ? await holder.hold(feedKey) : null
// Minted on first admission, read from disk on every one after — a key that changed would
// break every remote server that had cached it.
const keys = userKeys ? await userKeys.forUser(username) : null
const user = users.add({ username, root, feedKey, peer, keys })
log({ msg: 'admitted', username, held: !!peer, keyed: !!keys })
// Their observation feed exists from admission on, so a node can hold it before the first observation.
if (obsFeeds && registrations) {
registrations.get(username)
.then((reg) => {
if (reg && feedKey) feedRegs.set(String(feedKey).toLowerCase(), { username, generation: reg.generation })
return reg ? obsFeeds.open(username, reg.generation) : null
})
.then((e) => { if (e) log({ msg: 'observation feed', username, key: e.key }) })
.catch((err) => log({ msg: 'observation feed open failed', username, error: err.message }))
}
// ⛔ AND START PUBLISHING. Without this a registered user's posts reach the OUTBOX and nobody
// else: the outbox is a pull surface and no fediverse server backfills one. Every post
// ashatron made was visible at /outbox and delivered to zero followers, because the per-user
// Publisher was built lazily inside refreshUser and the only method ever called on it was
// refreshProfile. Nothing primed a watermark and nothing ever called onHead.
if (peer) await startPublishing(user).catch((err) => log({ msg: 'publisher start failed', username, error: err.message }))
}
ctx.keysFor = (u) => (u && u.keys) || null
/**
* A publisher per registered user, so a profile change reaches their followers.
*
* ⛔ Signed with THEIR key. deliver() takes a per-delivery signer (0.9.52), so this hands each
* user's own signer down — the primary's key against their key id would be rejected by every
* receiver, and the actor would resolve while nothing it sent was believed.
*/
const publishers = new Map()
const publisherFor = (u) => {
if (publishers.has(u.username)) return publishers.get(u.username)
const ids = users.ids(u)
const signer = perUser.signer(u)
const p = new Publisher({
records: perUser.records(u),
// ⛔ The SAME store instance the router writes an inbound Follow into — PerUser memoises per
// username, so a follow accepted at /ap/users/<n>/inbox is a recipient here immediately.
followers: perUser.followers(u),
deliver: (d) => outbound.deliver({ ...d, signer }),
actorId: ids.actor,
newId: () => `${Date.now().toString(36)}-${(seq++).toString(36)}`,
log,
state: new JsonState(join(dataDir, 'publisher-' + u.username + '.json')),
// ⚠ Inherited from the bridge's own setting deliberately. A registered user's history is
// one-shot and self-disarming exactly as the primary's is (see primeWatermark): those rkeys
// enter the delivered-set as they go out, so a restart repeats nothing.
backfill: config.backfill,
buildActorDoc: async () => actorDocFor(ctx, u)
})
publishers.set(u.username, p)
return p
}
/**
* Give a held user a publisher, and an EVENT to drive it.
*
* ⛔ The primary's trigger is DidPeer's `onHead`. A held favela feed had no equivalent, so this
* is it: the feed's writer cores, re-hooked, plus a peer returning. Never a poll — the standing
* rule is that events set the latency floor and a timer would only be expressing absence, which
* a `connection` event already does.
*/
const startPublishing = async (u) => {
const p = publisherFor(u)
// ⛔ PRIME ON EVERY ADVANCE, not once at start. A held feed's log replicates asynchronously,
// so a prime at boot routinely sees an empty listing — and 0.9.76 correctly declined to record
// that, but nothing ever called it again. The publisher stayed unprimed, the backfill never
// ran, and the delivered-set stayed empty with the outbox already serving 16 posts. Measured:
// primed=true/delivered=0 across two deploys.
//
// primeWatermark is cheap and returns early the moment it has genuinely primed, so running it
// per advance costs one listNotes on a feed that has already settled and is the only thing
// that makes "try again when the log arrives" true.
// ⛔ refreshProfile BELONGS IN THE ADVANCE, not before it. It used to run once here and never
// again, so a registered user could edit their name, bio or avatar and no follower would ever
// be told — posts kept flowing, because those ride onHead, which made the gap look like a
// profile-specific bug rather than a missing trigger. Same defect the primary had above.
const advance = async () => {
await p.primeWatermark()
p.onHead()
await p.refreshProfile()
}
// watch() fires once immediately, so this still covers the already-settled case at boot.
if (holder && u.feedKey) holder.watch(u.feedKey, advance)
else await advance()
}
ctx.refreshUser = async (username, force = false) => {
const u = users.get(username)
if (!u || !u.peer) return { ok: false, error: 'not a held user' }
await publisherFor(u).refreshProfile({ force: !!force })
return { ok: true, forced: !!force }
}
ctx.publishUser = async (username) => {
const u = users.get(username)
if (!u || !u.peer) return { ok: false, error: 'not a held user' }
publisherFor(u).onHead()
return { ok: true }
}
// A follow accepted at /ap/users/<n>/inbox drains THAT user's held posts immediately.
ctx.drainFor = (u) => { const p = u && publishers.get(u.username); if (p) p.onHead() }
ctx.onRegistered = admit
// ⛔ Releasing must STOP HOLDING. A bridge that kept replicating a feed whose owner had left
// would go on serving their posts under a name they no longer hold — which is the reuse hazard
// the generation counter exists to prevent, arriving by a different door.
ctx.onReleased = async (username) => {
const u = users.get(username)
if (u && u.feedKey && holder) await holder.release(u.feedKey)
if (userKeys) await userKeys.forget(username) // a burned name must not keep signing
users.remove(username)
log({ msg: 'released', username })
}
// ⛔ THE SWARM DOOR onto the same registration flow. A peer has no outbound HTTP — favela-peer
// has none at all — and both sides are already swarm members, so this is how a handset asks for a
// name without anybody shelling into anything. The rules are `admit.js`; this is transport.
//
// ⚠ On the FeedHolder's swarm, not a second one: it already runs one to hold registered peers'
// feeds, and two swarms in a process is two NAT mappings and two sets of connections for no gain.
if (registrations && holder && holder.swarm) {
try {
const { serveRegistration } = await import('./register-rpc.js')
// observerFor: the user's observation feed + this bridge's observer root, for the signed grant a claim
// returns (hyper-social #82, I1 S3). Opening is idempotent; a failure just omits the grant.
const observerFor = obsFeeds
? async ({ username, generation }) => ({ observationFeed: (await obsFeeds.open(username, generation)).key, observerRoot: obsFeeds.root })
: null
const reg = serveRegistration({ swarm: holder.swarm, registrations, admission, observations, config, onRegistered: ctx.onRegistered, onReleased: ctx.onReleased, observerFor, log })
await reg.ready()
open.reg = reg
// ⛔ PRINTED, because it is what a peer needs in order to find this bridge at all. A key that
// exists and is written nowhere is the favela-android device-key mistake one layer over.
log({ msg: 'registration open over the swarm', bridgeKey: reg.key })
} catch (err) {
// Non-fatal: the HTTP door still works, and a bridge that federates is worth more than one
// that refuses to boot because a second door would not open.
log({ msg: 'swarm registration unavailable', error: (err && err.message) || String(err) })
}
}
// ⛔ AND ON BOOT, or a restart silently unserves everyone who registered. The registry is
// durable; what was held is not, so it has to be re-held from what the registry remembers.
if (registrations) {
for (const reg of await registrations.list()) {
try { await admit(reg) } catch (err) { log({ msg: 'could not re-admit', username: reg.username, error: err.message }) }
}
}
log({ msg: 'multi-user', count: userEntries.length, primary: users.primary && users.primary.username })
}
const server = createServer(ctx)
await new Promise((resolve, reject) => {
server.on('error', reject)
server.listen(config.port, config.bind, resolve)
})
const port = server.address().port
// ⛔ THE OPERATOR SURFACE IS A SECOND LISTENER, ALWAYS LOOPBACK, AND OFF UNLESS ASKED FOR. It must
// not share the public port: that one is proxied to the internet, and this one has NO authentication
// in front of it — reach it over SSH. The same decision favela-surface states for itself: "exposing
// it must be a decision, never a default." It mints nothing; every action on it is a flow that
// already exists (invite, list, release).
//
// ⚠ `adminPort: 0` asks for any free port (tests); null/undefined leaves the surface off.
let adminPort = null
if (config.adminPort != null && config.adminPort !== false) {
const { serveAdmin } = await import('./admin-surface.js')
const { Sessions } = await import('./admin-auth.js')
const { parseForm } = await import('./admin-serve.js')
const { Browsers } = await import('favela-peer/lib/browsers.js')
// ⛔ A TOKEN PER BOOT, HELD ONLY IN MEMORY AND ONLY UNTIL CLAIMED. Writing it to disk would make a
// standing secret out of something needed exactly once; regenerating it every start costs nothing
// because it is meaningless the moment the bridge is claimed.
const bootstrapToken = crypto.randomBytes(16).toString('hex')
// GOTCHA, same as the record store above: openCorestore, never `new Corestore` from a separately
// installed copy — `instanceof Corestore` fails across duplicated module copies.
open.browsersStore = openCorestore(join(dataDir, 'admin-identity'))
const browsers = new Browsers(open.browsersStore, { bootstrap: bootstrapToken })
open.browsers = browsers
await browsers.ready()
// ⛔ PRINTED ONLY WHILE UNCLAIMED, because after that it grants nothing and logging a dead secret
// every boot teaches an operator to ignore the line that matters.
if (await browsers.unclaimed()) {
log({ msg: 'admin surface UNCLAIMED — claim it with this bootstrap token, then it closes for good', bootstrap: bootstrapToken })
}
// `pages`: admin pages another role of the process adds (the favela gateway's status page). A Map the
// caller may fill after start; served behind the same login (lib/admin.js).
const adminCtx = { browsers, sessions: new Sessions(), registrations, inbound, config, log, pages: adminPages, root: new URL('..', import.meta.url).pathname }
const adminServer = http.createServer((req, res) => {
parseForm(req).then((form) => serveAdmin(
{
method: req.method,
path: (req.url || '/').split('?')[0],
// ⚠ Parsed here rather than in the routes, so a handler never sees a raw URL. The timeline
// pages on `offset`, and a page control that reads its own query string is the one place a
// surface starts re-implementing a URL parser.
query: Object.fromEntries(new URLSearchParams((req.url || '').split('?')[1] || '')),
form,
headers: req.headers || {},
// ⛔ FROM THE SOCKET. The recovery route gates on being on this machine, and a header would
// let a caller assert that rather than demonstrate it.
remoteAddress: (req.socket && req.socket.remoteAddress) || null
}, adminCtx))
.then((out) => { res.writeHead(out.status, out.headers); res.end(out.body) })
.catch((err) => { res.writeHead(500, { 'content-type': 'text/plain' }); res.end(String((err && err.message) || err)) })
})
await new Promise((resolve, reject) => {
adminServer.on('error', reject)
// ⛔ ITS OWN BIND, never config.bind — this deployment sets that to 0.0.0.0 for the fediverse
// listener, and inheriting it would publish the operator surface as a side effect. Defaults to
// loopback; widening it is deliberate, and safe because the surface is authenticated.
adminServer.listen(config.adminPort, config.adminBind || '127.0.0.1', resolve)
})
open.adminServer = adminServer
adminPort = adminServer.address().port
// ⛔ STILL LOOPBACK EVEN THOUGH IT NOW HAS AUTH, and for a reason that is not caution:
// static/anproto-client.js hashes with `crypto.subtle`, which browsers expose only in a SECURE
// CONTEXT ("works on localhost + HTTPS" — its own comment). Bound to a public port over plain
// HTTP this would serve a login page whose crypto is undefined in the browser that loaded it.
// Publish it through the proxy that already terminates TLS for this host.
log({ msg: 'admin surface (authenticated)', bind: (config.adminBind || '127.0.0.1') + ':' + adminPort })
// ⛔ THE LOCAL CONTROL SOCKET — the door that cannot be locked out. Losing every operator browser
// otherwise shuts the bridge for good: the bootstrap token gates the FIRST claim only and the
// instance never reopens. Owner-gated by the filesystem, which is the fleet's pattern for exactly
// this (pgv3 gates on ctx.transport; hpp's UDS surface is "TRUSTED WITHOUT the writer-allowlist"
// because only a local process that can open the path connects).
try {
const { serveAdminControl } = await import('./admin-control.js')
const control = await serveAdminControl({ browsers, uds: join(dataDir, 'admin.sock'), log })
open.adminControl = control
log({ msg: 'admin control socket (operator recovery)', uds: control.uds })
} catch (err) {
// ⚠ NOT FATAL, and said out loud. A bridge that refuses to serve the fediverse because its
// recovery socket would not bind has turned a recovery aid into an outage.
log({ msg: 'admin control socket unavailable — recovery from the CLI will not work', error: (err && err.message) || String(err) })
}
}
// ★ The first pass picks up whatever the last run left owed — which is the entire point of the
// queue being durable. Started after listen so a backlog drain never delays the port opening.
outbound.start()
log({
msg: 'ap-bridge listening',
bind: `${config.bind}:${port}`,
actor: `https://${config.host}/ap/actor`,
handle: `@${config.username}@${config.host}`,
followers: await followers.count(),
queued: await outbound.pending()
})
// The callbacks lib/control.js takes, resolved at CALL time: the publisher is created lazily once
// the peer is fronted, so capturing it at boot would capture a null.
const callbacks = {
status: () => ({ ok: true, actor: `https://${config.host}/ap/actor`, did: config.did, port }),
// ⛔ The operator's half of registration. `/ap/register` handles the HOLDER's side; minting the
// token that gates it is not something an HTTP caller should be able to assert, which is the same
// split the release path already makes.
onInvite: registrations ? (() => registrations.invite()) : null,
// What has the fediverse actually sent us? Without a way to ASK, an inbound store is
// indistinguishable from the 202-and-drop it replaced.
// `items` is the raw stored shape, kept for existing callers. `notifications` is the same thing
// projected into favela-peer's notification row (peer.js:646) so a view can MERGE it with the
// peer's own rows instead of building a second, parallel notification screen. Each carries
// source:'fediverse' — its `author` is an actor URI, not a feed key, and a view must not try to
// resolve one as the other.
onNotifications: async ({ limit = 50 } = {}) => {
const items = await inbound.list({ limit })
return { ok: true, items, notifications: toNotifications(items, { limit }) }
},
onReproject: async (opts) => {
if (!publisher) return { ok: false, error: 'publisher not ready (peer not fronted yet)' }
return { ok: true, ...(await publisher.reproject(opts)) }
}
}
/**
* What the gateway's panel and CLI show for this role — read from what the bridge already keeps on
* disk and in memory. Nothing here talks to the network.
*
* `delivered` is the publisher's delivered-set size (rkeys sent at least once), read from the same
* state file the publisher writes. `failing` lists pending entries that have failed at least once,
* newest error first, with the inbox HOST only — an operator needs to see which server is refusing,
* not the activity body.
*/
const metrics = async () => {
const deliveredFor = (u) => {
const file = u.primary ? 'publisher.json' : 'publisher-' + u.username + '.json'
try {
const d = JSON.parse(fs.readFileSync(join(dataDir, file), 'utf8')).delivered
return Array.isArray(d) ? d.length : Object.keys(d || {}).length
} catch { return 0 }
}
const list = users ? users.all() : [{ username: config.username, primary: true }]
const userRows = []
for (const u of list) {
const store = u.primary || !perUser ? followers : perUser.followers(u)
userRows.push({ username: u.username, primary: !!u.primary, held: !!u.peer, followers: await store.count(), delivered: deliveredFor(u) })
}
const pending = await deliveryQueue.list()
const dead = await deliveryQueue.dead()
const hostOf = (inbox) => { try { return new URL(inbox).host } catch { return String(inbox || '') } }
const failing = pending.filter((e) => e.attempts > 0)
.sort((a, b) => (b.nextAt || 0) - (a.nextAt || 0))
.slice(0, 10)
.map((e) => ({ host: hostOf(e.inbox), attempts: e.attempts, lastError: e.lastError, nextAt: e.nextAt }))
const newest = await inbound.list({ limit: 1 })
return {
users: userRows,
queue: {
pending: pending.length,
dead: dead.length,
oldestPendingAt: pending.length ? Math.min(...pending.map((e) => e.createdAt || Date.now())) : null,
failing
},
inbound: { count: await inbound.count(), lastReceivedAt: newest.length ? (newest[0].receivedAt || null) : null },
registrations: registrations ? (await registrations.list()).length : 0
}
}
const closeServer = (s) => new Promise((resolve) => s.close(() => resolve()))
return {
port,
adminPort,
bridgeKey: open.reg ? open.reg.key : null,
// For other roles in this process (the gateway's bsky ingest, hyper-social #82 I2b): a held favela feed by key
// (read-only), and the observation-feed sink for the identity that feed belongs to — null unless it is
// registered here. Narrow on purpose: no registry, no keys, no write path into anyone's own feed.
heldFeed: (feedKey) => (holder ? holder.get(String(feedKey || '').toLowerCase()) : null),
observationsForFeed: (feedKey) => {
const reg = obsFeeds ? feedRegs.get(String(feedKey || '').toLowerCase()) : null
if (!reg) return null
return {
note: (obs) => obsFeeds.note(reg.username, reg.generation, obs),
remove: (activity) => obsFeeds.remove(reg.username, reg.generation, activity)
}
},
...callbacks,
metrics,
handlers: bridgeHandlers({ name: 'ap-bridge', version, ...callbacks }),
// The stop sequence. The delivery drain first — never the queue itself: anything still owed is on
// disk and is picked up by the next boot, which is the difference that module was built to make.
// Then the listeners, then the swarms and stores that hold locks.
steps: [
['outbound', () => outbound.stop()],
['server', () => closeServer(server)],
['admin-server', () => (open.adminServer ? closeServer(open.adminServer) : null)],
['admin-control', () => (open.adminControl ? open.adminControl.close() : null)],
['registration', () => (open.reg ? open.reg.close() : null)],
['observation feeds', () => (obsFeeds ? obsFeeds.close() : null)],
['holder', () => (holder ? holder.close() : null)],
['did-peer', () => (open.didPeer ? open.didPeer.close() : null)],
['did-swarm', () => (open.didSwarm ? open.didSwarm.destroy() : null)],
['did-store', () => (open.didStore ? open.didStore.close() : null)],
['admin-identity', async () => {
if (open.browsers) await open.browsers.close()
if (open.browsersStore) await open.browsersStore.close()
}]
]
}
}
export default startApBridge