Feat/media ai pipeline

apps/media-service/.env.example

@@ -30,3 +30,48 @@ MEDIA_DESCRIBE_TIMEOUT_MS=30000
 ORCHESTRATOR_URL=http://localhost:4001
 # Same value as orchestrator-service INTERNAL_SERVICE_SECRET (min 16 chars).
 INTERNAL_SERVICE_SECRET=
+
+# AI media: Site Blueprint + creative pipeline (docs/media-ai.md). Off by default.
+# server, orchestrator-service and media-service all read this block; keep them identical.
+MEDIA_ENABLED=false
+# Credits per generated image, per video second and per Site Blueprint call.
+# Required in production once MEDIA_ENABLED=true; 1 elsewhere when unset.
+MEDIA_CREDIT_IMAGE=
+MEDIA_CREDIT_VIDEO_PER_SECOND=
+MEDIA_CREDIT_BLUEPRINT=
+# images:videos per budget tier, plus hard caps.
+MEDIA_BUDGET_IMAGES_ONLY=8:0
+MEDIA_BUDGET_LIGHT=4:1
+MEDIA_BUDGET_STANDARD=8:3
+MEDIA_BUDGET_RICH=14:6
+MEDIA_HARD_MAX_IMAGES=24
+MEDIA_HARD_MAX_VIDEOS=10
+MEDIA_HARD_MAX_GENERATED_ASSETS=60
+VIDEO_MIN_SECONDS=5
+VIDEO_MAX_SECONDS=10
+# Circuit breaker: provider spend over the last 24 h above this pauses AI media. Empty = no cap.
+MEDIA_DAILY_COST_CAP_USD=
+# Provider cost estimates in USD, used when a provider does not report cost.
+MEDIA_COST_IMAGE_USD=0.25
+MEDIA_COST_VIDEO_USD_PER_SECOND=0.08
+MEDIA_COST_LLM_CALL_USD=0.05
+MEDIA_COST_BLUEPRINT_USD=0.4
+
+# fake = no provider calls or spend. live needs OPENAI_API_KEY, RUNWARE_API_KEY and INTERNAL_SERVICE_SECRET.
+MEDIA_AI_PROVIDER_MODE=fake
+# OPENAI_API_KEY=replace-with-provider-key
+OPENAI_IMAGE_MODEL=gpt-image-2.5-sunburst
+OPENAI_IMAGE_QUALITY=high
+VIDEO_PROVIDER=runware
+# RUNWARE_API_KEY=replace-with-provider-key
+MEDIA_VIDEO_MODEL=minimax:h3@max
+MEDIA_VIDEO_RESOLUTION=768p
+# Longest a single video render may take before the item falls back to a still.
+MEDIA_VIDEO_TIMEOUT_S=600
+MEDIA_VIDEO_PROMPT_EXPANSION=disabled
+MEDIA_VIDEO_PROMPT_MAX_CHARS=3000
+# Concurrent provider calls across all media-service replicas (Redis leases).
+MEDIA_IMAGE_CONCURRENCY=3
+MEDIA_VIDEO_CONCURRENCY=2
+# https hosts provider result downloads may come from, comma-separated.
+MEDIA_PROVIDER_HOST_ALLOWLIST=vm.runware.ai,im.runware.ai

apps/media-service/package.json

