// The atproto MEMBRANE transport — plain HTTP + WS (node edition). Keyless: it holds no Pds
// and no key. Every DID it serves is a remote self-custody hyper-proto-peer, fronted via
// PeerRouter/PeerFront (discovered + verified over the DHT). Reads route to the peer or 404;
// authed/custodial endpoints 404 (identity lives in the peer); subscribeRepos streams the
// aggregate relay. Read/write XRPC as JSON; getRepo as a CAR; subscribeRepos as a binary WS
// firehose. server-bare.js mirrors this endpoint-for-endpoint on bare-http1/bare-ws.
import { createServer } from 'node:http'
import { WebSocketServer } from 'ws'
import { proxyToAppView, isAppViewMethod } from './appview.js'
import { buildMembraneDidDoc } from './did-doc.js'
import { withDeadline } from './deadline.js'
// Twin of the bare edition's ceiling — kills hangs, not slow responses.
const DEFAULT_UPSTREAM_MS = 20000
// AppView upstream client (node): global fetch → { status, contentType, body:Buffer }.
// Deadline for the same reason as the bare edition (see server-bare.js): global fetch has no default
// timeout either, so an unreachable upstream hangs the proxied request instead of failing it. Twin of
// the guard there — changing one and not the other is the "two surfaces" bug this file has hit before.
async function doFetch (url, { method, headers, body, timeoutMs = DEFAULT_UPSTREAM_MS }) {
const ac = new AbortController()
const call = (async () => {
const r = await fetch(url, { method, headers, body, signal: ac.signal })
return { status: r.status, contentType: r.headers.get('content-type'), body: Buffer.from(await r.arrayBuffer()) }
})()
return withDeadline(call, timeoutMs, () => { try { ac.abort() } catch {} })
}
// Lexicon-correct JSON: top-level CIDs (format: cid) → strings; CIDs/bytes INSIDE a
// record `value` (dag) → {$link}/{$bytes}. `inValue` latches once we descend a value.
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
}
// atproto PDSes are browser-facing (the bsky web app is a cross-origin fetch client), so
// the XRPC surface must be CORS-open. Tokens ride in the Authorization header (not cookies),
// so a wildcard origin is correct and safe.
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')) : {}
}
// An HTTP header value must be ISO-8859-1; Node throws ERR_INVALID_CHAR on anything else and the whole
// response becomes a 500. Our own 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 — the failure mode
// this feature exists to remove, reintroduced one layer out. Any explanatory text bound for a header
// goes through here.
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).
export function createPdsServer ({ peers = null, serverDid = null, availableUserDomains = [] } = {}) {
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
}
// subscribeRepos streams the aggregated relay (all fronted peers' signed firehoses, one
// monotonic cursor). No peers/relay → an empty stream.
const firehose = (cursor, opts) => (peers && peers.relay) ? peers.relay.subscribe(cursor, opts) : (async function * () {})()
// Authed/custodial endpoints do not exist on a keyless membrane — identity lives in the peer.
const notHere = (res) => sendErr(res, 404, 'NotFound', 'not served by a keyless membrane (identity lives in the peer)')
const server = 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()
}
// atproto HTTP handle resolution: GET https://<handle>/.well-known/atproto-did → the DID
// of the fronted peer whose handle matches the Host. Host-routed off the fronted set.
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. With the lens-export on
// The Node edition serves no XRPC extensions, so it never advertises a feed generator.
if (req.method === 'GET' && url.pathname === '/.well-known/did.json') {
const doc = buildMembraneDidDoc(serverDid, {}) // the Node edition serves no extensions
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 — the full repo CAR, or a PARTIAL one when `since` is given and can be honoured.
//
// `since` was accepted and thrown away until 0.9.45: this called front.getRepo() with no
// arguments, so it never reached the peer and the identical full CAR came back (measured
// byte-for-byte, 45,563 bytes for wj62). A caller passing it got correct results BY ACCIDENT.
//
// ⛔ THE RESPONSE SAYS WHICH ONE IT IS. A partial CAR is NOT self-contained — the unchanged MST
// nodes are deliberately absent — so a client that walks one as if it were a repo gets a
// missing-block error, and a client that assumes it is complete is simply wrong. The
// distinction therefore travels in headers rather than being left for the reader to infer:
// atproto-repo-partial: true|false
// atproto-since-honoured: true|false (+ atproto-since-reason when false)
// A front too old to know about `since` returns no `partial` field at all, which reads as false
// — so an un-upgraded peer degrades to "full CAR, since not honoured", stated, never silent.
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)
}
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')
}
// authed/custodial reads — identity lives in the peer (session/write routing is deferred)
case 'com.atproto.sync.getBlob':
case 'com.atproto.server.getSession':
case 'com.atproto.server.getServiceAuth':
case 'app.bsky.actor.getPreferences':
return notHere(res)
default:
// hybrid seam: unhandled app.bsky.*/chat.bsky.* reads proxy (public) to the AppView
if (isAppViewMethod(nsid)) return proxyToAppView(req, res, { nsid, method: 'GET', search: url.search, cors: CORS, doFetch })
return sendErr(res, 404, 'MethodNotImplemented', nsid)
}
}
if (req.method === 'POST') {
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 / account admin — the peer's job (routing deferred)
case 'com.atproto.server.refreshSession':
case 'com.atproto.server.createAccount':
case 'com.atproto.repo.uploadBlob':
case 'com.atproto.repo.applyWrites':
case 'com.atproto.repo.createRecord':
case 'com.atproto.repo.putRecord':
case 'com.atproto.repo.deleteRecord':
case 'app.bsky.actor.putPreferences':
return notHere(res)
default:
if (isAppViewMethod(nsid)) return proxyToAppView(req, res, { nsid, method: 'POST', search: url.search, body, cors: CORS, doFetch })
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 (cursor=0 backfills history)
const wss = new WebSocketServer({ noServer: true })
server.on('upgrade', (req, socket, head) => {
const url = new URL(req.url, 'http://localhost')
if (url.pathname !== '/xrpc/com.atproto.sync.subscribeRepos') { socket.destroy(); return }
const cursor = parseInt(url.searchParams.get('cursor') || '0', 10) || 0
wss.handleUpgrade(req, socket, head, (ws) => {
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 })) {
if (ws.readyState !== ws.OPEN) break
ws.send(bytes, { binary: true })
}
} catch { /* client gone / aborted */ }
try { ws.close() } catch {}
})()
})
})
return server
}