// serveMembrane — Phase D (option A): expose the swarm's repo + firehose so the KEYLESS membrane
// (atproto-bridge PeerFront) fronts it UNCHANGED. Each device projects the shared, replicated
// __commit__ log → a local FirehoseSequencer Hypercore + the records → a RepoStore Hypercore, and
// PUSHES an announce (atproto/did-announce/v1) on connect. Because __commit__ is rev-ordered + deterministic,
// every device projects the SAME firehose (same seq), so the membrane can front any device and
// reconnect to another on hand-off. No shared-writer core, no fork hazard.
import b4a from 'b4a'
import c from 'compact-encoding'
import Protomux from 'protomux'
import ProtomuxRPC from 'protomux-rpc'
import { CID } from 'multiformats/cid'
import * as dagCbor from '@ipld/dag-cbor'
import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
import { buildMst } from 'atproto-repo/lib/mst.js'
import { RepoStore } from 'atproto-store/lib/repo-store.js'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
const RPC_PROTOCOL = 'hyper-proto-peer/rpc/v1' // MUST match atproto-bridge PeerFront + peer.serve
const ANNOUNCE_PROTOCOL = 'atproto/did-announce/v1' // MUST match atproto-bridge lib/did-peer.js (corrected discovery)
// A serialized __commit__ entry (swarm-device serializeEntry) → the `write` object commitFrame wants.
function deserializeEntry (e) {
const cid = (s) => (s ? CID.parse(s) : null)
const bytes = (h) => b4a.from(h, 'hex')
const commitBlock = { cid: cid(e.commitCid), bytes: bytes(e.commit) }
const signedCommit = dagCbor.decode(commitBlock.bytes)
return {
commitCid: cid(e.commitCid),
commitBlock,
signedCommit,
dataCid: signedCommit.data,
prevData: cid(e.prevData),
rev: e.rev,
since: e.since,
ops: (e.ops || []).map((o) => ({ action: o.action, path: o.path, cid: cid(o.cid), prev: cid(o.prev) })),
newBlocks: (e.blocks || []).map((b) => ({ cid: cid(b.cid), bytes: bytes(b.bytes) }))
}
}
// Project the device's current LIVE record set into a RepoStore (last-writer-wins by key).
//
// ⚠ NOT the membrane projection path — see projectRepoFrames below for why. This reads THIS NODE's
// view, which is a different source from the __commit__ frames the firehose projection consumes, so
// using it to fill the membrane repo core produces a repo/head pair that does not describe one state.
// Kept because it is the "project the live view" baseline the derive-vs-project equivalence proof
// compares exportCar against (hyper-proto-peer/test/derive-vs-project.test.mjs) — a legitimate,
// different job. It also never deletes, so it can only ever be used against a fresh store.
async function projectRepo (device, repo) {
for (const { collection, rkey, record } of await device.repo.listRecords()) {
await repo.put(collection, rkey, record)
}
}
// The MST data root for a record set — byte-for-byte the computation `materialize` (lib/signer.js)
// and read-view's getRepoFromCommit perform, so agreeing with it here means the consumer agrees too.
// byPath: Map<"collection/rkey", record>. Sorted, because the MST is defined over ordered keys.
function dataRoot (byPath) {
const mstRecords = [...byPath.entries()].map(([key, record]) => ({ key, value: cidForBytesSync(dagCbor.encode(record)) }))
mstRecords.sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0))
return buildMst(mstRecords).root
}
// The data root a serialized __commit__ entry's signed commit commits to.
function entryDataCid (e) {
return dagCbor.decode(b4a.from(e.commit, 'hex')).data
}
// Apply a run of __commit__ frames' ops onto a record set, resolving each create/update to the record
// block the frame ITSELF carries (buildCommitWrite puts them in newBlocks). Race-free: it depends on
// nothing but the frames. Returns the mutated map, or null if a referenced block is absent — a
// partially-replicated log we must NOT pretend to have reconstructed.
//
// ⚠ LOAD-BEARING ASSUMPTION: a #commit frame carries the record bytes for its own create/update ops,
// and __commit__ entries replicate whole (blocks included) as autobase values. That is what lets this
// reconstruct a complete record set with NO byte resolution and no live view — so a co-owner whose
// overlay bytes have not arrived can still serve a complete, self-consistent repo for any head whose
// frames it holds. SwarmDevice.restore/_recordsFromFirehose already depend on the same property.
// hyper-proto-peer's swarm-device.js:78 muses about a future step where "the firehose no longer
// carries them" (record bytes move wholly to the overlay files). Doing that would break BOTH this
// projection and restore(); it would need the projection to resolve bytes through the record store
// first, and would give up the no-byte-resolution property above. Not a plan of record anywhere —
// noted so it is a decision rather than a surprise.
function applyFrames (byPath, frames) {
for (const e of frames) {
const blocks = new Map((e.blocks || []).map((b) => [b.cid, b.bytes]))
for (const op of (e.ops || [])) {
if (op.action === 'delete') { byPath.delete(op.path); continue }
const hex = op.cid ? blocks.get(op.cid) : null
if (!hex) return null
byPath.set(op.path, dagCbor.decode(b4a.from(hex, 'hex')))
}
}
return byPath
}
// Bring the membrane repo core to EXACTLY the record set the log's head frame commits to.
//
// This is the fix for the projectRepo/projectFirehose DRIFT. The two projections are read by a
// keyless consumer as a PAIR: it takes the head off the firehose and asserts the MST rebuilt from
// the repo core equals that head's data root (read-view getRepoFromCommit). projectRepo sourced the
// repo half from `device.repo.listRecords()` — this NODE's live view — while projectFirehose sourced
// the other half from the SHARED __commit__ log, so the invariant held only by luck. Four ways it
// broke, all seen or provable (test/membrane-projection-drift.test.mjs):
// - LAGGING view: a co-owner whose bytes/pointers have not caught up projects a SUBSET. A
// non-signer never adopts the network head and only resolves bytes as a side effect of sign()
// (which returns at the election gate), so it sits there indefinitely — tabertron's device head
// stuck at s222 against a log head of s322, every getRepo 500 (hyper-social #14, incident 3).
// - DELETES: projectRepo only ever put(), never del(), so one signed delete broke the core
// PERMANENTLY regardless of lag.
// - Running AHEAD: an unsigned write in the view projects a SUPERSET.
// - POLLUTION: a core already holding leftovers from the above never self-corrects.
// Deriving BOTH halves from the frames makes the pair correct by construction, for signer and
// non-signer alike, and needs no byte resolution and no live view at all. It is the file's own
// premise ("__commit__ is rev-ordered + deterministic, so every device projects the SAME firehose")
// applied to the repo half.
//
// Incremental first (apply only the not-yet-projected frames onto what the core already holds), then
// VERIFY against the head frame before writing anything. On a mismatch, replay the whole log from
// genesis and re-verify — that heals a core polluted by the old live-view projection, which is the
// state every deployed node is in right now, with no operator step.
//
// Returns { ok: true } when the core now matches the head (the caller may advance the firehose), or
// { ok: false, reason } when it could not be made to match — the caller must then advance NEITHER core
// (an older CONSISTENT pair still serves; an inconsistent one 500s on every read). It never throws:
// a malformed frame is one of the conditions it exists to survive. `reason` distinguishes the faults
// rather than asserting one of them, because a message that names a cause it cannot tell apart from
// another is worse than one that says nothing (hyper-social #14, lesson 2 — "no connected peer served
// this oid" cost two days by calling a transport fault a content fault).
async function projectRepoFrames (repo, log, projected) {
let rows, curBytes, headData, target
try {
rows = await repo.records() // [{ key, bytes, record }], key order
curBytes = new Map(rows.map((r) => [r.key, r.bytes]))
headData = entryDataCid(log[log.length - 1])
target = applyFrames(new Map(rows.map((r) => [r.key, r.record])), log.slice(projected))
if (!target || !dataRoot(target).equals(headData)) {
target = applyFrames(new Map(), log) // full replay from genesis — heals a polluted/mis-seeded core
if (!target) return { ok: false, reason: 'a record block referenced by a #commit op is absent from the log (partial replication)' }
if (!dataRoot(target).equals(headData)) return { ok: false, reason: 'the replayed ops do not reproduce the head commit\'s data root (discontinuous log)' }
}
} catch (e) {
return { ok: false, reason: 'the log head could not be read as a signed #commit: ' + e.message }
}
// Write the DIFF only. Redundant puts are already no-ops (RepoStore opens the bee with
// alwaysDuplicate:false), but not walking them keeps an idle projection genuinely write-free.
for (const [key, record] of target) {
const bytes = dagCbor.encode(record)
const have = curBytes.get(key)
if (have && b4a.equals(b4a.from(have), b4a.from(bytes))) continue
const i = key.indexOf('/')
await repo.put(key.slice(0, i), key.slice(i + 1), record)
}
for (const key of curBytes.keys()) {
if (target.has(key)) continue
const i = key.indexOf('/')
await repo.del(key.slice(0, i), key.slice(i + 1)) // the delete projectRepo never performed
}
return { ok: true }
}
// Append any NOT-yet-projected __commit__ frames into the FirehoseSequencer, in seq order. Idempotent
// (only entries with seq > sequencer.length). Deterministic across devices (same __commit__ order).
async function projectFirehose (device, seq, log = null) {
if (!log) log = await device.firehoseLog() // seq-ordered
for (let i = seq.length; i < log.length; i++) {
const write = deserializeEntry(log[i])
// `time` comes from the frame the SIGNER recorded in the shared __commit__ log, so it is a real
// wall-clock instant AND identical on every device (all reading one replicated fact). Frames
// written before that field existed fall back to deriving it from the rev — correct for those,
// because their revs WERE wall-clock TIDs. Deriving it from a deterministic prev+1µs rev chain
// would freeze `time` near the first commit and drift ever further into the past, which relays
// read as staleness. Frame metadata only — never part of the signed commit.
await seq.sequenceCommit(write, { time: log[i].time || revTime(log[i].rev) })
}
}
// Deterministic ISO time from a TID rev (13-char base32-sortable, micros since epoch in the high bits).
function revTime (rev) {
try {
const A = '234567abcdefghijklmnopqrstuvwxyz'
let n = 0n
for (const ch of String(rev)) n = (n << 5n) | BigInt(A.indexOf(ch))
const micros = n >> 10n // drop the 10-bit clock id
return new Date(Number(micros / 1000n)).toISOString()
} catch { return new Date(0).toISOString() }
}
// serveMembrane — project the stores + PUSH the announce + replicate the corestore over `swarm`.
// Mirrors peer.serve: joins the DID discovery topic + the serverKey topic; on connect, replicates
// the corestore and pushes an announce. Call `sync()` after new signs to project the latest state.
// `projectRepoCore` (default true) — when FALSE the membrane-repo Hyperbee is never opened and
// projectRepo never runs, so record bytes exist ONLY as the content-addressed files. Consumers then
// read the repo via the `get-repo-car` RPC above (derived on demand) or, when co-located, straight off
// the shared overlay files. This is SPEC §6.3's "the virtual device holds nothing".
//
// It is an OPTION rather than a deletion because the announce contract still carries `repoKey`, and
// atproto-did-peer's onAnnounce/holds/_announceTo all gate on it — a peer that stops projecting is
// simply not fronted by an unpatched consumer. Flipping this to false is therefore a COORDINATED
// change with the consumer; shipping the flag first lets the derive-on-read surface land with zero
// behavioural change and be verified in place before anything is removed.
// `labels` — a LabelRoom (or anything with projectTo(core)) for a LABELER DID. Its signed __label__ log
// lives in the StateBase, which only co-owners replicate; a keyless consumer (the gateway answering bsky's
// queryLabels/subscribeLabels) reaches the DID only through this announce. So the log is projected into a
// plain `membrane-labels:<did>` hypercore and named as labelKey/labelLength, the firehose pattern again.
// Absent = no core opened and labelKey:null in the announce (the field exists, so a consumer can branch).
export async function serveMembrane (device, { corestore, swarm, did, handle, sessions = null, blobs = null, projectRepoCore = true, labels = null } = {}) {
if (!corestore || !swarm || !did) throw new Error('serveMembrane: corestore, swarm, did required')
// Step markers. This function is `await`ed during a DID's boot, so anything slow here stalls the whole
// node — and on a phone it stalled with NO output at all, leaving "somewhere inside serveMembrane" as
// the entire diagnosis. Each step is cheap and local on a workstation; the point is that when one is
// not, the log says WHICH. Timed, because "slow" and "wedged" look identical from outside.
const t0 = Date.now()
const step = (msg) => { try { console.log(JSON.stringify({ msg: 'membrane: ' + msg, did, ms: Date.now() - t0 })) } catch {} }
const repo = projectRepoCore ? await new RepoStore(corestore, { name: 'membrane-repo:' + did }).ready() : null
step(projectRepoCore ? 'repo core ready' : 'repo core SKIPPED (virtual device)')
const seq = await new FirehoseSequencer(corestore.get({ name: 'membrane-firehose:' + did })).ready()
step('firehose sequencer ready')
const labelCore = labels ? corestore.get({ name: 'membrane-labels:' + did }) : null
if (labelCore) { await labelCore.ready(); step('label core ready') }
// Incremental and idempotent (LabelRoom.projectTo appends only labels the core lacks). Called by the
// host right after the label signer's tick; returns the count appended.
const syncLabels = async () => (labels && labelCore ? labels.projectTo(labelCore) : 0)
// Project ONLY when the device's __commit__ log has advanced past what the membrane firehose already
// holds. WITHOUT this gate, the caller's sign loop (every signIntervalMs, ~2s) called sync()
// UNCONDITIONALLY, and projectRepo re-put ALL records into the APPEND-ONLY Hyperbee repo core on EVERY
// tick — even with zero new commits — growing it ~one record-set / 2s → 1.3M blocks / GB in hours, a
// runaway storage+CPU leak (favela then re-replicates the bloated core at ~100 MB/s). projectFirehose
// was already incremental; projectRepo was not. After a full projection seq.length === log.length, so
// `log.length <= seq.length` means there is nothing new to project — skip both (read-only, no writes).
// Freshness is a REV comparison, not a length comparison.
//
// The old gate was `if (log.length <= seq.length) return`, which silently assumes the __commit__ log
// only ever GROWS. After a base rebuild / resync the log RESTARTS SHORTER, so a node carrying an old
// membrane (say 83 frames) with a fresh log (say 49) satisfies 49 <= 83 and NEVER PROJECTS AGAIN — it
// advertises a pre-rebuild head forever while its device head is current. Observed live: a wj62
// co-owner stuck at rev 3mr436gzyts22 for four days, still announcing, so the bridge could pick it as
// a front and serve a REV REGRESSION to the relay (which parks a forward-only cursor).
//
// Deliberately NOT auto-repaired here. The membrane firehose is an append-only core that consumers
// replicate; appending the new log's frames after the stale ones would put a rev REGRESSION mid-stream
// — worse than freezing. A correct repair re-bases onto a NEW core (consumers already re-front on a
// key change) or emits #sync, and that is a deliberate change, not something to do inside a sync tick.
// What this DOES do is make it loud, once, naming the DID and both revs — the condition was previously
// invisible from every surface, which is why it survived four days.
let staleWarned = false
let driftWarnedAt = null // the head rev we last refused to project — re-report when the head moves
// The head rev at which this process last VERIFIED the repo core against the firehose head. The
// anti-balloon guard below used to skip on `log.length === seq.length` alone, which also skipped the
// only thing that can notice an ALREADY-INCONSISTENT pair — so a core polluted by the old live-view
// projection (every deployed node, today) would keep 500ing until a new frame happened to arrive, and
// an idle DID would never heal at all. Verifying once per head instead of once per new frame means the
// first sync after a deploy repairs it, and steady state stays free (the check is skipped the moment
// this head is known good). Nothing else writes the core, so a verified head cannot go bad underneath us.
let verifiedAt = null
const sync = async () => {
const log = await device.firehoseLog()
if (!log.length) return
const logHeadRev = log[log.length - 1].rev
if (log.length < seq.length) {
if (!staleWarned) {
staleWarned = true
console.error(JSON.stringify({
msg: 'membrane STALE — the __commit__ log restarted shorter than the projected firehose (base rebuild/resync). This node advertises a PRE-REBUILD head and must not be fronted; it needs a membrane re-base, not an append.',
did, logLength: log.length, membraneLength: seq.length, logHeadRev
}))
}
return
}
// Fully projected AND already verified at this head — nothing to do (the anti-balloon guard).
if (log.length === seq.length && verifiedAt === logHeadRev) return
// The repo core is brought to the head frame's record set FIRST, and the firehose is advanced ONLY
// if that succeeded — the two cores are a pair a consumer reads together, so they advance together
// or not at all. Skipped entirely when the repo core is off (the stateless virtual device derives
// its CAR on demand instead, so there is no pair to keep).
const projection = repo ? await projectRepoFrames(repo, log, seq.length) : { ok: true }
if (!projection.ok) {
if (driftWarnedAt !== logHeadRev) {
driftWarnedAt = logHeadRev
console.error(JSON.stringify({
msg: 'membrane projection REFUSED — the repo core cannot be made to match the record set the log head commits to, so NEITHER core advanced: the previously-served head stays CONSISTENT (degraded) instead of 500ing every read. A co-owner that can reconstruct this head should be fronted instead.',
did, reason: projection.reason, logLength: log.length, membraneLength: seq.length, logHeadRev
}))
}
return
}
verifiedAt = logHeadRev
// No-op when the firehose is already at the log head (the repo-only reconcile above) — its loop
// runs from seq.length to log.length.
await projectFirehose(device, seq, log)
}
await sync()
const serverKey = b4a.toString(swarm.keyPair.publicKey, 'hex')
const repoKey = repo ? b4a.toString(repo.core.key, 'hex') : null
const firehoseKey = b4a.toString(seq.core.key, 'hex')
const labelKey = labelCore ? b4a.toString(labelCore.key, 'hex') : null
// The announce payload, one place. repoKey/repoLength are null when the repo core is off — a consumer
// that requires them simply does not front this peer (it keeps using a co-owner that still projects),
// which is the graceful degradation that makes the flag safe to roll out one node at a time.
const announcement = () => ({ did, handle: handle || null, repoKey, firehoseKey, repoLength: repo ? repo.core.length : null, firehoseLength: seq.core.length, labelKey, labelLength: labelCore ? labelCore.length : null })
const onconn = (conn) => {
corestore.replicate(conn)
const rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json })
// ANNOUNCE-PUSH (atproto/did-announce/v1) is the SOLE discovery path — PUSH {repoKey,firehoseKey,
// lengths} on connect + on each new firehose commit, so a DidPeer consumer learns the cores with NO
// round-trip, NO serverKey directionality (get-coords was removed after the bridge cut over). The
// rpc above stays for the write/session/blob relay. Shares this connection's protomux (one mux per
// conn, keet model). See atproto-bridge lib/did-peer.js + lib/peer-router.js.
try {
const achan = Protomux.from(conn).createChannel({ protocol: ANNOUNCE_PROTOCOL })
if (achan) {
const amsg = achan.addMessage({ encoding: c.json, onmessage: () => {} })
achan.open()
const pushAnnounce = () => { try { amsg.send(announcement()) } catch {} }
pushAnnounce()
const onapp = () => pushAnnounce()
seq.core.on('append', onapp)
// A new label is news to the consumer just like a new commit: re-announce so it extends its replica.
if (labelCore) labelCore.on('append', onapp)
conn.on('close', () => { try { seq.core.off('append', onapp); if (labelCore) labelCore.off('append', onapp) } catch {} })
}
} catch {}
// get-repo-car (com.atproto.sync.getRepo, DERIVED) — the stateless replacement for the
// membrane-repo projection core. device.exportCar() rebuilds the MST + CAR from the record FILES on
// demand and asserts the rebuilt data root equals the signed head's, so it serves nothing that does
// not match its own signature. Nothing is stored to answer this.
//
// WHY: projectRepo copies EVERY record into an append-only Hyperbee (membrane-repo:<did>), which a
// consumer then replicates again — so record bytes live three times (files, this core, the
// consumer's replica) for data the files already hold. SPEC-atproto-overlay-storage §6.3: "the
// virtual device stores nothing of its own — it derives." This is that surface. PUBLIC, like
// get-blob: a repo CAR is public atproto data and is self-verifying against the DID key.
rpc.respond('get-repo-car', async () => {
const r = await device.exportCar()
return { car: b4a.toString(r.carBytes, 'hex'), commitCid: r.commitCid, dataCid: r.dataCid, rev: r.rev, recordCount: r.recordCount }
})
// get-blob (com.atproto.sync.getBlob) — PUBLIC: atproto blobs are public bytes addressed by CID.
// Blobs live in the PEER's BlobStore; the keyless bridge relays this RPC (peer-front.getBlob) and
// streams the bytes back. null → the bridge answers 404. bytes are hex in the c.json RPC value.
if (blobs) {
// list-blobs (com.atproto.sync.listBlobs) — PUBLIC, same reasoning as get-blob: it returns content
// addresses this peer already serves by CID, so it grants no access get-blob does not.
//
// ⛔ THIS SURFACE IS THE ONE THE BRIDGE TALKS TO. hpp's serve-device.js has a SEPARATE responder
// list, and adding the method only there produced exactly the confusing symptom you would expect:
// getBlob worked, listBlobs answered UNKNOWN_METHOD, and the deployed tree demonstrably contained
// the handler. Two surfaces, one patched. Keep them in step.
rpc.respond('list-blobs', async ({ limit, cursor } = {}) => {
if (typeof blobs.list !== 'function') return { cids: [], cursor: null, skipped: 'not-supported' }
return blobs.list({ limit, cursor })
})
rpc.respond('get-blob', async ({ cid } = {}) => {
if (!cid) return null
const b = await blobs.get(String(cid))
return b ? { mimeType: b.mimeType, size: b.size, bytes: b4a.toString(b.bytes, 'hex') } : null
})
}
// bsky-login relay: the bridge relays com.atproto.server.createSession/refreshSession here (it is
// keyless — the PEER verifies the app-password + mints/verifies tokens). Same channel + protocol
// the single-tenant peer served on (lib/peer.js _attachRpc). Gated on sessions being configured.
if (sessions && sessions.canLogin) {
rpc.respond('create-session', async (params) => sessions.createSession(params))
rpc.respond('refresh-session', async ({ refreshJwt } = {}) => sessions.refreshSession(refreshJwt))
// bsky write path: the bridge relays createRecord/putRecord/deleteRecord as apply-write, carrying
// the caller's access token. The PEER (not the keyless bridge) verifies the token, then appends
// the intent to the shared log (lease-holder signs later). Returns { uri, cid } for createRecord:
// the record content CID is computable immediately; the commit rev/seq are not (async sign).
rpc.respond('apply-write', async (op = {}) => {
sessions.verifyAccess(op.token) // throws → CHANNEL error → client 401; only a valid access JWT writes
const { token, action = 'create', collection, rkey, record } = op
const r = await device.applyWrite({ action, collection, rkey, record })
const uri = 'at://' + did + '/' + collection + '/' + rkey
let cid = null
if (record) cid = cidForBytesSync(dagCbor.encode(record)).toString()
return { ...r, uri, cid }
})
// batch write path (com.atproto.repo.applyWrites) — the bsky composer posts via applyWrites,
// not always createRecord. The bridge normalizes each write to { action, collection, rkey,
// record }; we apply them in order + shape the atproto result objects (uri/cid per write).
rpc.respond('apply-writes', async (op = {}) => {
sessions.verifyAccess(op.token)
const results = []
for (const w of (op.writes || [])) {
const { action = 'create', collection, rkey, record } = w
await device.applyWrite({ action, collection, rkey, record })
if (action === 'delete') { results.push({ $type: 'com.atproto.repo.applyWrites#deleteResult' }); continue }
const uri = 'at://' + did + '/' + collection + '/' + rkey
const cid = record ? cidForBytesSync(dagCbor.encode(record)).toString() : null
results.push({ $type: 'com.atproto.repo.applyWrites#' + (action === 'update' ? 'updateResult' : 'createResult'), uri, cid, validationStatus: 'unknown' })
}
return { results }
})
// com.atproto.server.getServiceAuth — the peer mints the ES256K token (it holds the key);
// the keyless bridge attaches it when proxying user-scoped AppView reads. Token-verified.
rpc.respond('get-service-auth', async (op = {}) => {
sessions.verifyAccess(op.token)
return device.getServiceAuth({ aud: op.aud, lxm: op.lxm })
})
// private preferences (birthdate etc.) — token-verified; stored on the peer, out of the public repo.
rpc.respond('get-preferences', async (op = {}) => { sessions.verifyAccess(op.token); return device.getPreferences() })
rpc.respond('put-preferences', async (op = {}) => { sessions.verifyAccess(op.token); return device.putPreferences(op) })
// upload-blob (com.atproto.repo.uploadBlob) — token-gated: a logged-in bsky client uploads an
// avatar/image, the PEER stores it content-addressed + returns the blob ref for the record. bytes
// are hex in the c.json RPC value (the bridge hex-encodes the raw POST body). See #133 pfp.
if (blobs) {
rpc.respond('upload-blob', async ({ bytes, mimeType, token } = {}) => {
sessions.verifyAccess(token)
if (!bytes) throw new Error('upload-blob: bytes (hex) required')
const { cid, size } = await blobs.put(b4a.from(bytes, 'hex'), mimeType || 'application/octet-stream')
return { cid: cid.toString(), mimeType: mimeType || 'application/octet-stream', size }
})
}
}
}
swarm.on('connection', onconn)
// client:true is load-bearing for holepunch, not just discovery: peer and bridge are both
// typically firewalled, and a client:false member never punches outbound so its NAT stays shut.
// Both sides must be client:true to punch simultaneously (see peer.js:serve for the full why).
const dKey = swarm.join(swarm.keyPair.publicKey, { server: true, client: true })
const dDid = swarm.join(discoveryTopic(did), { server: true, client: true })
step('topics joined — serveMembrane returning (the announce flush is the CALLER\'s to await)')
return {
repo, seq, serverKey, repoKey, firehoseKey, sync, labelCore, labelKey, syncLabels, announcement,
ready: () => Promise.all([dKey.flushed(), dDid.flushed()]),
close: () => swarm.off('connection', onconn)
}
}
export { deserializeEntry, projectRepo, projectRepoFrames, projectFirehose, revTime }