/**
* Hyperswarm transport for hyper-svc.
*
* Joins a topic, and on every connection:
* 1. corestore.replicate(conn) — replicates all hypercores in the corestore
* 2. for each mounted module exposing a `.base` (Autobase consumer), also
* base.replicate(conn) — belt-and-suspenders against the autobase
* replication gotcha (corestore.replicate alone is unreliable in tests
* and we don't want to chase that bug in production)
* 3. for each mounted module exposing `attachProtocol(mux)` (mux-based
* protocols like hyper-overlay), lazily build `Protomux.from(conn)`
* once per connection and call attachProtocol on each consumer.
* Protomux.from reuses any protomux already attached by corestore/autobase
* replication, so channels multiplex cleanly over the same transport.
*
* No RPC over swarm today — modules communicate either via replication or
* via their own mux channels. Adding a new mux-based protocol only requires
* exposing `attachProtocol(mux)` on the mounted module.
*/
import Hyperswarm from 'hyperswarm'
import Protomux from 'protomux'
import b4a from 'b4a'
import { attachRpcServer } from './swarm-rpc.js'
export class SwarmTransport {
/**
* @param {object} opts
* @param {Buffer|string} opts.topic - 32-byte topic to join (hex string or buffer)
* @param {object} opts.corestore - corestore to replicate
* @param {() => Iterable<{ name: string, instance: object }>} opts.mounts - returns the current mount list (called on each connection so late mounts work)
* @param {Function} [opts.dispatch] - RPC dispatch function. When provided, a swarm-RPC Protomux channel is attached to every connection.
* @param {object} [opts.swarm] - hyperswarm options (e.g., bootstrap nodes for testnet)
* @param {boolean} [opts.server=true]
* @param {boolean} [opts.client=true]
*/
constructor (opts = {}) {
if (!opts.topic) throw new Error('SwarmTransport requires opts.topic')
if (!opts.corestore) throw new Error('SwarmTransport requires opts.corestore')
if (typeof opts.mounts !== 'function') throw new Error('SwarmTransport requires opts.mounts (function)')
this._topic = typeof opts.topic === 'string' ? b4a.from(opts.topic, 'hex') : opts.topic
if (this._topic.length !== 32) throw new Error('topic must be 32 bytes')
this._corestore = opts.corestore
this._getMounts = opts.mounts
this._dispatch = opts.dispatch || null
this._swarmOpts = opts.swarm || {}
this._joinAsServer = opts.server !== false
this._joinAsClient = opts.client !== false
this._swarm = null
this._discovery = null
this._listening = false
this._connections = new Set()
}
get topic () { return this._topic }
get listening () { return this._listening }
get swarm () { return this._swarm }
get connections () { return this._connections.size }
async listen () {
if (this._listening) return
this._swarm = new Hyperswarm(this._swarmOpts)
this._swarm.on('connection', (conn, info) => this._onConnection(conn, info))
this._discovery = this._swarm.join(this._topic, {
server: this._joinAsServer,
client: this._joinAsClient
})
await this._discovery.flushed()
this._listening = true
}
/**
* Join an ADDITIONAL topic on this same shared swarm (Keet activateRoom
* pattern — one Hyperswarm, many topics). Used to rendezvous for a
* specific room (per-org / per-repo autobase) by its discoveryKey, so
* peers find each other for that room without needing the daemon's
* identity topic. Replication still rides the single `_onConnection`
* (corestore.replicate + per-mount replicate). Deduped + idempotent.
*
* @param {Buffer|string} topic - 32-byte discoveryKey (hex or buffer)
* @param {object} [opts] - { server=true, client=true }
* @returns {object|null} the PeerDiscovery handle (or existing one)
*/
joinTopic (topic, opts = {}) {
if (!this._swarm) return null // not listening yet
const t = typeof topic === 'string' ? b4a.from(topic, 'hex') : topic
if (!t || t.length !== 32) return null
const key = b4a.toString(t, 'hex')
if (!this._extraTopics) this._extraTopics = new Map()
if (this._extraTopics.has(key)) return this._extraTopics.get(key)
const { server = true, client = true } = opts
const disc = this._swarm.join(t, { server, client })
this._extraTopics.set(key, disc)
return disc
}
/** discoveryKeys of every additional room topic joined via joinTopic. */
get joinedRoomTopics () {
return this._extraTopics ? Array.from(this._extraTopics.keys()) : []
}
_onConnection (conn, info) {
this._connections.add(conn)
conn.on('close', () => this._connections.delete(conn))
conn.on('error', () => { /* connection errors are routine — ignore */ })
// Always replicate the corestore. Hypercore handles muxing internally.
try {
this._corestore.replicate(conn)
} catch (err) {
// Don't crash on replication setup errors — log via stderr
// eslint-disable-next-line no-console
console.error('[hyper-svc/swarm] corestore.replicate failed:', err.message)
}
// Belt and suspenders: also call base.replicate on every mounted Autobase
// consumer. This is the documented workaround for the test-time gotcha
// where corestore.replicate alone doesn't propagate writer cores.
//
// In the same pass, wire up mux-based protocols: any mount exposing
// `attachProtocol(mux)` (hyper-overlay and future consumers) gets attached
// to a single Protomux per connection. Protomux.from reuses any protomux
// corestore/autobase replication already set up.
let mux = null
for (const { name, instance } of this._getMounts()) {
if (!instance) continue
// A mount may expose `replicate(conn)` to replicate its own
// autobase(s) — required for MULTI-autobase mounts (e.g. per-org
// HyperOrgs) whose single `.base` getter is undefined or THROWS.
// Otherwise fall back to the single-autobase `.base.replicate`.
// The whole access is wrapped: a throwing `.base` getter must
// never abort the daemon on an incoming connection.
try {
if (typeof instance.replicate === 'function') {
instance.replicate(conn)
} else {
const base = instance.base
if (base && typeof base.replicate === 'function') base.replicate(conn)
}
} catch (err) {
// eslint-disable-next-line no-console
console.error(`[hyper-svc/swarm] ${name} replicate failed:`, err.message)
}
if (typeof instance.attachProtocol === 'function') {
if (!mux) {
try {
mux = Protomux.from(conn)
} catch (err) {
// eslint-disable-next-line no-console
console.error('[hyper-svc/swarm] Protomux.from failed:', err.message)
}
}
if (mux) {
try {
instance.attachProtocol(mux)
} catch (err) {
// eslint-disable-next-line no-console
console.error(`[hyper-svc/swarm] ${name}.attachProtocol failed:`, err.message)
}
}
}
}
// Swarm-RPC channel — attach only when dispatch is configured.
// Multiplexes cleanly over the same Noise stream as replication
// and module protocols above.
if (this._dispatch) {
if (!mux) {
try {
mux = Protomux.from(conn)
} catch (err) {
// eslint-disable-next-line no-console
console.error('[hyper-svc/swarm] Protomux.from failed (rpc):', err.message)
}
}
if (mux) {
try {
attachRpcServer(mux, this._dispatch, {
remotePublicKey: conn.remotePublicKey || null
})
} catch (err) {
// eslint-disable-next-line no-console
console.error('[hyper-svc/swarm] attachRpcServer failed:', err.message)
}
}
}
}
async close () {
if (!this._swarm) return
for (const conn of this._connections) {
try { conn.destroy() } catch {}
}
this._connections.clear()
try { await this._swarm.destroy() } catch {}
this._swarm = null
this._discovery = null
this._listening = false
}
}