atproto-bridge / pds / lib / membrane-reactor.js

// The MEMBRANE feed-registry reactor (community-ux.md §111): watch the durable P2P FeedRegistry and, for each
// feed entry's HOME repo DID, front it over hyperswarm (extend the allow-set + discover) and fire a bsky
// requestCrawl exactly ONCE per newly-fronted DID so the relay re-enumerates and picks up the new repo. No
// per-community config, no restart, no manual curl. This is the membrane counterpart to favela-web's
// lib/feed-reactor.js — favela-web SERVES the feed, the membrane makes it VISIBLE + crawlable to bsky.
//
// Pure + injected (addFront / front / requestCrawl / log stubs) so the reconcile is unit-testable without a
// live swarm or a live bsky relay. front() and requestCrawl() are invoked but NOT awaited: front() discovers
// in the background (a peer offline now is retried on demand, exactly like the boot-time fronts in
// pds-serve.js), and requestCrawl() is outward-facing. The PROD wiring passes requestCrawl=null unless a
// crawl relay host is configured, so this whole mechanism is INERT (no live bsky call) until deployment
// explicitly turns it on.
//
// The registry entry shape is favela-community feed-schema.js: { did, rkey, name, curator, params, roomKey }.
// The membrane keys on `did` only (the feed record's home repo) — curation/roster is favela-web's concern.
const msg = (e) => (e && e.message) || String(e)

export function makeMembraneReactor ({ addFront, front, requestCrawl = null, log = () => {} } = {}) {
  if (typeof addFront !== 'function' || typeof front !== 'function') throw new Error('makeMembraneReactor requires addFront + front')
  const fronted = new Set() // DIDs we've begun fronting (front() itself is a cache; this guards re-work)
  const crawled = new Set() // DIDs we've already requestCrawl'd (fire exactly once per new DID)

  async function reconcile (entries) {
    for (const e of entries || []) {
      const did = e && e.did
      if (!did) continue
      // Front once per DID (front() itself caches; addFront/join is idempotent). Independent of the crawl
      // guard below so a failed requestCrawl can still retry on a later re-fire.
      if (!fronted.has(did)) {
        fronted.add(did)
        addFront(did) // PeerRouter.add — extend the allow-set + join the topic (idempotent)
        // front() discovers + verifies in the BACKGROUND; never block the reactor or throw out of it.
        front(did).then(
          () => log(`[membrane] fronting ${did} (registry)`),
          (err) => log(`[membrane] front failed (retry on demand): ${did} - ${msg(err)}`)
        )
      }
      // federation: ask bsky's relay to re-enumerate so the new DID appears — exactly once per DID. On
      // failure, un-mark so a later reconcile re-fires (the registry watch re-emits its full set), which is
      // why this must NOT sit behind the fronted-guard above.
      if (requestCrawl && !crawled.has(did)) {
        crawled.add(did)
        requestCrawl(did).then(
          () => log(`[membrane] requestCrawl fired: ${did}`),
          (err) => { crawled.delete(did); log(`[membrane] requestCrawl failed (retry): ${did} - ${msg(err)}`) }
        )
      }
    }
  }

  reconcile.fronted = () => [...fronted]
  reconcile.crawled = () => [...crawled]
  return reconcile
}

export default makeMembraneReactor