Repository Public
1import { sql, j } from "./db";2import { audit } from "./audit";3import { getPolicy } from "./budget";4import { CreativeBrief } from "./contracts";5import { env } from "./env";6import { hashOf } from "./hash";7import { enqueueDirector, enqueueStage } from "./queue";8import type { User } from "./auth";9import { startStage } from "../pipeline/orchestrator";10import { approveProject, publishProject, recordApproval, rejectProject, setFilmStatus } from "../pipeline/publish";11import { bumpStageRev, getProject, getShots, type ShotRow, type Stage, STAGES } from "../pipeline/stages";1213/** Studio operations shared by server actions and the /api/studio routes. Every mutation is audited. */1415export class OpError extends Error {16 constructor(message: string, readonly status = 400) { super(message); this.name = "OpError"; }17}1819export async function createProject(user: User | null, input: unknown, o: { idempotencyKey?: string | null; origin?: "studio" | "api" } = {}) {20 const parsed = CreativeBrief.safeParse(input);21 if (!parsed.success) throw new OpError(`Invalid brief: ${parsed.error.issues.map((i) => `${i.path.join(".")} ${i.message}`).join("; ")}`);22 const brief = parsed.data;23 const dedupe = o.idempotencyKey ? hashOf({ key: o.idempotencyKey, user: user?.id }) : null;24 if (dedupe) {25 const [dup] = await sql<{ id: string }[]>`SELECT id FROM projects WHERE dedupe_hash = ${dedupe}`;26 if (dup) return { id: dup.id, duplicate: true };27 }28 const [series] = brief.seriesSlug ? await sql<{ id: string }[]>`SELECT id FROM series WHERE slug = ${brief.seriesSlug}` : [];29 if (brief.seriesSlug && !series) throw new OpError("Unknown series");30 const [p] = await sql<{ id: string }[]>`31 INSERT INTO projects (title, series_id, brief, language, mode, origin, video_model, created_by, dedupe_hash)32 VALUES (${brief.theme.slice(0, 90)}, ${series?.id ?? null}, ${j(brief)}, ${brief.language}, ${brief.mode}, ${o.origin ?? "studio"},33 ${brief.quality === "draft" ? env.videoDraftModel : env.videoModel}, ${user?.id ?? null}, ${dedupe})34 RETURNING id`;35 await audit(user?.id ?? null, "project.create", "project", p.id, { theme: brief.theme, mode: brief.mode, series: brief.seriesSlug });36 // Pitches cost a few cents: every mode starts with them so the operator has something to choose from.37 await enqueueStage(p.id, "concept");38 return { id: p.id, duplicate: false };39}4041/** Director mode: one idea, and Claude Code orchestrates the whole film (plan, keyframes, previews, finals, cut, QA). */42export async function startDirector(user: User, input: { idea: string; language: string; shots: number; budget: number }) {43 const idea = input.idea.trim();44 if (idea.length < 8) throw new OpError("Describe the idea in a sentence or two");45 if (idea.length > 2000) throw new OpError("The idea is too long (2000 characters at most)");46 const language = ["en", "fr", "bilingual"].includes(input.language) ? input.language : "en";47 const shots = Math.min(6, Math.max(3, Math.round(input.shots) || 5));48 const policy = await getPolicy();49 if (policy.kill_switch) throw new OpError("The kill switch is on: no job can start", 409);50 const budget = Math.min(Number(policy.per_film_usd), Math.max(3, Number(input.budget) || 15));51 const [busy] = await sql<{ n: number }[]>`SELECT COUNT(*)::int AS n FROM director_runs WHERE status IN ('queued','running')`;52 if (busy.n >= 2) throw new OpError("Two director runs are already in progress: wait for one to finish", 409);53 const brief = CreativeBrief.parse({ theme: idea.slice(0, 600), language, shots, mode: "manual", quality: "master" });54 const [p] = await sql<{ id: string }[]>`55 INSERT INTO projects (title, brief, language, mode, origin, video_model, created_by, running_stage)56 VALUES (${idea.slice(0, 80)}, ${j(brief)}, ${language}, 'manual', 'director', ${env.videoModel}, ${user.id}, 'director') RETURNING id`;57 const [run] = await sql<{ id: string }[]>`INSERT INTO director_runs (project_id, idea, budget_usd, created_by) VALUES (${p.id}, ${idea}, ${budget}, ${user.id}) RETURNING id`;58 await enqueueDirector(run.id);59 await audit(user.id, "director.start", "project", p.id, { runId: run.id, budget, language, shots });60 return { runId: run.id, projectId: p.id };61}62export async function stopDirector(user: User, runId: string) {63 const hit = await sql`UPDATE director_runs SET stop_requested = true WHERE id = ${runId} AND status IN ('queued','running') RETURNING id`;64 if (!hit.length) throw new OpError("This run is not in progress", 409);65 await audit(user.id, "director.stop", "director_run", runId);66}6768/** Run one stage. `fresh` asks the LLM stages for a new answer instead of reusing the stored one. */69export async function runStage(user: User, projectId: string, stage: string, o: { fresh?: boolean } = {}) {70 if (!(STAGES as readonly string[]).includes(stage)) throw new OpError("Unknown stage");71 const policy = await getPolicy();72 if (policy.kill_switch) throw new OpError("The kill switch is on: no job can start", 409);73 if (o.fresh && ["concept", "script", "storyboard", "qa"].includes(stage)) {74 await bumpStageRev(projectId, stage);75 if (stage === "script") await bumpStageRev(projectId, "research");76 }77 try {78 const r = await startStage(projectId, stage as Stage);79 await audit(user.id, "stage.run", "project", projectId, { stage, fresh: !!o.fresh, duplicate: r.duplicate });80 return r;81 } catch (e) { throw new OpError((e as Error).message, 409); }82}8384/** Next stage for a project, used by POST /generate without an explicit stage. */85export async function nextStageFor(projectId: string): Promise<Stage | null> {86 const { NEXT } = await import("../pipeline/orchestrator");87 return NEXT[(await getProject(projectId)).status] ?? null;88}8990export async function selectPitch(user: User, projectId: string, candidateId: string) {91 await sql.begin(async (tx) => {92 const hit = await tx`UPDATE concept_candidates SET selected = (id = ${candidateId}) WHERE project_id = ${projectId} RETURNING id`;93 if (!hit.length) throw new OpError("Pitch not found", 404);94 });95 await audit(user.id, "pitch.select", "project", projectId, { candidateId });96}9798async function settleStatus(projectId: string) {99 const shots = await getShots(projectId);100 const status = shots.some((s) => !s.image_asset_id) ? "storyboarded" : shots.some((s) => s.narration.trim() && !s.audio_asset_id) ? "keyframes" : shots.some((s) => !s.video_asset_id) ? "voiced" : "filmed";101 await sql`UPDATE projects SET status = ${status}, updated_at = now() WHERE id = ${projectId} AND status NOT IN ('published','archived')`;102 return status;103}104105async function shotOf(shotId: string): Promise<ShotRow> {106 const [s] = await sql<ShotRow[]>`SELECT * FROM shots WHERE id = ${shotId}`;107 if (!s) throw new OpError("Shot not found", 404);108 const p = await getProject(s.project_id);109 if (p.running_stage) throw new OpError(`Stage "${p.running_stage}" is running: wait for it to finish`, 409);110 if (p.status === "published") throw new OpError("A published film cannot be edited: unpublish it first", 409);111 return s;112}113async function newPromptVersion(shotId: string, kind: "image" | "video" | "voice", prompt: string, author: string) {114 await sql`INSERT INTO prompt_versions (shot_id, kind, version, prompt, author, parent_id)115 SELECT ${shotId}, ${kind}, COALESCE(MAX(version), 0) + 1, ${prompt}, ${author},116 (SELECT id FROM prompt_versions WHERE shot_id = ${shotId} AND kind = ${kind} ORDER BY version DESC LIMIT 1)117 FROM prompt_versions WHERE shot_id = ${shotId} AND kind = ${kind}`;118}119120/** Prompt Workbench: saving a prompt creates a new immutable version and invalidates what depends on it. */121export async function updateShot(user: User, shotId: string, patch: { narration?: string; narrationAlt?: string; imagePrompt?: string; videoPrompt?: string }) {122 const s = await shotOf(shotId);123 const clean = (v: string | undefined, max: number) => (v === undefined ? undefined : v.trim().slice(0, max));124 const narration = clean(patch.narration, 400), alt = clean(patch.narrationAlt, 400), image = clean(patch.imagePrompt, 4000), video = clean(patch.videoPrompt, 4000);125 const changed: string[] = [];126 if (image !== undefined && image !== s.image_prompt) {127 if (image.length < 20) throw new OpError("The image prompt is too short");128 await newPromptVersion(shotId, "image", image, user.email);129 await sql`UPDATE shots SET image_prompt = ${image}, image_rev = image_rev + 1, image_asset_id = NULL, video_asset_id = NULL, status = 'planned', updated_at = now() WHERE id = ${shotId}`;130 changed.push("image");131 }132 if (video !== undefined && video !== s.video_prompt) {133 if (video.length < 20) throw new OpError("The video prompt is too short");134 await newPromptVersion(shotId, "video", video, user.email);135 await sql`UPDATE shots SET video_prompt = ${video}, video_rev = video_rev + 1, video_asset_id = NULL, updated_at = now() WHERE id = ${shotId}`;136 changed.push("video");137 }138 if ((narration !== undefined && narration !== s.narration) || (alt !== undefined && alt !== s.narration_alt)) {139 const spokenChanged = narration !== undefined && narration !== s.narration;140 if (spokenChanged) {141 await newPromptVersion(shotId, "voice", narration!, user.email);142 // A new line has a new length, so the clip that was sized for the old line is invalidated too.143 await sql`UPDATE shots SET narration = ${narration!}, voice_rev = voice_rev + 1, audio_asset_id = NULL, narration_duration = NULL, video_asset_id = NULL, updated_at = now() WHERE id = ${shotId}`;144 }145 if (alt !== undefined) await sql`UPDATE shots SET narration_alt = ${alt}, updated_at = now() WHERE id = ${shotId}`;146 changed.push(spokenChanged ? "voice" : "captions");147 }148 if (changed.length) {149 const status = changed.every((c) => c === "captions") ? null : await settleStatus(s.project_id);150 if (!status) await sql`UPDATE projects SET status = 'filmed', updated_at = now() WHERE id = ${s.project_id} AND status IN ('assembled','qa_passed','qa_failed','approved')`;151 await audit(user.id, "shot.edit", "shot", shotId, { changed });152 }153 return changed;154}155156/** Regenerate one element of one shot. Capped: beyond the policy limit an owner must approve explicitly. */157export async function regenerateShot(user: User, shotId: string, what: "image" | "video" | "voice", o: { approve?: boolean } = {}) {158 const s = await shotOf(shotId);159 const policy = await getPolicy();160 if (policy.kill_switch) throw new OpError("The kill switch is on: no job can start", 409);161 const paid = what !== "voice";162 if (paid && s.regen_count >= policy.max_regenerations && !s.regen_approved) {163 if (!o.approve) {164 await sql`UPDATE shots SET status = 'needs_approval', updated_at = now() WHERE id = ${shotId}`;165 throw new OpError(`This shot was already regenerated ${s.regen_count} times (limit ${policy.max_regenerations}). An owner must approve further attempts.`, 409);166 }167 if (user.role !== "owner") throw new OpError("Only an owner can approve regeneration beyond the limit", 403);168 await recordApproval({ projectId: s.project_id, userId: user.id, decision: "approve_regeneration", note: `Shot ${s.idx + 1}: regeneration beyond the limit of ${policy.max_regenerations}` });169 await sql`UPDATE shots SET regen_approved = true WHERE id = ${shotId}`;170 }171 if (what === "image") await sql`UPDATE shots SET image_rev = image_rev + 1, image_asset_id = NULL, video_asset_id = NULL, regen_count = regen_count + 1, status = 'planned', error = NULL, updated_at = now() WHERE id = ${shotId}`;172 if (what === "video") await sql`UPDATE shots SET video_rev = video_rev + 1, video_asset_id = NULL, regen_count = regen_count + 1, status = 'voiced', error = NULL, updated_at = now() WHERE id = ${shotId}`;173 if (what === "voice") await sql`UPDATE shots SET voice_rev = voice_rev + 1, audio_asset_id = NULL, narration_duration = NULL, video_asset_id = NULL, error = NULL, updated_at = now() WHERE id = ${shotId}`;174 await settleStatus(s.project_id);175 const stage: Stage = what === "image" ? "keyframes" : what === "voice" ? "voice" : "video";176 const run = await enqueueStage(s.project_id, stage);177 await audit(user.id, "shot.regenerate", "shot", shotId, { what, attempt: s.regen_count + 1 });178 return { ...run, stage };179}180181export async function moveShot(user: User, shotId: string, dir: -1 | 1) {182 const s = await shotOf(shotId);183 const other = s.idx + dir;184 await sql.begin(async (tx) => {185 const [o] = await tx<{ id: string }[]>`SELECT id FROM shots WHERE project_id = ${s.project_id} AND idx = ${other}`;186 if (!o) throw new OpError("The shot is already at the edge");187 await tx`UPDATE shots SET idx = -1 WHERE id = ${s.id}`;188 await tx`UPDATE shots SET idx = ${s.idx} WHERE id = ${o.id}`;189 await tx`UPDATE shots SET idx = ${other} WHERE id = ${s.id}`;190 await tx`UPDATE projects SET status = 'filmed', updated_at = now() WHERE id = ${s.project_id} AND status IN ('assembled','qa_passed','qa_failed','approved')`;191 });192 await audit(user.id, "shot.reorder", "shot", shotId, { from: s.idx + 1, to: other + 1 });193}194195export async function approve(user: User, projectId: string, o: { override?: boolean; note?: string }) {196 if (o.override && user.role !== "owner") throw new OpError("Only an owner can override a failed QA", 403);197 try { await approveProject(projectId, user.id, o); } catch (e) { throw new OpError((e as Error).message, 409); }198 await audit(user.id, o.override ? "film.override_qa" : "film.approve", "project", projectId, { note: o.note ?? "" });199}200export async function reject(user: User, projectId: string, note: string) {201 await rejectProject(projectId, user.id, note.slice(0, 1000));202 await audit(user.id, "film.reject", "project", projectId, { note });203}204export async function publish(user: User, projectId: string, o: { scheduleFor?: Date | null; featured?: boolean } = {}) {205 let filmId: string;206 try { filmId = await publishProject(projectId, user.id, o); } catch (e) { throw new OpError((e as Error).message, 409); }207 if (o.featured) await sql`UPDATE films SET featured = (id = ${filmId})`;208 await audit(user.id, o.scheduleFor ? "film.schedule" : "film.publish", "film", filmId, { projectId, scheduleFor: o.scheduleFor ?? null });209 return filmId;210}211export async function changeFilmStatus(user: User, filmId: string, status: "unpublished" | "hold" | "published", note = "") {212 try { await setFilmStatus(filmId, status, user.id, note); } catch (e) { throw new OpError((e as Error).message, 404); }213 if (status !== "published") await sql`UPDATE projects SET status = 'approved' WHERE id = (SELECT project_id FROM films WHERE id = ${filmId}) AND status = 'published'`;214 else await sql`UPDATE projects SET status = 'published' WHERE id = (SELECT project_id FROM films WHERE id = ${filmId})`;215 await audit(user.id, `film.${status}`, "film", filmId, { note });216}217218export async function setKillSwitch(user: User | null, on: boolean) {219 await sql`UPDATE budget_policy SET kill_switch = ${on}, updated_at = now() WHERE id = 1`;220 await audit(user?.id ?? null, on ? "kill_switch.on" : "kill_switch.off", "policy", "1");221 // Jobs already submitted to the provider cannot be cancelled through the API: they are tracked to completion.222 const [{ n }] = await sql<{ n: number }[]>`SELECT COUNT(*)::int AS n FROM provider_jobs WHERE status = 'submitted'`;223 return { killSwitch: on, inFlight: n };224}225226export async function costSummary() {227 const policy = await getPolicy();228 const [tot] = await sql<{ day: number; month: number; all_time: number; failed: number }[]>`229 SELECT COALESCE(SUM(usd) FILTER (WHERE at >= date_trunc('day', now())), 0)::float8 AS day,230 COALESCE(SUM(usd) FILTER (WHERE at >= date_trunc('month', now())), 0)::float8 AS month,231 COALESCE(SUM(usd), 0)::float8 AS all_time,232 COALESCE(SUM(usd) FILTER (WHERE failed), 0)::float8 AS failed FROM cost_entries`;233 const byModel = await sql<{ kind: string; model: string; calls: number; usd: number; est: number | null }[]>`234 SELECT kind, model, COUNT(*)::int AS calls, COALESCE(SUM(cost_usd), 0)::float8 AS usd, SUM(est_usd)::float8 AS est235 FROM provider_jobs WHERE status IN ('done','failed') GROUP BY kind, model ORDER BY usd DESC`;236 const byDay = await sql<{ day: string; usd: number }[]>`237 SELECT to_char(date_trunc('day', at), 'YYYY-MM-DD') AS day, SUM(usd)::float8 AS usd FROM cost_entries WHERE at > now() - interval '30 days' GROUP BY 1 ORDER BY 1`;238 const byProject = await sql<{ id: string; title: string; status: string; usd: number; est: number | null }[]>`239 SELECT p.id, p.title, p.status, COALESCE((SELECT SUM(usd) FROM cost_entries c WHERE c.project_id = p.id), 0)::float8 AS usd,240 (SELECT SUM(est_usd) FROM provider_jobs x WHERE x.project_id = p.id AND x.status IN ('done','failed'))::float8 AS est241 FROM projects p ORDER BY p.created_at DESC LIMIT 40`;242 const [storage] = await sql<{ bytes: number; files: number }[]>`SELECT COALESCE(SUM(bytes), 0)::float8 AS bytes, COUNT(*)::int AS files FROM (SELECT DISTINCT sha256, bytes FROM assets) a`;243 const [fx] = await sql<{ value: { rate: number; at: string } }[]>`SELECT value FROM settings WHERE key = 'fx_usd_cad'`;244 return { policy, totals: tot, byModel, byDay, byProject, storage, fx: fx?.value ?? null };245}246