import { join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { Store } from './store.mjs'; import { Broker, BusError } from './broker.mjs'; import { Credentials } from './credentials.mjs'; import { serve } from './server.mjs'; import { verifyHuman } from './human.mjs'; // Trusted host API. S1 supplies resolved definitions, S6 supplies launch records. // Neither a business-file writer nor a socket-accessible configuration endpoint. // `tasks` is S3's adapter factory: ({broker, credentials, businesses}) => {handle, timeout, close}. export async function startBroker({ dataRoot, businesses, launches = [], readers = [], repoRoots = [], tasks = null, }) { let store, credentials, server, adapter; try { const references = {}; for (const [business, b] of Object.entries(businesses)) { for (const [role, r] of Object.entries(b.roles)) if (r.credentials) references[business + '/' + role] = r.credentials; if (b.tracker?.sync?.credentials) references[business + '/@sync'] = b.tracker.sync.credentials; } const protectedRoots = [ fileURLToPath(new URL('../../../', import.meta.url)), ...repoRoots, ...Object.values(businesses).flatMap((b) => Object.values(b.projects ?? {}).map((p) => p.root)), ]; credentials = new Credentials({ references, repoRoots: protectedRoots, dataRoot }); store = new Store(dataRoot); const broker = new Broker({ store, businesses, secretCheck: (value) => credentials.assertClean(value) }); // Process identities must be supplied for every managed run when using the human transport. const launchRecords = []; function bindLaunch(r) { if ( !Number.isSafeInteger(r.pid) || r.pid < 2 || typeof r.startTime !== 'string' || !/^\d+$/.test(r.startTime) ) throw new BusError('invalid-launch-process'); const cap = broker.bindLaunch(r); launchRecords.push({ ...r }); return { business: r.business, run: r.run, cap }; } const bound = launches.map(bindLaunch); const readCaps = readers.map((business) => ({ business, cap: broker.bindReader({ business }) })); if (tasks) { adapter = await tasks({ broker, credentials, businesses }); if (typeof adapter?.handle !== 'function' || !Number.isSafeInteger(adapter.timeout) || adapter.timeout < 1) throw new BusError('invalid-adapter'); } const path = join(store.directory, 'broker.sock'); server = await serve({ tasks: adapter ? { handle: adapter.handle, timeout: adapter.timeout } : null, broker, path, authenticateHuman: (proof) => { const { business } = verifyHuman(proof, { cliPath: fileURLToPath(new URL('./human-cli.mjs', import.meta.url)), launches: launchRecords, }); const human = businesses[business]?.human; if (!human) throw new BusError('unknown-business'); return broker.bindHuman({ business, human, via: 'cli', outsideAgent: true }); }, }); return { broker, credentials, path, bindLaunch, launches: bound, readers: readCaps, async close() { await server.close(); await adapter?.close?.(); store.close(); credentials.close(); }, }; } catch (e) { await server?.close(); await adapter?.close?.(); store?.close(); credentials?.close(); throw e; } }