/**
* PearDeployer - Wraps pear-ipc for stage/release/seed/info operations
*
* Connects to the shared Pear sidecar via Unix socket.
* All stream methods use the pear-ipc async iterable pattern:
* - Each message is { tag, data }
* - Terminal: tag 'final' with data.success
* - Errors: tag 'error' with data.stack
*/
import path from 'bare-path'
import os from 'bare-os'
const PEAR_DIR = path.join(os.homedir(), '.config', 'pear')
const SOCKET_PATH = path.join(PEAR_DIR, 'pear.sock')
export class PearDeployer {
constructor (opts = {}) {
this.ipc = null
this._socketPath = opts.socketPath || SOCKET_PATH
this._seedStreams = new Map() // channel → active seed stream
this._ready = false
}
async ready () {
if (this._ready) return
// pear-ipc is CommonJS. Dynamic import works for CJS under Bare/Pear
// and avoids createRequire(import.meta.url) which fails under pear://
// URLs (the virtual protocol can't resolve node_modules).
const mod = await import('pear-ipc')
const IPC = mod.default || mod
this.ipc = new IPC.Client({
socketPath: this._socketPath,
connectTimeout: 20_000
})
await this.ipc.ready()
this._ready = true
}
/**
* Consume a pear-ipc stream until a target tag or 'final'
* Returns collected data for the target tag
*/
async _consumeStream (stream, targetTag) {
let result = null
for await (const msg of stream) {
if (msg.tag === 'error') {
const err = new Error(msg.data?.message || msg.data?.stack || 'IPC error')
err.code = msg.data?.code
err.sidecarStack = msg.data?.stack
throw err
}
if (msg.tag === targetTag) {
result = msg.data
}
if (msg.tag === 'final') {
if (msg.data && !msg.data.success) {
throw new Error('Operation failed: ' + JSON.stringify(msg.data))
}
break
}
}
return result
}
/**
* Stage a Pear app directory
* @param {string} dir - Path to app directory
* @param {string} channel - Pear channel name
* @param {object} opts - { dryRun, compact }
* @returns {{ key, version }}
*/
async stage (dir, channel, opts = {}) {
if (!this._ready) throw new Error('PearDeployer not ready')
const stream = this.ipc.stage({
channel,
name: channel,
dir,
dryRun: opts.dryRun || false,
compact: opts.compact || false,
purge: opts.purge || false
})
// Collect addendum (has key + version) and wait for final
let addendum = null
let finalData = null
for await (const msg of stream) {
if (msg.tag === 'error') {
const err = new Error(msg.data?.message || msg.data?.stack || 'Stage error')
err.sidecarStack = msg.data?.stack
throw err
}
if (msg.tag === 'addendum') {
addendum = msg.data
}
if (msg.tag === 'final') {
finalData = msg.data
break
}
}
return {
key: addendum?.key || null,
version: addendum?.version || null,
channel,
success: finalData?.success !== false
}
}
/**
* Release a staged version as the default
* @param {string} channel - Pear channel name
* @returns {{ version, channel, success }}
*/
async release (channel) {
if (!this._ready) throw new Error('PearDeployer not ready')
const stream = this.ipc.release({
channel,
name: channel
})
let releaseData = null
for await (const msg of stream) {
if (msg.tag === 'error') {
const err = new Error(msg.data?.message || msg.data?.stack || 'Release error')
err.sidecarStack = msg.data?.stack
throw err
}
if (msg.tag === 'released' || msg.tag === 'releasing') {
releaseData = msg.data
}
if (msg.tag === 'final') {
break
}
}
return {
version: releaseData?.version || releaseData?.length || null,
channel,
success: true
}
}
/**
* Start seeding a channel (long-running)
* Waits for 'key' and 'announced' tags, then keeps stream open
* @param {string} channel - Pear channel name
* @param {object} opts - { dir, key }
* @returns {{ key, channel }}
*/
async seed (channel, opts = {}) {
if (!this._ready) throw new Error('PearDeployer not ready')
// End any existing seed for this channel
await this.unseed(channel)
const stream = this.ipc.seed({
channel,
name: channel,
dir: opts.dir || null,
key: opts.key || null,
cmdArgs: []
})
this._seedStreams.set(channel, stream)
// Wait for key and announced, but don't consume the whole stream
let key = null
return new Promise((resolve, reject) => {
let announced = false
const timeout = setTimeout(() => {
reject(new Error('Seed timed out waiting for announcement'))
}, 30_000)
const ondata = (msg) => {
if (!msg || typeof msg !== 'object') return
if (msg.tag === 'error') {
clearTimeout(timeout)
stream.off('data', ondata)
const data = msg.data || {}
const errMsg = (data && typeof data === 'object')
? (data.message || data.stack || JSON.stringify(data))
: String(data || 'Seed error')
const err = new Error(errMsg)
reject(err)
return
}
if (msg.tag === 'key') {
key = msg.data
}
if (msg.tag === 'announced') {
announced = true
clearTimeout(timeout)
stream.off('data', ondata)
resolve({ key, channel })
}
}
stream.on('data', ondata)
stream.on('error', (err) => {
clearTimeout(timeout)
reject(err)
})
stream.on('close', () => {
if (!announced) {
clearTimeout(timeout)
reject(new Error('Seed stream closed before announcement'))
}
})
})
}
/**
* Stop seeding a channel
*/
async unseed (channel) {
const stream = this._seedStreams.get(channel)
if (stream) {
if (!stream.destroyed) stream.end()
this._seedStreams.delete(channel)
}
}
/**
* Get info about a channel
* @param {string} channel - Pear channel name
* @returns {{ channel, release, length, name }}
*/
async info (channel) {
if (!this._ready) throw new Error('PearDeployer not ready')
const stream = this.ipc.info({
channel,
name: channel
})
const data = await this._consumeStream(stream, 'info')
return {
channel,
release: data?.release || null,
length: data?.length || null,
name: data?.name || channel
}
}
/**
* Full deploy: stage → release → seed
* @param {string} dir - App directory
* @param {string} channel - Pear channel name
* @param {object} opts - { release: bool, seed: bool }
* @returns {{ staged, released, seeded, duration }}
*/
async deploy (dir, channel, opts = {}) {
const start = Date.now()
const staged = await this.stage(dir, channel)
let released = null
if (opts.release !== false) {
released = await this.release(channel)
}
let seeded = null
if (opts.seed !== false) {
seeded = await this.seed(channel, { dir, key: staged.key })
}
return {
staged,
released,
seeded,
duration: Date.now() - start
}
}
/**
* Check if a channel is being seeded
*/
isSeeding (channel) {
const stream = this._seedStreams.get(channel)
return stream != null && !stream.destroyed
}
/**
* Get list of actively seeded channels
*/
getSeededChannels () {
const channels = []
for (const [channel, stream] of this._seedStreams) {
if (!stream.destroyed) channels.push(channel)
}
return channels
}
async close () {
for (const [channel, stream] of this._seedStreams) {
if (!stream.destroyed) stream.end()
}
this._seedStreams.clear()
if (this.ipc && !this.ipc.closed && !this.ipc.closing) {
await this.ipc.close()
}
this._ready = false
}
}