// The client half of the worklet protocol — the ONE place that knows the wire format.
//
// Deliberately runtime-agnostic and dependency-free: it takes a `send(string)` and is fed bytes via
// `receive(chunk)`. React Native hands it `react-native-bare-kit`'s IPC stream; a test harness hands it a
// child process's stdio. Neither is imported here, so this file can be tested without a device, an
// emulator, or Metro — which is the point.
//
// ⛔ THE UI MUST NOT SPEAK THE PROTOCOL DIRECTLY. Every screen goes through this client, so the framing,
// the id correlation and the error shape exist once. The moment a component builds its own request object
// the contract has two definitions and they will drift.
//
// Wire format (lib/node.js): line-delimited JSON.
// → { id, op, params }
// ← { id, result } | { id, error }
// ← { t: 'log'|'ready'|'fatal', ... } unsolicited; no id
const DEFAULT_TIMEOUT = 30_000
export class WorkletClient {
/**
* @param {(line: string) => void} send write a line to the worklet
* @param {object} [opts]
* @param {number} [opts.timeout] per-request ms (default 30s)
*/
constructor (send, { timeout = DEFAULT_TIMEOUT } = {}) {
if (typeof send !== 'function') throw new Error('WorkletClient requires send(line)')
this._send = send
this._timeout = timeout
this._seq = 0
this._pending = new Map()
this._buf = ''
this._listeners = { log: [], ready: [], fatal: [], close: [] }
this.ready = false
}
on (event, fn) {
if (!this._listeners[event]) this._listeners[event] = []
this._listeners[event].push(fn)
return this
}
/**
* Remove ONE registration of `fn`, so a screen that subscribes per attempt can unsubscribe.
*
* Removes a single entry rather than every match: `on` is additive and a caller that registered the
* same function twice asked for two subscriptions. The pairing screen subscribes to `log` for each
* pairing attempt; without this, a failed pair followed by a retry leaves the previous listener
* attached and every later line fans out to all of them.
*/
off (event, fn) {
const list = this._listeners[event]
if (!list) return this
const i = list.indexOf(fn)
if (i >= 0) list.splice(i, 1)
return this
}
_emit (event, payload) {
for (const fn of this._listeners[event] || []) {
// A throwing listener must not take down the transport or strand every pending request.
try { fn(payload) } catch { /* a UI callback is not the protocol's problem */ }
}
}
/** Feed bytes from the transport. Partial lines are buffered — chunk boundaries are not message boundaries. */
receive (chunk) {
this._buf += typeof chunk === 'string' ? chunk : String(chunk)
let i
while ((i = this._buf.indexOf('\n')) >= 0) {
const line = this._buf.slice(0, i)
this._buf = this._buf.slice(i + 1)
if (!line.trim()) continue
let msg
// A malformed line is the worklet's bug, not a reason to kill the channel — surface and continue.
try { msg = JSON.parse(line) } catch { this._emit('log', { msg: 'unparseable line from worklet', line }); continue }
this._dispatch(msg)
}
}
_dispatch (msg) {
if (msg && msg.id !== undefined && this._pending.has(msg.id)) {
const { resolve, reject, timer } = this._pending.get(msg.id)
this._pending.delete(msg.id)
clearTimeout(timer)
if (msg.error) reject(new Error(msg.error))
else resolve(msg.result)
return
}
if (msg && msg.t === 'ready') { this.ready = true; this._emit('ready', msg); return }
if (msg && msg.t === 'fatal') { this._emit('fatal', msg); this._failAll(new Error(msg.error || 'worklet fatal')); return }
if (msg && msg.t === 'log') { this._emit('log', msg); return }
this._emit('log', msg)
}
/** Reject everything outstanding. A dead worklet must not leave a screen spinning forever. */
_failAll (err) {
for (const [, p] of this._pending) { clearTimeout(p.timer); p.reject(err) }
this._pending.clear()
}
/** The transport closed. */
close (reason = 'worklet closed') {
this._failAll(new Error(reason))
this.ready = false
this._emit('close', { reason })
}
/** Issue a request. Rejects on worklet error, transport death, or timeout. */
request (op, params = {}) {
const id = ++this._seq
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
this._pending.delete(id)
// Name the op. "timeout" alone sends someone reading a bug report to the wrong place.
reject(new Error(`worklet timeout after ${this._timeout}ms: ${op}`))
}, this._timeout)
this._pending.set(id, { resolve, reject, timer })
try {
this._send(JSON.stringify({ id, op, params }) + '\n')
} catch (e) {
this._pending.delete(id)
clearTimeout(timer)
reject(e)
}
})
}
// Typed wrappers. Thin on purpose — they exist so a screen never types an op string, not to add logic.
status () { return this.request('status') }
/** Mint a fresh favela-web session. Used when /auth/handoff refuses a token the server has forgotten. */
resession () { return this.request('resession') }
collections () { return this.request('collections') }
// App updates. `checkForUpdate` stages at most — the restart that runs a staged bundle is the shell's
// (app/worklet.js), because only the side that owns the runtime can start a different one.
updateStatus () { return this.request('updateStatus') }
checkForUpdate () { return this.request('checkForUpdate') }
// The ported surface. Thin by design — they exist so a screen never types an op string, not to add
// logic; every one of them maps to an appview query or a device write in the worklet.
compose (params) { return this.request('compose', params) }
deletePost (params) { return this.request('deletePost', params) }
follow (params) { return this.request('follow', params) }
invite () { return this.request('invite') }
// Media. Bytes cross as base64 on the same wire (there is no local origin to serve from), so the
// caller gets { found, mimeType, base64 } and builds a data: URI — see app/Post.jsx BlobImage.
blob (params) { return this.request('blob', params) }
// Identities. This node hosts N DIDs (your own + each co-owned community); everything else defaults
// to the first, so a caller that wants another must say which — see `viewer` on the read ops.
hostedDids () { return this.request('hostedDids') }
// Viewing DIDs this node does not host, and the people around them.
viewDid (params) { return this.request('viewDid', params) }
trackedDids () { return this.request('trackedDids') }
untrackDid (params) { return this.request('untrackDid', params) }
syncFollows (params) { return this.request('syncFollows', params) }
devices () { return this.request('devices') }
revokeDevice (params) { return this.request('revokeDevice', params) }
/** Resolve once the worklet says it is up. Boot is slow on a phone — a screen needs to await it. */
whenReady ({ timeout = 120_000 } = {}) {
if (this.ready) return Promise.resolve()
return new Promise((resolve, reject) => {
const timer = setTimeout(() => reject(new Error('worklet did not become ready in ' + timeout + 'ms')), timeout)
this.on('ready', () => { clearTimeout(timer); resolve() })
this.on('fatal', (m) => { clearTimeout(timer); reject(new Error(m.error || 'worklet fatal')) })
})
}
}