// favela's XRPC-over-P2P handler — the backend for an atcute Client({ handler }). It reads atproto
// repos over GENUINE P2P via the atproto-did-peer module (announce + replicate + read-view). This is
// the keystone spike's SWAP POINT, filled with real P2P instead of the HTTP stand-in.
//
// MODULE NOTE (per feedback_ask-should-this-be-a-module): the multi-DID orchestration below (a swarm +
// a Map<did,DidPeer>) parallels atproto-bridge's PeerRouter. Kept favela-LOCAL for now; extract a shared
// `DidPeerRouter` module when we migrate the live bridge onto atproto-did-peer (the natural unification).
//
// RUNTIME NOTE: uses URL / Response / fetch (WHATWG). Present under Node; under Bare they need shims
// (lib/bare-globals + bare-fetch) — OR atcute runs UI-side (§4.1 of DESIGN.md, still open). Node-first
// here to prove the P2P integration; the Bare-placement decision is separate.
import Hyperswarm from 'hyperswarm'
import { DidPeer } from 'atproto-did-peer'
import { openCorestore } from 'atproto-store/lib/corestore.js' // GOTCHA: use this, not `new Corestore`
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { decodeDidKey } from 'atproto-identity/lib/did-key.js'
// Default verifyKey resolver: the PLC directory (#atproto key). Injectable for tests / offline.
export async function resolvePlcVerifyKey (did, { plcUrl = 'https://plc.directory', doFetch = fetch } = {}) {
const res = await doFetch(`${plcUrl}/${did}`)
if (!res.ok) throw new Error(`PLC ${res.status} for ${did}`)
const doc = await res.json()
const vm = (doc.verificationMethod || []).find((m) => m.id && m.id.endsWith('#atproto')) || (doc.verificationMethod || [])[0]
if (!vm || !vm.publicKeyMultibase) throw new Error(`no verificationMethod for ${did}`)
return decodeDidKey(`did:key:${vm.publicKeyMultibase}`).pubkey
}
const json = (body, status = 200) => new Response(JSON.stringify(body), { status, headers: { 'content-type': 'application/json' } })
const xrpcErr = (status, error, message) => json({ error, message }, status)
export class P2PSource {
// storage: a dir for the Corestore. resolveVerifyKey: async (did) -> secp256k1 pubkey (default PLC).
// `swarm` is INJECTABLE. A host that already runs a swarm (the hpp daemon on a phone, which is
// holding live connections to the very machines serving these DIDs) must not stand up a second one:
// rediscovering and re-punching from behind a mobile NAT is how "nobody is serving it right now"
// happens for a DID whose front is already connected on the other socket. An injected swarm is NOT
// destroyed by close() — we did not open it.
constructor ({ storage, resolveVerifyKey = resolvePlcVerifyKey, swarm = null } = {}) {
this.corestore = openCorestore(storage)
this._ownSwarm = !swarm
this.swarm = swarm || new Hyperswarm()
this._resolveVerifyKey = resolveVerifyKey
this._peers = new Map() // did -> DidPeer
this._joined = new Set()
// one replicate per connection; every DidPeer attaches to learn its cores (keet model).
this.swarm.on('connection', (conn) => { for (const p of this._peers.values()) p.attach(conn) })
}
// Ensure a live DidPeer for a DID: resolve verifyKey, join its topic, wait for a verified head.
async peer (did, { timeout = 20000 } = {}) {
let p = this._peers.get(did)
if (!p) {
const verifyKey = await this._resolveVerifyKey(did)
p = new DidPeer(this.corestore, { did, verifyKey })
this._peers.set(did, p)
if (!this._joined.has(did)) {
this._joined.add(did)
this.swarm.join(discoveryTopic(did), { client: true, server: true })
await this.swarm.flush()
}
}
await p.ready({ timeout })
return p
}
// The atcute FetchHandler: (pathname, init) => Response. Routes XRPC repo reads to the DidPeer.
handler () {
return async (pathname, init) => {
try {
const url = new URL(pathname, 'http://favela.local')
const nsid = url.pathname.replace(/^\/xrpc\//, '')
const q = url.searchParams
const repo = q.get('repo') || q.get('did')
if (!repo) return xrpcErr(400, 'InvalidRequest', 'repo/did required')
const p = await this.peer(repo)
if (!p.head) return xrpcErr(404, 'NotFound', `no verified head for ${repo}`)
switch (nsid) {
case 'com.atproto.repo.getRecord': {
const rec = await p.getRecord(q.get('collection'), q.get('rkey'))
return rec ? json(rec) : xrpcErr(404, 'RecordNotFound', 'record not found')
}
case 'com.atproto.repo.listRecords': {
// ⛔ DOUBLE-NESTING. `listRecords` already returns the lexicon response shape
// { records, cursor } (atproto-did-peer 0.2.10); wrapping it again emitted
// `{"records":{"records":[…],"cursor":…}}`, so a conformant client read an object where the
// array should be. It never threw — precisely the non-conformance-that-looks-like-success
// the 0.2.10 change existed to remove, reintroduced by its caller.
//
// Pass the paging params through too, or they are dropped on the floor here exactly as they
// were inside listRecords before 0.2.10.
const listed = await p.listRecords(q.get('collection'), {
limit: q.get('limit') ? Number(q.get('limit')) : 50,
reverse: q.get('reverse') === 'true',
cursor: q.get('cursor') || null
})
return json(Array.isArray(listed) ? { records: listed } : listed)
}
case 'com.atproto.repo.describeRepo':
return json({ did: repo, collections: await p.collections(), handleIsCorrect: false })
case 'com.atproto.sync.getLatestCommit': {
const lc = p.getLatestCommit()
return lc ? json(lc) : xrpcErr(404, 'NotFound', 'no commit')
}
default:
return xrpcErr(501, 'MethodNotImplemented', `${nsid} not served by favela's P2P source yet`)
}
} catch (e) {
return xrpcErr(500, 'InternalError', e.message)
}
}
}
async close () {
if (this._ownSwarm) { try { await this.swarm.destroy() } catch {} }
for (const p of this._peers.values()) { try { await p.close() } catch {} }
this._peers.clear()
}
}