/**
* Client — RPC client for hyper-svc daemons.
*
* Local-only. Connects via UDS (preferred) or HTTP. Both transports speak the
* same JSON-RPC-ish protocol; the wire format differs (length-prefixed frames
* for UDS, POST /rpc/<method> for HTTP) but the call signature is identical.
*
* Discovery: try UDS first (if a path is provided and the socket is reachable),
* fall back to HTTP. Explicit transport selection is also supported.
*
* Usage:
*
* const client = await Client.discover({
* uds: '/tmp/pear-git.sock',
* http: 'http://127.0.0.1:4040'
* })
*
* const orgs = await client.call('list-orgs', { limit: 10 })
*
* await client.close()
*/
import http from '#node:http'
import { URL } from '#node:url'
import b4a from 'b4a'
import { stat } from '#node:fs/promises'
import { udsBackend } from './uds-backend.js'
import Hyperswarm from 'hyperswarm'
import Protomux from 'protomux'
import crypto from 'hypercore-crypto'
import { encodeFrame, FrameDecoder } from './frame.js'
import { attachRpcClient } from './transports/swarm-rpc.js'
export class Client {
/**
* Discover and connect to a running hyper-svc daemon.
*
* @param {object} opts
* @param {string} [opts.uds] - UDS socket path. Tried first.
* @param {string} [opts.http] - HTTP base URL (e.g. 'http://127.0.0.1:4040'). Fallback.
* @param {'uds'|'http'} [opts.prefer] - explicit override
* @returns {Promise<Client>}
*/
static async discover (opts = {}) {
const prefer = opts.prefer
if (prefer === 'uds') {
if (!opts.uds) throw new Error('discover({ prefer: "uds" }) requires opts.uds')
return Client.connectUds(opts.uds)
}
if (prefer === 'http') {
if (!opts.http) throw new Error('discover({ prefer: "http" }) requires opts.http')
return Client.connectHttp(opts.http)
}
// Auto: UDS first if reachable, then HTTP
if (opts.uds) {
const reachable = await udsReachable(opts.uds)
if (reachable) return Client.connectUds(opts.uds)
}
if (opts.http) {
return Client.connectHttp(opts.http)
}
const tried = []
if (opts.uds) tried.push(`uds:${opts.uds}`)
if (opts.http) tried.push(`http:${opts.http}`)
throw new Error(`no daemon reachable (tried: ${tried.join(', ') || 'nothing — pass uds and/or http'})`)
}
static async connectUds (path) {
const { backend, name } = await udsBackend(path)
const sock = backend.createConnection(name)
await new Promise((resolve, reject) => {
sock.once('connect', () => {
sock.removeListener('error', reject)
resolve()
})
sock.once('error', reject)
})
return new Client({ kind: 'uds', sock })
}
static connectHttp (base) {
return new Client({ kind: 'http', base: base.replace(/\/+$/, '') })
}
/**
* Connect to a remote daemon via Hyperswarm. Joins the topic derived from
* `serverKey`, waits for a connection, attaches a Protomux RPC channel, and
* (when `opts.verify` is given) confirms the peer that answered is the one
* that was asked for.
*
* ⚠ `serverKey` IS NOT AN IDENTITY — IT IS ONLY A RENDEZVOUS ADDRESS.
*
* The topic is `hash(serverKey + salt)`, and SwarmTransport.listen() creates
* its Hyperswarm with NO keyPair, so a server's Noise key is random per
* process and never equals the `serverKey` clients use to find it. Anything
* joined to that topic as a server can therefore answer, and there is no
* cryptographic binding to check `conn.remotePublicKey` against.
*
* That is not hypothetical. In a pear-git-v3 fleet, a mirror joins its
* ORIGIN's topic (the topic travels in the deployment bundle), so it is a
* server on the exact topic clients use to reach the origin. Measured
* 2026-08-08: `connectSwarm({serverKey: <tabertron>})` issued ON stortron was
* answered by STORTRON ITSELF 6 times out of 6. It asked itself for the
* origin's release index, got its own, and every artifact fetch then failed
* with "no holder" — a wrong ANSWER, not a transport error, so nothing looked
* broken. resync-from-origin on the same path hung twice for 300 s.
*
* `opts.verify` closes that hole at the application layer, where identity
* actually lives. It receives a connected Client and returns truthy to accept.
* Rejected peers are destroyed and the wait continues until `timeout`, so a
* caller reaches the intended daemon or fails loudly, never silently talks to
* the wrong one. It is layered as a callback rather than a built-in `status`
* check because hyper-svc is generic — which handler proves identity is the
* consumer's business, and this stays compatible with servers of any version.
*
* @param {object} opts
* @param {string|Buffer} opts.serverKey - rendezvous key; derives the topic
* @param {string} [opts.topicSalt='pear-git-v3'] - salt for topic derivation
* @param {object} [opts.swarm] - Hyperswarm constructor options (e.g. testnet bootstrap)
* @param {number} [opts.timeout=30000] - total ms to find an ACCEPTED peer
* @param {(client: Client) => Promise<boolean>} [opts.verify] - identity gate
* @returns {Promise<Client>}
*/
static async connectSwarm (opts = {}) {
if (!opts.serverKey) throw new Error('connectSwarm requires opts.serverKey')
const serverKey = typeof opts.serverKey === 'string'
? b4a.from(opts.serverKey, 'hex')
: opts.serverKey
const salt = opts.topicSalt || 'pear-git-v3'
const topic = crypto.hash(b4a.concat([serverKey, b4a.from(salt)]))
const timeout = opts.timeout || 30_000
const verify = typeof opts.verify === 'function' ? opts.verify : null
const swarm = new Hyperswarm(opts.swarm || {})
const discovery = swarm.join(topic, { server: false, client: true })
let rejected = 0
let client = null
try {
client = await new Promise((resolve, reject) => {
let settled = false
// NOTE: onConnection stays attached for the swarm's whole life — it is
// deliberately NOT removed here. Connections can still arrive after a
// peer is accepted, and a NoiseSecretStream with no 'error' listener
// throws on reset, so detaching left later arrivals unguarded and
// crashed the process at teardown with ECONNRESET. The listener dies
// with the swarm when the returned Client is closed.
const finish = (fn, arg) => {
if (settled) return
settled = true
clearTimeout(timer)
fn(arg)
}
const timer = setTimeout(() => {
finish(reject, new Error('swarm-rpc: connection timeout' +
(rejected ? ' (' + rejected + ' peer(s) answered but failed verification — ' +
'the requested serverKey is a topic, not an identity, so other peers on ' +
'that topic can answer)' : '')))
}, timeout)
const onConnection = (c) => {
// Routine, and load-bearing: a NoiseSecretStream with no 'error'
// listener throws on ECONNRESET, and this path DESTROYS connections
// by design (every peer that fails verification, plus any that arrive
// after one is accepted). Without this the first rejected peer takes
// the whole process down. SwarmTransport._onConnection takes the same
// posture server-side.
c.on('error', () => {})
if (settled) { try { c.destroy() } catch {} ; return }
const candidate = new Client({
kind: 'swarm',
swarm,
conn: c,
rpc: attachRpcClient(Protomux.from(c))
})
if (!verify) return finish(resolve, candidate)
// Verified off the critical path: keep listening while this runs, so a
// slow or lying peer cannot stall a good one behind it.
Promise.resolve()
.then(() => verify(candidate))
.then((ok) => {
if (ok) return finish(resolve, candidate)
rejected++
try { c.destroy() } catch {}
})
.catch(() => {
rejected++
try { c.destroy() } catch {}
})
}
swarm.on('connection', onConnection)
discovery.flushed().catch((err) => finish(reject, err))
})
} catch (err) {
// Destroy the swarm on EVERY failure path. It used to be left running: a
// Hyperswarm holds the event loop open, so each failed connect leaked one
// and the process never exited. That is what hung pear-git-v3's
// basic.test.js when a mirror's origin was unreachable — 123/123 tests
// passed, then the run never terminated.
try { await swarm.destroy() } catch {}
throw err
}
return client
}
constructor (opts) {
this._kind = opts.kind
this._sock = opts.sock || null
this._base = opts.base || null
this._swarm = opts.swarm || null
this._conn = opts.conn || null
this._rpc = opts.rpc || null
this._closed = false
// UDS pending request map (id → {resolve, reject})
this._pending = new Map()
this._nextId = 1
if (this._kind === 'uds') {
this._decoder = new FrameDecoder((frame) => this._onFrame(frame))
this._sock.on('data', (chunk) => {
try {
this._decoder.push(chunk)
} catch (err) {
this._failAll(err)
this._down()
}
})
// EVENT-DRIVEN disconnect detection (fleet reconciliation contract). When the peer daemon
// restarts (the common case: an OTA/`pgh app restart` of hpp/pgv3/etc.), the cached client's
// socket dies. Previously 'close' only `_failAll`'d the pending map but NEVER set `_closed`,
// and there was no 'end' handler at all — so `client.closed` stayed FALSE and long-lived
// consumers (favela-web Writer, pgh sidecar lookups) kept handing back a DEAD client, whose
// calls hang or reject "connection closed" forever with nothing to trigger a reconnect. Mark
// the client CLOSED on close/end/error so `.closed` is truthy → consumers reconcile (reconnect
// a fresh Client) on their next call. 'end' is load-bearing under bare-net (allowHalfOpen: on
// peer-end it fires 'end' but not 'close'), so we destroy to force teardown — same posture as
// the server-side half-open fix (hyper-svc 0.1.3).
this._sock.on('close', () => this._down())
this._sock.on('end', () => this._down())
this._sock.on('error', () => this._down())
}
}
get kind () { return this._kind }
get closed () { return this._closed }
/**
* Make an RPC call. Throws on transport errors or remote handler errors.
* @param {string} method
* @param {any} [params]
*/
async call (method, params = {}) {
if (this._closed) throw new Error('client closed')
if (this._kind === 'uds') return this._callUds(method, params)
if (this._kind === 'http') return this._callHttp(method, params)
if (this._kind === 'swarm') return this._rpc.call(method, params)
throw new Error('unknown client kind: ' + this._kind)
}
_callUds (method, params) {
const id = this._nextId++
return new Promise((resolve, reject) => {
this._pending.set(id, { resolve, reject })
try {
this._sock.write(encodeFrame({ id, method, params }))
} catch (err) {
this._pending.delete(id)
reject(err)
}
})
}
_onFrame (frame) {
if (frame == null || (typeof frame.id !== 'number' && typeof frame.id !== 'string')) {
// Malformed response, can't match it to a caller
return
}
const pending = this._pending.get(frame.id)
if (!pending) return
this._pending.delete(frame.id)
if (frame.error) {
pending.reject(new Error(frame.error))
} else {
pending.resolve(frame.result)
}
}
_failAll (err) {
for (const [, pending] of this._pending) pending.reject(err)
this._pending.clear()
}
// The socket died (peer restarted / errored / half-open end). Mark the client CLOSED so any
// consumer's `.closed` check reconnects, fail every in-flight call, and force teardown. Idempotent.
_down () {
if (this._closed) return
this._closed = true
this._failAll(new Error('connection closed'))
if (this._sock) { try { this._sock.destroy() } catch {} }
}
async _callHttp (method, params) {
const url = new URL(`${this._base}/rpc/${encodeURIComponent(method)}`)
return new Promise((resolve, reject) => {
const body = JSON.stringify(params)
const req = http.request({
method: 'POST',
hostname: url.hostname,
port: url.port || 80,
path: url.pathname + url.search,
headers: {
'Content-Type': 'application/json',
'Content-Length': b4a.byteLength(body)
}
}, (res) => {
const parts = []
res.on('data', (c) => parts.push(c))
res.on('end', () => {
const text = b4a.concat(parts).toString('utf8')
let parsed
try {
parsed = JSON.parse(text)
} catch {
return reject(new Error(`bad JSON response (status ${res.statusCode}): ${text.slice(0, 200)}`))
}
if (res.statusCode >= 200 && res.statusCode < 300) {
return resolve(parsed.result)
}
return reject(new Error(parsed.error || `HTTP ${res.statusCode}`))
})
res.on('error', reject)
})
req.on('error', reject)
req.write(body)
req.end()
})
}
async close () {
if (this._closed) return
this._closed = true
if (this._sock) {
this._failAll(new Error('client closed'))
try { this._sock.destroy() } catch {}
this._sock = null
}
if (this._rpc) {
try { this._rpc.close() } catch {}
this._rpc = null
}
if (this._conn) {
try { this._conn.destroy() } catch {}
this._conn = null
}
if (this._swarm) {
try { await this._swarm.destroy() } catch {}
this._swarm = null
}
}
}
async function udsReachable (path) {
const { backend, name, abstract } = await udsBackend(path)
// Pathname sockets: verify the file exists and is actually a socket first.
// Abstract sockets have no filesystem entry — skip straight to the connect probe.
if (!abstract) {
try {
const s = await stat(path)
if (!s.isSocket()) return false
} catch {
return false
}
}
// Try a real connect — covers stale sockets where the daemon crashed
return new Promise((resolve) => {
const sock = backend.createConnection(name)
sock.once('connect', () => {
sock.destroy()
resolve(true)
})
sock.once('error', () => {
sock.destroy()
resolve(false)
})
})
}