atproto-bridge / pds / lib / control.js

// Control surface — a hyper-svc Service on the UDS ONLY (no HTTP transport). This is the
// pgh control plane: the readiness-probe target and the `pgh app`/operator API. The
// bsky-facing HTTP+WS membrane (server.js / server-bare.js) is entirely separate.
//
// The membrane is keyless — there are no accounts to administer here. Handlers report what
// the membrane fronts (the self-custody peers) and resolve their handles.
//
// ⛔ The verbs are a plain map (membraneHandlers), separate from the socket that serves them. The
// favela gateway (hyper-social #80) serves them on ITS control socket, beside ap-bridge's verbs, and
// must not bind atproto-bridge's socket or register its name: a release of atproto-bridge pushed to
// a path the gateway served would restart the wrong process.

import { Service } from 'hyper-svc'
import { maintainSidecarRegistration } from './sidecar.js'

export function membraneHandlers ({ peers = null, relay = null } = {}) {
  return {
    status: async () => ({
      ok: true,
      membrane: !relay,          // the blind-relay mode passes `relay` instead of `peers`
      relay: relay ? relay.status() : undefined,
      fronted: peers ? peers.list() : []
    }),

    'list-repos': async () => ({ repos: peers ? peers.list().map((did) => ({ did, active: true })) : [] }),
    'list-fronted': async () => ({ fronted: peers ? peers.list() : [] }),

    // Introspection: a live snapshot of the membrane's swarm + per-front state (see it, don't
    // guess). `diagnose` is read-only + secret-free. `probe-front` forces an on-demand front for
    // one DID and returns the outcome/error — the fast way to see WHY a peer isn't being served.
    diagnose: async () => (peers ? peers.diagnose() : { membrane: true, peers: null }),
    // Recovery: lift the relay's emitted-seq floor above a downstream relay's stored cursor (read via
    // com.atproto.sync.getHostStatus), so it stops treating us as a future-cursor. Monotonic. Then
    // emit-sync + the relay's requestCrawl re-baselines us. See the indigo audit / the post-mortem.
    'set-seq-base': async (params = {}) => {
      if (!peers || !peers.relay) return { ok: false, error: 'no relay' }
      const n = Number(params.seq)
      if (!Number.isFinite(n)) return { ok: false, error: 'numeric seq required' }
      const seqBase = await peers.relay.setSeqBase(n)
      return { ok: true, seqBase, length: peers.relay.length }
    },
    'emit-sync': async (params = {}) => {
      if (!peers) return { ok: false, error: 'no peers router' }
      if (!params.did) return { ok: false, error: 'did required' }
      try { return { ok: true, ...(await peers.emitSync(params.did)) } } catch (e) { return { ok: false, did: params.did, error: e.message } }
    },
    'probe-front': async (params = {}) => {
      if (!peers) return { ok: false, error: 'no peers router' }
      const did = params.did
      if (!did) return { ok: false, error: 'did required' }
      try {
        const f = await peers.front(did)
        const lc = await f.getLatestCommit()
        return { ok: true, did, handle: f.handle || null, firehoseLen: (f.seq && f.seq.core) ? f.seq.core.length : null, latest: lc }
      } catch (e) {
        return { ok: false, did, error: e.message }
      }
    },
    // Eviction: stop serving a wrong head WITHOUT bouncing the process. A cached front outlives its peer's
    // connection by design (its cores keep replicating over the next one), so before these handlers the only
    // way to drop a front announcing a FORKED line — one whose rev sorts higher, so freshest-by-rev keeps
    // picking it — was a restart. `forget: true` also DENIES that firehose key, which is required whenever
    // the reason to evict is what the front is SERVING: peers re-announce on every append, so a plain drop
    // would be undone milliseconds later. `allow-front` is the undo.
    'drop-front': async (params = {}) => {
      if (!peers) return { ok: false, error: 'no peers router' }
      if (!params.did) return { ok: false, error: 'did required' }
      try {
        return { ok: true, ...(await peers.dropFront(params.did, { firehoseKey: params.firehoseKey || null, forget: !!params.forget })) }
      } catch (e) { return { ok: false, did: params.did, error: e.message } }
    },
    'allow-front': async (params = {}) => {
      if (!peers) return { ok: false, error: 'no peers router' }
      if (!params.did) return { ok: false, error: 'did required' }
      try { return { ok: true, ...peers.allowFront(params.did, params.firehoseKey || null) } } catch (e) { return { ok: false, did: params.did, error: e.message } }
    },
    // Drop EVERY cached front + resolved key for a DID so the next request re-fronts from scratch. Nothing is
    // denied and the DID stays fronted — this is the "re-derive, don't restart" lever.
    'forget-did': async (params = {}) => {
      if (!peers) return { ok: false, error: 'no peers router' }
      if (!params.did) return { ok: false, error: 'did required' }
      try { return { ok: true, ...(await peers.forgetDid(params.did)) } } catch (e) { return { ok: false, did: params.did, error: e.message } }
    },

    'resolve-handle': async (params = {}) => ({ did: peers ? peers.resolveHandle(params.handle) : null })
  }
}

export async function startControl ({ uds, name = 'atproto-bridge', version = null, peers = null, relay = null } = {}) {
  const svc = new Service({ name, uds }) // UDS only — omit `http` so we don't bind a port

  for (const [verb, fn] of Object.entries(membraneHandlers({ peers, relay }))) {
    svc.handle(verb, async (ctx, params) => fn(params))
  }

  await svc.listen()

  const stopRegistration = maintainSidecarRegistration({ udsPath: uds, name, runtime: 'bare', version })

  return {
    svc,
    async close () { stopRegistration(); try { await svc.close() } catch {} }
  }
}