/**
* 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 the server's public key, waits for the first connection, then
* attaches a Protomux RPC channel.
*
* @param {object} opts
* @param {string|Buffer} opts.serverKey - server's Noise public key (hex or Buffer)
* @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] - connection timeout ms
* @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 swarm = new Hyperswarm(opts.swarm || {})
const discovery = swarm.join(topic, { server: false, client: true })
const conn = await new Promise((resolve, reject) => {
const timer = setTimeout(() => {
reject(new Error('swarm-rpc: connection timeout'))
}, timeout)
swarm.on('connection', (c) => {
clearTimeout(timer)
resolve(c)
})
discovery.flushed().catch(reject)
})
const mux = Protomux.from(conn)
const rpc = attachRpcClient(mux)
return new Client({ kind: 'swarm', swarm, conn, rpc })
}
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)
})
})
}