atproto-bridge / pds / lib / fleet-serve.js

// Fleet serve (§11 job 2, B-serve): the HTTP↔Hyperswarm read path. PROTOTYPE.
//
// bsky relays can only reach us over HTTP; home instances are NAT'd/certless and reachable
// ONLY over Hyperswarm. The edge bridges: given a DID → serverKey (from FleetDirectory), it
// opens a Hyperswarm connection to that home instance and pulls the repo as a CAR — which it
// then serves to bsky over HTTP. A P2P-backed CDN: HTTP front, Hyperswarm back,
// content-addressed so any replica is equivalent.
//
// Transport = the ecosystem's daemon↔daemon RPC (hyper-svc swarm-rpc over Protomux), the
// same primitive pear-git-v3 uses for `--peer`. Rendezvous = the home joins a topic equal to
// its own swarm public key (its "serverKey"); the edge joins the same 32 bytes and connects
// straight to it. (json-encoded RPC → the CAR rides base64 for the prototype; production
// would stream binary over a dedicated protomux channel — noted, not blocking.)

import Protomux from 'protomux'
import b4a from 'b4a'
import { attachRpcServer, attachRpcClient } from 'hyper-svc'

// Home side: answer repo reads for the DIDs this instance hosts, over every swarm connection.
// Returns { serverKey, ready(), close() }. `swarm` is a Hyperswarm this instance owns.
export function serveRepos (pds, swarm) {
  const dispatch = async (method, params = {}) => {
    if (method === 'get-repo') {
      const { carBytes } = await pds.getRepo({ did: params.did })
      return { car: b4a.toString(b4a.from(carBytes), 'base64') }
    }
    if (method === 'list-repos') return pds.listRepos()
    throw new Error(`unknown method: ${method}`)
  }

  const servers = new Set()
  const onconn = (conn) => {
    const s = attachRpcServer(Protomux.from(conn), dispatch)
    if (s) { servers.add(s); conn.once('close', () => servers.delete(s)) }
  }
  swarm.on('connection', onconn)

  const topic = swarm.keyPair.publicKey
  const discovery = swarm.join(topic, { server: true, client: false })

  return {
    serverKey: b4a.toString(topic, 'hex'),
    ready: () => discovery.flushed(),
    async close () { swarm.off('connection', onconn); for (const s of servers) s.close() }
  }
}

// One persistent RPC client per connection — the channel stays open for the connection's
// life and multiplexes many request/response pairs by id (closing it per-call would tear
// down the shared channel and break the next call).
const clients = new WeakMap()
function clientFor (conn) {
  let c = clients.get(conn)
  if (!c) {
    c = attachRpcClient(Protomux.from(conn))
    clients.set(conn, c)
    conn.once('close', () => clients.delete(conn))
  }
  return c
}

// Edge side: fetch a repo (as CAR bytes) from the home instance identified by serverKey.
export async function fetchRepo (swarm, serverKeyHex, did, { timeout = 15000 } = {}) {
  const conn = await connectToServer(swarm, b4a.from(serverKeyHex, 'hex'), timeout)
  const { car } = await clientFor(conn).call('get-repo', { did })
  return new Uint8Array(b4a.from(car, 'base64'))
}

// Hot failover (§11): try candidate home instances (from the directory + known replicas) in
// preference order; succeed as long as ONE has the repo. Repos are content-addressed, so
// every replica returns a BYTE-IDENTICAL CAR — which one answers is irrelevant. The edge
// holds no authoritative state, so a home going offline just means fetching from another
// replica. (Sequential preserves preference; production could race candidates for latency.)
export async function fetchRepoFailover (swarm, serverKeys, did, { timeout = 8000 } = {}) {
  if (!serverKeys || !serverKeys.length) throw new Error('fetchRepoFailover: no candidate serverKeys')
  let lastErr
  for (const serverKey of serverKeys) {
    try { return await fetchRepo(swarm, serverKey, did, { timeout }) } catch (e) { lastErr = e }
  }
  throw new Error(`fetchRepoFailover: all ${serverKeys.length} candidates failed (${lastErr && lastErr.message})`)
}

// Connect straight to a known peer by its swarm public key (== the rendezvous topic).
export function connectToServer (swarm, serverKey, timeout) {
  for (const conn of swarm.connections) {
    if (b4a.equals(conn.remotePublicKey, serverKey)) return Promise.resolve(conn)
  }
  return new Promise((resolve, reject) => {
    const to = setTimeout(() => { cleanup(); reject(new Error('fetchRepo: connect timeout')) }, timeout)
    const onconn = (conn) => { if (b4a.equals(conn.remotePublicKey, serverKey)) { cleanup(); resolve(conn) } }
    const cleanup = () => { clearTimeout(to); swarm.off('connection', onconn) }
    swarm.on('connection', onconn)
    // symmetric join: both endpoints are typically firewalled, and a client:false member never
    // punches outbound (see peer-front.js:discover for the full why). Match the proven mesh.
    swarm.join(serverKey, { client: true, server: true })
    swarm.flush().catch(() => {})
  })
}