Repository Public
1import { Queue } from "bullmq";2import IORedis from "ioredis";3import { env } from "./env";4import { sql } from "./db";5import type { Stage } from "../pipeline/stages";67export const QUEUE = "pipeline";8export const PREFIX = "clip";9declare global { var __clipRedis: IORedis | undefined; var __clipQueue: Queue | undefined; }1011export const redis = () => globalThis.__clipRedis ?? (globalThis.__clipRedis = new IORedis(env.redisUrl, { maxRetriesPerRequest: null, lazyConnect: false }));12export const queue = () => globalThis.__clipQueue ?? (globalThis.__clipQueue = new Queue(QUEUE, { connection: redis(), prefix: PREFIX }));1314export interface StageJob { projectId: string; stage: Stage; runId: string }15export interface DirectorJob { runId: string }1617/** Queue a Claude Code director run (one attempt: a half-directed film is resumed by a person, not blindly retried). */18export async function enqueueDirector(runId: string) {19 await queue().add("director", { runId } satisfies DirectorJob, { jobId: `director-${runId}`, attempts: 1, removeOnComplete: { count: 100 }, removeOnFail: { count: 200 } });20}2122/**23 * Queue one stage for a project. The project row is the lock: a second request while a stage is24 * queued or running returns the existing run instead of creating a duplicate.25 */26export async function enqueueStage(projectId: string, stage: Stage): Promise<{ runId: string; duplicate: boolean }> {27 const locked = await sql<{ id: string }[]>`UPDATE projects SET running_stage = ${stage}, error = NULL, blocked_reason = NULL, updated_at = now() WHERE id = ${projectId} AND running_stage IS NULL RETURNING id`;28 if (!locked.length) {29 const [run] = await sql<{ id: string }[]>`SELECT id FROM stage_runs WHERE project_id = ${projectId} AND status IN ('queued','running') ORDER BY queued_at DESC LIMIT 1`;30 if (run) return { runId: run.id, duplicate: true };31 // Lock without a live run (e.g. Redis was flushed): take it over.32 await sql`UPDATE projects SET running_stage = ${stage} WHERE id = ${projectId}`;33 }34 const [run] = await sql<{ id: string }[]>`INSERT INTO stage_runs (project_id, stage, status) VALUES (${projectId}, ${stage}, 'queued') RETURNING id`;35 await queue().add(stage, { projectId, stage, runId: run.id } satisfies StageJob, {36 jobId: run.id, attempts: 3, backoff: { type: "exponential", delay: 20_000 }, removeOnComplete: { count: 300 }, removeOnFail: { count: 500 },37 });38 return { runId: run.id, duplicate: false };39}4041export async function queueDepth() {42 try { return await queue().getJobCounts("waiting", "active", "delayed", "failed"); } catch { return null; }43}44