atproto-bridge / pds / test / membrane.test.js

// The HTTP membrane routes a fronted DID to its remote hyper-proto-peer (keyless), over real HTTP.
// A bsky-shaped client hits getRepo/getRecord/listRecords/getLatestCommit/getRepoStatus/
// createSession for a DID the bridge fronts; each is served from the replicated peer (or the
// peer mints, for createSession) — the local Pds has no such account. This is the membrane.

import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import b4a from 'b4a'
import c from 'compact-encoding'
import Hyperswarm from 'hyperswarm'
import createTestnet from '@hyperswarm/testnet'
import ProtomuxRPC from 'protomux-rpc'
import { pushAnnounce } from './serve-announce.js'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { RepoStore } from 'atproto-store/lib/repo-store.js'
import { RepoManager } from 'atproto-store/lib/repo-manager.js'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { keyFromPrivate } from 'atproto-identity'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { createPdsServer } from '../lib/server.js'
import { DidPeerRouter } from 'atproto-did-peer'

const DID = 'did:plc:membrane'
const HANDLE = 'membrane.7rnx.net'
const PRIV = new Uint8Array(32).fill(7)
const NOW = '2026-07-03T00:00:00.000Z'

test('HTTP membrane routes a fronted DID to its peer (reads + createSession)', async (t) => {
  const testnet = await createTestnet(3)

  // a discoverable peer: signed repo + firehose + announce push + create-session responder
  const peerDir = mkdtempSync(join(tmpdir(), 'mem-peer-'))
  const peerStore = openCorestore(peerDir)
  const repo = await new RepoStore(peerStore, { name: `atproto-repo:${DID}` }).ready()
  let micros = 1_700_000_000_000_000n
  const mgr = await new RepoManager(repo, { did: DID, privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init()
  const seq = await new FirehoseSequencer(peerStore.get({ name: 'firehose' })).ready()
  const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3a', record: { $type: 'app.bsky.feed.post', text: 'via membrane', createdAt: NOW } })
  await seq.sequenceCommit(w, { time: NOW })
  const peerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
  peerSwarm.on('connection', (conn) => {
    peerStore.replicate(conn)
    const rpc = new ProtomuxRPC(conn, { protocol: 'hyper-proto-peer/rpc/v1', valueEncoding: c.json })
    pushAnnounce(conn, { did: DID, handle: HANDLE, repo, seq })
    rpc.respond('create-session', async ({ identifier }) => ({ did: DID, handle: HANDLE, accessJwt: `access-${identifier}`, refreshJwt: 'refresh' }))
  })
  await peerSwarm.join(discoveryTopic(DID), { server: true, client: false }).flushed()

  // bridge: a keyless membrane — the shared DidPeerRouter fronting DID + the HTTP membrane (no Pds).
  // Constructed EXACTLY as pds-serve.js does (owns-swarm + join + rpc), so this verifies the prod repoint.
  const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'mem-router-')))
  const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
  const router = new DidPeerRouter(routerStore, {
    swarm: routerSwarm,
    join: (did) => { try { routerSwarm.join(discoveryTopic(did), { server: true, client: true }); routerSwarm.flush().catch(() => {}) } catch {} },
    fronted: [DID],
    resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey,
    rpc: true
  })
  const server = createPdsServer({ peers: router, serverDid: 'did:web:pds.7rnx.net' })
  await new Promise((r) => server.listen(0, '127.0.0.1', r))
  const base = `http://127.0.0.1:${server.address().port}`
  const xrpc = (m) => `${base}/xrpc/${m}`

  t.teardown(async () => {
    server.close(); await router.close()
    await peerSwarm.destroy(); await routerSwarm.destroy()
    await seq.close(); await repo.close(); await peerStore.close(); await routerStore.close()
    await testnet.destroy()
    rmSync(peerDir, { recursive: true, force: true })
  })

  const carBuf = new Uint8Array(await (await fetch(xrpc('com.atproto.sync.getRepo') + '?did=' + DID)).arrayBuffer())
  t.ok(carBuf.length > 0, 'membrane getRepo returns the fronted peer CAR (keyless)')

  const rec = await (await fetch(xrpc('com.atproto.repo.getRecord') + '?repo=' + DID + '&collection=app.bsky.feed.post&rkey=3a')).json()
  t.is(rec.value.text, 'via membrane', 'membrane getRecord routed to the peer')

  const list = await (await fetch(xrpc('com.atproto.repo.listRecords') + '?repo=' + DID + '&collection=app.bsky.feed.post')).json()
  t.ok(list.records.some((r) => r.value.text === 'via membrane'), 'membrane listRecords routed to the peer')

  const status = await (await fetch(xrpc('com.atproto.sync.getRepoStatus') + '?did=' + DID)).json()
  t.is(status.active, true, 'membrane getRepoStatus routed to the peer')

  const latest = await (await fetch(xrpc('com.atproto.sync.getLatestCommit') + '?did=' + DID)).json()
  t.ok(latest.cid && latest.rev, 'membrane getLatestCommit routed to the peer')

  const sess = await (await fetch(xrpc('com.atproto.server.createSession'), { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ identifier: DID, password: 'x' }) })).json()
  t.is(sess.accessJwt, `access-${DID}`, 'membrane createSession routed to the peer (peer minted)')
})