@@ -17,7 +17,9 @@
   "dependencies": {
     "@aws-sdk/client-s3": "3.1144.0",
     "@aws-sdk/lib-storage": "3.1144.0",
+    "@aws-sdk/s3-request-presigner": "3.1144.0",
     "@repo/events": "workspace:*",
+    "@repo/media-ai": "workspace:*",
     "@repo/schemas": "workspace:*",
     "@useframe/db": "workspace:*",
     "bullmq": "^5.28.0",

apps/media-service/src/config/schema.ts

@@ -41,8 +41,46 @@ export const envSchema = z
     ORCHESTRATOR_URL: z.string().url().default("http://localhost:4001"),
     INTERNAL_SERVICE_SECRET: optional(z.string().min(16, "INTERNAL_SERVICE_SECRET must be at least 16 chars")),
     MEDIA_DESCRIBE_TIMEOUT_MS: int(30_000),
+
+    // AI creative pipeline. Budgets, caps and credit prices are read by @repo/media-ai.
+    MEDIA_AI_PROVIDER_MODE: z.enum(["fake", "live"]).default("fake"),
+    OPENAI_API_KEY: optional(z.string().min(1)),
+    OPENAI_IMAGE_MODEL: z.string().min(1).default("gpt-image-2.5-sunburst"),
+    OPENAI_IMAGE_QUALITY: z.enum(["low", "medium", "high", "xhigh", "max", "auto"]).default("high"),
+    VIDEO_PROVIDER: z.enum(["runware", "fal", "minimax"]).default("runware"),
+    RUNWARE_API_KEY: optional(z.string().min(1)),
+    MEDIA_VIDEO_MODEL: z.string().min(1).default("minimax:h3@max"),
+    MEDIA_VIDEO_RESOLUTION: z.enum(["480p", "768p"]).default("768p"),
+    MEDIA_VIDEO_TIMEOUT_S: int(600),
+    MEDIA_VIDEO_PROMPT_EXPANSION: z.enum(["disabled", "balanced", "quality"]).default("disabled"),
+    MEDIA_VIDEO_PROMPT_MAX_CHARS: int(3000),
+    MEDIA_IMAGE_CONCURRENCY: int(3),
+    MEDIA_VIDEO_CONCURRENCY: int(2),
+    // Hosts provider result downloads may come from (https only).
+    MEDIA_PROVIDER_HOST_ALLOWLIST: z
+      .string()
+      .default("vm.runware.ai,im.runware.ai")
+      .transform((v) =>
+        v
+          .split(",")
+          .map((h) => h.trim().toLowerCase())
+          .filter(Boolean),
+      ),
   })
   .superRefine((value, ctx) => {
+    if (value.MEDIA_AI_PROVIDER_MODE === "live") {
+      if (!value.OPENAI_API_KEY) {
+        ctx.addIssue({ code: z.ZodIssueCode.custom, path: ["OPENAI_API_KEY"], message: "OPENAI_API_KEY is required in live mode" })
+      }
+      if (value.VIDEO_PROVIDER !== "runware") {
+        ctx.addIssue({ code: z.ZodIssueCode.custom, path: ["VIDEO_PROVIDER"], message: "Only the runware video provider is built" })
+      } else if (!value.RUNWARE_API_KEY) {
+        ctx.addIssue({ code: z.ZodIssueCode.custom, path: ["RUNWARE_API_KEY"], message: "RUNWARE_API_KEY is required in live mode" })
+      }
+      if (!value.INTERNAL_SERVICE_SECRET) {
+        ctx.addIssue({ code: z.ZodIssueCode.custom, path: ["INTERNAL_SERVICE_SECRET"], message: "INTERNAL_SERVICE_SECRET is required in live mode" })
+      }
+    }
     if (!value.MEDIA_R2_ENDPOINT && !value.CLOUDFLARE_ACCOUNT_ID) {
       ctx.addIssue({
         code: z.ZodIssueCode.custom,

apps/media-service/src/createAsset.ts

@@ -92,3 +92,46 @@ export async function createAssetFromBytes(deps: Deps, input: CreateAssetInput):
   await deps.enqueue({ assetId: asset.id, ...(input.options ? { options: input.options } : {}) }, `${asset.id}-0`)
   return { assetId: asset.id }
 }
+
+// Direct provider delivery: the row and key exist first, the provider PUTs the
+// original, then finishReservedAsset verifies it and starts processing.
+export async function reserveGeneratedAsset(
+  deps: Deps,
+  input: { projectId: string; userId: string; kind: "IMAGE" | "VIDEO"; mime: string; purpose?: string; title?: string; generation?: Prisma.InputJsonValue },
+): Promise<{ assetId: string; key: string }> {
+  return deps.db.$transaction(async (tx) => {
+    const created = await tx.mediaAsset.create({
+      data: {
+        projectId: input.projectId,
+        userId: input.userId,
+        kind: input.kind,
+        origin: "GENERATED",
+        status: "UPLOADING",
+        title: input.title?.slice(0, 120) ?? null,
+        purpose: input.purpose ?? null,
+        aiGenerated: true,
+        generation: input.generation,
+        originalKey: "",
+        originalMime: input.mime,
+        originalBytes: 0,
+      },
+      select: { id: true },
+    })
+    const key = `${mediaPrefix(input.projectId, created.id)}orig`
+    await tx.mediaAsset.update({ where: { id: created.id }, data: { originalKey: key } })
+    return { assetId: created.id, key }
+  })
+}
+
+export async function finishReservedAsset(
+  deps: Deps,
+  asset: { assetId: string; key: string; kind: "IMAGE" | "VIDEO" },
+  options?: MediaProcessOptions,
+): Promise<boolean> {
+  const head = await deps.store.head(asset.key)
+  const max = asset.kind === "VIDEO" ? deps.config.maxVideoBytes : deps.config.maxImageBytes
+  if (!head || head.size <= 0 || head.size > max) return false
+  await deps.db.mediaAsset.update({ where: { id: asset.assetId }, data: { status: "PROCESSING", originalBytes: head.size } })
+  await deps.enqueue({ assetId: asset.assetId, ...(options ? { options } : {}) }, `${asset.assetId}-0`)
+  return true
+}

apps/media-service/src/generate/assets.ts

@@ -0,0 +1,118 @@
+import sharp from "sharp"
+import type { Prisma } from "@useframe/db"
+import {
+  deltaE76,
+  hexToRgb,
+  minDeltaE,
+  nearestStandardSize,
+  ProviderError,
+  requestSizeFor,
+  sha256,
+  type Rgb,
+  type Size,
+} from "@repo/media-ai"
+import { log } from "@/lib/logger.js"
+import type { GenerateDeps } from "./deps.js"
+
+export type Analysis = { paletteDeltaE: number; textZoneLuminance: number; textZoneVariance: number }
+
+// Crops to the target ratio and resizes, so every frame matches the video canvas exactly.
+export async function fitToSize(bytes: Buffer, target: Size): Promise<Buffer> {
+  return sharp(bytes).resize(target.width, target.height, { fit: "cover", position: "attention" }).png().toBuffer()
+}
+
+type ImageCall = {
+  prompt: string
+  target: Size
+  requestKey: string
+  edit?: Buffer[]
+}
+
+// Asks for the target size (or the smallest valid one above it); a rejected
+// size falls back to the nearest standard size, then everything is cropped to the target.
+export async function generateImage(deps: GenerateDeps, call: ImageCall): Promise<{ bytes: Buffer; costUsd: number; model: string; size: string }> {
+  return deps.semaphores.image.run(async () => {
+    const attempt = async (size: Size) => {
+      const s = `${size.width}x${size.height}`
+      const req = { prompt: call.prompt, size: s, quality: deps.config.imageQuality, format: "png" as const, requestKey: call.requestKey }
+      const result = call.edit?.length ? await deps.images.edit({ ...req, images: call.edit }) : await deps.images.generate(req)
+      return { ...result, size: s }
+    }
+    let result
+    try {
+      result = await attempt(requestSizeFor(call.target))
+    } catch (err) {
+      if (!(err instanceof ProviderError) || err.kind !== "INVALID_SIZE") throw err
+      const fallback = nearestStandardSize(call.target)
+      log.warn("image size rejected; using a standard size", { requested: requestSizeFor(call.target), fallback })
+      result = await attempt(fallback)
+    }
+    return { bytes: await fitToSize(result.bytes, call.target), costUsd: result.costUsd ?? deps.config.ai.providerCostUsd.image, model: result.model, size: result.size }
+  })
+}
+
+// Soft signals only: palette distance of the dominant colours and how calm the headline area is.
+export async function analyzeImage(bytes: Buffer, palette: string[], zone: string): Promise<Analysis> {
+  const { data, info } = await sharp(bytes).resize(64, 64, { fit: "fill" }).removeAlpha().raw().toBuffer({ resolveWithObject: true })
+  const buckets = new Map<string, { rgb: Rgb; count: number }>()
+  for (let i = 0; i < data.length; i += 3) {
+    const rgb: Rgb = [data[i]! & 0xe0, data[i + 1]! & 0xe0, data[i + 2]! & 0xe0]
+    const key = rgb.join(",")
+    const b = buckets.get(key)
+    if (b) b.count++
+    else buckets.set(key, { rgb, count: 1 })
+  }
+  const dominant = [...buckets.values()].sort((a, b) => b.count - a.count).slice(0, 5).map((b) => b.rgb)
+  const pal = palette.flatMap((h) => (hexToRgb(h) ? [hexToRgb(h)!] : []))
+  const paletteDeltaE = pal.length ? Math.round(Math.min(...dominant.map((c) => minDeltaE(c, pal))) * 10) / 10 : 0
+
+  const third = Math.floor(info.width / 3)
+  const [x0, x1] = zone === "right" ? [info.width - third, info.width] : zone === "center" ? [third, 2 * third] : [0, third]
+  const lums: number[] = []
+  for (let y = 0; y < info.height; y++) {
+    for (let x = x0; x < x1; x++) {
+      const i = (y * info.width + x) * 3
+      lums.push((0.2126 * data[i]! + 0.7152 * data[i + 1]! + 0.0722 * data[i + 2]!) / 255)
+    }
+  }
+  const mean = lums.reduce((a, b) => a + b, 0) / lums.length
+  const variance = lums.reduce((a, b) => a + (b - mean) ** 2, 0) / lums.length
+  return { paletteDeltaE, textZoneLuminance: Math.round(mean * 1000) / 1000, textZoneVariance: Math.round(variance * 10000) / 10000 }
+}
+
+// Scrim hint for background slots: needed when the headline area is busy or mid-grey.
+export function overlayHint(analysis: Analysis, slotDefault: "none" | "light" | "dark" | undefined): "none" | "light" | "dark" {
+  const wanted = slotDefault ?? "dark"
+  if (wanted === "none") return "none"
+  const busy = analysis.textZoneVariance > 0.02
+  if (!busy && wanted === "dark" && analysis.textZoneLuminance < 0.35) return "none"
+  if (!busy && wanted === "light" && analysis.textZoneLuminance > 0.7) return "none"
+  return wanted
+}
+
+export async function waitForAsset(deps: GenerateDeps, assetId: string): Promise<"READY" | "FAILED"> {
+  const deadline = Date.now() + deps.config.assetTimeoutMs
+  let wait = 1000
+  for (;;) {
+    const asset = await deps.db.mediaAsset.findUnique({ where: { id: assetId }, select: { status: true } })
+    if (asset?.status === "READY" || asset?.status === "FAILED") return asset.status
+    if (!asset || Date.now() > deadline) return "FAILED"
+    await deps.sleep(wait)
+    wait = Math.min(wait * 1.5, 5000)
+  }
+}
+
+export async function variantBytes(deps: GenerateDeps, assetId: string, role: string): Promise<Buffer | null> {
+  const asset = await deps.db.mediaAsset.findUnique({ where: { id: assetId }, select: { variants: true } })
+  const variants = (Array.isArray(asset?.variants) ? asset.variants : []) as { role?: string; key?: string }[]
+  const key = variants.find((v) => v.role === role)?.key
+  return key ? deps.store.getBuffer(key, 200 * 1024 * 1024) : null
+}
+
+export async function updateAsset(deps: GenerateDeps, assetId: string, data: Prisma.MediaAssetUpdateInput): Promise<void> {
+  await deps.db.mediaAsset.update({ where: { id: assetId }, data })
+}
+
+export const promptHash = (prompt: string) => sha256(prompt).slice(0, 12)
+
+export { deltaE76 }

apps/media-service/src/generate/context.ts

@@ -0,0 +1,99 @@
+import { randomUUID } from "node:crypto"
+import { readFile, rm, writeFile } from "node:fs/promises"
+import path from "node:path"
+import { prisma } from "@useframe/db"
+import {
+  FakeImageProvider,
+  FakeLlm,
+  FakeVideoProvider,
+  loadMediaAiConfig,
+  OpenAiImageProvider,
+  RunwareVideoProvider,
+} from "@repo/media-ai"
+import { env } from "@/config/env.js"
+import { deps as processDeps } from "@/context.js"
+import { createAssetFromBytes, finishReservedAsset, reserveGeneratedAsset } from "@/createAsset.js"
+import { createOrchestratorLlm } from "@/lib/llmClient.js"
+import { redis } from "@/lib/redis.js"
+import { safeDownload } from "@/lib/safeDownload.js"
+import { localSemaphore, redisSemaphore } from "@/lib/semaphore.js"
+import { createS3MediaStore } from "@/lib/storage.js"
+import { runTool } from "@/processing/ffmpeg.js"
+import type { GenerateDeps } from "./deps.js"
+
+const live = env.MEDIA_AI_PROVIDER_MODE === "live"
+const ffmpeg = processDeps.config.ffmpeg
+
+async function scratch<T>(fn: (dir: (name: string) => string) => Promise<T>): Promise<T> {
+  const id = randomUUID()
+  const file = (name: string) => path.join(env.MEDIA_WORK_DIR, `${id}-${name}`)
+  try {
+    return await fn(file)
+  } finally {
+    await Promise.all(["in.mp4", "out.png", "fake.mp4"].map((n) => rm(file(n), { force: true })))
+  }
+}
+
+async function extractFrame(video: Buffer, atSec: number): Promise<Buffer> {
+  return scratch(async (file) => {
+    await writeFile(file("in.mp4"), video)
+    await runTool(
+      ffmpeg.ffmpegPath,
+      ["-nostdin", "-loglevel", "error", "-y", "-ss", String(atSec), "-protocol_whitelist", "file", "-i", file("in.mp4"), "-frames:v", "1", "-f", "image2", "-c:v", "png", file("out.png")],
+      ffmpeg.timeoutMs,
+    )
+    return readFile(file("out.png"))
+  })
+}
+
+// Fake mode stands in for a provider download with a short, silent test clip.
+async function fakeVideo(): Promise<Buffer> {
+  return scratch(async (file) => {
+    await runTool(
+      ffmpeg.ffmpegPath,
+      ["-nostdin", "-loglevel", "error", "-y", "-f", "lavfi", "-i", "testsrc2=size=1344x768:rate=24:duration=5", "-c:v", "libx264", "-pix_fmt", "yuv420p", file("fake.mp4")],
+      ffmpeg.timeoutMs,
+    )
+    return readFile(file("fake.mp4"))
+  })
+}
+
+const store = createS3MediaStore({
+  endpoint: env.MEDIA_R2_ENDPOINT,
+  accessKeyId: env.MEDIA_R2_ACCESS_KEY_ID,
+  secretAccessKey: env.MEDIA_R2_SECRET_ACCESS_KEY,
+  bucket: env.MEDIA_R2_BUCKET,
+})
+
+export const generateDeps: GenerateDeps = {
+  db: prisma,
+  store,
+  llm: live ? createOrchestratorLlm({ url: env.ORCHESTRATOR_URL, secret: env.INTERNAL_SERVICE_SECRET! }) : new FakeLlm(),
+  images: live ? new OpenAiImageProvider({ apiKey: env.OPENAI_API_KEY!, model: env.OPENAI_IMAGE_MODEL }) : new FakeImageProvider(),
+  video: live ? new RunwareVideoProvider({ apiKey: env.RUNWARE_API_KEY!, model: env.MEDIA_VIDEO_MODEL }) : new FakeVideoProvider(),
+  createAsset: (input) => createAssetFromBytes(processDeps, input),
+  reserveAsset: (input) => reserveGeneratedAsset(processDeps, input),
+  finishReserved: (asset) => finishReservedAsset(processDeps, { ...asset, kind: "VIDEO" }, { stripAudio: true, skipDescribe: true }),
+  download: (url) =>
+    url.startsWith("fake://") && !live
+      ? fakeVideo()
+      : safeDownload(url, { allowedHosts: env.MEDIA_PROVIDER_HOST_ALLOWLIST, maxBytes: 200 * 1024 * 1024 }),
+  extractFrame,
+  semaphores: {
+    image: redisSemaphore(redis, "image", env.MEDIA_IMAGE_CONCURRENCY),
+    video: redisSemaphore(redis, "video", env.MEDIA_VIDEO_CONCURRENCY),
+    // Replaced per job by the runner: concepts are written one video at a time.
+    concepts: localSemaphore(1),
+  },
+  config: {
+    ai: loadMediaAiConfig(process.env),
+    imageQuality: env.OPENAI_IMAGE_QUALITY,
+    videoResolution: env.MEDIA_VIDEO_RESOLUTION,
+    videoTimeoutS: env.MEDIA_VIDEO_TIMEOUT_S,
+    promptExpansion: env.MEDIA_VIDEO_PROMPT_EXPANSION,
+    promptMaxChars: env.MEDIA_VIDEO_PROMPT_MAX_CHARS,
+    assetTimeoutMs: env.MEDIA_FFMPEG_TIMEOUT_MS * 2 + 60_000,
+    directUpload: live,
+  },
+  sleep: (ms) => new Promise((r) => setTimeout(r, ms)),
+}