diff --git a/.github/workflows/helm-ci.yml b/.github/workflows/helm-ci.yml index 90e9ab8..0cf0f76 100644 --- a/.github/workflows/helm-ci.yml +++ b/.github/workflows/helm-ci.yml @@ -66,7 +66,7 @@ jobs: run: | set -euo pipefail crds="https://raw.githubusercontent.com/datreeio/CRDs-catalog/${CRDS_CATALOG_REF}/{{.Group}}/{{.ResourceKind}}_{{.ResourceAPIVersion}}.json" - all_on="--set services.deploy-service-worker.enabled=true --set services.billing-service.enabled=true --set services.scoring-service.enabled=true --set networkPolicy.enabled=true --set migration.enabled=true" + all_on="--set services.deploy-service-worker.enabled=true --set services.media-service.enabled=true --set services.billing-service.enabled=true --set services.scoring-service.enabled=true --set networkPolicy.enabled=true --set migration.enabled=true" render() { echo "::group::$1" shift diff --git a/AGENTS.md b/AGENTS.md index dcf331b..e27ccc3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,6 +42,7 @@ Read the implementation and package manifests as the source of truth. See `READM | `apps/worker` | BullMQ processors for scans, SEO, scoring, domains, PDFs, billing jobs, and cache expiry | | `apps/deploy-service` | Site builds, R2 uploads, KV publishing (`syncSiteToKvs`), rollback, GC, reaper, reconcile; see `RUNBOOK.md` | | `apps/site-edge` | Cloudflare Worker serving generated sites from KV + R2; deployed with Wrangler, never from CI | +| `apps/media-service` | BullMQ worker for the media library: upload processing (sharp/ffmpeg), vision descriptions, media GC | | `infra/terraform`, `infra/k8s` | Cloudflare sites-hosting Terraform module and GKE manifests (applied by a human, never by agents) | | `infra/helm/useframe` | Helm chart for the backend workloads on GKE; agents run only offline checks (lint/template/unittest) | | `packages/db` | Shared Prisma client, schema, migrations, seed data, and scan cache helpers | @@ -155,7 +156,7 @@ Run these from the repository root unless otherwise noted: - There is no root `test` script. Run relevant existing tests/evaluations where available and report coverage gaps instead of inventing a passing test command. - Shared packages such as db, schemas, events, and site-builder export `dist`; rebuild them after changes before restarting consumers (`@useframe/db` runs `prisma generate` then `tsc`). A running watch process does not prove shared output is current. - Backend services build with `tsc && tsc-alias` (the `@/` alias becomes relative paths) and run in production with plain `node dist/...`; never ship `tsx` in an image. Runtime workspace packages declare `files` (normally `dist`) because `pnpm deploy` only copies published files. -- Service default ports: API 4000, orchestrator 4001, billing 4002, scoring 4003, research 4004, deploy 4005. Confirm actual configured ports and probe their `/health` routes. Check the frontend HTTP response and worker startup/Redis connectivity separately; the worker has no HTTP health endpoint. +- Service default ports: API 4000, orchestrator 4001, billing 4002, scoring 4003, research 4004, deploy 4005. Confirm actual configured ports and probe their `/health` routes. Check the frontend HTTP response and worker startup/Redis connectivity separately; the worker and media-service have no HTTP endpoint beyond an optional `HEALTH_PORT` listener (media-service). - Identify project-owned processes before restarting them. Do not kill every Node process. Leave the updated development services running after verification; on Windows, launch background helpers without visible windows. - Check required PostgreSQL, Redis, and external-service configuration before starting dependent flows. If configuration or infrastructure prevents startup, report the exact blocker and what could be verified. diff --git a/CLAUDE.md b/CLAUDE.md index dcf331b..e27ccc3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -42,6 +42,7 @@ Read the implementation and package manifests as the source of truth. See `READM | `apps/worker` | BullMQ processors for scans, SEO, scoring, domains, PDFs, billing jobs, and cache expiry | | `apps/deploy-service` | Site builds, R2 uploads, KV publishing (`syncSiteToKvs`), rollback, GC, reaper, reconcile; see `RUNBOOK.md` | | `apps/site-edge` | Cloudflare Worker serving generated sites from KV + R2; deployed with Wrangler, never from CI | +| `apps/media-service` | BullMQ worker for the media library: upload processing (sharp/ffmpeg), vision descriptions, media GC | | `infra/terraform`, `infra/k8s` | Cloudflare sites-hosting Terraform module and GKE manifests (applied by a human, never by agents) | | `infra/helm/useframe` | Helm chart for the backend workloads on GKE; agents run only offline checks (lint/template/unittest) | | `packages/db` | Shared Prisma client, schema, migrations, seed data, and scan cache helpers | @@ -155,7 +156,7 @@ Run these from the repository root unless otherwise noted: - There is no root `test` script. Run relevant existing tests/evaluations where available and report coverage gaps instead of inventing a passing test command. - Shared packages such as db, schemas, events, and site-builder export `dist`; rebuild them after changes before restarting consumers (`@useframe/db` runs `prisma generate` then `tsc`). A running watch process does not prove shared output is current. - Backend services build with `tsc && tsc-alias` (the `@/` alias becomes relative paths) and run in production with plain `node dist/...`; never ship `tsx` in an image. Runtime workspace packages declare `files` (normally `dist`) because `pnpm deploy` only copies published files. -- Service default ports: API 4000, orchestrator 4001, billing 4002, scoring 4003, research 4004, deploy 4005. Confirm actual configured ports and probe their `/health` routes. Check the frontend HTTP response and worker startup/Redis connectivity separately; the worker has no HTTP health endpoint. +- Service default ports: API 4000, orchestrator 4001, billing 4002, scoring 4003, research 4004, deploy 4005. Confirm actual configured ports and probe their `/health` routes. Check the frontend HTTP response and worker startup/Redis connectivity separately; the worker and media-service have no HTTP endpoint beyond an optional `HEALTH_PORT` listener (media-service). - Identify project-owned processes before restarting them. Do not kill every Node process. Leave the updated development services running after verification; on Windows, launch background helpers without visible windows. - Check required PostgreSQL, Redis, and external-service configuration before starting dependent flows. If configuration or infrastructure prevents startup, report the exact blocker and what could be verified. diff --git a/GEMINI.md b/GEMINI.md index dcf331b..e27ccc3 100644 --- a/GEMINI.md +++ b/GEMINI.md @@ -42,6 +42,7 @@ Read the implementation and package manifests as the source of truth. See `READM | `apps/worker` | BullMQ processors for scans, SEO, scoring, domains, PDFs, billing jobs, and cache expiry | | `apps/deploy-service` | Site builds, R2 uploads, KV publishing (`syncSiteToKvs`), rollback, GC, reaper, reconcile; see `RUNBOOK.md` | | `apps/site-edge` | Cloudflare Worker serving generated sites from KV + R2; deployed with Wrangler, never from CI | +| `apps/media-service` | BullMQ worker for the media library: upload processing (sharp/ffmpeg), vision descriptions, media GC | | `infra/terraform`, `infra/k8s` | Cloudflare sites-hosting Terraform module and GKE manifests (applied by a human, never by agents) | | `infra/helm/useframe` | Helm chart for the backend workloads on GKE; agents run only offline checks (lint/template/unittest) | | `packages/db` | Shared Prisma client, schema, migrations, seed data, and scan cache helpers | @@ -155,7 +156,7 @@ Run these from the repository root unless otherwise noted: - There is no root `test` script. Run relevant existing tests/evaluations where available and report coverage gaps instead of inventing a passing test command. - Shared packages such as db, schemas, events, and site-builder export `dist`; rebuild them after changes before restarting consumers (`@useframe/db` runs `prisma generate` then `tsc`). A running watch process does not prove shared output is current. - Backend services build with `tsc && tsc-alias` (the `@/` alias becomes relative paths) and run in production with plain `node dist/...`; never ship `tsx` in an image. Runtime workspace packages declare `files` (normally `dist`) because `pnpm deploy` only copies published files. -- Service default ports: API 4000, orchestrator 4001, billing 4002, scoring 4003, research 4004, deploy 4005. Confirm actual configured ports and probe their `/health` routes. Check the frontend HTTP response and worker startup/Redis connectivity separately; the worker has no HTTP health endpoint. +- Service default ports: API 4000, orchestrator 4001, billing 4002, scoring 4003, research 4004, deploy 4005. Confirm actual configured ports and probe their `/health` routes. Check the frontend HTTP response and worker startup/Redis connectivity separately; the worker and media-service have no HTTP endpoint beyond an optional `HEALTH_PORT` listener (media-service). - Identify project-owned processes before restarting them. Do not kill every Node process. Leave the updated development services running after verification; on Windows, launch background helpers without visible windows. - Check required PostgreSQL, Redis, and external-service configuration before starting dependent flows. If configuration or infrastructure prevents startup, report the exact blocker and what could be verified. diff --git a/apps/deploy-service/.env.example b/apps/deploy-service/.env.example index 6571008..0260e7b 100644 --- a/apps/deploy-service/.env.example +++ b/apps/deploy-service/.env.example @@ -26,6 +26,14 @@ SITES_R2_SECRET_ACCESS_KEY= # Defaults to https://.r2.cloudflarestorage.com SITES_R2_ENDPOINT= +# Media library (read-only token on useframe-media). Needed only to deploy sites that use media; +# such deploys fail with a clear message while these are unset. +MEDIA_R2_BUCKET=useframe-media +MEDIA_R2_READ_ACCESS_KEY_ID= +MEDIA_R2_READ_SECRET_ACCESS_KEY= +# Defaults to https://.r2.cloudflarestorage.com +MEDIA_R2_ENDPOINT= + DEPLOY_RETAIN_COUNT=10 DEPLOY_MAX_FILES=5000 DEPLOY_MAX_TOTAL_MB=100 diff --git a/apps/deploy-service/src/config/schema.ts b/apps/deploy-service/src/config/schema.ts index 0e73e1a..413e2bf 100644 --- a/apps/deploy-service/src/config/schema.ts +++ b/apps/deploy-service/src/config/schema.ts @@ -43,6 +43,13 @@ export const envSchema = z SITES_R2_SECRET_ACCESS_KEY: z.string().min(1, "SITES_R2_SECRET_ACCESS_KEY is required"), SITES_R2_ENDPOINT: optional(z.string().url()), + // Read-only access to the private media library, used to copy referenced + // media into each deployment. Deploys of sites without media work without it. + MEDIA_R2_BUCKET: z.string().min(1).default("useframe-media"), + MEDIA_R2_READ_ACCESS_KEY_ID: optional(z.string().min(1)), + MEDIA_R2_READ_SECRET_ACCESS_KEY: optional(z.string().min(1)), + MEDIA_R2_ENDPOINT: optional(z.string().url()), + DEPLOY_RETAIN_COUNT: int(10), DEPLOY_MAX_FILES: int(5000), DEPLOY_MAX_TOTAL_MB: int(100), @@ -88,6 +95,8 @@ export const envSchema = z ...value, SITES_R2_ENDPOINT: value.SITES_R2_ENDPOINT ?? `https://${value.CLOUDFLARE_ACCOUNT_ID}.r2.cloudflarestorage.com`, + MEDIA_R2_ENDPOINT: + value.MEDIA_R2_ENDPOINT ?? `https://${value.CLOUDFLARE_ACCOUNT_ID}.r2.cloudflarestorage.com`, SITES_EDGE_CNAME_TARGET: value.SITES_EDGE_CNAME_TARGET ?? `cname.${value.SITES_BASE_DOMAIN}`, DEPLOY_WORK_DIR: value.DEPLOY_WORK_DIR ?? diff --git a/apps/deploy-service/src/context.ts b/apps/deploy-service/src/context.ts index a4f54b8..0aa04bb 100644 --- a/apps/deploy-service/src/context.ts +++ b/apps/deploy-service/src/context.ts @@ -31,6 +31,15 @@ export const deps: Deps = { secretAccessKey: env.SITES_R2_SECRET_ACCESS_KEY, bucket: env.SITES_R2_BUCKET, }), + mediaSource: + env.MEDIA_R2_READ_ACCESS_KEY_ID && env.MEDIA_R2_READ_SECRET_ACCESS_KEY + ? createR2({ + endpoint: env.MEDIA_R2_ENDPOINT, + accessKeyId: env.MEDIA_R2_READ_ACCESS_KEY_ID, + secretAccessKey: env.MEDIA_R2_READ_SECRET_ACCESS_KEY, + bucket: env.MEDIA_R2_BUCKET, + }) + : null, redis, runQueue, validationQueue: env.DEPLOY_ENQUEUE_VALIDATION diff --git a/apps/deploy-service/src/deps.ts b/apps/deploy-service/src/deps.ts index 686f6b0..c2ac080 100644 --- a/apps/deploy-service/src/deps.ts +++ b/apps/deploy-service/src/deps.ts @@ -37,6 +37,8 @@ export type Deps = { db: Db; kv: KvClient; r2: R2; + // Read-only client for the private media library; absent until configured. + mediaSource?: R2 | null; redis: LockClient; runQueue: Pick, "add" | "getJob">; validationQueue?: Pick; diff --git a/apps/deploy-service/src/pipeline/files.ts b/apps/deploy-service/src/pipeline/files.ts index 5ed7b76..b06f56f 100644 --- a/apps/deploy-service/src/pipeline/files.ts +++ b/apps/deploy-service/src/pipeline/files.ts @@ -1,26 +1,42 @@ import { z } from "zod"; -import { SiteSpecSchema } from "@repo/schemas"; +import { parseMediaVariants, SiteSpecSchema, type ResolvedMediaAsset } from "@repo/schemas"; import { buildRobotsTxt, buildSitemapXml, - buildSiteFiles, + buildSite, withScaffold, type GeneratedFile, } from "@repo/site-builder"; import type { Db } from "@/site/sync.js"; import { DeployError } from "./errors.js"; +import type { MediaCopyItem } from "./media.js"; const NextFilesSchema = z.array(z.object({ path: z.string().min(1), content: z.string() })).min(1); export type VersionSource = { snapshot: unknown; nextFiles: unknown }; +// Media as loaded from the library: resolved assets for the builder plus the +// source key of every variant, so the copy step can find what the markup uses. +export type VersionMedia = { + assets: ResolvedMediaAsset[]; + sources: Map; +}; + +export type BuildableVersion = { files: GeneratedFile[]; media: MediaCopyItem[]; mediaWarnings: string[] }; + +const NO_MEDIA: VersionMedia = { assets: [], sources: new Map() }; + // Replicate versions carry a raw Next.js tree; generated versions carry a SiteSpec. -export function filesFromVersion(version: VersionSource, siteUrl: string): GeneratedFile[] { +export function filesFromVersion( + version: VersionSource, + siteUrl: string, + media: VersionMedia = NO_MEDIA, +): BuildableVersion { if (version.nextFiles !== null && version.nextFiles !== undefined) { const parsed = NextFilesSchema.safeParse(version.nextFiles); if (!parsed.success) throw new DeployError("This version's files are not in a deployable format."); try { - return withScaffold(parsed.data); + return { files: withScaffold(parsed.data), media: [], mediaWarnings: [] }; } catch (err) { throw new DeployError( "This version's files are not in a deployable format.", @@ -31,20 +47,92 @@ export function filesFromVersion(version: VersionSource, siteUrl: string): Gener const spec = SiteSpecSchema.safeParse(version.snapshot); if (!spec.success) throw new DeployError("This version's site data is not in a deployable format."); - return buildSiteFiles(spec.data, { - robotsTxt: buildRobotsTxt(siteUrl), - sitemapXml: buildSitemapXml(spec.data.pages, siteUrl), + const built = buildSite( + spec.data, + { robotsTxt: buildRobotsTxt(siteUrl), sitemapXml: buildSitemapXml(spec.data.pages, siteUrl) }, + { media: { target: "export", assets: media.assets } }, + ); + const items = built.mediaFiles.map((file) => { + const source = media.sources.get(`${file.assetId}/${file.role}`); + if (!source) throw new DeployError("A media file on this site could not be found.", `${file.assetId}/${file.role}`); + const variant = media.assets.find((a) => a.id === file.assetId)!.variants.find((v) => v.role === file.role)!; + return { + assetId: file.assetId, + title: source.title, + sourceKey: source.key, + destRelPath: file.path, + bytes: file.bytes, + mime: variant.mime, + }; + }); + return { files: built.files, media: items, mediaWarnings: built.mediaWarnings }; +} + +export function referencedAssetIds(snapshot: unknown): string[] { + const media = (snapshot as { media?: unknown } | null)?.media; + if (!media || typeof media !== "object") return []; + const ids = Object.values(media as Record).flatMap((b) => + typeof b?.assetId === "string" ? [b.assetId] : [], + ); + return [...new Set(ids)]; +} + +export async function loadVersionMedia(db: Db, projectId: string, assetIds: string[]): Promise { + if (assetIds.length === 0) return NO_MEDIA; + // Scoped to the project, so a binding can never pull another project's media. + const rows = await db.mediaAsset.findMany({ + where: { id: { in: assetIds }, projectId }, + select: { + id: true, + kind: true, + origin: true, + status: true, + deletedAt: true, + title: true, + width: true, + height: true, + durationMs: true, + dominantColor: true, + lqip: true, + altText: true, + decorative: true, + variants: true, + }, + }); + const sources = new Map(); + const assets = rows.map((row): ResolvedMediaAsset => { + const variants = parseMediaVariants(row.variants); + for (const v of variants) sources.set(`${row.id}/${v.role}`, { key: v.key, title: row.title ?? "Untitled" }); + return { + id: row.id, + kind: row.kind, + origin: row.origin, + status: row.status, + deleted: row.deletedAt !== null, + title: row.title, + width: row.width, + height: row.height, + durationMs: row.durationMs, + dominantColor: row.dominantColor, + lqip: row.lqip, + altText: row.altText, + decorative: row.decorative, + variants: variants.map(({ role, mime, width, height, bytes, sha256 }) => ({ role, mime, width, height, bytes, sha256 })), + }; }); + return { assets, sources }; } export async function getBuildableFiles( db: Db, versionId: string, siteUrl: string, -): Promise { + projectId: string, +): Promise { const version = await db.projectVersion.findUniqueOrThrow({ where: { id: versionId }, select: { snapshot: true, nextFiles: true }, }); - return filesFromVersion(version, siteUrl); + const media = await loadVersionMedia(db, projectId, referencedAssetIds(version.snapshot)); + return filesFromVersion(version, siteUrl, media); } diff --git a/apps/deploy-service/src/pipeline/media.ts b/apps/deploy-service/src/pipeline/media.ts new file mode 100644 index 0000000..a166cda --- /dev/null +++ b/apps/deploy-service/src/pipeline/media.ts @@ -0,0 +1,125 @@ +import { CopyObjectCommand, GetObjectCommand, HeadObjectCommand, PutObjectCommand } from "@aws-sdk/client-s3"; +import type { Readable } from "node:stream"; +import type { R2 } from "@/lib/r2.js"; +import { log } from "@/lib/logger.js"; +import { mapWithConcurrency, withRetry } from "@/lib/retry.js"; +import { DeployError } from "./errors.js"; + +export type MediaCopyItem = { + assetId: string; + // User-visible name for error messages; raw storage keys are never shown. + title: string; + sourceKey: string; + // Relative to the deployment prefix, e.g. media//w960.1a2b3c4d.webp + destRelPath: string; + bytes: number; + mime: string; +}; + +const IMMUTABLE = "public, max-age=31536000, immutable"; + +export class MissingMediaError extends DeployError { + constructor(title: string) { + super(`The media "${title}" is missing from storage. Remove it from the site or upload it again, then redeploy.`); + } +} + +function isNotFound(err: unknown): boolean { + if (!(err instanceof Error)) return false; + const status = (err as { $metadata?: { httpStatusCode?: number } }).$metadata?.httpStatusCode; + return err.name === "NoSuchKey" || err.name === "NotFound" || status === 404; +} + +// Scoped tokens may refuse cross-bucket CopyObject; after one refusal this process streams instead. +let preferStream = false; +const loggedPaths = new Set(); + +function logPath(path: "copy" | "stream"): void { + if (loggedPaths.has(path)) return; + loggedPaths.add(path); + log.info("media copy path in use", { path }); +} + +export function resetMediaCopyStateForTests(): void { + preferStream = false; + loggedPaths.clear(); +} + +async function sourceExists(source: R2, key: string): Promise { + try { + await source.client.send(new HeadObjectCommand({ Bucket: source.bucket, Key: key })); + return true; + } catch (err) { + if (isNotFound(err)) return false; + throw err; + } +} + +async function streamCopy(source: R2, dest: R2, item: MediaCopyItem, destKey: string): Promise { + let body: Readable; + try { + const res = await source.client.send(new GetObjectCommand({ Bucket: source.bucket, Key: item.sourceKey })); + if (!res.Body) throw new MissingMediaError(item.title); + body = res.Body as Readable; + } catch (err) { + if (isNotFound(err)) throw new MissingMediaError(item.title); + throw err; + } + await dest.client.send( + new PutObjectCommand({ + Bucket: dest.bucket, + Key: destKey, + Body: body, + ContentLength: item.bytes, + ContentType: item.mime, + CacheControl: IMMUTABLE, + }), + ); + logPath("stream"); +} + +async function copyOne(source: R2, dest: R2, item: MediaCopyItem, destKey: string): Promise { + if (!preferStream) { + try { + await dest.client.send( + new CopyObjectCommand({ + Bucket: dest.bucket, + Key: destKey, + CopySource: `${source.bucket}/${item.sourceKey.split("/").map(encodeURIComponent).join("/")}`, + MetadataDirective: "REPLACE", + ContentType: item.mime, + CacheControl: IMMUTABLE, + }), + ); + logPath("copy"); + return; + } catch (err) { + // Only the read token can tell a refused copy from a missing object. + if (!(await sourceExists(source, item.sourceKey))) throw new MissingMediaError(item.title); + preferStream = true; + log.warn("media CopyObject failed; streaming instead", { assetId: item.assetId, error: (err as Error).name }); + } + } + await streamCopy(source, dest, item, destKey); +} + +// Deployed sites keep their own copy, so later library changes never break them. +export async function copyMediaToSite( + source: R2 | null | undefined, + dest: R2, + prefix: string, + items: MediaCopyItem[], +): Promise { + if (items.length === 0) return 0; + if (!source) { + throw new DeployError("This site uses media, but media storage is not configured for deploys.", "MEDIA_R2_READ_* unset"); + } + await mapWithConcurrency(items, 4, (item) => + withRetry(() => copyOne(source, dest, item, prefix + item.destRelPath), { + attempts: 3, + baseDelayMs: 500, + shouldRetry: (err) => !(err instanceof DeployError), + }), + ); + return items.reduce((sum, item) => sum + item.bytes, 0); +} diff --git a/apps/deploy-service/src/pipeline/runDeployment.ts b/apps/deploy-service/src/pipeline/runDeployment.ts index 4dce7df..142db88 100644 --- a/apps/deploy-service/src/pipeline/runDeployment.ts +++ b/apps/deploy-service/src/pipeline/runDeployment.ts @@ -10,6 +10,7 @@ import { buildChildEnv, INSTALL_ARGS, runChild, type ChildResult } from "./build import { dependencyHash, detectFramework, readPackageJson } from "./detect.js"; import { DeployError, publicReason } from "./errors.js"; import { getBuildableFiles } from "./files.js"; +import { copyMediaToSite } from "./media.js"; import { failDeployment, goLive, tail, transition } from "./lifecycle.js"; import { copyTemplate, prepareWorkDir, removeWorkDir, writeTree, type WorkDirs } from "./materialize.js"; import { probeDeployment } from "./probe.js"; @@ -58,7 +59,8 @@ export async function runDeployment(deps: Deps, job: DeployRunJobPayload): Promi const siteUrl = `https://${site.primaryHost}`; let t = Date.now(); - const files = await getBuildableFiles(deps.db, job.versionId, siteUrl); + const { files, media, mediaWarnings } = await getBuildableFiles(deps.db, job.versionId, siteUrl, projectId); + if (mediaWarnings.length) log.warn("media warnings", { deploymentId, warnings: mediaWarnings }); const detected = detectFramework(files); dirs = await prepareWorkDir(config.workDir, deploymentId, owner); await writeTree(dirs.src, files, owner); @@ -103,14 +105,22 @@ export async function runDeployment(deps: Deps, job: DeployRunJobPayload): Promi maxTotalBytes: config.maxTotalBytes, }); if (output.skipped.length) log.warn("skipped denied output files", { deploymentId, files: output.skipped }); + // Media is copied next to the build output and counts toward the same limits. + const mediaBytes = media.reduce((sum, item) => sum + item.bytes, 0); + const fileCount = output.fileCount + media.length; + const totalBytes = output.totalBytes + mediaBytes; + if (fileCount > config.maxFiles) throw new DeployError(`Build output has more than ${config.maxFiles} files.`); + if (totalBytes > config.maxTotalBytes) { + throw new DeployError(`Build output is larger than ${Math.round(config.maxTotalBytes / 1048576)} MB.`); + } const prefix = storagePrefixFor(projectId, deploymentId); const toUploading = await transition(deps, deploymentId, ["BUILDING"], { status: "UPLOADING", framework: detected.framework, storagePrefix: prefix, - fileCount: output.fileCount, - totalBytes: BigInt(output.totalBytes), + fileCount, + totalBytes: BigInt(totalBytes), rootHtmlSha256: output.rootHtmlSha256, siteUrl, buildLog: tail(buildLog), @@ -118,8 +128,9 @@ export async function runDeployment(deps: Deps, job: DeployRunJobPayload): Promi if (!toUploading) return; t = Date.now(); + await copyMediaToSite(deps.mediaSource, deps.r2, prefix, media); await uploadOutput(deps.r2, prefix, output.files); - await verifyUpload(deps.r2, prefix, output.fileCount, output.rootHtmlSha256); + await verifyUpload(deps.r2, prefix, fileCount, output.rootHtmlSha256); metric("deploy.upload_duration_ms", Date.now() - t, { deploymentId, files: output.fileCount }); step("upload", t); diff --git a/apps/deploy-service/test/isolation/run.mjs b/apps/deploy-service/test/isolation/run.mjs index a79022b..81c3725 100644 --- a/apps/deploy-service/test/isolation/run.mjs +++ b/apps/deploy-service/test/isolation/run.mjs @@ -89,7 +89,7 @@ await removeWorkDir(dirs.base); // A real Vite build of a generated site, offline, from the baked template. const site = await prepareWorkDir("/work", "vite-build", owner); -const files = filesFromVersion({ snapshot: FIXTURE_SPEC(2), nextFiles: null }, "https://acme-x7k.useframe.in"); +const { files } = filesFromVersion({ snapshot: FIXTURE_SPEC(2), nextFiles: null }, "https://acme-x7k.useframe.in"); const detected = detectFramework(files); await writeTree(site.src, files, owner); const usedTemplate = await copyTemplate( diff --git a/apps/deploy-service/test/media.test.ts b/apps/deploy-service/test/media.test.ts new file mode 100644 index 0000000..0f8bad9 --- /dev/null +++ b/apps/deploy-service/test/media.test.ts @@ -0,0 +1,182 @@ +import { Readable } from "node:stream"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { + CopyObjectCommand, + GetObjectCommand, + HeadObjectCommand, + PutObjectCommand, +} from "@aws-sdk/client-s3"; +import type { R2 } from "@/lib/r2.js"; +import { filesFromVersion, loadVersionMedia, referencedAssetIds } from "@/pipeline/files.js"; +import { copyMediaToSite, MissingMediaError, resetMediaCopyStateForTests, type MediaCopyItem } from "@/pipeline/media.js"; +import { DeployError } from "@/pipeline/errors.js"; + +type Store = Map; + +function notFound() { + return Object.assign(new Error("NoSuchKey"), { name: "NoSuchKey", $metadata: { httpStatusCode: 404 } }); +} + +// Two buckets behind two tokens; `canCrossCopy` decides whether the sites token +// may read the media bucket (the scoped tokens normally cannot). +function fakeR2(buckets: Record, opts: { canCrossCopy?: boolean; failGets?: number } = {}) { + const calls: string[] = []; + let failGets = opts.failGets ?? 0; + const make = (bucket: string): R2 => ({ + bucket, + client: { + send: vi.fn(async (cmd: unknown) => { + if (cmd instanceof CopyObjectCommand) { + calls.push("copy"); + if (!opts.canCrossCopy) throw Object.assign(new Error("AccessDenied"), { name: "AccessDenied", $metadata: { httpStatusCode: 403 } }); + const [srcBucket, ...rest] = cmd.input.CopySource!.split("/"); + const src = buckets[srcBucket!]!.get(rest.map(decodeURIComponent).join("/")); + if (!src) throw notFound(); + buckets[bucket]!.set(cmd.input.Key!, { body: src.body, contentType: cmd.input.ContentType, cacheControl: cmd.input.CacheControl }); + return {}; + } + if (cmd instanceof HeadObjectCommand) { + calls.push("head"); + if (!buckets[bucket]!.has(cmd.input.Key!)) throw notFound(); + return {}; + } + if (cmd instanceof GetObjectCommand) { + calls.push("get"); + if (failGets > 0) { + failGets--; + throw Object.assign(new Error("Service Unavailable"), { $metadata: { httpStatusCode: 503 } }); + } + const obj = buckets[bucket]!.get(cmd.input.Key!); + if (!obj) throw notFound(); + return { Body: Readable.from([obj.body]) }; + } + if (cmd instanceof PutObjectCommand) { + calls.push("put"); + const chunks: Buffer[] = []; + for await (const c of cmd.input.Body as Readable) chunks.push(Buffer.from(c as Uint8Array)); + buckets[bucket]!.set(cmd.input.Key!, { body: Buffer.concat(chunks), contentType: cmd.input.ContentType, cacheControl: cmd.input.CacheControl }); + return {}; + } + throw new Error("unexpected command"); + }), + } as unknown as R2["client"], + }); + return { media: make("useframe-media"), sites: make("useframe-sites"), calls }; +} + +const item = (id: string, overrides: Partial = {}): MediaCopyItem => ({ + assetId: id, + title: `Photo ${id}`, + sourceKey: `media/p1/${id}/v/w960.webp`, + destRelPath: `media/${id}/w960.abcdef12.webp`, + bytes: 5, + mime: "image/webp", + ...overrides, +}); + +let buckets: Record; + +beforeEach(() => { + resetMediaCopyStateForTests(); + buckets = { "useframe-media": new Map(), "useframe-sites": new Map() }; + for (const id of ["a1", "a2", "a3"]) buckets["useframe-media"]!.set(`media/p1/${id}/v/w960.webp`, { body: Buffer.from(id) }); +}); + +describe("copyMediaToSite", () => { + it("uses CopyObject when the token allows it, with immutable caching", async () => { + const r2 = fakeR2(buckets, { canCrossCopy: true }); + const bytes = await copyMediaToSite(r2.media, r2.sites, "sites/p1/d1/", [item("a1"), item("a2")]); + expect(bytes).toBe(10); + expect(buckets["useframe-sites"]!.get("sites/p1/d1/media/a1/w960.abcdef12.webp")).toMatchObject({ + body: Buffer.from("a1"), + contentType: "image/webp", + cacheControl: "public, max-age=31536000, immutable", + }); + expect(r2.calls.filter((c) => c === "get")).toHaveLength(0); + }); + + it("falls back to a streamed GET→PUT once CopyObject is refused, and stays on it", async () => { + const r2 = fakeR2(buckets, { canCrossCopy: false }); + await copyMediaToSite(r2.media, r2.sites, "sites/p1/d1/", [item("a1")]); + await copyMediaToSite(r2.media, r2.sites, "sites/p1/d2/", [item("a2"), item("a3")]); + expect(r2.calls.filter((c) => c === "copy")).toHaveLength(1); + expect(buckets["useframe-sites"]!.get("sites/p1/d2/media/a3/w960.abcdef12.webp")?.body).toEqual(Buffer.from("a3")); + expect([...buckets["useframe-sites"]!.keys()]).toHaveLength(3); + }); + + it("fails closed, naming the asset title, when a referenced variant is missing", async () => { + for (const canCrossCopy of [true, false]) { + resetMediaCopyStateForTests(); + const r2 = fakeR2(buckets, { canCrossCopy }); + const err = await copyMediaToSite(r2.media, r2.sites, "sites/p1/d1/", [item("a1"), item("gone", { title: "Team photo" })]).catch( + (e: unknown) => e, + ); + expect(err).toBeInstanceOf(MissingMediaError); + expect((err as DeployError).publicMessage).toContain('"Team photo"'); + expect((err as DeployError).publicMessage).not.toContain("media/p1"); + } + }); + + it("retries transient source errors", async () => { + const r2 = fakeR2(buckets, { canCrossCopy: false, failGets: 1 }); + await copyMediaToSite(r2.media, r2.sites, "sites/p1/d1/", [item("a1")]); + expect(buckets["useframe-sites"]!.has("sites/p1/d1/media/a1/w960.abcdef12.webp")).toBe(true); + }); + + it("refuses to deploy media without a configured source and is a no-op without media", async () => { + const r2 = fakeR2(buckets); + await expect(copyMediaToSite(null, r2.sites, "sites/p1/d1/", [item("a1")])).rejects.toBeInstanceOf(DeployError); + expect(await copyMediaToSite(null, r2.sites, "sites/p1/d1/", [])).toBe(0); + }); +}); + +describe("version media manifest", () => { + const sha = (c: string) => c.repeat(64); + const spec = { + siteType: "SINGLE_PAGE", + copyFramework: "AIDA", + citations: [], + designSystem: { primaryColor: "#000000", secondaryColor: "#000000", accentColor: "#000000", fontPrimary: "Inter", fontSecondary: "Inter", spacing: "", borderRadius: "md", animationStyle: "" }, + pages: [{ type: "HOME", slug: "home", title: "Home", sections: [{ type: "HERO", index: 0, content: { headline: "Hi" } }] }], + media: { "home/hero-0/visual": { assetId: "img1" }, "home/hero-0/background": { assetId: "purged" } }, + }; + + it("lists only the variants the export markup uses, with their library keys", async () => { + const db = { + mediaAsset: { + findMany: vi.fn(async () => [ + { + id: "img1", + kind: "IMAGE", + origin: "UPLOADED", + status: "READY", + deletedAt: null, + title: "Studio", + width: 1920, + height: 1080, + durationMs: null, + dominantColor: null, + lqip: null, + altText: "Studio", + decorative: false, + variants: [ + { role: "w960", key: "media/p1/img1/v/w960.webp", mime: "image/webp", width: 960, height: 540, bytes: 100, sha256: sha("a") }, + { role: "fallback", key: "media/p1/img1/v/fallback.jpg", mime: "image/jpeg", width: 1600, height: 900, bytes: 200, sha256: sha("b") }, + { role: "thumb", key: "media/p1/img1/v/thumb.webp", mime: "image/webp", width: 320, height: 180, bytes: 10, sha256: sha("c") }, + ], + }, + ]), + }, + }; + expect(referencedAssetIds(spec)).toEqual(["img1", "purged"]); + const media = await loadVersionMedia(db as never, "p1", referencedAssetIds(spec)); + expect(db.mediaAsset.findMany).toHaveBeenCalledWith(expect.objectContaining({ where: { id: { in: ["img1", "purged"] }, projectId: "p1" } })); + const built = filesFromVersion({ snapshot: spec, nextFiles: null }, "https://acme.useframe.in", media); + expect(built.media).toEqual([ + { assetId: "img1", title: "Studio", sourceKey: "media/p1/img1/v/w960.webp", destRelPath: "media/img1/w960.aaaaaaaa.webp", bytes: 100, mime: "image/webp" }, + { assetId: "img1", title: "Studio", sourceKey: "media/p1/img1/v/fallback.jpg", destRelPath: "media/img1/fallback.bbbbbbbb.jpg", bytes: 200, mime: "image/jpeg" }, + ]); + expect(built.mediaWarnings).toEqual(["home/hero-0/background: media dropped (asset not found)"]); + expect(built.files.find((f) => f.path === "src/App.jsx")!.content).toContain('"/media/img1/fallback.bbbbbbbb.jpg"'); + }); +}); diff --git a/apps/deploy-service/test/pipeline.test.ts b/apps/deploy-service/test/pipeline.test.ts index 0206202..0a7191e 100644 --- a/apps/deploy-service/test/pipeline.test.ts +++ b/apps/deploy-service/test/pipeline.test.ts @@ -87,7 +87,7 @@ const spec = { describe("buildable files", () => { it("builds a Vite SPA with robots/sitemap on the real site URL", () => { - const files = filesFromVersion({ snapshot: spec, nextFiles: null }, "https://acme-x7k.useframe.in"); + const { files } = filesFromVersion({ snapshot: spec, nextFiles: null }, "https://acme-x7k.useframe.in"); const robots = files.find((f) => f.path === "public/robots.txt")!.content; const sitemap = files.find((f) => f.path === "public/sitemap.xml")!.content; expect(robots).toContain("Sitemap: https://acme-x7k.useframe.in/sitemap.xml"); @@ -97,7 +97,7 @@ describe("buildable files", () => { }); it("scaffolds a Replicate Next tree as a static export", () => { - const files = filesFromVersion( + const { files } = filesFromVersion( { snapshot: {}, nextFiles: [{ path: "app/page.tsx", content: "export default () => null" }] }, "https://x.useframe.in", ); diff --git a/apps/deploy-service/test/runDeployment.test.ts b/apps/deploy-service/test/runDeployment.test.ts index 76e2068..ccf21a7 100644 --- a/apps/deploy-service/test/runDeployment.test.ts +++ b/apps/deploy-service/test/runDeployment.test.ts @@ -12,6 +12,8 @@ const m = vi.hoisted(() => ({ verifyOutput: vi.fn(), copyTemplate: vi.fn(), remove: vi.fn(), + copyMedia: vi.fn(), + media: [] as { bytes: number }[], })); vi.mock("@useframe/db", () => ({ ensureValidationRun: vi.fn() })); @@ -41,10 +43,13 @@ vi.mock("@/pipeline/materialize.js", () => ({ removeWorkDir: m.remove, })); vi.mock("@/pipeline/files.js", () => ({ - getBuildableFiles: async () => [ - { path: "package.json", content: JSON.stringify({ devDependencies: { vite: "5.4.0" } }) }, - ], + getBuildableFiles: async () => ({ + files: [{ path: "package.json", content: JSON.stringify({ devDependencies: { vite: "5.4.0" } }) }], + media: m.media, + mediaWarnings: m.media.length ? ["home/hero-0/visual: hero image is larger than 400 KB"] : [], + }), })); +vi.mock("@/pipeline/media.js", () => ({ copyMediaToSite: m.copyMedia })); const { runDeployment } = await import("@/pipeline/runDeployment.js"); @@ -82,7 +87,11 @@ function deps() { } beforeEach(() => { - for (const fn of Object.values(m)) fn.mockReset(); + for (const fn of Object.values(m)) if (typeof fn === "function") fn.mockReset(); + m.media = []; + m.copyMedia.mockImplementation(async (_s: unknown, _d: unknown, _p: string, items: { bytes: number }[]) => + items.reduce((sum, i) => sum + i.bytes, 0), + ); m.transition.mockResolvedValue(true); m.copyTemplate.mockResolvedValue(true); m.runChild.mockResolvedValue({ code: 0, signal: null, timedOut: false, output: "built" }); @@ -167,4 +176,34 @@ describe("runDeployment", () => { await runDeployment(deps(), job); expect(m.sync).toHaveBeenLastCalledWith(expect.anything(), "s1"); }); + + it("copies referenced media into the deployment before the build upload, counting its files and bytes", async () => { + m.media = [{ bytes: 300 }, { bytes: 200 }]; + const d = deps() as unknown as { mediaSource: unknown; r2: unknown }; + d.mediaSource = { bucket: "useframe-media" }; + await runDeployment(d as never, job); + expect(m.copyMedia).toHaveBeenCalledWith(d.mediaSource, d.r2, "sites/p1/d1/", m.media); + expect(m.transition.mock.calls[1]![3]).toMatchObject({ fileCount: 3, totalBytes: BigInt(510) }); + expect(m.verifyUpload).toHaveBeenCalledWith(expect.anything(), "sites/p1/d1/", 3, "abc"); + expect(m.copyMedia.mock.invocationCallOrder[0]).toBeLessThan(m.upload.mock.invocationCallOrder[0]!); + expect(m.fail).not.toHaveBeenCalled(); + }); + + it("fails before uploading when media would push the deployment over the size limit", async () => { + m.media = [{ bytes: 995 }]; + await runDeployment(deps(), job); + expect(m.fail.mock.calls[0]![2]).toMatch(/larger than/); + expect(m.copyMedia).not.toHaveBeenCalled(); + expect(m.upload).not.toHaveBeenCalled(); + }); + + it("fails closed when a referenced media file is missing", async () => { + m.media = [{ bytes: 10 }]; + const { DeployError } = await import("@/pipeline/errors.js"); + m.copyMedia.mockRejectedValue(new DeployError('The media "Team photo" is missing from storage.')); + await runDeployment(deps(), job); + expect(m.fail.mock.calls[0]![2]).toBe('The media "Team photo" is missing from storage.'); + expect(m.sync).not.toHaveBeenCalled(); + expect(m.goLive).not.toHaveBeenCalled(); + }); }); diff --git a/apps/media-service/.env.example b/apps/media-service/.env.example new file mode 100644 index 0000000..558df8e --- /dev/null +++ b/apps/media-service/.env.example @@ -0,0 +1,32 @@ +# apps/media-service — every variable is validated at boot (src/config/schema.ts). +NODE_ENV=development +# Health listener port (optional; the worker has no other HTTP). +HEALTH_PORT= +DATABASE_URL= +REDIS_URL=redis://localhost:6379 + +# R2 token: Object Read & Write on useframe-media only. +MEDIA_R2_BUCKET=useframe-media +MEDIA_R2_ACCESS_KEY_ID= +MEDIA_R2_SECRET_ACCESS_KEY= +# Set one of these two. +MEDIA_R2_ENDPOINT= +CLOUDFLARE_ACCOUNT_ID= + +# Scratch space; defaults to /work in production and /useframe-media elsewhere. +MEDIA_WORK_DIR= +MEDIA_PROCESS_CONCURRENCY=2 +MEDIA_FFMPEG_TIMEOUT_MS=180000 +MEDIA_FFMPEG_THREADS=2 +MEDIA_MAX_INPUT_PIXELS=60000000 +MEDIA_MAX_VIDEO_SECONDS=60 +# Hard ceilings for any original, uploaded or generated. +MEDIA_PAID_MAX_IMAGE_BYTES=20971520 +MEDIA_PAID_MAX_VIDEO_BYTES=157286400 + +# Vision descriptions via orchestrator-service (/internal/media/describe). +MEDIA_DESCRIBE_ENABLED=true +MEDIA_DESCRIBE_TIMEOUT_MS=30000 +ORCHESTRATOR_URL=http://localhost:4001 +# Same value as orchestrator-service INTERNAL_SERVICE_SECRET (min 16 chars). +INTERNAL_SERVICE_SECRET= diff --git a/apps/media-service/.gitignore b/apps/media-service/.gitignore new file mode 100644 index 0000000..9c97bbd --- /dev/null +++ b/apps/media-service/.gitignore @@ -0,0 +1,3 @@ +node_modules +dist +.env diff --git a/apps/media-service/Dockerfile b/apps/media-service/Dockerfile new file mode 100644 index 0000000..ca6292a --- /dev/null +++ b/apps/media-service/Dockerfile @@ -0,0 +1,91 @@ +# syntax=docker/dockerfile:1.7 +# Build from the repo root: docker build -f apps/media-service/Dockerfile . + +ARG NODE_VERSION=24 +ARG NODE_IMAGE_DIGEST= + +# Stage 1 — builder +FROM node:${NODE_VERSION}-bookworm-slim${NODE_IMAGE_DIGEST:+@${NODE_IMAGE_DIGEST}} AS builder +ARG PNPM_VERSION=11.1.3 +ARG APP_DIR=apps/media-service +ARG APP_PACKAGE=@useframe/media-service +ENV PNPM_HOME=/pnpm \ + PATH=/pnpm:$PATH \ + CI=1 \ + pnpm_config_store_dir=/pnpm/store \ + pnpm_config_verify_deps_before_run=false \ + pnpm_config_minimum_release_age=0 \ + PLAYWRIGHT_SKIP_BROWSER_DOWNLOAD=1 +# The release-age policy is enforced when the lockfile is resolved; re-checking every entry +# here only repeats it (one registry request per package) for a --frozen-lockfile install. +SHELL ["/bin/bash", "-o", "pipefail", "-c"] + +# Optional BuildKit secret "extra_ca": a CA for TLS-intercepting proxies on build machines. +# Download steps trust it; it is never written to a layer. +RUN --mount=type=secret,id=extra_ca,required=false \ + if [ -s /run/secrets/extra_ca ]; then export NODE_EXTRA_CA_CERTS=/run/secrets/extra_ca; fi; \ + corepack enable && corepack prepare "pnpm@${PNPM_VERSION}" --activate + +WORKDIR /repo + +COPY pnpm-lock.yaml pnpm-workspace.yaml package.json .npmrc* ./ +RUN --mount=type=cache,id=pnpm-store,target=/pnpm/store \ + --mount=type=secret,id=extra_ca,required=false \ + if [ -s /run/secrets/extra_ca ]; then export NODE_EXTRA_CA_CERTS=/run/secrets/extra_ca; fi; \ + pnpm fetch + +COPY . . +RUN --mount=type=cache,id=pnpm-store,target=/pnpm/store \ + --mount=type=secret,id=extra_ca,required=false \ + if [ -s /run/secrets/extra_ca ]; then export NODE_EXTRA_CA_CERTS=/run/secrets/extra_ca; fi; \ + pnpm install --offline --frozen-lockfile + +# prisma.config.ts resolves DATABASE_URL eagerly; `prisma generate` never connects. +RUN DATABASE_URL=postgresql://build:build@localhost:5432/build pnpm --filter "${APP_PACKAGE}..." build + +# pnpm 11 deploys non-injected workspaces only with --legacy. +RUN --mount=type=cache,id=pnpm-store,target=/pnpm/store \ + --mount=type=secret,id=extra_ca,required=false \ + if [ -s /run/secrets/extra_ca ]; then export NODE_EXTRA_CA_CERTS=/run/secrets/extra_ca; fi; \ + pnpm --filter "${APP_PACKAGE}" deploy --prod --legacy /out \ + && node docker/prune-dev-peers.mjs /out \ + && mkdir /runtime \ + && cp -a /out/package.json /out/dist /out/node_modules /runtime/ + +# Stage 2 — runner +FROM node:${NODE_VERSION}-bookworm-slim${NODE_IMAGE_DIGEST:+@${NODE_IMAGE_DIGEST}} AS runner +ARG GIT_SHA=unknown +ARG BUILD_DATE=unknown +ARG VERSION=unknown +ARG SOURCE_URL=unknown +LABEL org.opencontainers.image.title="useframe-media-service" \ + org.opencontainers.image.source="${SOURCE_URL}" \ + org.opencontainers.image.revision="${GIT_SHA}" \ + org.opencontainers.image.created="${BUILD_DATE}" \ + org.opencontainers.image.version="${VERSION}" + +ENV NODE_ENV=production \ + TMPDIR=/tmp \ + MEDIA_WORK_DIR=/work + +ARG DEBIAN_FRONTEND=noninteractive +# ffmpeg/ffprobe transcode and probe uploaded video. Package versions follow the +# pinned base image's Debian security updates. +# hadolint ignore=DL3008 +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates tini ffmpeg \ + && rm -rf /var/lib/apt/lists/* \ + && groupadd --gid 10001 app \ + && useradd --uid 10001 --gid 10001 --no-create-home --shell /usr/sbin/nologin app \ + && mkdir -p /work \ + && chown 10001:10001 /work \ + && ffmpeg -hide_banner -version | head -1 + +WORKDIR /app +COPY --from=builder --chown=10001:10001 /runtime ./ + +USER 10001:10001 +# Health listener only (HEALTH_PORT); there is no public HTTP. +EXPOSE 8080 +ENTRYPOINT ["/usr/bin/tini", "--"] +CMD ["node", "dist/worker.js"] diff --git a/apps/media-service/package.json b/apps/media-service/package.json new file mode 100644 index 0000000..4538eb0 --- /dev/null +++ b/apps/media-service/package.json @@ -0,0 +1,37 @@ +{ + "name": "@useframe/media-service", + "version": "0.1.0", + "private": true, + "type": "module", + "files": [ + "dist" + ], + "scripts": { + "dev": "tsx --watch src/worker.ts", + "build": "tsc -p tsconfig.json && tsc-alias -p tsconfig.json", + "start": "node dist/worker.js", + "check-types": "tsc --noEmit", + "test": "vitest run", + "media:gc": "tsx src/cli/gc.ts" + }, + "dependencies": { + "@aws-sdk/client-s3": "3.1144.0", + "@aws-sdk/lib-storage": "3.1144.0", + "@repo/events": "workspace:*", + "@repo/schemas": "workspace:*", + "@useframe/db": "workspace:*", + "bullmq": "^5.28.0", + "dotenv": "^16.6.1", + "ioredis": "^5.4.1", + "sharp": "^0.34.3", + "zod": "^3.25.76" + }, + "devDependencies": { + "@types/node": "^22.0.0", + "@useframe/typescript-config": "workspace:*", + "tsc-alias": "1.9.7", + "tsx": "^4.22.1", + "typescript": "5.9.2", + "vitest": "^2.1.9" + } +} diff --git a/apps/media-service/src/cli/gc.ts b/apps/media-service/src/cli/gc.ts new file mode 100644 index 0000000..bcbb176 --- /dev/null +++ b/apps/media-service/src/cli/gc.ts @@ -0,0 +1,6 @@ +import { closeContext, deps } from "@/context.js" +import { runMediaGc } from "@/jobs/gc.js" + +const result = await runMediaGc(deps) +console.log(JSON.stringify(result)) +await closeContext() diff --git a/apps/media-service/src/config/env.ts b/apps/media-service/src/config/env.ts new file mode 100644 index 0000000..aea3ec7 --- /dev/null +++ b/apps/media-service/src/config/env.ts @@ -0,0 +1,14 @@ +import dotenv from "dotenv" +import { parseEnv } from "./schema.js" + +dotenv.config() + +const parsed = parseEnv(process.env) + +if (!parsed.ok) { + console.error("Invalid environment variables:") + console.error(parsed.errors) + process.exit(1) +} + +export const env = parsed.env diff --git a/apps/media-service/src/config/schema.ts b/apps/media-service/src/config/schema.ts new file mode 100644 index 0000000..bed5e06 --- /dev/null +++ b/apps/media-service/src/config/schema.ts @@ -0,0 +1,76 @@ +import os from "node:os" +import path from "node:path" +import { z } from "zod" + +const int = (fallback: number) => z.coerce.number().int().positive().default(fallback) + +// dotenv turns `NAME=` into "", which must mean "not set" for optional values. +const optional = (schema: T) => + z.preprocess((v) => (v === "" ? undefined : v), schema.optional()) + +const bool = (fallback: "true" | "false") => + z + .enum(["true", "false"]) + .default(fallback) + .transform((v) => v === "true") + +export const envSchema = z + .object({ + NODE_ENV: z.enum(["development", "production", "test"]).default("development"), + HEALTH_PORT: optional(z.coerce.number().int().min(1).max(65535)), + DATABASE_URL: z.string().min(1, "DATABASE_URL is required"), + REDIS_URL: z.string().url("REDIS_URL must be a redis:// or rediss:// URL"), + + CLOUDFLARE_ACCOUNT_ID: optional(z.string()), + MEDIA_R2_BUCKET: z.string().min(1).default("useframe-media"), + MEDIA_R2_ACCESS_KEY_ID: z.string().min(1, "MEDIA_R2_ACCESS_KEY_ID is required"), + MEDIA_R2_SECRET_ACCESS_KEY: z.string().min(1, "MEDIA_R2_SECRET_ACCESS_KEY is required"), + MEDIA_R2_ENDPOINT: optional(z.string().url()), + + MEDIA_WORK_DIR: optional(z.string()), + MEDIA_PROCESS_CONCURRENCY: int(2), + MEDIA_FFMPEG_TIMEOUT_MS: int(180_000), + MEDIA_FFMPEG_THREADS: int(2), + MEDIA_MAX_INPUT_PIXELS: int(60_000_000), + MEDIA_MAX_VIDEO_SECONDS: int(60), + // Hard ceilings for any original, uploaded or generated (tier limits are the API's job). + MEDIA_PAID_MAX_IMAGE_BYTES: int(20_971_520), + MEDIA_PAID_MAX_VIDEO_BYTES: int(157_286_400), + + MEDIA_DESCRIBE_ENABLED: bool("true"), + 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), + }) + .superRefine((value, ctx) => { + if (!value.MEDIA_R2_ENDPOINT && !value.CLOUDFLARE_ACCOUNT_ID) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ["MEDIA_R2_ENDPOINT"], + message: "Set MEDIA_R2_ENDPOINT or CLOUDFLARE_ACCOUNT_ID", + }) + } + if (value.MEDIA_DESCRIBE_ENABLED && !value.INTERNAL_SERVICE_SECRET) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ["INTERNAL_SERVICE_SECRET"], + message: "INTERNAL_SERVICE_SECRET is required while MEDIA_DESCRIBE_ENABLED=true", + }) + } + }) + .transform((value) => ({ + ...value, + MEDIA_R2_ENDPOINT: value.MEDIA_R2_ENDPOINT ?? `https://${value.CLOUDFLARE_ACCOUNT_ID}.r2.cloudflarestorage.com`, + MEDIA_WORK_DIR: + value.MEDIA_WORK_DIR ?? (value.NODE_ENV === "production" ? "/work" : path.join(os.tmpdir(), "useframe-media")), + })) + +export type Env = z.infer + +export function parseEnv( + source: NodeJS.ProcessEnv, +): { ok: true; env: Env } | { ok: false; errors: Record } { + const parsed = envSchema.safeParse(source) + if (parsed.success) return { ok: true, env: parsed.data } + return { ok: false, errors: parsed.error.flatten().fieldErrors as Record } +} diff --git a/apps/media-service/src/context.ts b/apps/media-service/src/context.ts new file mode 100644 index 0000000..81b3b44 --- /dev/null +++ b/apps/media-service/src/context.ts @@ -0,0 +1,50 @@ +import { Queue } from "bullmq" +import { prisma } from "@useframe/db" +import { QUEUES, type MediaProcessJobPayload } from "@repo/events" +import { env } from "@/config/env.js" +import { PROCESS_JOB_OPTIONS, type Deps } from "@/deps.js" +import { redis } from "@/lib/redis.js" +import { createS3MediaStore } from "@/lib/storage.js" +import { createOrchestratorDescriber } from "@/processing/describe.js" + +export const processQueue = new Queue(QUEUES.MEDIA_PROCESS, { connection: redis }) + +export const deps: Deps = { + db: prisma, + 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, + }), + describe: + env.MEDIA_DESCRIBE_ENABLED && env.INTERNAL_SERVICE_SECRET + ? createOrchestratorDescriber({ + url: env.ORCHESTRATOR_URL, + secret: env.INTERNAL_SERVICE_SECRET, + timeoutMs: env.MEDIA_DESCRIBE_TIMEOUT_MS, + }) + : null, + enqueue: async (payload, jobId) => { + await processQueue.add("process", payload, { jobId, ...PROCESS_JOB_OPTIONS }) + }, + config: { + workDir: env.MEDIA_WORK_DIR, + maxInputPixels: env.MEDIA_MAX_INPUT_PIXELS, + maxVideoSeconds: env.MEDIA_MAX_VIDEO_SECONDS, + maxImageBytes: env.MEDIA_PAID_MAX_IMAGE_BYTES, + maxVideoBytes: env.MEDIA_PAID_MAX_VIDEO_BYTES, + ffmpeg: { + ffmpegPath: "ffmpeg", + ffprobePath: "ffprobe", + timeoutMs: env.MEDIA_FFMPEG_TIMEOUT_MS, + threads: env.MEDIA_FFMPEG_THREADS, + }, + }, +} + +export async function closeContext(): Promise { + await processQueue.close() + await redis.quit().catch(() => undefined) + await prisma.$disconnect() +} diff --git a/apps/media-service/src/createAsset.ts b/apps/media-service/src/createAsset.ts new file mode 100644 index 0000000..f22c279 --- /dev/null +++ b/apps/media-service/src/createAsset.ts @@ -0,0 +1,94 @@ +import { Readable } from "node:stream" +import { sniffAgrees, sniffMediaBytes, UPLOAD_IMAGE_MIMES, UPLOAD_VIDEO_MIMES } from "@repo/schemas" +import type { MediaProcessOptions } from "@repo/events" +import type { Prisma } from "@useframe/db" +import { mediaPrefix, type Deps } from "@/deps.js" +import { MediaRejectedError } from "@/processing/errors.js" + +export type CreateAssetInput = { + projectId: string + userId: string + kind: "IMAGE" | "VIDEO" + origin: "UPLOADED" | "GENERATED" + mime: string + purpose?: string + title?: string + generation?: Prisma.InputJsonValue + options?: MediaProcessOptions +} & ({ bytes: Buffer } | { stream: Readable; size: number }) + +const SNIFF_BYTES = 4096 + +async function peek(stream: Readable, size: number): Promise<{ head: Buffer; body: Readable }> { + const iterator = stream[Symbol.asyncIterator]() as AsyncIterator + const chunks: Buffer[] = [] + let length = 0 + while (length < Math.min(SNIFF_BYTES, size)) { + const next = await iterator.next() + if (next.done) break + const buf = Buffer.from(next.value) + chunks.push(buf) + length += buf.length + } + const head = Buffer.concat(chunks) + async function* rest() { + yield head + for (;;) { + const next = await iterator.next() + if (next.done) return + yield next.value + } + } + return { head: head.subarray(0, SNIFF_BYTES), body: Readable.from(rest()) } +} + +// System-created media (Part B): upload type/size checks, but no user quota or tier rules. +export async function createAssetFromBytes(deps: Deps, input: CreateAssetInput): Promise<{ assetId: string }> { + const allowed: readonly string[] = input.kind === "VIDEO" ? UPLOAD_VIDEO_MIMES : UPLOAD_IMAGE_MIMES + if (!allowed.includes(input.mime)) throw new MediaRejectedError("Unsupported media type", input.mime) + const size = "bytes" in input ? input.bytes.length : input.size + const max = input.kind === "VIDEO" ? deps.config.maxVideoBytes : deps.config.maxImageBytes + if (size <= 0 || size > max) throw new MediaRejectedError("The file is larger than allowed", `${size} bytes`) + + const { head, body } = + "bytes" in input + ? { head: input.bytes.subarray(0, SNIFF_BYTES), body: input.bytes } + : await peek(input.stream, size) + if (!sniffAgrees(input.mime, sniffMediaBytes(head))) { + throw new MediaRejectedError("The file's contents don't match its type", "create sniff") + } + + const asset = await deps.db.$transaction(async (tx) => { + const created = await tx.mediaAsset.create({ + data: { + projectId: input.projectId, + userId: input.userId, + kind: input.kind, + origin: input.origin, + status: "PROCESSING", + title: input.title?.slice(0, 120) ?? null, + purpose: input.purpose ?? null, + aiGenerated: input.origin === "GENERATED", + generation: input.generation, + originalKey: "", + originalMime: input.mime, + originalBytes: size, + }, + select: { id: true }, + }) + return tx.mediaAsset.update({ + where: { id: created.id }, + data: { originalKey: `${mediaPrefix(input.projectId, created.id)}orig` }, + select: { id: true, originalKey: true }, + }) + }) + + try { + await deps.store.put(asset.originalKey, body, size, input.mime) + } catch (err) { + await deps.db.mediaAsset.delete({ where: { id: asset.id } }).catch(() => undefined) + throw err + } + await deps.enqueue({ assetId: asset.id, ...(input.options ? { options: input.options } : {}) }, `${asset.id}-0`) + return { assetId: asset.id } +} diff --git a/apps/media-service/src/deps.ts b/apps/media-service/src/deps.ts new file mode 100644 index 0000000..4561700 --- /dev/null +++ b/apps/media-service/src/deps.ts @@ -0,0 +1,37 @@ +import type { prisma } from "@useframe/db" +import type { MediaProcessJobPayload } from "@repo/events" +import type { MediaStore } from "@/lib/storage.js" +import type { Describer } from "@/processing/describe.js" +import type { FfmpegConfig } from "@/processing/ffmpeg.js" + +export type Db = Pick + +export type MediaConfig = { + workDir: string + maxInputPixels: number + maxVideoSeconds: number + maxImageBytes: number + maxVideoBytes: number + ffmpeg: FfmpegConfig +} + +export type Deps = { + db: Db + store: MediaStore + describe: Describer | null + enqueue: (payload: MediaProcessJobPayload, jobId: string) => Promise + config: MediaConfig +} + +export const PROCESS_JOB_OPTIONS = { + attempts: 3, + backoff: { type: "exponential", delay: 5000 }, + removeOnComplete: 1000, + removeOnFail: 1000, +} as const + +export const FLAGGED_REASON = "This file can't be used" + +export function mediaPrefix(projectId: string, assetId: string): string { + return `media/${projectId}/${assetId}/` +} diff --git a/apps/media-service/src/jobs/gc.ts b/apps/media-service/src/jobs/gc.ts new file mode 100644 index 0000000..7466bb3 --- /dev/null +++ b/apps/media-service/src/jobs/gc.ts @@ -0,0 +1,33 @@ +import { mediaPrefix, type Deps } from "@/deps.js" +import { log, metric } from "@/lib/logger.js" + +const ABANDONED_UPLOAD_MS = 60 * 60 * 1000 +const PURGE_AFTER_MS = 30 * 24 * 60 * 60 * 1000 +const BATCH = 200 + +type AssetWhere = NonNullable[0]>["where"] + +async function purge(deps: Deps, where: AssetWhere): Promise { + let purged = 0 + for (;;) { + const batch = await deps.db.mediaAsset.findMany({ where, select: { id: true, projectId: true }, take: BATCH }) + for (const asset of batch) { + await deps.store.deletePrefix(mediaPrefix(asset.projectId, asset.id)) + await deps.db.mediaAsset.delete({ where: { id: asset.id } }) + purged++ + } + if (batch.length < BATCH) return purged + } +} + +// Deployed sites hold their own copies, so purging never breaks a live site. +export async function runMediaGc(deps: Deps, now = new Date()): Promise<{ abandoned: number; purged: number }> { + const abandoned = await purge(deps, { + status: "UPLOADING", + createdAt: { lt: new Date(now.getTime() - ABANDONED_UPLOAD_MS) }, + }) + const purged = await purge(deps, { deletedAt: { lt: new Date(now.getTime() - PURGE_AFTER_MS) } }) + metric("media.gc_purged", purged, { abandoned }) + log.info("media gc finished", { abandoned, purged }) + return { abandoned, purged } +} diff --git a/apps/media-service/src/jobs/process.ts b/apps/media-service/src/jobs/process.ts new file mode 100644 index 0000000..34fab38 --- /dev/null +++ b/apps/media-service/src/jobs/process.ts @@ -0,0 +1,186 @@ +import { randomUUID } from "node:crypto" +import { mkdir, open, rm } from "node:fs/promises" +import path from "node:path" +import type { MediaProcessJobPayload } from "@repo/events" +import { sniffAgrees, sniffMediaBytes, type MediaVariant } from "@repo/schemas" +import { FLAGGED_REASON, mediaPrefix, type Deps } from "@/deps.js" +import { errorMessage, log, metric } from "@/lib/logger.js" +import { mapWithConcurrency, withRetry } from "@/lib/retry.js" +import { isTransientStorageError, TooLargeError } from "@/lib/storage.js" +import { MediaRejectedError } from "@/processing/errors.js" +import { processImage } from "@/processing/image.js" +import type { ProcessedMedia } from "@/processing/types.js" +import { processVideo } from "@/processing/video.js" + +export type Attempt = { attemptsMade: number; attempts: number } + +const EXT: Record = { "image/webp": "webp", "image/jpeg": "jpg", "video/mp4": "mp4" } + +async function readHead(file: string): Promise { + const handle = await open(file, "r") + try { + const buf = Buffer.alloc(4096) + const { bytesRead } = await handle.read(buf, 0, buf.length, 0) + return buf.subarray(0, bytesRead) + } finally { + await handle.close() + } +} + +async function userTier(deps: Deps, userId: string): Promise<"free" | "paid"> { + const balance = await deps.db.creditBalance.findUnique({ where: { userId }, select: { balance: true } }) + return (balance?.balance ?? 0) > 0 ? "paid" : "free" +} + +async function markFailed(deps: Deps, assetId: string, reason: string): Promise { + await deps.db.mediaAsset.updateMany({ + where: { id: assetId, status: "PROCESSING" }, + data: { status: "FAILED", failureReason: reason }, + }) +} + +export async function processAsset(deps: Deps, payload: MediaProcessJobPayload, attempt: Attempt): Promise { + const started = Date.now() + const asset = await deps.db.mediaAsset.findUnique({ + where: { id: payload.assetId }, + select: { + id: true, + projectId: true, + userId: true, + kind: true, + status: true, + originalKey: true, + originalMime: true, + altText: true, + deletedAt: true, + }, + }) + if (!asset || asset.deletedAt || asset.status !== "PROCESSING") { + log.info("media job skipped", { assetId: payload.assetId, status: asset?.status ?? "missing" }) + return + } + const options = payload.options ?? {} + const prefix = mediaPrefix(asset.projectId, asset.id) + const dir = path.join(deps.config.workDir, `${asset.id}-${randomUUID()}`) + + try { + await mkdir(path.join(dir, "out"), { recursive: true, mode: 0o700 }) + const input = path.join(dir, "orig") + const maxBytes = asset.kind === "VIDEO" ? deps.config.maxVideoBytes : deps.config.maxImageBytes + let original: { bytes: number; sha256: string } | null + try { + original = await deps.store.download(asset.originalKey, input, maxBytes) + } catch (err) { + if (err instanceof TooLargeError) + throw new MediaRejectedError("The file is larger than allowed", "download limit") + throw err + } + if (!original) throw new MediaRejectedError("The upload is missing. Please upload it again.", "no original") + if (!sniffAgrees(asset.originalMime, sniffMediaBytes(await readHead(input)))) { + throw new MediaRejectedError("The file's contents don't match its type", "worker sniff") + } + + const out = path.join(dir, "out") + const processed: ProcessedMedia = + asset.kind === "IMAGE" + ? await processImage(input, out, { maxInputPixels: deps.config.maxInputPixels }) + : await processVideo(deps.config.ffmpeg, input, out, { + maxSeconds: deps.config.maxVideoSeconds, + stripAudio: !!options.stripAudio, + }) + + const variants: MediaVariant[] = await mapWithConcurrency(processed.variants, 4, async (v) => { + const key = `${prefix}v/${v.role}.${EXT[v.mime] ?? "bin"}` + await withRetry(() => deps.store.uploadFile(key, v.path, v.bytes, v.mime), { + attempts: 3, + baseDelayMs: 500, + shouldRetry: isTransientStorageError, + }) + return { role: v.role, key, mime: v.mime, width: v.width, height: v.height, bytes: v.bytes, sha256: v.sha256 } + }) + + let described: Awaited>> = null + if (!options.skipDescribe && deps.describe && processed.describeImage) { + described = await deps.describe({ + imagePath: processed.describeImage, + tier: await userTier(deps, asset.userId), + assetId: asset.id, + }) + if (described?.flagged) { + // Quarantined: kept in the bucket for review, never served (only READY assets are). + await deps.db.mediaAsset.updateMany({ + where: { id: asset.id, status: "PROCESSING" }, + data: { status: "FAILED", failureReason: FLAGGED_REASON, sha256: original.sha256, variants }, + }) + await deps.db.pipelineLog.upsert({ + where: { jobId: `media-flag-${asset.id}` }, + create: { + projectId: asset.projectId, + jobId: `media-flag-${asset.id}`, + stage: "FAILED", + status: "MEDIA_FLAGGED", + }, + update: {}, + }) + log.warn("media flagged for review", { assetId: asset.id }) + return + } + } + + // updateMany so a delete that raced this job is never undone. + const updated = await deps.db.mediaAsset.updateMany({ + where: { id: asset.id, status: "PROCESSING", deletedAt: null }, + data: { + status: "READY", + sha256: original.sha256, + width: processed.width, + height: processed.height, + durationMs: processed.durationMs ?? null, + hasAudio: processed.hasAudio ?? null, + variants, + dominantColor: processed.dominantColor, + lqip: processed.lqip, + failureReason: null, + ...(described + ? { + description: described.description, + suggestedPurposes: described.suggestedPurposes, + ...(asset.altText ? {} : { altText: described.altText }), + } + : {}), + }, + }) + if (updated.count === 0) await deps.store.deletePrefix(`${prefix}v/`) + metric("media.processed", Date.now() - started, { + assetId: asset.id, + kind: asset.kind, + variants: variants.length, + bytes: original.bytes, + described: !!described, + warnings: processed.warnings.length, + }) + } catch (err) { + if (err instanceof MediaRejectedError) { + log.warn("media rejected", { assetId: asset.id, reason: err.publicReason, detail: err.detail }) + await deps.store + .deletePrefix(prefix) + .catch((e) => log.warn("media cleanup failed", { assetId: asset.id, error: errorMessage(e) })) + await markFailed(deps, asset.id, err.publicReason) + return + } + log.error("media processing error", { + assetId: asset.id, + attempt: attempt.attemptsMade + 1, + error: errorMessage(err), + }) + if (attempt.attemptsMade + 1 >= attempt.attempts) { + await deps.store.deletePrefix(`${prefix}v/`).catch(() => undefined) + await markFailed(deps, asset.id, "We couldn't process this file. Please try again.") + } + throw err + } finally { + await rm(dir, { recursive: true, force: true, maxRetries: 3 }).catch((err) => + log.warn("media scratch cleanup failed", { assetId: asset.id, error: errorMessage(err) }), + ) + } +} diff --git a/apps/media-service/src/lib/health.ts b/apps/media-service/src/lib/health.ts new file mode 100644 index 0000000..13e83c4 --- /dev/null +++ b/apps/media-service/src/lib/health.ts @@ -0,0 +1,9 @@ +import { createServer, type Server } from "node:http" + +export function startHealthServer(port: number, isHealthy: () => boolean): Server { + return createServer((_req, res) => { + const ok = isHealthy() + res.writeHead(ok ? 200 : 503, { "content-type": "application/json" }) + res.end(JSON.stringify({ status: ok ? "ok" : "unavailable", service: "media-worker" })) + }).listen(port) +} diff --git a/apps/media-service/src/lib/logger.ts b/apps/media-service/src/lib/logger.ts new file mode 100644 index 0000000..9a34823 --- /dev/null +++ b/apps/media-service/src/lib/logger.ts @@ -0,0 +1,24 @@ +type Fields = Record + +function emit(level: "info" | "warn" | "error", msg: string, fields: Fields): void { + const line = JSON.stringify({ level, msg, time: new Date().toISOString(), ...fields }) + if (level === "error") console.error(line) + else if (level === "warn") console.warn(line) + else console.log(line) +} + +export const log = { + info: (msg: string, fields: Fields = {}) => emit("info", msg, fields), + warn: (msg: string, fields: Fields = {}) => emit("warn", msg, fields), + error: (msg: string, fields: Fields = {}) => emit("error", msg, fields), +} + +// The repo has no metrics library; metrics are structured log lines that the +// log pipeline can turn into counters/histograms. +export function metric(name: string, value: number, tags: Fields = {}): void { + emit("info", "metric", { metric: name, value, ...tags }) +} + +export function errorMessage(err: unknown): string { + return err instanceof Error ? err.message : String(err) +} diff --git a/apps/media-service/src/lib/redis.ts b/apps/media-service/src/lib/redis.ts new file mode 100644 index 0000000..d7f5701 --- /dev/null +++ b/apps/media-service/src/lib/redis.ts @@ -0,0 +1,8 @@ +import { Redis as IORedis } from "ioredis" +import { env } from "@/config/env.js" + +export const redis = new IORedis(env.REDIS_URL, { + maxRetriesPerRequest: null, + // Matches the other services: Upstash's rediss:// chain fails verification on dev machines. + tls: env.REDIS_URL.startsWith("rediss://") ? { rejectUnauthorized: false } : undefined, +}) diff --git a/apps/media-service/src/lib/retry.ts b/apps/media-service/src/lib/retry.ts new file mode 100644 index 0000000..412300b --- /dev/null +++ b/apps/media-service/src/lib/retry.ts @@ -0,0 +1,46 @@ +export type RetryOptions = { + attempts: number + baseDelayMs: number + maxDelayMs?: number + shouldRetry: (err: unknown) => boolean + retryAfterMs?: (err: unknown) => number | undefined + sleep?: (ms: number) => Promise +} + +export const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) + +export function jitteredDelay(attempt: number, baseDelayMs: number, maxDelayMs: number): number { + const exp = Math.min(maxDelayMs, baseDelayMs * 2 ** attempt) + return Math.round(exp / 2 + Math.random() * (exp / 2)) +} + +export async function withRetry(fn: () => Promise, options: RetryOptions): Promise { + const wait = options.sleep ?? sleep + const maxDelay = options.maxDelayMs ?? 30_000 + for (let attempt = 0; ; attempt++) { + try { + return await fn() + } catch (err) { + if (attempt + 1 >= options.attempts || !options.shouldRetry(err)) throw err + const hinted = options.retryAfterMs?.(err) + await wait(hinted ?? jitteredDelay(attempt, options.baseDelayMs, maxDelay)) + } + } +} + +export async function mapWithConcurrency( + items: readonly T[], + concurrency: number, + fn: (item: T, index: number) => Promise, +): Promise { + const results = new Array(items.length) + let next = 0 + const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => { + while (next < items.length) { + const index = next++ + results[index] = await fn(items[index]!, index) + } + }) + await Promise.all(workers) + return results +} diff --git a/apps/media-service/src/lib/storage.ts b/apps/media-service/src/lib/storage.ts new file mode 100644 index 0000000..88aecac --- /dev/null +++ b/apps/media-service/src/lib/storage.ts @@ -0,0 +1,140 @@ +import { createHash } from "node:crypto" +import { createReadStream, createWriteStream } from "node:fs" +import { Readable, Transform } from "node:stream" +import { pipeline } from "node:stream/promises" +import { + DeleteObjectsCommand, + GetObjectCommand, + HeadObjectCommand, + ListObjectsV2Command, + PutObjectCommand, + S3Client, +} from "@aws-sdk/client-s3" +import { Upload } from "@aws-sdk/lib-storage" + +const MULTIPART_THRESHOLD = 5 * 1024 * 1024 + +export class TooLargeError extends Error { + constructor() { + super("object exceeds the allowed size") + this.name = "TooLargeError" + } +} + +export interface MediaStore { + // Streams an object to disk, refusing to write more than maxBytes. + download(key: string, destPath: string, maxBytes: number): Promise<{ bytes: number; sha256: string } | null> + uploadFile(key: string, absPath: string, size: number, contentType: string): Promise + put(key: string, body: Buffer | Readable, size: number, contentType: string): Promise + head(key: string): Promise<{ size: number } | null> + deletePrefix(prefix: string): Promise +} + +function isNotFound(err: unknown): boolean { + if (!(err instanceof Error)) return false + const status = (err as { $metadata?: { httpStatusCode?: number } }).$metadata?.httpStatusCode + return err.name === "NoSuchKey" || err.name === "NotFound" || status === 404 +} + +// 5xx and network failures are worth retrying; everything else is final. +export function isTransientStorageError(err: unknown): boolean { + if (!(err instanceof Error)) return false + const status = (err as { $metadata?: { httpStatusCode?: number } }).$metadata?.httpStatusCode + if (status !== undefined) return status >= 500 || status === 429 + return /ECONNRESET|ETIMEDOUT|EAI_AGAIN|ENOTFOUND|socket hang up|TimeoutError/i.test(`${err.name} ${err.message}`) +} + +export function hashingLimiter(maxBytes: number) { + const hash = createHash("sha256") + let bytes = 0 + const transform = new Transform({ + transform(chunk: Buffer, _enc, callback) { + bytes += chunk.length + if (bytes > maxBytes) return callback(new TooLargeError()) + hash.update(chunk) + callback(null, chunk) + }, + }) + return { transform, result: () => ({ bytes, sha256: hash.digest("hex") }) } +} + +export function createS3MediaStore(config: { + endpoint: string + accessKeyId: string + secretAccessKey: string + bucket: string +}): MediaStore { + const client = new S3Client({ + region: "auto", + endpoint: config.endpoint, + credentials: { accessKeyId: config.accessKeyId, secretAccessKey: config.secretAccessKey }, + // Default CRC32 checksums from recent SDKs have broken R2 uploads before. + requestChecksumCalculation: "WHEN_REQUIRED", + responseChecksumValidation: "WHEN_REQUIRED", + }) + const Bucket = config.bucket + + const put = async (key: string, body: Buffer | Readable, size: number, contentType: string) => { + if (size > MULTIPART_THRESHOLD && !Buffer.isBuffer(body)) { + await new Upload({ + client, + params: { Bucket, Key: key, Body: body, ContentType: contentType }, + queueSize: 4, + partSize: MULTIPART_THRESHOLD, + }).done() + return + } + await client.send( + new PutObjectCommand({ Bucket, Key: key, Body: body, ContentLength: size, ContentType: contentType }), + ) + } + + return { + async download(key, destPath, maxBytes) { + let res + try { + res = await client.send(new GetObjectCommand({ Bucket, Key: key })) + } catch (err) { + if (isNotFound(err)) return null + throw err + } + if (!res.Body) return null + if ((res.ContentLength ?? 0) > maxBytes) throw new TooLargeError() + const limiter = hashingLimiter(maxBytes) + await pipeline(res.Body as Readable, limiter.transform, createWriteStream(destPath, { mode: 0o600 })) + return limiter.result() + }, + + uploadFile: (key, absPath, size, contentType) => put(key, createReadStream(absPath), size, contentType), + + put, + + async head(key) { + try { + const res = await client.send(new HeadObjectCommand({ Bucket, Key: key })) + return { size: res.ContentLength ?? 0 } + } catch (err) { + if (isNotFound(err)) return null + throw err + } + }, + + async deletePrefix(prefix) { + if (!prefix.endsWith("/")) throw new Error("Refusing to delete a non-directory prefix") + let deleted = 0 + let token: string | undefined + do { + const page = await client.send( + new ListObjectsV2Command({ Bucket, Prefix: prefix, ContinuationToken: token, MaxKeys: 1000 }), + ) + const objects = (page.Contents ?? []).flatMap((o) => (o.Key ? [{ Key: o.Key }] : [])) + if (objects.length) { + await client.send(new DeleteObjectsCommand({ Bucket, Delete: { Objects: objects, Quiet: true } })) + deleted += objects.length + } + token = page.IsTruncated ? page.NextContinuationToken : undefined + } while (token) + return deleted + }, + } +} diff --git a/apps/media-service/src/processing/describe.ts b/apps/media-service/src/processing/describe.ts new file mode 100644 index 0000000..444e0a2 --- /dev/null +++ b/apps/media-service/src/processing/describe.ts @@ -0,0 +1,38 @@ +import { readFile } from "node:fs/promises" +import { MediaDescribeOutputSchema, type MediaDescribeOutput } from "@repo/schemas" +import { errorMessage, log } from "@/lib/logger.js" + +export type DescribeInput = { imagePath: string; tier: "free" | "paid"; assetId: string } + +export type Describer = (input: DescribeInput) => Promise + +// Only pixels and the tier leave this service; never file names, titles or user ids. +export function createOrchestratorDescriber(config: { + url: string + secret: string + timeoutMs: number + fetchImpl?: typeof fetch +}): Describer { + const doFetch = config.fetchImpl ?? fetch + return async ({ imagePath, tier, assetId }) => { + const image = (await readFile(imagePath)).toString("base64") + for (let attempt = 1; attempt <= 2; attempt++) { + try { + const res = await doFetch(`${config.url.replace(/\/$/, "")}/internal/media/describe`, { + method: "POST", + headers: { "content-type": "application/json", "x-internal-secret": config.secret }, + body: JSON.stringify({ image, mime: "image/webp", tier }), + signal: AbortSignal.timeout(config.timeoutMs), + }) + const body = (await res.json().catch(() => null)) as { success?: boolean; data?: unknown } | null + const parsed = MediaDescribeOutputSchema.safeParse(body?.data) + if (res.ok && parsed.success) return parsed.data + log.warn("media describe returned an unusable response", { assetId, attempt, status: res.status }) + if (res.status >= 400 && res.status < 500 && res.status !== 422) return null + } catch (err) { + log.warn("media describe failed", { assetId, attempt, error: errorMessage(err) }) + } + } + return null + } +} diff --git a/apps/media-service/src/processing/errors.ts b/apps/media-service/src/processing/errors.ts new file mode 100644 index 0000000..83f1e92 --- /dev/null +++ b/apps/media-service/src/processing/errors.ts @@ -0,0 +1,10 @@ +// Final and user-facing (never retried); `detail` is for logs and never holds file contents. +export class MediaRejectedError extends Error { + constructor( + public readonly publicReason: string, + public readonly detail?: string, + ) { + super(detail ? `${publicReason} (${detail})` : publicReason) + this.name = "MediaRejectedError" + } +} diff --git a/apps/media-service/src/processing/ffmpeg.ts b/apps/media-service/src/processing/ffmpeg.ts new file mode 100644 index 0000000..c7ee25c --- /dev/null +++ b/apps/media-service/src/processing/ffmpeg.ts @@ -0,0 +1,71 @@ +import { spawn } from "node:child_process" + +export type FfmpegConfig = { + ffmpegPath: string + ffprobePath: string + timeoutMs: number + threads: number +} + +export const ALLOWED_DEMUXERS = "mov,mp4,m4a,3gp,3g2,mj2,matroska,webm" + +// One local file per run: no network protocols, playlists, concat lists or image sequences. +export const INPUT_GUARD = ["-protocol_whitelist", "file", "-format_whitelist", ALLOWED_DEMUXERS] + +export class FfmpegError extends Error { + constructor( + message: string, + public readonly timedOut: boolean, + ) { + super(message) + this.name = "FfmpegError" + } +} + +const MAX_OUTPUT = 64 * 1024 + +export function runTool(command: string, args: string[], timeoutMs: number): Promise { + return new Promise((resolve, reject) => { + const child = spawn(command, args, { stdio: ["ignore", "pipe", "pipe"], shell: false, windowsHide: true }) + let stdout = "" + let stderr = "" + let timedOut = false + const timer = setTimeout(() => { + timedOut = true + child.kill("SIGKILL") + }, timeoutMs) + child.stdout.on("data", (chunk: Buffer) => { + if (stdout.length < MAX_OUTPUT * 16) stdout += chunk.toString() + }) + child.stderr.on("data", (chunk: Buffer) => { + stderr = (stderr + chunk.toString()).slice(-MAX_OUTPUT) + }) + child.on("error", (err) => { + clearTimeout(timer) + reject(new FfmpegError(`${command} failed to start: ${err.message}`, false)) + }) + child.on("close", (code) => { + clearTimeout(timer) + if (timedOut) reject(new FfmpegError(`${command} timed out`, true)) + else if (code !== 0) reject(new FfmpegError(`${command} exited ${code}: ${stderr.trim().slice(-500)}`, false)) + else resolve(stdout) + }) + }) +} + +export function ffmpeg(config: FfmpegConfig, args: string[]): Promise { + return runTool( + config.ffmpegPath, + ["-nostdin", "-hide_banner", "-loglevel", "error", "-y", "-threads", String(config.threads), ...args], + config.timeoutMs, + ) +} + +export async function ffprobe(config: FfmpegConfig, input: string): Promise { + const out = await runTool( + config.ffprobePath, + ["-v", "error", ...INPUT_GUARD, "-print_format", "json", "-show_format", "-show_streams", "-i", input], + config.timeoutMs, + ) + return JSON.parse(out) +} diff --git a/apps/media-service/src/processing/image.ts b/apps/media-service/src/processing/image.ts new file mode 100644 index 0000000..01a03c5 --- /dev/null +++ b/apps/media-service/src/processing/image.ts @@ -0,0 +1,113 @@ +import { createHash } from "node:crypto" +import { readFile, stat } from "node:fs/promises" +import path from "node:path" +import sharp, { type Sharp } from "sharp" +import type { MediaVariantRole } from "@repo/schemas" +import { MediaRejectedError } from "./errors.js" +import type { ProcessedMedia, ProducedVariant } from "./types.js" + +const WEBP_QUALITY = 80 +const WIDTHS: [MediaVariantRole, number][] = [ + ["w480", 480], + ["w960", 960], + ["w1600", 1600], +] +const LQIP_MAX_BYTES = 1536 +const ALLOWED_FORMATS = new Set(["jpeg", "png", "webp", "heif"]) + +export type ImageLimits = { maxInputPixels: number } + +async function fileVariant(role: MediaVariantRole, file: string, mime: string): Promise { + const [meta, info, data] = await Promise.all([sharp(file).metadata(), stat(file), readFile(file)]) + return { + role, + path: file, + mime, + width: meta.width, + height: meta.height, + bytes: info.size, + sha256: createHash("sha256").update(data).digest("hex"), + } +} + +export async function imageExtras(base: Sharp): Promise<{ dominantColor: string | null; lqip: string | null }> { + const { dominant } = await base.clone().stats() + const hex = (n: number) => n.toString(16).padStart(2, "0") + const dominantColor = `#${hex(dominant.r)}${hex(dominant.g)}${hex(dominant.b)}` + const tiny = await base.clone().resize({ width: 16 }).webp({ quality: 40 }).toBuffer() + const lqip = `data:image/webp;base64,${tiny.toString("base64")}` + return { dominantColor, lqip: lqip.length <= LQIP_MAX_BYTES ? lqip : null } +} + +// Every served file is a re-encode, which drops EXIF/GPS, ICC and any appended bytes. +export async function processImage(input: string, outDir: string, limits: ImageLimits): Promise { + const open = () => sharp(input, { limitInputPixels: limits.maxInputPixels, failOn: "error", sequentialRead: true }) + + let meta: sharp.Metadata + try { + meta = await open().metadata() + } catch (err) { + if (err instanceof Error && /pixel limit/i.test(err.message)) { + throw new MediaRejectedError("This image is too large to process", "pixel limit") + } + throw new MediaRejectedError("Couldn't read this image", "metadata decode failed") + } + if (!meta.format || !ALLOWED_FORMATS.has(meta.format)) { + throw new MediaRejectedError("Couldn't read this image", `format ${meta.format ?? "unknown"}`) + } + if (meta.format === "heif" && meta.compression !== "av1") { + throw new MediaRejectedError("Please export as JPEG or PNG", "heic") + } + if ((meta.pages ?? 1) > 1) throw new MediaRejectedError("Animated images aren't supported", "multi-page") + + let base: Sharp + let width: number + let height: number + try { + // Full decode up front, so a truncated or corrupt file fails here rather than mid-way. + const { data, info } = await open().rotate().toColourspace("srgb").raw().toBuffer({ resolveWithObject: true }) + width = info.width + height = info.height + base = sharp(data, { raw: { width: info.width, height: info.height, channels: info.channels } }) + } catch (err) { + if (err instanceof Error && /pixel limit/i.test(err.message)) { + throw new MediaRejectedError("This image is too large to process", "pixel limit") + } + throw new MediaRejectedError("Couldn't read this image", "decode failed") + } + + const variants: ProducedVariant[] = [] + const targets = WIDTHS.filter(([, w]) => w <= width) + if (targets.length === 0) targets.push(["w480", width]) + if (width >= 2400) targets.push(["w2400", 2400]) + for (const [role, w] of targets) { + const file = path.join(outDir, `${role}.webp`) + await base + .clone() + .resize({ width: Math.min(w, width), withoutEnlargement: true }) + .webp({ quality: WEBP_QUALITY }) + .toFile(file) + variants.push(await fileVariant(role, file, "image/webp")) + } + + const fallback = path.join(outDir, "fallback.jpg") + await base + .clone() + .resize({ width: Math.min(1600, width), withoutEnlargement: true }) + .flatten({ background: "#ffffff" }) + .jpeg({ quality: 82, mozjpeg: true }) + .toFile(fallback) + variants.push(await fileVariant("fallback", fallback, "image/jpeg")) + + const thumb = path.join(outDir, "thumb.webp") + await base + .clone() + .resize({ width: Math.min(320, width), withoutEnlargement: true }) + .webp({ quality: 70 }) + .toFile(thumb) + variants.push(await fileVariant("thumb", thumb, "image/webp")) + + const extras = await imageExtras(base) + const describe = variants.find((v) => v.role === "w960") ?? variants.find((v) => v.role === "w480") + return { width, height, ...extras, variants, describeImage: describe?.path ?? null, warnings: [] } +} diff --git a/apps/media-service/src/processing/types.ts b/apps/media-service/src/processing/types.ts new file mode 100644 index 0000000..469392b --- /dev/null +++ b/apps/media-service/src/processing/types.ts @@ -0,0 +1,24 @@ +import type { MediaVariantRole } from "@repo/schemas" + +export type ProducedVariant = { + role: MediaVariantRole + path: string + mime: string + width?: number + height?: number + bytes: number + sha256: string +} + +export type ProcessedMedia = { + width: number + height: number + durationMs?: number + hasAudio?: boolean + dominantColor: string | null + lqip: string | null + variants: ProducedVariant[] + // The image handed to the describe step (w960 or the poster). + describeImage: string | null + warnings: string[] +} diff --git a/apps/media-service/src/processing/video.ts b/apps/media-service/src/processing/video.ts new file mode 100644 index 0000000..4c0b781 --- /dev/null +++ b/apps/media-service/src/processing/video.ts @@ -0,0 +1,241 @@ +import { createHash } from "node:crypto" +import { readFile, stat } from "node:fs/promises" +import path from "node:path" +import sharp from "sharp" +import { z } from "zod" +import type { MediaVariantRole } from "@repo/schemas" +import { MediaRejectedError } from "./errors.js" +import { ffmpeg, FfmpegError, ffprobe, INPUT_GUARD, type FfmpegConfig } from "./ffmpeg.js" +import { imageExtras } from "./image.js" +import type { ProcessedMedia, ProducedVariant } from "./types.js" + +const ALLOWED_CONTAINERS = new Set(["mov,mp4,m4a,3gp,3g2,mj2", "matroska,webm"]) +const MAX_DIMENSION = 3840 +const LIGHT_VIDEO_BYTES = 8 * 1024 * 1024 + +export type VideoOptions = { maxSeconds: number; stripAudio: boolean } + +const ProbeSchema = z.object({ + format: z.object({ + format_name: z.string(), + duration: z.string().optional(), + filename: z.string().optional(), + }), + streams: z.array( + z.object({ + codec_type: z.string().optional(), + width: z.number().optional(), + height: z.number().optional(), + disposition: z.object({ attached_pic: z.number().optional() }).partial().optional(), + side_data_list: z.array(z.object({ rotation: z.number().optional() }).passthrough()).optional(), + }), + ), +}) + +export type VideoProbe = { width: number; height: number; durationMs: number; hasAudio: boolean } + +export function checkProbe(raw: unknown, input: string, maxSeconds: number): VideoProbe { + const parsed = ProbeSchema.safeParse(raw) + if (!parsed.success) throw new MediaRejectedError("Couldn't read this video", "unparseable probe") + const { format, streams } = parsed.data + if (!ALLOWED_CONTAINERS.has(format.format_name)) { + throw new MediaRejectedError("Use an MP4, MOV or WebM video", `container ${format.format_name}`) + } + if (format.filename && path.resolve(format.filename) !== path.resolve(input)) { + throw new MediaRejectedError("Couldn't read this video", "probe referenced another input") + } + const video = streams.find((s) => s.codec_type === "video" && !s.disposition?.attached_pic) + if (!video?.width || !video.height) throw new MediaRejectedError("This file has no video track", "no video stream") + const durationMs = Math.round(Number(format.duration ?? "0") * 1000) + if (!Number.isFinite(durationMs) || durationMs <= 0) { + throw new MediaRejectedError("Couldn't read this video", "unknown duration") + } + if (durationMs > maxSeconds * 1000) { + throw new MediaRejectedError(`Videos can be up to ${maxSeconds} seconds long`, `duration ${durationMs}ms`) + } + if (Math.max(video.width, video.height) > MAX_DIMENSION) { + throw new MediaRejectedError("Videos can be up to 4K (3840 px)", `${video.width}x${video.height}`) + } + const rotation = Math.abs(video.side_data_list?.find((d) => d.rotation !== undefined)?.rotation ?? 0) % 180 + const [width, height] = rotation === 90 ? [video.height, video.width] : [video.width, video.height] + return { width, height, durationMs, hasAudio: streams.some((s) => s.codec_type === "audio") } +} + +const even = (n: number) => Math.max(2, Math.floor(n / 2) * 2) + +async function fileVariant( + role: MediaVariantRole, + file: string, + mime: string, + width?: number, + height?: number, +): Promise { + const [info, data] = await Promise.all([stat(file), readFile(file)]) + return { + role, + path: file, + mime, + width, + height, + bytes: info.size, + sha256: createHash("sha256").update(data).digest("hex"), + } +} + +function scaledSize(probe: VideoProbe, shortSide: number): { width: number; height: number } { + const landscape = probe.width >= probe.height + const short = Math.min(probe.width, probe.height) + const ratio = shortSide / short + return landscape + ? { width: even(probe.width * ratio), height: shortSide } + : { width: shortSide, height: even(probe.height * ratio) } +} + +async function transcode( + cfg: FfmpegConfig, + input: string, + out: string, + shortSide: number, + audio: boolean, +): Promise { + const scale = `scale='if(gte(iw,ih),-2,${shortSide})':'if(gte(iw,ih),${shortSide},-2)':flags=lanczos` + await ffmpeg(cfg, [ + ...INPUT_GUARD, + "-i", + input, + "-map", + "0:v:0", + ...(audio ? ["-map", "0:a:0?"] : []), + "-vf", + scale, + "-fpsmax", + "30", + "-c:v", + "libx264", + "-profile:v", + "high", + "-pix_fmt", + "yuv420p", + "-preset", + "veryfast", + "-crf", + "23", + "-movflags", + "+faststart", + "-map_metadata", + "-1", + "-map_chapters", + "-1", + "-sn", + "-dn", + ...(audio ? ["-c:a", "aac", "-b:a", "128k"] : ["-an"]), + "-f", + "mp4", + out, + ]) +} + +async function frame(cfg: FfmpegConfig, input: string, out: string, position: string[]): Promise { + try { + await ffmpeg(cfg, [ + ...position, + ...INPUT_GUARD, + "-i", + input, + "-frames:v", + "1", + "-update", + "1", + "-f", + "image2", + "-c:v", + "png", + out, + ]) + return (await stat(out)).size > 0 + } catch (err) { + if (err instanceof FfmpegError && err.timedOut) throw err + return false + } +} + +export async function processVideo( + cfg: FfmpegConfig, + input: string, + outDir: string, + options: VideoOptions, +): Promise { + let probe: VideoProbe + try { + probe = checkProbe(await ffprobe(cfg, input), input, options.maxSeconds) + } catch (err) { + if (err instanceof MediaRejectedError) throw err + throw new MediaRejectedError( + "Couldn't read this video", + err instanceof Error ? err.message.slice(0, 200) : "probe failed", + ) + } + + const audio = probe.hasAudio && !options.stripAudio + const short = Math.min(probe.width, probe.height) + const variants: ProducedVariant[] = [] + try { + const s720 = even(Math.min(720, short)) + const mp4720 = path.join(outDir, "mp4_720.mp4") + await transcode(cfg, input, mp4720, s720, audio) + const size720 = scaledSize(probe, s720) + variants.push(await fileVariant("mp4_720", mp4720, "video/mp4", size720.width, size720.height)) + if (short >= 1080) { + const mp41080 = path.join(outDir, "mp4_1080.mp4") + await transcode(cfg, input, mp41080, 1080, audio) + const size1080 = scaledSize(probe, 1080) + variants.push(await fileVariant("mp4_1080", mp41080, "video/mp4", size1080.width, size1080.height)) + } + + const firstPng = path.join(outDir, "first.png") + const at = probe.durationMs >= 1000 ? "0.5" : "0" + if (!(await frame(cfg, input, firstPng, ["-ss", at]))) { + throw new MediaRejectedError("Couldn't read this video", "no decodable frame") + } + const lastPng = path.join(outDir, "last.png") + const hasLast = await frame(cfg, input, lastPng, ["-sseof", "-0.25"]) + + const poster = path.join(outDir, "poster.webp") + await sharp(firstPng).webp({ quality: 80 }).toFile(poster) + variants.push(await fileVariant("poster", poster, "image/webp", probe.width, probe.height)) + const posterLast = path.join(outDir, "poster_last.webp") + await sharp(hasLast ? lastPng : firstPng) + .webp({ quality: 80 }) + .toFile(posterLast) + variants.push(await fileVariant("poster_last", posterLast, "image/webp", probe.width, probe.height)) + const thumb = path.join(outDir, "thumb.webp") + const thumbInfo = await sharp(firstPng) + .resize({ width: Math.min(320, probe.width), withoutEnlargement: true }) + .webp({ quality: 70 }) + .toFile(thumb) + variants.push(await fileVariant("thumb", thumb, "image/webp", thumbInfo.width, thumbInfo.height)) + + const extras = await imageExtras(sharp(firstPng)) + const warnings = + variants[0]!.bytes > LIGHT_VIDEO_BYTES ? ["mp4_720 is over 8 MB; background videos should stay light"] : [] + return { + width: probe.width, + height: probe.height, + durationMs: probe.durationMs, + hasAudio: audio, + ...extras, + variants, + describeImage: poster, + warnings, + } + } catch (err) { + if (err instanceof MediaRejectedError) throw err + if (err instanceof FfmpegError) { + throw new MediaRejectedError( + err.timedOut ? "This video took too long to process" : "Couldn't convert this video", + err.message.slice(0, 200), + ) + } + throw err + } +} diff --git a/apps/media-service/src/worker.ts b/apps/media-service/src/worker.ts new file mode 100644 index 0000000..40c4365 --- /dev/null +++ b/apps/media-service/src/worker.ts @@ -0,0 +1,67 @@ +import { mkdir } from "node:fs/promises" +import { Queue, Worker } from "bullmq" +import { prisma } from "@useframe/db" +import { QUEUES, type MediaProcessJobPayload } from "@repo/events" +import { env } from "@/config/env.js" +import { closeContext, deps } from "@/context.js" +import { runMediaGc } from "@/jobs/gc.js" +import { processAsset } from "@/jobs/process.js" +import { startHealthServer } from "@/lib/health.js" +import { errorMessage, log } from "@/lib/logger.js" +import { redis } from "@/lib/redis.js" + +const start = async () => { + await prisma.$connect() + await mkdir(env.MEDIA_WORK_DIR, { recursive: true, mode: 0o700 }) + + const gcQueue = new Queue(QUEUES.MEDIA_GC, { connection: redis }) + await gcQueue.upsertJobScheduler("media-gc-daily", { pattern: "15 4 * * *" }, { name: "gc" }) + + const workers = [ + new Worker( + QUEUES.MEDIA_PROCESS, + (job) => processAsset(deps, job.data, { attemptsMade: job.attemptsMade, attempts: job.opts.attempts ?? 1 }), + { + connection: redis, + concurrency: env.MEDIA_PROCESS_CONCURRENCY, + // A transcode keeps a child process busy for minutes; the lock must outlast it. + lockDuration: env.MEDIA_FFMPEG_TIMEOUT_MS + 60_000, + }, + ), + new Worker(QUEUES.MEDIA_GC, () => runMediaGc(deps), { connection: redis }), + ] + for (const worker of workers) { + worker.on("failed", (job, err) => + log.error("job failed", { queue: worker.name, jobId: job?.id, error: errorMessage(err) }), + ) + } + + let shuttingDown = false + if (env.HEALTH_PORT) { + // Draining after SIGTERM is expected, so it must not fail liveness. + startHealthServer( + env.HEALTH_PORT, + () => redis.status === "ready" && (shuttingDown || workers.every((w) => w.isRunning())), + ) + } + log.info("media worker started", { + concurrency: env.MEDIA_PROCESS_CONCURRENCY, + describe: !!deps.describe, + workDir: env.MEDIA_WORK_DIR, + }) + + const shutdown = () => { + shuttingDown = true + log.info("media worker shutting down; waiting for in-flight jobs") + void Promise.all(workers.map((w) => w.close())) + .then(() => gcQueue.close()) + .finally(() => closeContext().finally(() => process.exit(0))) + } + process.once("SIGTERM", shutdown) + process.once("SIGINT", shutdown) +} + +start().catch((err) => { + console.error("Failed to start media worker:", err) + process.exit(1) +}) diff --git a/apps/media-service/test/fakes.ts b/apps/media-service/test/fakes.ts new file mode 100644 index 0000000..cb0cfc1 --- /dev/null +++ b/apps/media-service/test/fakes.ts @@ -0,0 +1,132 @@ +import { mkdtemp, writeFile } from "node:fs/promises" +import os from "node:os" +import path from "node:path" +import { Readable } from "node:stream" +import { createHash } from "node:crypto" +import type { Deps } from "@/deps.js" +import { TooLargeError, type MediaStore } from "@/lib/storage.js" + +type Row = Record + +function matches(row: Row, where: Row | undefined): boolean { + if (!where) return true + return Object.entries(where).every(([key, cond]) => { + const value = row[key] + if (cond === null) return value === null || value === undefined + if (cond && typeof cond === "object" && !(cond instanceof Date)) { + const c = cond as { lt?: Date; not?: unknown } + if (c.lt) return value instanceof Date && value < c.lt + if ("not" in c) return value !== c.not + } + return value === cond + }) +} + +export function createFakeDb() { + const assets: Row[] = [] + const logs: Row[] = [] + const balances = new Map() + let seq = 0 + const mediaAsset = { + findUnique: async ({ where }: { where: { id: string } }) => assets.find((a) => a.id === where.id) ?? null, + findMany: async ({ where, take }: { where?: Row; take?: number }) => + assets.filter((a) => matches(a, where)).slice(0, take ?? Infinity), + updateMany: async ({ where, data }: { where: Row; data: Row }) => { + const rows = assets.filter((a) => matches(a, where)) + for (const row of rows) Object.assign(row, data) + return { count: rows.length } + }, + create: async ({ data }: { data: Row }) => { + const row: Row = { id: `asset${++seq}`, deletedAt: null, altText: null, createdAt: new Date(), ...data } + assets.push(row) + return row + }, + update: async ({ where, data }: { where: { id: string }; data: Row }) => + Object.assign(assets.find((a) => a.id === where.id)!, data), + delete: async ({ where }: { where: { id: string } }) => { + const i = assets.findIndex((a) => a.id === where.id) + return assets.splice(i, 1)[0] + }, + } + const db = { + mediaAsset, + creditBalance: { + findUnique: async ({ where }: { where: { userId: string } }) => ({ balance: balances.get(where.userId) ?? 0 }), + }, + pipelineLog: { + upsert: async ({ create }: { create: Row }) => { + logs.push(create) + return create + }, + }, + $transaction: async (fn: (tx: unknown) => Promise) => fn(db), + } + return { db: db as unknown as Deps["db"], assets, logs, balances } +} + +export function createFakeStore(): MediaStore & { objects: Map; failUploads: number } { + const objects = new Map() + const store = { + objects, + failUploads: 0, + async download(key: string, dest: string, maxBytes: number) { + const body = objects.get(key) + if (!body) return null + if (body.length > maxBytes) throw new TooLargeError() + await writeFile(dest, body) + return { bytes: body.length, sha256: createHash("sha256").update(body).digest("hex") } + }, + async uploadFile(key: string, absPath: string) { + if (store.failUploads > 0) { + store.failUploads-- + throw Object.assign(new Error("Service Unavailable"), { $metadata: { httpStatusCode: 503 } }) + } + const { readFile } = await import("node:fs/promises") + objects.set(key, await readFile(absPath)) + }, + async put(key: string, body: Buffer | Readable) { + if (Buffer.isBuffer(body)) { + objects.set(key, body) + return + } + const chunks: Buffer[] = [] + for await (const chunk of body) chunks.push(Buffer.from(chunk as Uint8Array)) + objects.set(key, Buffer.concat(chunks)) + }, + async head(key: string) { + const obj = objects.get(key) + return obj ? { size: obj.length } : null + }, + async deletePrefix(prefix: string) { + let n = 0 + for (const key of [...objects.keys()]) if (key.startsWith(prefix) && objects.delete(key)) n++ + return n + }, + } + return store +} + +export async function createDeps(overrides: Partial = {}) { + const fake = createFakeDb() + const store = createFakeStore() + const workDir = await mkdtemp(path.join(os.tmpdir(), "media-test-")) + const enqueued: { payload: unknown; jobId: string }[] = [] + const deps: Deps = { + db: fake.db, + store, + describe: null, + enqueue: async (payload, jobId) => { + enqueued.push({ payload, jobId }) + }, + config: { + workDir, + maxInputPixels: 60_000_000, + maxVideoSeconds: 60, + maxImageBytes: 20 * 1024 * 1024, + maxVideoBytes: 150 * 1024 * 1024, + ffmpeg: { ffmpegPath: "ffmpeg", ffprobePath: "ffprobe", timeoutMs: 60_000, threads: 2 }, + }, + ...overrides, + } + return { deps, fake, store, workDir, enqueued } +} diff --git a/apps/media-service/test/image.test.ts b/apps/media-service/test/image.test.ts new file mode 100644 index 0000000..0de95d6 --- /dev/null +++ b/apps/media-service/test/image.test.ts @@ -0,0 +1,150 @@ +import { deflateSync, crc32 } from "node:zlib" +import { mkdtemp, readFile, writeFile } from "node:fs/promises" +import os from "node:os" +import path from "node:path" +import sharp from "sharp" +import { beforeEach, describe, expect, it } from "vitest" +import { MediaRejectedError } from "@/processing/errors.js" +import { processImage } from "@/processing/image.js" + +let dir = "" +const limits = { maxInputPixels: 60_000_000 } + +beforeEach(async () => { + dir = await mkdtemp(path.join(os.tmpdir(), "media-img-")) +}) + +async function fixture(name: string, data: Buffer): Promise { + const file = path.join(dir, name) + await writeFile(file, data) + return file +} + +const photo = (width: number, height: number) => + sharp({ create: { width, height, channels: 3, background: { r: 200, g: 80, b: 40 } } }) + +function chunk(type: string, data: Buffer): Buffer { + const len = Buffer.alloc(4) + len.writeUInt32BE(data.length) + const body = Buffer.concat([Buffer.from(type, "ascii"), data]) + const crc = Buffer.alloc(4) + crc.writeUInt32BE(crc32(body)) + return Buffer.concat([len, body, crc]) +} + +// A valid PNG header that claims 30000x30000 pixels, with almost no data behind it. +function pngBomb(): Buffer { + const ihdr = Buffer.alloc(13) + ihdr.writeUInt32BE(30_000, 0) + ihdr.writeUInt32BE(30_000, 4) + ihdr[8] = 8 + ihdr[9] = 2 + return Buffer.concat([ + Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]), + chunk("IHDR", ihdr), + chunk("IDAT", deflateSync(Buffer.alloc(64))), + chunk("IEND", Buffer.alloc(0)), + ]) +} + +describe("processImage", () => { + it("strips EXIF/GPS metadata from every variant and auto-orients", async () => { + const input = await photo(2000, 1000) + .withExif({ + IFD0: { Make: "SpyCam", Copyright: "secret" }, + IFD3: { GPSLatitudeRef: "N", GPSLatitude: "51/1 30/1 0/1" }, + }) + .withMetadata({ orientation: 6 }) + .jpeg() + .toBuffer() + expect((await sharp(input).metadata()).exif).toBeDefined() + const out = await processImage(await fixture("orig", input), dir, limits) + // Orientation 6 means "rotate 90°", so the stored landscape becomes portrait. + expect([out.width, out.height]).toEqual([1000, 2000]) + for (const v of out.variants) { + const meta = await sharp(v.path).metadata() + expect(meta.exif, v.role).toBeUndefined() + expect(meta.xmp, v.role).toBeUndefined() + expect(meta.orientation ?? 1, v.role).toBe(1) + const bytes = await readFile(v.path) + expect(bytes.includes(Buffer.from("SpyCam")), v.role).toBe(false) + } + }) + + it("produces WebP widths, a JPEG fallback, a thumb, LQIP and a dominant colour without upscaling", async () => { + const out = await processImage(await fixture("orig", await photo(1700, 900).png().toBuffer()), dir, limits) + expect(out.variants.map((v) => v.role)).toEqual(["w480", "w960", "w1600", "fallback", "thumb"]) + expect(out.variants.find((v) => v.role === "w1600")).toMatchObject({ mime: "image/webp", width: 1600 }) + expect(out.variants.find((v) => v.role === "fallback")).toMatchObject({ mime: "image/jpeg", width: 1600 }) + expect(out.dominantColor).toMatch(/^#[0-9a-f]{6}$/) + expect(out.lqip).toMatch(/^data:image\/webp;base64,/) + expect(out.lqip!.length).toBeLessThanOrEqual(1536) + for (const v of out.variants) expect(v.sha256).toMatch(/^[0-9a-f]{64}$/) + expect(out.describeImage).toBe(out.variants.find((v) => v.role === "w960")!.path) + }) + + it("adds w2400 only for sources at least 2400 px wide", async () => { + const out = await processImage(await fixture("orig", await photo(2600, 1300).jpeg().toBuffer()), dir, limits) + expect(out.variants.find((v) => v.role === "w2400")?.width).toBe(2400) + }) + + it("never upscales a tiny image", async () => { + const out = await processImage(await fixture("orig", await photo(120, 60).png().toBuffer()), dir, limits) + expect(out.variants.map((v) => [v.role, v.width])).toEqual([ + ["w480", 120], + ["fallback", 120], + ["thumb", 120], + ]) + }) + + it("keeps alpha in WebP and flattens the JPEG fallback", async () => { + const png = await sharp({ + create: { width: 600, height: 400, channels: 4, background: { r: 0, g: 0, b: 255, alpha: 0.3 } }, + }) + .png() + .toBuffer() + const out = await processImage(await fixture("orig", png), dir, limits) + expect((await sharp(out.variants.find((v) => v.role === "w480")!.path).metadata()).hasAlpha).toBe(true) + expect((await sharp(out.variants.find((v) => v.role === "fallback")!.path).metadata()).hasAlpha).toBe(false) + }) + + it("rejects a 30000x30000 decompression bomb by its pixel count", async () => { + await expect(processImage(await fixture("orig", pngBomb()), dir, limits)).rejects.toMatchObject({ + publicReason: "This image is too large to process", + }) + }) + + it("rejects SVG renamed to .png, GIF and garbage", async () => { + const svg = Buffer.from( + '', + ) + await expect(processImage(await fixture("logo.png", svg), dir, limits)).rejects.toBeInstanceOf(MediaRejectedError) + const gif = await sharp({ create: { width: 10, height: 10, channels: 3, background: "#fff" } }) + .gif() + .toBuffer() + await expect(processImage(await fixture("a.gif", gif), dir, limits)).rejects.toMatchObject({ + publicReason: "Couldn't read this image", + }) + await expect(processImage(await fixture("junk", Buffer.from("not an image")), dir, limits)).rejects.toBeInstanceOf( + MediaRejectedError, + ) + }) + + it("decodes a JPEG with HTML appended and emits clean variants", async () => { + const polyglot = Buffer.concat([ + await photo(800, 600).jpeg().toBuffer(), + Buffer.from(""), + ]) + const out = await processImage(await fixture("orig", polyglot), dir, limits) + for (const v of out.variants) expect((await readFile(v.path)).includes(Buffer.from("') diff --git a/apps/server/src/modules/projects/projects.routes.ts b/apps/server/src/modules/projects/projects.routes.ts index c246290..45938d4 100644 --- a/apps/server/src/modules/projects/projects.routes.ts +++ b/apps/server/src/modules/projects/projects.routes.ts @@ -12,6 +12,7 @@ import deployRoutes from "@/modules/deploy/deploy.routes.js" import pipelineRoutes from "@/modules/pipeline/pipeline.routes.js" import domainsRoutes from "@/modules/domains/domains.routes.js" import { projectBriefRouter } from "@/modules/briefs/briefs.routes.js" +import mediaRoutes from "@/modules/media/media.routes.js" const router: Router = Router() @@ -35,5 +36,6 @@ router.use("/:slug/seo-step", seoStepRoutes) router.use("/:slug", deployRoutes) router.use("/:slug/pipeline", pipelineRoutes) router.use("/:slug/domains", domainsRoutes) +router.use("/:slug/media", mediaRoutes) export default router diff --git a/apps/server/tsconfig.json b/apps/server/tsconfig.json index 5af9534..9857d31 100644 --- a/apps/server/tsconfig.json +++ b/apps/server/tsconfig.json @@ -18,5 +18,5 @@ } }, "include": ["src"], - "exclude": ["node_modules", "dist", "**/*.test.ts"] + "exclude": ["node_modules", "dist", "**/*.test.ts", "**/*.testkit.ts"] } diff --git a/apps/site-edge/src/index.ts b/apps/site-edge/src/index.ts index 37a4e87..49c8685 100644 --- a/apps/site-edge/src/index.ts +++ b/apps/site-edge/src/index.ts @@ -115,33 +115,81 @@ function isMedia(contentType: string): boolean { return contentType.startsWith("video/") || contentType.startsWith("audio/"); } -async function serveRange( +// The Cache API never stores 206s, so video and ranged requests go straight to R2. +const DIRECT_EXTENSIONS = /\.(mp4|webm|mov)$/i; +// Deployed media files carry a content hash in their name and never change. +const HASHED_MEDIA = /^\/media\/[A-Za-z0-9_-]+\/[A-Za-z0-9_-]+\.[0-9a-f]{8}\.[a-z0-9]+$/; +const IMMUTABLE = "public, max-age=31536000, immutable"; +const BYTE_RANGE = /^bytes=(\d*)-(\d*)$/; + +// Malformed ranges get a full 200; Miniflare ignores out-of-bounds ranges, so satisfiability is decided here. +function rangeIntent(header: string | null, size: number): "none" | "partial" | "unsatisfiable" { + const match = header ? BYTE_RANGE.exec(header.trim()) : null; + if (!match || (match[1] === "" && match[2] === "")) return "none"; + if (match[1] !== "") return Number(match[1]) >= size ? "unsatisfiable" : "partial"; + return Number(match[2]) === 0 ? "unsatisfiable" : "partial"; +} + +function objectHeaders(headers: Headers, obj: R2Object, cacheControl: string | undefined): void { + headers.set("content-type", obj.httpMetadata?.contentType ?? "application/octet-stream"); + headers.set("etag", obj.httpEtag); + headers.set("cache-control", cacheControl ?? obj.httpMetadata?.cacheControl ?? "public, max-age=3600"); + headers.set("accept-ranges", "bytes"); + headers.set("x-useframe-cache", "bypass"); +} + +async function serveDirect( env: Env, request: Request, r2Key: string, headers: Headers, + cacheControl: string | undefined, ): Promise { + const ranged = rangeIntent(request.headers.get("range"), Infinity) !== "none"; + if (request.method === "HEAD" && !ranged) { + const head = await env.SITES_BUCKET.head(r2Key); + if (!head) return null; + objectHeaders(headers, head, cacheControl); + headers.set("content-length", String(head.size)); + const fresh = etagMatches(request.headers.get("if-none-match"), head.httpEtag); + return new Response(null, { status: fresh ? 304 : 200, headers }); + } + let obj: R2ObjectBody | R2Object | null; try { - obj = await env.SITES_BUCKET.get(r2Key, { range: request.headers }); + obj = await env.SITES_BUCKET.get(r2Key, { range: ranged ? request.headers : undefined, onlyIf: request.headers }); } catch { const head = await env.SITES_BUCKET.head(r2Key); if (!head) return null; + objectHeaders(headers, head, cacheControl); headers.set("content-range", `bytes */${head.size}`); return new Response(null, { status: 416, headers }); } if (!obj) return null; + objectHeaders(headers, obj, cacheControl); - headers.set("content-type", obj.httpMetadata?.contentType ?? "application/octet-stream"); - headers.set("etag", obj.httpEtag); - headers.set("cache-control", obj.httpMetadata?.cacheControl ?? "public, max-age=3600"); - headers.set("accept-ranges", "bytes"); - headers.set("x-useframe-cache", "bypass"); + // A failed precondition returns metadata without a body. + if (!("body" in obj)) { + const conditional = request.headers.has("if-none-match") || request.headers.has("if-modified-since"); + return new Response(null, { status: conditional ? 304 : 412, headers }); + } + // R2 compares strong ETags only; browsers often revalidate with weak ones. + if (etagMatches(request.headers.get("if-none-match"), obj.httpEtag)) { + await obj.body.cancel(); + return new Response(null, { status: 304, headers }); + } - const body = "body" in obj ? obj.body : null; - const range = obj.range; + const intent = rangeIntent(request.headers.get("range"), obj.size); + if (intent === "unsatisfiable") { + await obj.body.cancel(); + headers.set("content-range", `bytes */${obj.size}`); + return new Response(null, { status: 416, headers }); + } + const range = intent === "partial" ? obj.range : undefined; if (!range) { - return new Response(request.method === "HEAD" ? null : body, { status: 200, headers }); + headers.set("content-length", String(obj.size)); + if (request.method === "HEAD") await obj.body.cancel(); + return new Response(request.method === "HEAD" ? null : obj.body, { status: 200, headers }); } // The runtime may return every key with undefined values, so check values, not keys. const r = range as { offset?: number; length?: number; suffix?: number }; @@ -152,11 +200,12 @@ async function serveRange( offset = obj.size - length; } else { offset = r.offset ?? 0; - length = r.length ?? obj.size - offset; + length = Math.min(r.length ?? obj.size - offset, obj.size - offset); } headers.set("content-range", `bytes ${offset}-${offset + length - 1}/${obj.size}`); headers.set("content-length", String(length)); - return new Response(request.method === "HEAD" ? null : body, { status: 206, headers }); + if (request.method === "HEAD") await obj.body.cancel(); + return new Response(request.method === "HEAD" ? null : obj.body, { status: 206, headers }); } async function serve( @@ -177,9 +226,10 @@ async function serve( return redirect(url.pathname + "/" + url.search, headers); } - if (request.headers.has("range")) { - const ranged = await serveRange(env, request, `sites/${deploymentKey}/${plan.key}`, headers); - if (ranged) return ranged; + const immutable = HASHED_MEDIA.test(path) ? IMMUTABLE : undefined; + if (request.headers.has("range") || DIRECT_EXTENSIONS.test(plan.key)) { + const direct = await serveDirect(env, request, `sites/${deploymentKey}/${plan.key}`, headers, immutable); + if (direct) return direct; } let status = 200; @@ -195,7 +245,7 @@ async function serve( headers.set("content-type", loaded.contentType); if (loaded.etag) headers.set("etag", loaded.etag); - headers.set("cache-control", status === 404 ? HTML_404_CACHE_CONTROL : loaded.cacheControl); + headers.set("cache-control", status === 404 ? HTML_404_CACHE_CONTROL : (immutable ?? loaded.cacheControl)); headers.set("x-useframe-cache", loaded.cache); if (isMedia(loaded.contentType)) headers.set("accept-ranges", "bytes"); diff --git a/apps/site-edge/test/edge.worker.test.ts b/apps/site-edge/test/edge.worker.test.ts index fb4df58..e752fd9 100644 --- a/apps/site-edge/test/edge.worker.test.ts +++ b/apps/site-edge/test/edge.worker.test.ts @@ -250,3 +250,110 @@ describe("responses", () => { expect(r2).toHaveBeenCalledTimes(1); }); }); + +describe("media and ranges", () => { + const CLIP = `https://${NEXT_HOST}/media/clip.mp4`; + const HASHED = `https://${NEXT_HOST}/media/a1b2/mp4_720.0123abcd.mp4`; + + beforeEach(async () => { + await put("sites/proj1/dep1/media/a1b2/mp4_720.0123abcd.mp4", "abcdefghij", "video/mp4", "public, max-age=3600"); + await put("sites/proj1/dep1/media/a1b2/w960.0123abcd.webp", "webp-bytes", "image/webp", "public, max-age=3600"); + }); + + it("serves a full video straight from R2 with length and range support", async () => { + const r2 = vi.spyOn(env.SITES_BUCKET, "get"); + const res = await call(CLIP); + expect(res.status).toBe(200); + expect(res.headers.get("content-length")).toBe("10"); + expect(res.headers.get("accept-ranges")).toBe("bytes"); + expect(res.headers.get("x-useframe-cache")).toBe("bypass"); + expect(res.headers.get("x-useframe-deployment")).toBe("dep1"); + expect(res.headers.get("x-content-type-options")).toBe("nosniff"); + expect(await res.text()).toBe("0123456789"); + expect(r2.mock.calls[0]![1]).toMatchObject({ range: undefined, onlyIf: expect.any(Headers) }); + const ranged = await call(CLIP, { headers: { range: "bytes=0-1" } }); + expect(r2.mock.calls[1]![1]).toMatchObject({ range: expect.any(Headers), onlyIf: expect.any(Headers) }); + await ranged.text(); + }); + + it("answers open-ended and suffix ranges", async () => { + const open = await call(CLIP, { headers: { range: "bytes=6-" } }); + expect(open.status).toBe(206); + expect(open.headers.get("content-range")).toBe("bytes 6-9/10"); + expect(open.headers.get("content-length")).toBe("4"); + expect(await open.text()).toBe("6789"); + + const suffix = await call(CLIP, { headers: { range: "bytes=-3" } }); + expect(suffix.status).toBe(206); + expect(suffix.headers.get("content-range")).toBe("bytes 7-9/10"); + expect(await suffix.text()).toBe("789"); + }); + + it("answers 416 for an unsatisfiable range", async () => { + const res = await call(CLIP, { headers: { range: "bytes=50-60" } }); + expect(res.status).toBe(416); + expect(res.headers.get("content-range")).toBe("bytes */10"); + expect(await res.text()).toBe(""); + }); + + it("ignores a malformed range and serves the whole file", async () => { + const res = await call(CLIP, { headers: { range: "pages=1-2" } }); + expect(res.status).toBe(200); + expect(await res.text()).toBe("0123456789"); + }); + + it("answers HEAD with the full length and no body", async () => { + const res = await call(CLIP, { method: "HEAD" }); + expect(res.status).toBe(200); + expect(res.headers.get("content-length")).toBe("10"); + expect(res.headers.get("accept-ranges")).toBe("bytes"); + expect(await res.text()).toBe(""); + + const ranged = await call(CLIP, { method: "HEAD", headers: { range: "bytes=0-1" } }); + expect(ranged.status).toBe(206); + expect(ranged.headers.get("content-range")).toBe("bytes 0-1/10"); + expect(await ranged.text()).toBe(""); + }); + + it("answers 304 for a matching If-None-Match, strong or weak", async () => { + const first = await call(CLIP); + const etag = first.headers.get("etag")!; + await first.text(); + for (const tag of [etag, `W/${etag}`]) { + const res = await call(CLIP, { headers: { "if-none-match": tag } }); + expect(res.status).toBe(304); + expect(res.headers.get("etag")).toBe(etag); + expect(await res.text()).toBe(""); + } + const head = await call(CLIP, { method: "HEAD", headers: { "if-none-match": etag } }); + expect(head.status).toBe(304); + }); + + it("marks hashed media paths immutable on both the direct and the cached path", async () => { + const video = await call(HASHED, { headers: { range: "bytes=0-3" } }); + expect(video.status).toBe(206); + expect(video.headers.get("cache-control")).toBe("public, max-age=31536000, immutable"); + expect(await video.text()).toBe("abcd"); + + const image = await call(`https://${NEXT_HOST}/media/a1b2/w960.0123abcd.webp`); + expect(image.status).toBe(200); + expect(image.headers.get("cache-control")).toBe("public, max-age=31536000, immutable"); + expect(image.headers.get("x-useframe-cache")).toBe("miss"); + expect(await image.text()).toBe("webp-bytes"); + + const unhashed = await call(CLIP); + expect(unhashed.headers.get("cache-control")).toBe("public, max-age=3600"); + await unhashed.text(); + }); + + it("serves media in SPA mode and 404s a missing video", async () => { + await put("sites/proj2/dep2/media/x/clip.0123abcd.mp4", "spa-video", "video/mp4"); + const res = await call(`https://${SPA_HOST}/media/x/clip.0123abcd.mp4`, { headers: { range: "bytes=0-2" } }); + expect(res.status).toBe(206); + expect(await res.text()).toBe("spa"); + + const missing = await call(`https://${NEXT_HOST}/media/nope.mp4`); + expect(missing.status).toBe(404); + await missing.text(); + }); +}); diff --git a/apps/web/src/components/brief/siteBuilderXss.test.ts b/apps/web/src/components/brief/siteBuilderXss.test.ts index fa29a0a..04b37d4 100644 --- a/apps/web/src/components/brief/siteBuilderXss.test.ts +++ b/apps/web/src/components/brief/siteBuilderXss.test.ts @@ -4,7 +4,7 @@ import { afterAll, describe, expect, it } from "vitest" import * as React from "react" import { renderToStaticMarkup } from "react-dom/server" import { buildSiteFiles } from "@repo/site-builder" -import type { SiteSpec } from "@repo/schemas" +import type { ResolvedMediaAsset, SiteSpec } from "@repo/schemas" const PAYLOADS = [ "", @@ -89,4 +89,60 @@ describe("site-builder output encoding", () => { expect(css).not.toContain("javascript:") expect(css).not.toContain(" { + const asset = (id: string, kind: "IMAGE" | "VIDEO"): ResolvedMediaAsset => ({ + id, + kind, + origin: "UPLOADED", + status: "READY", + deleted: false, + title: evil, + width: 1600, + height: 900, + durationMs: kind === "VIDEO" ? 5000 : null, + dominantColor: 'red;background:url(javascript:alert(1))', + lqip: null, + altText: evil, + decorative: false, + variants: + kind === "IMAGE" + ? [ + { role: "w960", mime: "image/webp", width: 960, height: 540, bytes: 1, sha256: "a".repeat(64), url: "https://api.test/m/w960?sig=1" }, + { role: "fallback", mime: "image/jpeg", width: 1600, height: 900, bytes: 1, sha256: "b".repeat(64), url: 'javascript:alert(1)//"' }, + ] + : [ + { role: "mp4_720", mime: "video/mp4", width: 1280, height: 720, bytes: 1, sha256: "c".repeat(64), url: "https://api.test/m/v?sig=1" }, + { role: "poster", mime: "image/webp", width: 1280, height: 720, bytes: 1, sha256: "d".repeat(64), url: "https://api.test/m/p?sig=1" }, + ], + }) + const mediaSpec: SiteSpec = { + ...spec, + media: { + "home/hero-0/visual": { assetId: "img", alt: evil }, + "home/how_it_works-2/item-0": { assetId: "vid", alt: evil, objectPosition: "center" }, + }, + } + const exported = buildSiteFiles(mediaSpec, undefined, { media: { target: "export", assets: [asset("img", "IMAGE"), asset("vid", "VIDEO")] } }) + const preview = buildSiteFiles(mediaSpec, undefined, { media: { target: "preview", assets: [asset("img", "IMAGE"), asset("vid", "VIDEO")] } }) + + for (const [name, files] of [["export", exported], ["preview", preview]] as const) { + const app = files.find((f) => f.path === "src/App.jsx")!.content + writeFileSync(join(tmpDir, `Media-${name}.jsx`), `import * as React from "react" +${app}`) + const mod = (await import(/* @vite-ignore */ join(tmpDir, `Media-${name}.jsx`))) as { default: React.ComponentType } + const html = renderToStaticMarkup(React.createElement(mod.default)) + + expect((globalThis as Record).__pwned).toBeUndefined() + expect(html).not.toMatch(/