Repository Public
1import "../../scripts/_env";2import { Worker, UnrecoverableError, type Job } from "bullmq";3import { execFile } from "node:child_process";4import { hostname } from "node:os";5import { readdir, rm, stat } from "node:fs/promises";6import path from "node:path";7import { sql, j } from "../lib/db";8import { getPolicy } from "../lib/budget";9import { CreativeBrief } from "../lib/contracts";10import { hashOf } from "../lib/hash";11import { log } from "../lib/log";12import { PREFIX, QUEUE, enqueueStage, redis, type DirectorJob, type StageJob } from "../lib/queue";13import { runDirector } from "../pipeline/director";14import { tmpDir } from "../lib/storage";15import { SERIES } from "../lib/taxonomy";16import { afterStage, releaseStaleLocks } from "../pipeline/orchestrator";17import { releaseScheduled } from "../pipeline/publish";18import { releaseOrphans, setClaimer } from "../pipeline/provider-job";19import { STAGE_FN } from "../pipeline/stages";2021/** CLIP worker: pipeline stages, programming scheduler, housekeeping. */2223async function process(job: Job<StageJob | DirectorJob>) {24 if (job.name === "director") { await runDirector((job.data as DirectorJob).runId); return; }25 const { projectId, stage, runId } = job.data as StageJob;26 await sql`UPDATE stage_runs SET status = 'running', started_at = COALESCE(started_at, now()), attempts = ${job.attemptsMade + 1} WHERE id = ${runId}`;27 log("info", "stage.start", { projectId, stage, attempt: job.attemptsMade + 1 });28 try {29 await STAGE_FN[stage](projectId);30 } catch (e) {31 const err = e as Error & { retryable?: boolean };32 const budget = err.name === "BudgetError";33 const busy = err.name === "JobInProgressError";34 const final = budget || (!busy && err.retryable === false) || (!busy && err.name !== "ProviderError") || job.attemptsMade + 1 >= (job.opts.attempts ?? 1);35 if (final) {36 await sql`UPDATE stage_runs SET status = ${budget ? "blocked" : "failed"}, error = ${err.message.slice(0, 1000)}, finished_at = now() WHERE id = ${runId}`;37 await sql`UPDATE projects SET running_stage = NULL, updated_at = now(), error = ${budget ? null : err.message.slice(0, 1000)}, blocked_reason = ${budget ? err.message : null} WHERE id = ${projectId}`;38 log("error", "stage.failed", { projectId, stage, error: err.message, budget });39 throw new UnrecoverableError(err.message);40 }41 await sql`UPDATE stage_runs SET error = ${`retrying: ${err.message}`.slice(0, 1000)} WHERE id = ${runId}`;42 log("warn", "stage.retry", { projectId, stage, error: err.message });43 throw e;44 }45 await sql`UPDATE stage_runs SET status = 'done', finished_at = now(), error = NULL WHERE id = ${runId}`;46 await sql`UPDATE projects SET running_stage = NULL, updated_at = now() WHERE id = ${projectId}`;47 log("info", "stage.done", { projectId, stage });48 try { await afterStage(projectId); } catch (e) { log("error", "after_stage.failed", { projectId, error: (e as Error).message }); }49}5051/** Programming scheduler: bounded by the policy quota; never starts more than `films_per_day`. */52export async function runScheduler(now = new Date()) {53 const policy = await getPolicy();54 if (!policy.scheduler_enabled || policy.kill_switch) return 0;55 const [{ n }] = await sql<{ n: number }[]>`SELECT COUNT(*)::int AS n FROM projects WHERE origin = 'scheduler' AND created_at >= date_trunc('day', now())`;56 let started = 0;57 const due = await sql<{ id: string; series_id: string; slug: string; theme: string; language: "en" | "fr" | "bilingual"; mode: "manual" | "assisted" | "autonomous" }[]>`58 SELECT s.id, s.series_id, se.slug, s.theme, s.language, s.mode FROM schedules s JOIN series se ON se.id = s.series_id59 WHERE s.enabled AND ${now.getDay()} = ANY(s.weekdays) AND s.hour <= ${now.getHours()} AND (s.last_run_on IS NULL OR s.last_run_on < CURRENT_DATE)60 ORDER BY s.hour`;61 for (const s of due) {62 if (n + started >= policy.films_per_day) break;63 const tpl = SERIES.find((x) => x.slug === s.slug);64 const brief = CreativeBrief.parse({ theme: s.theme || `A new film for the series "${tpl?.titleEn ?? s.slug}"`, language: s.language, seriesSlug: s.slug, mode: s.mode, shots: 5, quality: "master" });65 const [p] = await sql<{ id: string }[]>`66 INSERT INTO projects (title, series_id, brief, language, mode, origin, video_model, dedupe_hash)67 VALUES (${`${tpl?.titleEn ?? s.slug} — ${now.toISOString().slice(0, 10)}`}, ${s.series_id}, ${j(brief)}, ${s.language}, ${s.mode}, 'scheduler', ${null}, ${hashOf({ schedule: s.id, day: now.toISOString().slice(0, 10) })})68 RETURNING id`;69 await sql`UPDATE schedules SET last_run_on = CURRENT_DATE WHERE id = ${s.id}`;70 await enqueueStage(p.id, "concept");71 started++;72 log("info", "scheduler.started", { project: p.id, series: s.slug });73 }74 return started;75}7677async function housekeeping() {78 await releaseStaleLocks();79 await releaseScheduled();80 await sql`DELETE FROM sessions WHERE expires_at < now()`;81 await sql`DELETE FROM rate_limits WHERE window_start < now() - interval '1 day'`;82 await sql`DELETE FROM analytics_events WHERE at < now() - interval '90 days'`;83 // Intermediates: work directories older than a day are leftovers from a crashed assembly.84 try {85 for (const name of await readdir(tmpDir())) {86 const dir = path.join(tmpDir(), name);87 if (Date.now() - (await stat(dir)).mtimeMs > 24 * 3600_000) await rm(dir, { recursive: true, force: true });88 }89 } catch { /* work dir may not exist yet */ }90}9192let lastBackupDay = "";93async function maybeBackup(now = new Date()) {94 const day = now.toISOString().slice(0, 10);95 if (now.getHours() !== 4 || now.getMinutes() < 20 || lastBackupDay === day) return;96 lastBackupDay = day;97 execFile("/bin/bash", [path.join(globalThis.process.cwd(), "scripts/backup.sh")], { timeout: 30 * 60_000 }, (err, stdout, stderr) => {98 if (err) log("error", "backup.failed", { error: (stderr || err.message).slice(-400) });99 else log("info", "backup.done", { out: stdout.trim().split("\n").at(-1) });100 void sql`INSERT INTO settings (key, value) VALUES ('last_backup', ${j({ at: new Date().toISOString(), ok: !err, detail: err ? (stderr || err.message).slice(-200) : stdout.trim().split("\n").at(-1) })}) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`;101 });102}103104async function main() {105 const id = `worker:${hostname()}`;106 setClaimer(id);107 const released = await releaseOrphans(id);108 // A director run cannot survive a worker restart (its Claude Code process is gone): mark it so the studio shows the truth.109 await sql`UPDATE director_runs SET status = 'failed', error = 'The worker restarted during this run', finished_at = now() WHERE status = 'running'`;110 await sql`UPDATE projects SET running_stage = NULL WHERE running_stage = 'director'`;111 if (released) log("warn", "worker.orphans_released", { released });112 const worker = new Worker<StageJob | DirectorJob>(QUEUE, process, { connection: redis(), prefix: PREFIX, concurrency: 2, lockDuration: 120_000, stalledInterval: 60_000, maxStalledCount: 3 });113 worker.on("error", (e) => log("error", "worker.error", { error: e.message }));114 const beat = async () => { await sql`INSERT INTO settings (key, value) VALUES ('worker_heartbeat', ${j({ at: new Date().toISOString(), pid: globalThis.process.pid })}) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`; };115 const tick = async () => {116 try { await beat(); await housekeeping(); await runScheduler(); await maybeBackup(); } catch (e) { log("error", "tick.failed", { error: (e as Error).message }); }117 };118 await tick();119 const hb = setInterval(() => void beat().catch(() => {}), 30_000);120 const tk = setInterval(() => void tick(), 5 * 60_000);121 log("info", "worker.ready", { queue: QUEUE });122 const stop = async () => { clearInterval(hb); clearInterval(tk); await worker.close(); await sql.end({ timeout: 5 }); globalThis.process.exit(0); };123 globalThis.process.on("SIGTERM", stop);124 globalThis.process.on("SIGINT", stop);125}126127if (import.meta.url === `file://${globalThis.process.argv[1]}`) void main();128