import { Worker } from 'bullmq'; import { Redis } from 'ioredis'; import { builds, createDb, eq, mcpServers } from '@bmm/db'; import { config } from './config.js'; import { generateSpec } from './lib/claude.js'; import { renderServerCode } from './lib/render.js'; import { dockerBuild, prepareBuildContext, staticCheck } from './lib/build.js'; import { allocatePort, deployContainer, dockerAvailable } from './lib/deploy.js'; import { emitDone, emitError, emitLog, emitStatus } from './lib/emit.js'; const db = createDb(); const connection = new Redis(config.REDIS_URL, { maxRetriesPerRequest: null }); interface JobData { buildId: string; serverId: string; orgId: string; prompt: string; version: number; slug: string; serverName: string; secrets: Record; } export const worker = new Worker( 'build', async (job) => { const { buildId, serverId, prompt, version, slug, secrets } = job.data; const log = (level: 'info' | 'warn' | 'error', msg: string) => emitLog(buildId, level, msg); try { await db.update(builds).set({ status: 'generating', startedAt: new Date() }).where(eq(builds.id, buildId)); await db.update(mcpServers).set({ status: 'generating', updatedAt: new Date() }).where(eq(mcpServers.id, serverId)); await emitStatus(buildId, 'generating'); await log('info', 'Generating MCP server spec...'); const { spec, source } = await generateSpec(prompt); await log('info', `Spec generated via ${source} (${spec.tools.length} tool(s))`); const generatedCode = renderServerCode(spec); await db .update(builds) .set({ generatedSpec: spec, generatedCode }) .where(eq(builds.id, buildId)); await db.update(builds).set({ status: 'building' }).where(eq(builds.id, buildId)); await db.update(mcpServers).set({ status: 'building', toolsSchema: spec.tools, updatedAt: new Date() }).where(eq(mcpServers.id, serverId)); await emitStatus(buildId, 'building'); await log('info', 'Preparing build context...'); const { contextDir, imageTag } = await prepareBuildContext(serverId, version, slug, generatedCode, spec); await log('info', `Build context at ${contextDir}`); await log('info', 'Running static checks...'); await staticCheck(contextDir); await log('info', 'Static checks passed.'); const hasDocker = await dockerAvailable(); if (!hasDocker) { await log('warn', 'Docker not available — skipping build/deploy. Server marked draft.'); await db.update(builds).set({ status: 'failed', errorMessage: 'docker_unavailable', finishedAt: new Date() }).where(eq(builds.id, buildId)); await db.update(mcpServers).set({ status: 'failed', updatedAt: new Date() }).where(eq(mcpServers.id, serverId)); await emitDone(buildId, 'failed', serverId, null); return; } await log('info', `Building Docker image ${imageTag}...`); await dockerBuild(contextDir, imageTag, (line) => { emitLog(buildId, 'info', line).catch(() => undefined); }); await log('info', 'Image built.'); await db.update(builds).set({ status: 'deploying' }).where(eq(builds.id, buildId)); await db.update(mcpServers).set({ status: 'deploying', updatedAt: new Date() }).where(eq(mcpServers.id, serverId)); await emitStatus(buildId, 'deploying'); const port = await allocatePort(); const publicUrl = `http://${config.RUNNER_HOST}:${port}`; const envVars: Record = { ...secrets, PUBLIC_URL: publicUrl, CONTROL_PLANE_URL: config.CONTROL_PLANE_URL, PORT: '3000', SERVER_ID: serverId, }; const handle = await deployContainer({ serverId, slug, hostPort: port, imageTag, envVars }); await log('info', `Container ${handle.containerId.slice(0, 12)} running at ${handle.publicUrl}`); await db .update(builds) .set({ status: 'success', finishedAt: new Date() }) .where(eq(builds.id, buildId)); await db .update(mcpServers) .set({ status: 'live', currentVersion: version, publicUrl: handle.publicUrl, updatedAt: new Date() }) .where(eq(mcpServers.id, serverId)); await emitStatus(buildId, 'success'); await emitDone(buildId, 'success', serverId, handle.publicUrl); } catch (err) { const msg = err instanceof Error ? err.message : String(err); console.error('[worker] build failed:', err); await db .update(builds) .set({ status: 'failed', errorMessage: msg, finishedAt: new Date() }) .where(eq(builds.id, buildId)); await db .update(mcpServers) .set({ status: 'failed', updatedAt: new Date() }) .where(eq(mcpServers.id, serverId)); await emitError(buildId, msg); await emitDone(buildId, 'failed', serverId, null); } }, { connection, concurrency: 2 }, ); worker.on('ready', () => console.log('[generator] worker ready')); worker.on('failed', (job, err) => console.error('[generator] job failed', job?.id, err?.message)); worker.on('error', (err) => console.error('[generator] worker error', err.message));