import * as fs from "node:fs"; import * as path from "node:path"; import cors from "@fastify/cors"; import fastifyStatic from "@fastify/static"; import { Queue, QueueEvents } from "bullmq"; import Fastify from "fastify"; import IORedis from "ioredis"; import { PACING_CONFIG } from "./agents/creative-director.js"; import { getArchetype, listArchetypes } from "./config/archetype-registry.js"; import { PLATFORMS } from "./config/platforms.js"; import { DirectorScore } from "./schema/director-score.js"; import type { SearchProviderKey } from "./schema/providers.js"; const REDIS_URL = process.env["REDIS_URL"] ?? "redis://localhost:6379"; const PORT = Number(process.env["PORT"] ?? 3000); const HOST = process.env["HOST"] ?? "0.0.0.0"; const JOBS_DIR = process.env["JOBS_DIR"] ?? path.join(process.cwd(), "jobs"); const MAX_JOBS = process.env["MAX_JOBS"] ? Number(process.env["MAX_JOBS"]) : 0; const WEB_DIST = path.join(process.cwd(), "web", "dist"); // Ensure jobs directory exists fs.mkdirSync(JOBS_DIR, { recursive: true }); /** Validate job ID to prevent path traversal — must be alphanumeric/hyphen/underscore only */ function isValidJobId(id: string): boolean { return /^[\w-]+$/.test(id); } const redis = new IORedis(REDIS_URL, { maxRetriesPerRequest: null }); const queue = new Queue("openreels", { connection: redis }); const queueEvents = new QueueEvents("openreels", { connection: redis.duplicate() }); const app = Fastify({ logger: true }); await app.register(cors, { origin: true }); // Serve job artifacts from the jobs directory await app.register(fastifyStatic, { root: JOBS_DIR, prefix: "/api/v1/jobs/", serve: false, // we handle serving manually for path traversal protection }); // --- Health check --- app.get("/api/v1/health", async () => { let redisOk = false; try { await redis.ping(); redisOk = true; } catch {} const jobsDirStats = fs.statSync(JOBS_DIR, { throwIfNoEntry: false }); return { status: redisOk ? "healthy" : "degraded", redis: redisOk ? "connected" : "disconnected", jobsDir: jobsDirStats ? "exists" : "missing", keys: { ANTHROPIC_API_KEY: !!process.env["ANTHROPIC_API_KEY"], OPENAI_API_KEY: !!process.env["OPENAI_API_KEY"], GOOGLE_API_KEY: !!process.env["GOOGLE_API_KEY"], ELEVENLABS_API_KEY: !!process.env["ELEVENLABS_API_KEY"], INWORLD_TTS_API_KEY: !!process.env["INWORLD_TTS_API_KEY"], PEXELS_API_KEY: !!process.env["PEXELS_API_KEY"], PIXABAY_API_KEY: !!process.env["PIXABAY_API_KEY"], }, }; }); // --- Aggregate stats --- app.get("/api/v1/stats", async () => { if (!fs.existsSync(JOBS_DIR)) { return { totalJobs: 0, completedJobs: 0, failedJobs: 0, activeJobs: 0, totalCost: 0 }; } let totalJobs = 0; let completedJobs = 0; let failedJobs = 0; let activeJobs = 0; let totalCost = 0; const dirs = fs.readdirSync(JOBS_DIR, { withFileTypes: true }).filter((d) => d.isDirectory()); await Promise.all( dirs.map(async (d) => { const metaPath = path.join(JOBS_DIR, d.name, "meta.json"); try { const raw = await fs.promises.readFile(metaPath, "utf-8"); const meta = JSON.parse(raw); totalJobs++; if (meta.status === "completed") { completedJobs++; const cost = meta.actualCost?.totalCost ?? meta.costEstimate?.totalCost ?? 0; totalCost += cost; } else if (meta.status === "failed" || meta.status === "cancelled") { failedJobs++; } else if (meta.status === "running" || meta.status === "queued") { activeJobs++; } } catch {} }), ); return { totalJobs, completedJobs, failedJobs, activeJobs, totalCost: Math.round(totalCost * 100) / 100, }; }); // --- Archetypes --- app.get("/api/v1/archetypes", async () => { const names = listArchetypes(); return names.map((name) => { const config = getArchetype(name); return { name, captionStyle: config.captionStyle, artStyle: config.artStyle, mood: config.mood, colorPalette: config.colorPalette, visualColorPalette: config.visualColorPalette, scenePacing: config.scenePacing, motionIntensity: config.motionIntensity, }; }); }); // --- Platforms --- app.get("/api/v1/platforms", async () => { return Object.entries(PLATFORMS).map(([name, config]) => ({ name, width: config.width, height: config.height, fps: config.fps, maxDurationSeconds: config.maxDurationSeconds, })); }); // --- Providers list --- app.get("/api/v1/providers", async () => ({ llm: [ { key: "anthropic", label: "Anthropic (Claude)" }, { key: "openai", label: "OpenAI (GPT)" }, { key: "gemini", label: "Google Gemini" }, { key: "openrouter", label: "OpenRouter" }, { key: "openai-compatible", label: "Custom (OpenAI-compatible)" }, ], search: [ { key: "native", label: "Native (provider built-in)" }, { key: "tavily", label: "Tavily" }, { key: "none", label: "None (parametric knowledge)" }, ], tts: [ { key: "elevenlabs", label: "ElevenLabs" }, { key: "inworld", label: "Inworld" }, { key: "kokoro", label: "Kokoro (Local)" }, { key: "gemini-tts", label: "Gemini TTS" }, { key: "openai-tts", label: "OpenAI TTS" }, ], image: [ { key: "gemini", label: "Google Gemini" }, { key: "openai", label: "OpenAI (GPT Image)" }, ], video: [ { key: "gemini", label: "Google Veo" }, { key: "fal", label: "fal.ai (Kling, Wan, etc.)" }, ], })); // --- Job creation --- interface CreateJobBody { topic: string; archetype?: string; pacing?: string; platform?: string; dryRun?: boolean; noMusic?: boolean; noVideo?: boolean; direction?: string; score?: Record; providers?: { llm?: string; tts?: string; image?: string; stock?: string; video?: string; videoModel?: string; music?: string; llmModel?: string; llmBaseUrl?: string; searchProvider?: SearchProviderKey; }; keys?: Record; } app.post<{ Body: CreateJobBody }>("/api/v1/jobs", async (request, reply) => { const { topic, archetype, pacing, platform, dryRun, noMusic, noVideo, direction, score, providers, keys } = request.body ?? {}; if (!topic || typeof topic !== "string" || topic.trim().length === 0) { return reply.status(400).send({ error: "topic is required" }); } if (topic.trim().length > 500) { return reply.status(400).send({ error: "topic must be 500 characters or fewer" }); } // Validate archetype if provided if (archetype) { try { getArchetype(archetype); } catch { return reply.status(400).send({ error: `Unknown archetype: ${archetype}` }); } } // Validate pacing tier if provided const validPacingTiers = Object.keys(PACING_CONFIG); if (pacing && !validPacingTiers.includes(pacing)) { return reply .status(400) .send({ error: `Unknown pacing tier: ${pacing}. Available: ${validPacingTiers.join(", ")}` }); } // Validate platform if provided const validPlatforms = Object.keys(PLATFORMS); if (platform && !validPlatforms.includes(platform)) { return reply .status(400) .send({ error: `Unknown platform: ${platform}. Available: ${validPlatforms.join(", ")}` }); } // Validate direction text size if provided if (direction != null) { if (typeof direction !== "string") { return reply.status(400).send({ error: "direction must be a string" }); } if (Buffer.byteLength(direction, "utf-8") > 10240) { return reply.status(400).send({ error: "direction exceeds 10KB limit" }); } } // Validate score (DirectorScore) if provided for replay let validatedScore: unknown | undefined; if (score != null) { const result = DirectorScore.safeParse(score); if (!result.success) { return reply.status(400).send({ error: `Invalid DirectorScore: ${result.error.message}` }); } // Validate archetype exists in the registry (Zod accepts any string, but getArchetype throws for unknown names) try { getArchetype((result.data as { archetype: string }).archetype); } catch { return reply.status(400).send({ error: `Score references unknown archetype: ${(result.data as { archetype: string }).archetype}`, }); } validatedScore = result.data; } const job = await queue.add("render", { topic: topic.trim(), archetype, pacing, platform: platform ?? "youtube", dryRun: dryRun ?? false, noMusic: noMusic === true, noVideo: noVideo === true, ...(direction?.trim() ? { direction: direction.trim() } : {}), ...(validatedScore ? { score: validatedScore } : {}), providers: { llm: providers?.llm ?? "anthropic", tts: providers?.tts ?? "elevenlabs", image: providers?.image ?? "gemini", stock: providers?.stock ?? "pexels", video: providers?.video, videoModel: providers?.videoModel, music: providers?.music ?? "bundled", llmModel: providers?.llmModel, llmBaseUrl: providers?.llmBaseUrl, searchProvider: providers?.searchProvider, }, keys: keys ?? {}, jobsDir: JOBS_DIR, }); // Create placeholder meta.json so GET /jobs/:id never 404s for a queued job const jobDir = path.join(JOBS_DIR, job.id!); fs.mkdirSync(jobDir, { recursive: true }); const placeholderMeta = { id: job.id, topic: topic.trim(), archetype, pacing, status: "queued", createdAt: new Date().toISOString(), stages: Object.fromEntries( ["research", "director", "tts", "visuals", "assembly", "critic"].map((s) => [ s, { status: "pending" }, ]), ), }; fs.writeFileSync(path.join(jobDir, "meta.json"), JSON.stringify(placeholderMeta, null, 2)); return reply.status(201).send({ id: job.id, topic: topic.trim(), archetype, status: "queued", }); }); // --- Job listing --- app.get("/api/v1/jobs", async (request) => { const { limit = "20", offset = "0" } = request.query as Record; const limitNum = Math.min(Number(limit) || 20, 100); const offsetNum = Number(offset) || 0; if (!fs.existsSync(JOBS_DIR)) return { jobs: [], total: 0 }; const dirents = fs.readdirSync(JOBS_DIR, { withFileTypes: true }).filter((d) => d.isDirectory()); const entries = await Promise.all( dirents.map(async (d) => { const metaPath = path.join(JOBS_DIR, d.name, "meta.json"); try { const raw = await fs.promises.readFile(metaPath, "utf-8"); const meta = JSON.parse(raw); return { id: d.name, ...meta }; } catch { return { id: d.name, status: "unknown" }; } }), ); entries.sort((a, b) => (b.createdAt ?? "").localeCompare(a.createdAt ?? "")); return { jobs: entries.slice(offsetNum, offsetNum + limitNum), total: entries.length, }; }); // --- Job detail --- app.get<{ Params: { id: string } }>("/api/v1/jobs/:id", async (request, reply) => { if (!isValidJobId(request.params.id)) { return reply.status(400).send({ error: "Invalid job ID" }); } const jobDir = path.join(JOBS_DIR, request.params.id); const metaPath = path.join(jobDir, "meta.json"); if (!fs.existsSync(metaPath)) { return reply.status(404).send({ error: "Job not found" }); } const meta = JSON.parse(fs.readFileSync(metaPath, "utf-8")); return meta; }); // --- Job SSE events --- app.get<{ Params: { id: string } }>("/api/v1/jobs/:id/events", async (request, reply) => { if (!isValidJobId(request.params.id)) { return reply.status(400).send({ error: "Invalid job ID" }); } const jobId = request.params.id; // Check if job exists in BullMQ const jobDir = path.join(JOBS_DIR, jobId); const metaPath = path.join(jobDir, "meta.json"); const job = await queue.getJob(jobId); if (!job) { // Job may have been cleaned from BullMQ (Redis restart, etc.) — fallback to meta.json if (fs.existsSync(metaPath)) { try { const meta = JSON.parse(fs.readFileSync(metaPath, "utf-8")); if ( meta.status === "completed" || meta.status === "failed" || meta.status === "cancelled" ) { reply.raw.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive", }); reply.raw.write(`event: job:snapshot\ndata: ${JSON.stringify(meta)}\n\n`); const terminalEvent = meta.status === "completed" ? "job:completed" : "job:failed"; reply.raw.write( `event: ${terminalEvent}\ndata: ${JSON.stringify({ state: meta.status })}\n\n`, ); reply.raw.end(); return; } } catch {} } return reply.status(404).send({ error: "Job not found" }); } reply.raw.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive", }); // Send current state as snapshot if (fs.existsSync(metaPath)) { try { const meta = JSON.parse(fs.readFileSync(metaPath, "utf-8")); reply.raw.write(`event: job:snapshot\ndata: ${JSON.stringify(meta)}\n\n`); } catch {} } // If job is already finished, send completion and close const state = await job.getState(); if (state === "completed" || state === "failed") { reply.raw.write(`event: job:${state}\ndata: ${JSON.stringify({ state })}\n\n`); reply.raw.end(); return; } // Guard against double-cleanup (client disconnect + job complete can race) let cleaned = false; // Listen for progress updates const onProgress = ({ jobId: progressJobId, data }: { jobId: string; data: unknown }) => { if (cleaned || progressJobId !== jobId) return; try { const eventData = data as Record; const stage = eventData.stage as string; reply.raw.write(`event: stage:${stage}\ndata: ${JSON.stringify(eventData)}\n\n`); } catch {} }; const onCompleted = ({ jobId: completedJobId }: { jobId: string }) => { if (completedJobId === jobId) { try { reply.raw.write(`event: job:completed\ndata: {}\n\n`); } catch {} cleanup(); } }; const onFailed = ({ jobId: failedJobId, failedReason, }: { jobId: string; failedReason: string; }) => { if (failedJobId === jobId) { try { reply.raw.write(`event: job:failed\ndata: ${JSON.stringify({ error: failedReason })}\n\n`); } catch {} cleanup(); } }; const cleanup = () => { if (cleaned) return; cleaned = true; queueEvents.off("progress", onProgress); queueEvents.off("completed", onCompleted); queueEvents.off("failed", onFailed); try { reply.raw.end(); } catch {} }; queueEvents.on("progress", onProgress); queueEvents.on("completed", onCompleted); queueEvents.on("failed", onFailed); // Cleanup on client disconnect request.raw.on("close", cleanup); }); // --- Artifact serving (with path traversal protection) --- app.get<{ Params: { id: string; "*": string } }>( "/api/v1/jobs/:id/artifacts/*", async (request, reply) => { const jobId = request.params.id; if (!isValidJobId(jobId)) { return reply.status(400).send({ error: "Invalid job ID" }); } const artifactPath = request.params["*"]; const jobDir = path.join(JOBS_DIR, jobId); const fullPath = path.resolve(jobDir, artifactPath); // Path traversal protection: ensure resolved path is within the job directory if ( !fullPath.startsWith(path.resolve(jobDir) + path.sep) && fullPath !== path.resolve(jobDir) ) { return reply.status(403).send({ error: "Access denied" }); } if (!fs.existsSync(fullPath)) { return reply.status(404).send({ error: "Artifact not found" }); } return reply.sendFile(path.relative(JOBS_DIR, fullPath), JOBS_DIR); }, ); // --- Job cancellation --- app.post<{ Params: { id: string } }>("/api/v1/jobs/:id/cancel", async (request, reply) => { if (!isValidJobId(request.params.id)) { return reply.status(400).send({ error: "Invalid job ID" }); } const job = await queue.getJob(request.params.id); if (!job) { return reply.status(404).send({ error: "Job not found" }); } const state = await job.getState(); if (state === "completed" || state === "failed") { return reply.status(409).send({ error: `Job already ${state}` }); } // Update meta to mark as cancelling (atomic write: tmp + rename) const jobDir = path.join(JOBS_DIR, request.params.id); const metaPath = path.join(jobDir, "meta.json"); if (fs.existsSync(metaPath)) { try { const meta = JSON.parse(fs.readFileSync(metaPath, "utf-8")); meta.cancelRequested = true; const tmpPath = path.join(jobDir, ".meta.tmp"); fs.writeFileSync(tmpPath, JSON.stringify(meta, null, 2)); fs.renameSync(tmpPath, metaPath); } catch {} } // moveToFailed handles queued jobs; for active jobs the token won't match // the worker's lock, so we catch and rely on cancelRequested flag instead try { await job.moveToFailed(new Error("Cancelled by user"), "0", true); } catch {} return { status: "cancelled" }; }); // --- Job deletion --- app.delete<{ Params: { id: string } }>("/api/v1/jobs/:id", async (request, reply) => { const jobId = request.params.id; if (!isValidJobId(jobId)) { return reply.status(400).send({ error: "Invalid job ID" }); } const jobDir = path.join(JOBS_DIR, jobId); if (!fs.existsSync(jobDir)) { return reply.status(404).send({ error: "Job not found" }); } // Don't delete active jobs const job = await queue.getJob(jobId); if (job) { const state = await job.getState(); if (state === "active" || state === "waiting") { return reply.status(409).send({ error: "Cannot delete an active job" }); } await job.remove(); } fs.rmSync(jobDir, { recursive: true, force: true }); return { status: "deleted" }; }); // --- Auto-pruning helper --- async function pruneOldJobs() { if (MAX_JOBS <= 0) return; const dirs = fs .readdirSync(JOBS_DIR, { withFileTypes: true }) .filter((d) => d.isDirectory()) .map((d) => { const metaPath = path.join(JOBS_DIR, d.name, "meta.json"); try { const meta = JSON.parse(fs.readFileSync(metaPath, "utf-8")); return { id: d.name, status: meta.status, createdAt: meta.createdAt ?? "" }; } catch { return { id: d.name, status: "unknown", createdAt: "" }; } }) .filter( (j) => j.status === "completed" || j.status === "failed" || j.status === "cancelled" || j.status === "unknown", ) .sort((a, b) => a.createdAt.localeCompare(b.createdAt)); while (dirs.length > MAX_JOBS) { const oldest = dirs.shift(); if (oldest) { fs.rmSync(path.join(JOBS_DIR, oldest.id), { recursive: true, force: true }); } } } // Export for worker to call after job completion export { JOBS_DIR, MAX_JOBS, pruneOldJobs }; // --- Serve frontend SPA --- if (fs.existsSync(WEB_DIST)) { await app.register(fastifyStatic, { root: WEB_DIST, prefix: "/", decorateReply: false, wildcard: false, }); // SPA fallback: serve index.html for all non-API routes app.setNotFoundHandler((request, reply) => { if (request.url.startsWith("/api/")) { return reply.status(404).send({ error: "Not found" }); } return reply.sendFile("index.html", WEB_DIST); }); } // --- Start server --- try { await app.listen({ port: PORT, host: HOST }); console.log(`OpenReels API server running on http://${HOST}:${PORT}`); } catch (err) { app.log.error(err); process.exit(1); }