// The atproto MEMBRANE transport, BARE edition (bare-http1 + bare-ws). Keyless: holds no Pds
// and no key; every DID it serves is a remote self-custody hyper-proto-peer, fronted via
// PeerRouter/PeerFront. Mirrors lib/server.js (the node edition) endpoint-for-endpoint; the
// only differences are the transport primitives. Runtime is selected in bin/pds-serve.js.
//
// SCOPE: this transport exists ONLY so bsky relays can crawl us over HTTP/WS. It is NOT the
// P2P sync path — peers replicate their repo + firehose Hypercores directly over Hyperswarm.
import http from 'bare-http1'
import https from 'bare-https'
import bareWs from 'bare-ws'
import { tidNow } from 'atproto-store/lib/tid.js'
import { buildRecordProofCar } from './sync-get-record.js'
import { proxyToAppView, isAppViewMethod } from './appview.js'
import { buildMembraneDidDoc } from './did-doc.js'
import { withDeadline } from './deadline.js'
// Ceiling for ANY upstream call (AppView proxy, describeRepo's PLC read). Generous on purpose — this
// exists to kill hangs, not slow-but-live responses. The lens proxy sets its own, much tighter, value.
const DEFAULT_UPSTREAM_MS = 20000
// The caller's bearer token (access/refresh JWT) — passed to the peer, which is the auth authority.
const bearer = (req) => (req.headers.authorization || '').replace(/^Bearer\s+/i, '')
// Decode a JWT's `sub` UNVERIFIED, only to ROUTE (e.g. refreshSession has no body.repo). The peer verifies.
const jwtSub = (jwt) => { try { const p = String(jwt).split('.')[1].replace(/-/g, '+').replace(/_/g, '/'); return JSON.parse(Buffer.from(p, 'base64').toString('utf8')).sub } catch { return null } }
// Upstream client (bare): request → { status, contentType, body:Buffer }. Picks http vs https by the URL
// scheme — the AppView proxy targets https://api.bsky.app, but an upstream may be plain http://, and
// https.request on an http URL hangs on a TLS handshake (the proxy-hang bug).
// ⛔ The deadline is NOT optional. Without it this rejects only on the request's 'error' event, which
// an unreachable HOST never produces: the agent times the socket out and destroys it, and
// client-connection turns that into a bare `req.destroy()` with no error — so the promise dangles and
// the proxied XRPC hangs until the caller gives up (measured: >30s against a blackholed host, vs ~5ms
// to reject a REFUSED connection). A per-request `timeout` option cannot fix it either — the global
// agent merges its own opts last and overwrites it.
function doFetch (url, { method, headers, body, timeoutMs = DEFAULT_UPSTREAM_MS }) {
const mod = String(url).startsWith('http:') ? http : https
let req = null
const call = new Promise((resolve, reject) => {
req = mod.request(url, { method, headers }, (res) => {
const chunks = []
res.on('data', (c) => chunks.push(c))
res.on('end', () => resolve({ status: res.statusCode, contentType: res.headers['content-type'], body: Buffer.concat(chunks) }))
})
req.on('error', reject)
if (body) req.write(body)
req.end()
})
return withDeadline(call, timeoutMs, () => { try { req.destroy() } catch {} })
}
// bare-ws (CJS) exports { Server, Socket }; default-import the exports object, then take each.
const WebSocketServer = bareWs.Server
const WebSocket = bareWs.Socket
// Lexicon-correct JSON: top-level CIDs (format: cid) → strings; CIDs/bytes inside a
// record `value` (dag) → {$link}/{$bytes}.
function serialize (v, inValue = false) {
if (v === null || v === undefined) return v
if (v.asCID === v) return inValue ? { $link: v.toString() } : v.toString()
if (v instanceof Uint8Array) return { $bytes: Buffer.from(v).toString('base64') }
if (Array.isArray(v)) return v.map(x => serialize(x, inValue))
if (typeof v === 'object') {
const o = {}
for (const k of Object.keys(v)) o[k] = serialize(v[k], inValue || k === 'value')
return o
}
return v
}
const CORS = {
'access-control-allow-origin': '*',
'access-control-allow-methods': 'GET,POST,OPTIONS',
'access-control-allow-headers': 'Content-Type,Authorization,atproto-accept-labelers,atproto-proxy,dpop',
'access-control-max-age': '86400'
}
function sendJson (res, status, obj) {
const body = Buffer.from(JSON.stringify(serialize(obj)))
res.writeHead(status, { 'content-type': 'application/json', 'content-length': body.length, ...CORS })
res.end(body)
}
const sendErr = (res, status, error, message) => sendJson(res, status, { error, message })
async function readJson (req) {
const chunks = []
for await (const c of req) chunks.push(c)
return chunks.length ? JSON.parse(Buffer.concat(chunks).toString('utf8')) : {}
}
// Raw request body as a Buffer (uploadBlob posts opaque bytes, not JSON).
async function readBytes (req) {
const chunks = []
for await (const c of req) chunks.push(c)
return Buffer.concat(chunks)
}
// Binary response (getBlob streams the blob bytes back with its stored mime type).
function sendBytes (res, status, contentType, buf) {
res.writeHead(status, { 'content-type': contentType || 'application/octet-stream', 'content-length': buf.length, ...CORS })
res.end(buf)
}
// An HTTP header value must be ISO-8859-1; a non-ASCII character makes the whole response a 500. Our
// diagnostic strings are written for humans and carry em-dashes, so a reason explaining WHY a request
// degraded would itself have taken the response down. Twin of the guard in server.js.
const headerSafe = (s) => String(s).replace(/[^\x20-\x7e]/g, ' ').replace(/\s+/g, ' ').trim().slice(0, 300)
// The membrane serves ONLY fronted peers. `peers` (PeerRouter) owns the fronted set + live
// PeerFronts. serverDid/availableUserDomains answer describeServer (there is no local account).
// `extensions`: XRPC GET methods and WebSocket paths another role of the same process serves through this
// membrane (the gateway's lens: getFeedSkeleton, describeFeedGenerator, queryLabels, subscribeLabels).
// Mutable Maps, exposed as `server.extensions`, so a role can register after the server is listening.
// xrpc: nsid -> async ({ query, req }) => ({ status, body, headers? })
// ws: pathname -> (ws, url, req) => void
// http: pathname -> async ({ req, url, host }) => ({ status, headers, body }) | null (plain GET pages;
// null declines and the request falls through unchanged — e.g. a page served on one host only)
export function createPdsServer ({ peers = null, serverDid = null, availableUserDomains = [], extensions = null } = {}) {
const ext = extensions || { xrpc: new Map(), ws: new Map(), http: new Map() }
if (!ext.http) ext.http = new Map()
const resolveHandle = (h) => (peers ? peers.resolveHandle(h) : null)
const resolveDid = async (idOrHandle) => {
if (!idOrHandle) return null
return idOrHandle.startsWith('did:') ? idOrHandle : await resolveHandle(idOrHandle)
}
const frontIfPeer = async (idOrHandle) => {
if (!peers) return null
const did = await resolveDid(idOrHandle)
return did && peers.isFronted(did) ? await peers.front(did) : null
}
const firehose = (cursor, opts) => (peers && peers.relay) ? peers.relay.subscribe(cursor, opts) : (async function * () {})()
const notHere = (res) => sendErr(res, 404, 'NotFound', 'not served by a keyless membrane (identity lives in the peer)')
// Service-auth token cache — the feed makes many AppView calls per load; without this each would
// round-trip to the peer to mint a token. Keyed did|aud|lxm; reused while ≥30s of life remains.
const svcAuthCache = new Map()
const serviceAuthFor = async (did, aud, lxm, token) => {
const key = did + '|' + aud + '|' + (lxm || '')
const now = Math.floor(Date.now() / 1000)
const hit = svcAuthCache.get(key)
if (hit && hit.exp - now > 30) return hit.token
const front = await frontIfPeer(did)
if (!front) return null
const r = await front.getServiceAuth({ aud, lxm, token })
if (r && r.token) { svcAuthCache.set(key, { token: r.token, exp: r.exp || (now + 300) }); return r.token }
return null
}
// A serviceAuth(aud, lxm) closure bound to THIS request's bearer → its sub (the account being
// viewed). proxyToAppView calls it to authenticate user-scoped reads; null = public fallback.
const appViewServiceAuth = (req) => async (aud, lxm) => {
const tok = bearer(req)
const sub = jwtSub(tok)
if (!sub) return null
return serviceAuthFor(sub, aud, lxm, tok)
}
const server = http.createServer(async (req, res) => {
try {
const url = new URL(req.url, 'http://localhost')
const nsid = url.pathname.replace(/^\/xrpc\//, '')
const q = url.searchParams
if (req.method === 'OPTIONS') {
const reqHeaders = req.headers['access-control-request-headers']
res.writeHead(204, { ...CORS, 'access-control-allow-headers': reqHeaders || CORS['access-control-allow-headers'] })
return res.end()
}
if (req.method === 'GET' && url.pathname === '/.well-known/atproto-did') {
const host = (req.headers.host || '').split(':')[0].toLowerCase()
const did = await resolveHandle(host)
if (!did) return sendErr(res, 404, 'HandleNotFound', 'no such handle')
const body = Buffer.from(did)
res.writeHead(200, { 'content-type': 'text/plain', 'content-length': body.length, ...CORS })
return res.end(body)
}
// did:web resolution: GET /.well-known/did.json → the membrane's own did doc, advertising #bsky_fg
// exactly when a feed skeleton is served here (a registered extension).
// A plain page another role serves (the gateway's public status page). Checked before the XRPC
// dispatch; a handler that returns null declines, and the request continues exactly as before.
if (req.method === 'GET' && ext.http.has(url.pathname)) {
const host = (req.headers.host || '').split(':')[0].toLowerCase()
const out = await ext.http.get(url.pathname)({ req, url, host })
if (out) {
const b = Buffer.from(String(out.body == null ? '' : out.body))
res.writeHead(out.status || 200, { 'content-type': 'text/html; charset=utf-8', ...(out.headers || {}), 'content-length': b.length })
return res.end(b)
}
}
if (req.method === 'GET' && url.pathname === '/.well-known/did.json') {
const doc = buildMembraneDidDoc(serverDid, { feedGenerator: ext.xrpc.has('app.bsky.feed.getFeedSkeleton') })
if (!doc) return sendErr(res, 404, 'NotFound', 'no did:web served here')
return sendJson(res, 200, doc)
}
if (req.method === 'GET') {
switch (nsid) {
case 'com.atproto.server.describeServer': return sendJson(res, 200, { did: serverDid, availableUserDomains })
case 'com.atproto.sync.listRepos': return sendJson(res, 200, { repos: peers ? peers.list().map((did) => ({ did, active: true })) : [] })
case 'com.atproto.sync.getRepoStatus': {
const front = await frontIfPeer(q.get('did'))
if (!front) return sendErr(res, 404, 'RepoNotFound', 'repo not fronted here')
return sendJson(res, 200, front.getRepoStatus())
}
case 'com.atproto.sync.getLatestCommit': {
const front = await frontIfPeer(q.get('did'))
if (!front) return sendErr(res, 404, 'RepoNotFound', 'repo not fronted here')
return sendJson(res, 200, await front.getLatestCommit())
}
// listBlobs: the DID's published blob INVENTORY, relayed from the peer. The referenced set can
// be derived by walking records, so this exists for exactly one thing that cannot: an ORPHAN,
// a blob uploaded and referenced by no record — the class that permanently lost the
// 2026-07-03 image. `since` is accepted and NOT honoured (pointers carry no rev); saying so is
// better than filtering wrongly, and it is reported in the response.
case 'com.atproto.sync.listBlobs': {
const front = await frontIfPeer(q.get('did'))
if (!front) return notHere(res)
if (typeof front.listBlobs !== 'function') return sendErr(res, 501, 'MethodNotImplemented', 'peer front too old to serve list-blobs')
// ⛔ Report why, never a bare InternalError. A peer with no live RPC channel, one too old to
// know the method, and a genuine fault are three different operator problems, and collapsing
// them into 500 is the same silence that hid the OTA bug for four versions.
let r
try {
r = await front.listBlobs({ limit: q.get('limit'), cursor: q.get('cursor') })
} catch (e) {
const msg = (e && e.message) || String(e)
console.error(JSON.stringify({ msg: 'listBlobs: peer call failed', did: q.get('did'), error: msg, code: e && e.code }))
return sendErr(res, 502, 'PeerUnavailable', 'peer could not answer list-blobs: ' + msg)
}
if (!r || r.skipped) return sendErr(res, 501, 'MethodNotImplemented', 'peer has no blob inventory' + (r && r.skipped ? ' (' + r.skipped + ')' : ''))
if (r.error) return sendErr(res, 502, 'PeerUnavailable', 'peer inventory read failed: ' + r.error)
const body = { cids: r.cids || [] }
if (r.cursor) body.cursor = r.cursor
if (q.get('since')) body.sinceIgnored = true
return sendJson(res, 200, body)
}
// getRepo — full CAR, or a PARTIAL one when `since` is given and can be honoured. See the twin
// in server.js for the full reasoning.
//
// ⛔ TWO SURFACES, BOTH PATCHED. `listBlobs` shipped into hpp's serve-device.js while the
// bridge talks to a DIFFERENT responder list, and the symptom was a deployed handler that the
// caller could not reach. This file and server.js are that same pair for the XRPC switch:
// changing one and not the other reproduces the bug exactly.
case 'com.atproto.sync.getRepo': {
const front = await frontIfPeer(q.get('did') || undefined)
if (!front) return sendErr(res, 404, 'RepoNotFound', 'repo not fronted here')
const since = q.get('since') || null
const r = await front.getRepo(since ? { since } : undefined)
const buf = Buffer.from(r.carBytes)
const headers = { 'content-type': 'application/vnd.ipld.car', 'content-length': buf.length, ...CORS }
if (since) {
const honoured = r.partial === true && r.sinceHonoured === true
headers['atproto-repo-partial'] = honoured ? 'true' : 'false'
headers['atproto-since-honoured'] = honoured ? 'true' : 'false'
if (honoured) {
if (r.fromRev) headers['atproto-since-from'] = String(r.fromRev)
if (r.toRev) headers['atproto-since-to'] = String(r.toRev)
} else {
headers['atproto-since-reason'] = headerSafe(r.reason || 'peer front too old to honour since')
}
}
res.writeHead(200, headers)
return res.end(buf)
}
// The PROOF-CARRYING single-record read. Distinct from com.atproto.repo.getRecord below, which
// returns hydrated JSON: no commit, no signature, no MST proof, so nothing about it can be
// verified. This returns a CAR holding the signed commit + the MST blocks along the covering
// path + the record block, which is what lets a consumer verify a record WITHOUT pulling the
// whole repo.
//
// We were missing it entirely, and it is not a niche gap: any client doing verified inbound
// ingest against us — including our own favela lens — got a 404 and could only fall back to
// fetching the ENTIRE repo. Found by pointing our own verifier at our own membrane.
//
// Per the lexicon, the response proves existence OR NON-EXISTENCE, so an absent record still
// returns 200 with the exclusion proof rather than a 404.
case 'com.atproto.sync.getRecord': {
const front = await frontIfPeer(q.get('did') || undefined)
if (!front) return sendErr(res, 404, 'RepoNotFound', 'repo not fronted here')
const collection = q.get('collection')
const rkey = q.get('rkey')
if (!collection || !rkey) return sendErr(res, 400, 'InvalidRequest', 'collection and rkey are required')
try {
const { carBytes } = await front.getRepo()
const proof = await buildRecordProofCar(carBytes, collection, rkey)
const out = Buffer.from(proof.carBytes)
res.writeHead(200, { 'content-type': 'application/vnd.ipld.car', 'content-length': out.length, ...CORS })
return res.end(out)
} catch (err) {
return sendErr(res, 500, 'InternalServerError', 'proof build failed: ' + ((err && err.message) || String(err)))
}
}
case 'com.atproto.repo.getRecord': {
const front = await frontIfPeer(q.get('repo') || q.get('did') || undefined)
if (!front) return sendErr(res, 404, 'RecordNotFound', 'record not found')
const r = await front.getRecord(q.get('collection'), q.get('rkey'))
return r ? sendJson(res, 200, r) : sendErr(res, 404, 'RecordNotFound', 'record not found')
}
case 'com.atproto.repo.listRecords': {
const front = await frontIfPeer(q.get('repo') || q.get('did') || undefined)
if (!front) return sendErr(res, 404, 'RepoNotFound', 'repo not fronted here')
// The lexicon's params were parsed by nobody: limit/reverse/cursor arrived here and were
// dropped, so a paging client looped on a cursor that never advanced. `records` and
// `cursor` are spread through verbatim — cursor absent means "last page".
return sendJson(res, 200, await front.listRecords(q.get('collection'), {
limit: q.get('limit') ? Number(q.get('limit')) : 50,
reverse: q.get('reverse') === 'true',
cursor: q.get('cursor') || null
}))
}
case 'com.atproto.repo.describeRepo': {
const front = await frontIfPeer(q.get('repo') || q.get('did') || undefined)
if (!front) return sendErr(res, 404, 'RepoNotFound', 'repo not fronted here')
return sendJson(res, 200, await front.describeRepo({ doFetch }))
}
case 'com.atproto.identity.resolveHandle': {
const did = await resolveHandle(q.get('handle'))
return did ? sendJson(res, 200, { did }) : sendErr(res, 404, 'HandleNotFound', 'handle not found')
}
case 'app.bsky.actor.getPreferences': {
const front = await frontIfPeer(jwtSub(bearer(req)))
if (!front) return sendErr(res, 401, 'ExpiredToken', 'cannot route getPreferences')
return sendJson(res, 200, await front.getPreferences({ token: bearer(req) }))
}
case 'com.atproto.server.getSession': {
const sub = jwtSub(bearer(req))
const front = sub ? await frontIfPeer(sub) : null
if (!front) return sendErr(res, 401, 'ExpiredToken', 'invalid session')
return sendJson(res, 200, { did: sub, handle: front.handle || null, active: true })
}
case 'com.atproto.server.getServiceAuth': {
// The peer (key-holder) mints the ES256K token; the keyless bridge only relays. Some
// clients call this directly (chat, video); the AppView proxy uses it internally too.
const sub = jwtSub(bearer(req))
const front = sub ? await frontIfPeer(sub) : null
if (!front) return sendErr(res, 401, 'ExpiredToken', 'cannot route getServiceAuth')
const r = await front.getServiceAuth({ aud: q.get('aud'), lxm: q.get('lxm') || undefined, token: bearer(req) })
return sendJson(res, 200, { token: r.token })
}
case 'com.atproto.sync.getBlob': {
// Blobs live in the PEER (self-custody); the keyless bridge relays the read (peer-front.getBlob)
// and streams the bytes back with their stored mime type. 404 if the peer has no such blob.
const front = await frontIfPeer(q.get('did'))
if (!front) return notHere(res)
const blob = await front.getBlob(q.get('cid'))
if (!blob || !blob.bytes) return sendErr(res, 404, 'BlobNotFound', 'blob not found')
return sendBytes(res, 200, blob.mimeType, Buffer.from(blob.bytes, 'hex'))
}
default: {
// Methods another role serves through this membrane (the gateway's lens). ⛔ MUST precede
// isAppViewMethod: getFeedSkeleton/describeFeedGenerator start with app.bsky. and would otherwise
// be proxied to bsky's AppView, which would answer for OUR feed. Always no-store: these are
// dynamic, and a cached skeleton is the "stuck days behind" feed.
const handler = ext.xrpc.get(nsid)
if (handler) {
const out = await handler({ query: q, req })
const b = Buffer.from(JSON.stringify(out.body))
res.writeHead(out.status || 200, { 'content-type': 'application/json', 'content-length': b.length, ...CORS, 'cache-control': 'no-store, no-cache, max-age=0, must-revalidate', ...(out.headers || {}) })
return res.end(b)
}
if (isAppViewMethod(nsid)) return proxyToAppView(req, res, { nsid, method: 'GET', search: url.search, cors: CORS, doFetch, serviceAuth: appViewServiceAuth(req) })
return sendErr(res, 404, 'MethodNotImplemented', nsid)
}
}
}
if (req.method === 'POST') {
// uploadBlob posts opaque BYTES, not JSON — handle it BEFORE readJson (which would choke on
// binary). Blobs live in the peer: relay the raw bytes (hex) + the caller's token; the peer
// verifies + stores content-addressed, returns the blob ref for the record.
if (nsid === 'com.atproto.repo.uploadBlob') {
const sub = jwtSub(bearer(req))
const front = sub ? await frontIfPeer(sub) : null
if (!front) return sendErr(res, 401, 'AuthRequired', 'uploadBlob requires a fronted peer session')
const raw = await readBytes(req)
const mimeType = req.headers['content-type'] || 'application/octet-stream'
const r = await front.uploadBlob({ bytes: raw.toString('hex'), mimeType, token: bearer(req) })
return sendJson(res, 200, { blob: { $type: 'blob', ref: { $link: r.cid }, mimeType: r.mimeType, size: r.size } })
}
const body = await readJson(req)
switch (nsid) {
case 'com.atproto.server.createSession': {
const front = await frontIfPeer(body.identifier)
if (!front) return sendErr(res, 404, 'NotFound', 'no such account on this membrane')
return sendJson(res, 200, await front.createSession(body))
}
case 'com.atproto.server.deleteSession': return sendJson(res, 200, {}) // stateless JWTs
// Writes/sessions relay to the peer (peer-signs-over-RPC): the bridge is keyless, it forwards
// the caller's access token and the PEER verifies it + signs. Routes by body.repo (or the
// token's sub for refresh). See hyper-proto-peer serve-membrane.js apply-write/refresh-session.
case 'com.atproto.server.refreshSession': {
const sub = jwtSub(bearer(req))
const front = sub ? await frontIfPeer(sub) : null
if (!front) return sendErr(res, 401, 'ExpiredToken', 'cannot route refresh token')
return sendJson(res, 200, await front.refreshSession(bearer(req)))
}
case 'com.atproto.repo.createRecord': {
const front = await frontIfPeer(body.repo)
if (!front) return sendErr(res, 404, 'NotFound', 'no such account on this membrane')
const rkey = body.rkey || tidNow()
const r = await front.applyWrite({ action: 'create', collection: body.collection, rkey, record: body.record, token: bearer(req) })
return sendJson(res, 200, { uri: r.uri, cid: r.cid, validationStatus: 'unknown' })
}
case 'com.atproto.repo.putRecord': {
const front = await frontIfPeer(body.repo)
if (!front) return sendErr(res, 404, 'NotFound', 'no such account on this membrane')
const r = await front.applyWrite({ action: 'put', collection: body.collection, rkey: body.rkey, record: body.record, token: bearer(req) })
return sendJson(res, 200, { uri: r.uri, cid: r.cid, validationStatus: 'unknown' })
}
case 'com.atproto.repo.deleteRecord': {
const front = await frontIfPeer(body.repo)
if (!front) return sendErr(res, 404, 'NotFound', 'no such account on this membrane')
await front.applyWrite({ action: 'delete', collection: body.collection, rkey: body.rkey, token: bearer(req) })
return sendJson(res, 200, {})
}
case 'app.bsky.actor.putPreferences': {
// private prefs (birthdate etc.) — routed by the token's sub (no body.repo), stored ON the peer.
const front = await frontIfPeer(jwtSub(bearer(req)))
if (!front) return sendErr(res, 401, 'ExpiredToken', 'cannot route putPreferences')
await front.putPreferences({ preferences: body.preferences, token: bearer(req) })
return sendJson(res, 200, {})
}
case 'com.atproto.repo.applyWrites': {
// The bsky composer posts via applyWrites (batch), not always createRecord. Normalize each
// write ($type suffix → action; value → record; mint a TID rkey for create) then relay to
// the peer (peer-signs). Returns atproto result objects (uri/cid per write).
const front = await frontIfPeer(body.repo)
if (!front) return sendErr(res, 404, 'NotFound', 'no such account on this membrane')
const writes = (body.writes || []).map((w) => {
const kind = String(w.$type || '').split('#')[1] || 'create'
const action = kind === 'delete' ? 'delete' : kind === 'update' ? 'update' : 'create'
const rkey = w.rkey || (action === 'create' ? tidNow() : w.rkey)
return { action, collection: w.collection, rkey, record: w.value }
})
const r = await front.applyWrites({ writes, token: bearer(req) })
return sendJson(res, 200, { results: r.results || [] })
}
case 'com.atproto.server.createAccount':
case 'com.atproto.repo.uploadBlob':
return notHere(res)
default:
if (isAppViewMethod(nsid)) return proxyToAppView(req, res, { nsid, method: 'POST', search: url.search, body, cors: CORS, doFetch, serviceAuth: appViewServiceAuth(req) })
return sendErr(res, 404, 'MethodNotImplemented', nsid)
}
}
return sendErr(res, 405, 'MethodNotAllowed', req.method)
} catch (e) {
return sendErr(res, e.status || 500, e.error || 'InternalError', e.message)
}
})
// subscribeRepos — binary WS firehose from the aggregate relay (bsky-facing only).
// bare-ws sockets are streamx streams: .write(Buffer) emits a BINARY frame.
const wss = new WebSocketServer({ server }, (ws, req) => {
const url = new URL(req.url, 'http://localhost')
// A WebSocket path another role serves (the gateway's lens: subscribeLabels).
const wsHandler = ext.ws.get(url.pathname)
if (wsHandler) { try { wsHandler(ws, url, req) } catch { try { ws.destroy() } catch {} } ; return }
if (url.pathname !== '/xrpc/com.atproto.sync.subscribeRepos') { try { ws.destroy() } catch {} ; return }
const cursor = parseInt(url.searchParams.get('cursor') || '0', 10) || 0
const ac = new AbortController()
ws.on('close', () => ac.abort())
ws.on('error', () => ac.abort())
;(async () => {
try {
for await (const { bytes } of firehose(cursor, { signal: ac.signal })) {
ws.write(Buffer.from(bytes)) // streamx write → BINARY frame
}
} catch { /* client gone / aborted */ }
try { ws.end() } catch {}
})()
})
server.extensions = ext
return server
}