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 and ends them. // 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 }); }, }); // S6 launcher operations. The host decides capacity and the instance list; the broker // authorizes role.launch and keeps the evidence, refusals included. function authorizeLaunch(cap, instance) { try { return broker.authorize(cap, 'role.launch', { target: instance }); } catch (e) { if (!(e instanceof BusError)) throw e; broker.refuse(cap, e.code); throw e; } } // Metadata only (service, date, state); the token stays in Credentials. function credentialStatus(business, role) { return credentials .status() .filter((c) => c.instance === business + '/' + role) .map(({ service, date, state }) => ({ service, date, state })); } return { broker, credentials, path, bindLaunch, endLaunch: (r) => broker.endLaunch(r), identity: (cap) => broker.identity(cap), authorizeLaunch, refuse: (cap, code) => broker.refuse(cap, code), credentialStatus, 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; } }