// 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)
}
})
}
/**
* The only typed wrapper kept. Every other op this client can reach is named by the caller.
*
* ⚠ TRIMMED from favela-android's copy, which carried ~15 more — `compose`, `follow`,
* `hostedDids`, `viewDid`, `syncFollows`, `revokeDevice` and the rest. They are a typed API for
* THAT worklet's ops, and calling one here returns "unknown op". The audit that brought this file
* over called it zero-coupling because it imports nothing atproto-shaped, which was true of its
* imports and false of a third of its body — a reminder that "no imports" is not "no coupling".
*/
status () { return this.request('status') }
/** 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')) })
})
}
}