diff --git a/docs/packages/cli.mdx b/docs/packages/cli.mdx index d072def67..435a20edd 100644 --- a/docs/packages/cli.mdx +++ b/docs/packages/cli.mdx @@ -934,6 +934,46 @@ hyperframes lambda render ./my-template --site-id=abc1234deadbeef0 \ Variables travel inside the Step Functions Standard execution input, which AWS caps at 256 KiB for the entire payload. Pass typed data (strings, numbers, structured records) through variables; URL-reference media assets (images, audio, video) the composition resolves at render time rather than inlining bytes. The SDK validates the size client-side and rejects oversize inputs with a clear error before any AWS call runs — see the [templates-on-lambda guide](/deploy/templates-on-lambda) for the URL-your-assets convention. +#### `lambda render-batch ` + +Fans out N personalised renders from a JSONL batch file — the headline ergonomic for automated template-rendering pipelines. Deploys the site once (or skips with `--site-id`), then invokes `renderToLambda` per batch row with per-entry `variables` and `outputKey`. Concurrent Step Functions starts are capped at `--max-concurrent` (default 50) so a 10 000-entry batch doesn't try to spawn 10 000 executions at once and trip AWS account limits. + +Batch file format (JSONL — one JSON object per line): + +```jsonl +{"outputKey": "renders/alice.mp4", "variables": {"name": "Alice", "accent": "#ff0000"}} +{"outputKey": "renders/bob.mp4", "variables": {"name": "Bob", "accent": "#0000ff"}} +{"outputKey": "renders/carol.mp4", "variables": {"name": "Carol"}, "executionName": "hf-carol-001"} +``` + +```bash +hyperframes lambda render-batch ./my-template \ + --batch ./users.jsonl \ + --width 1920 --height 1080 \ + --max-concurrent 10 +``` + +The verb prints a manifest — one row per input line — with `executionArn` + status: + +``` +Batch dispatched: 3 started, 0 failed-to-start. + + ✓ line 1 renders/alice.mp4 arn:aws:states:us-east-1:1234:execution:hf:hf-render-... + ✓ line 2 renders/bob.mp4 arn:aws:states:us-east-1:1234:execution:hf:hf-render-... + ✓ line 3 renders/carol.mp4 arn:aws:states:us-east-1:1234:execution:hf:hf-carol-001 +``` + +Pass `--json` for the machine-readable form. Poll each execution via `hyperframes lambda progress ` (or use the returned `executionArn`). + +`--dry-run` skips the AWS calls and prints the manifest with `status: "would-invoke"` for every entry — use it to lint the batch file before committing to N billable executions: + +```bash +hyperframes lambda render-batch ./my-template --batch ./users.jsonl \ + --width 1920 --height 1080 --dry-run --json +``` + +`--max-concurrent` is orchestrator-side only: it caps how many `StartExecution` calls run simultaneously, not how many Lambda invocations the account can run. AWS account-level Lambda concurrency limits live one level up and `render-batch` cannot enforce them; pick `--max-concurrent` based on your account's `concurrent-execution` quota and the Lambda reserved concurrency you provisioned via `lambda deploy --concurrency=`. + #### `lambda progress ` Prints one progress snapshot — overall percent, frames rendered, Lambda invocations, accrued cost, and any errors. Accepts either a bare `renderId` (resolved against the stack's state-machine ARN) or a full SFN execution ARN. diff --git a/packages/cli/src/commands/lambda.ts b/packages/cli/src/commands/lambda.ts index e7aa8b8c3..75004fabd 100644 --- a/packages/cli/src/commands/lambda.ts +++ b/packages/cli/src/commands/lambda.ts @@ -31,6 +31,10 @@ export const examples: Example[] = [ "Render with variables from a JSON file", "hyperframes lambda render ./my-template --site-id abc1234deadbeef0 --width 1920 --height 1080 --variables-file ./alice.json", ], + [ + "Batch-render N personalised videos from a JSONL file (deploys the site once)", + "hyperframes lambda render-batch ./my-template --batch ./users.jsonl --width 1920 --height 1080 --max-concurrent 10", + ], ["Check progress for a started render", "hyperframes lambda progress hf-render-abcd1234"], [ "Pre-upload a project so multiple renders share the upload", @@ -53,6 +57,7 @@ ${c.bold("SUBCOMMANDS:")} ${c.accent("deploy")} ${c.dim("Provision the Lambda + Step Functions + S3 stack via SAM")} ${c.accent("sites create")} ${c.dim("Tar + upload a project to S3 (reusable across renders)")} ${c.accent("render")} ${c.dim("Start a distributed render (returns a renderId)")} + ${c.accent("render-batch")} ${c.dim("Fan out N personalised renders from a JSONL batch file")} ${c.accent("progress")} ${c.dim("Print progress + cost for an in-flight or finished render")} ${c.accent("destroy")} ${c.dim("Tear the stack down (S3 bucket is retained)")} ${c.accent("policies")} ${c.dim("Print or validate the IAM permissions the CLI needs")} @@ -141,6 +146,23 @@ export default defineCommand({ "Fail the render command if any --variables key is undeclared or has a wrong type vs the composition's data-composition-variables. Without this flag, mismatches are warnings.", default: false, }, + // render-batch + batch: { + type: "string", + description: + 'Path to a JSONL batch file for `render-batch`. Each line: {"outputKey":"...","variables":{...}}', + }, + "max-concurrent": { + type: "string", + description: + "Max in-flight Step Functions executions for `render-batch` (default: 50). Distinct from --max-parallel-chunks (which caps chunks per render).", + }, + "dry-run": { + type: "boolean", + description: + "For `render-batch`: parse the batch file and print the manifest without invoking AWS. Every entry's status becomes `would-invoke`.", + default: false, + }, wait: { type: "boolean", description: "Block until the render finishes" }, "wait-interval-ms": { type: "string", @@ -179,7 +201,14 @@ export default defineCommand({ // dep) so the published CLI install stays small for users who don't // deploy to Lambda. Subverbs other than `policies` need aws-lambda; // catch the missing-module error here and turn it into a friendly hint. - const verbsNeedingSDK = new Set(["deploy", "sites", "render", "progress", "destroy"]); + const verbsNeedingSDK = new Set([ + "deploy", + "sites", + "render", + "render-batch", + "progress", + "destroy", + ]); if (verbsNeedingSDK.has(subcommand)) { try { await import("@hyperframes/aws-lambda/sdk"); @@ -278,6 +307,53 @@ export default defineCommand({ }); return; } + case "render-batch": { + const projectDir = args.target as string | undefined; + if (!projectDir) { + console.error( + "[lambda render-batch] usage: hyperframes lambda render-batch --batch --width --height ", + ); + process.exit(1); + } + const batch = args.batch as string | undefined; + if (!batch) { + console.error( + "[lambda render-batch] --batch is required. Each line is a JSON object with at least { outputKey: '...' }.", + ); + process.exit(1); + } + const width = parsePositiveInt(args.width, "--width"); + const height = parsePositiveInt(args.height, "--height"); + if (width === undefined || height === undefined) { + console.error("[lambda render-batch] --width and --height are required."); + process.exit(1); + } + const fpsRaw = parseIntFlag(args.fps) ?? 30; + if (fpsRaw !== 24 && fpsRaw !== 30 && fpsRaw !== 60) { + console.error(`[lambda render-batch] --fps must be 24, 30, or 60; got ${fpsRaw}.`); + process.exit(1); + } + const { runRenderBatch } = await import("./lambda/render-batch.js"); + await runRenderBatch({ + projectDir, + stackName, + batch, + siteId: args["site-id"] as string | undefined, + fps: fpsRaw, + width, + height, + format: parseFormat(args.format), + codec: parseCodec(args.codec), + quality: parseQuality(args.quality), + chunkSize: parsePositiveInt(args["chunk-size"], "--chunk-size"), + maxParallelChunks: parsePositiveInt(args["max-parallel-chunks"], "--max-parallel-chunks"), + maxConcurrent: parsePositiveInt(args["max-concurrent"], "--max-concurrent"), + strictVariables: Boolean(args["strict-variables"]), + dryRun: Boolean(args["dry-run"]), + json: Boolean(args.json), + }); + return; + } case "progress": { const target = args.target as string | undefined; if (!target) { diff --git a/packages/cli/src/commands/lambda/render-batch.test.ts b/packages/cli/src/commands/lambda/render-batch.test.ts new file mode 100644 index 000000000..283ca0088 --- /dev/null +++ b/packages/cli/src/commands/lambda/render-batch.test.ts @@ -0,0 +1,139 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { parseBatchFile, runWithConcurrencyLimit } from "./render-batch.js"; + +let tmpDir: string; + +beforeEach(() => { + tmpDir = mkdtempSync(join(tmpdir(), "hf-render-batch-")); +}); + +afterEach(() => { + rmSync(tmpDir, { recursive: true, force: true }); +}); + +function writeBatch(content: string): string { + const p = join(tmpDir, "batch.jsonl"); + writeFileSync(p, content, "utf8"); + return p; +} + +describe("runWithConcurrencyLimit", () => { + it("preserves input order in the output array regardless of completion order", async () => { + // First input takes longest to resolve; output array still positional. + const delays = [40, 10, 20]; + const out = await runWithConcurrencyLimit(delays, 3, async (ms, i) => { + await new Promise((r) => setTimeout(r, ms)); + return `done-${i}`; + }); + expect(out).toEqual(["done-0", "done-1", "done-2"]); + }); + + it("caps simultaneous in-flight work to the limit", async () => { + let inFlight = 0; + let peak = 0; + const inputs = Array.from({ length: 12 }, (_, i) => i); + const worker = async (i: number): Promise => { + inFlight++; + peak = Math.max(peak, inFlight); + await new Promise((r) => setTimeout(r, 5)); + inFlight--; + return i; + }; + await runWithConcurrencyLimit(inputs, 3, worker); + expect(peak).toBe(3); + }); + + it("does not exceed the input length even when limit > inputs.length", async () => { + let inFlight = 0; + let peak = 0; + const inputs = [1, 2]; + await runWithConcurrencyLimit(inputs, 50, async (n) => { + inFlight++; + peak = Math.max(peak, inFlight); + await new Promise((r) => setTimeout(r, 2)); + inFlight--; + return n; + }); + // Only 2 inputs → only 2 concurrent workers, even with limit=50. + expect(peak).toBe(2); + }); + + it("rejects a limit < 1", async () => { + await expect(runWithConcurrencyLimit([1, 2], 0, async (n) => n)).rejects.toThrow( + /limit must be/, + ); + }); + + it("returns immediately for an empty input array", async () => { + const out = await runWithConcurrencyLimit([], 10, async (n: number) => n * 2); + expect(out).toEqual([]); + }); + + it("propagates the first worker rejection", async () => { + await expect( + runWithConcurrencyLimit([1, 2, 3], 2, async (n) => { + if (n === 2) throw new Error("boom"); + return n; + }), + ).rejects.toThrow(/boom/); + }); +}); + +describe("parseBatchFile", () => { + it("parses a JSONL file into ordered entries (line numbers preserve source order)", () => { + const path = writeBatch( + [ + '{"outputKey":"renders/alice.mp4","variables":{"name":"Alice"}}', + '{"outputKey":"renders/bob.mp4","variables":{"name":"Bob"},"executionName":"hf-bob-001"}', + ].join("\n") + "\n", + ); + const out = parseBatchFile(path); + expect(out).toHaveLength(2); + expect(out[0]?.entry.outputKey).toBe("renders/alice.mp4"); + expect(out[0]?.entry.variables).toEqual({ name: "Alice" }); + expect(out[0]?.lineNumber).toBe(1); + expect(out[1]?.entry.executionName).toBe("hf-bob-001"); + expect(out[1]?.lineNumber).toBe(2); + }); + + it("skips blank lines and preserves line numbers", () => { + const path = writeBatch( + ["", '{"outputKey":"renders/a.mp4"}', "", "", '{"outputKey":"renders/b.mp4"}'].join("\n") + + "\n", + ); + const out = parseBatchFile(path); + expect(out).toHaveLength(2); + expect(out[0]?.lineNumber).toBe(2); + expect(out[1]?.lineNumber).toBe(5); + }); + + // Helper: stub `process.exit` to throw a sentinel, run the parser, and + // verify it called exit(1). Dedupes the 3 error-path tests so each one + // is a single readable assertion. + function expectExitOne(content: string): void { + const exitSpy = vi.spyOn(process, "exit").mockImplementation(() => { + throw new Error("EXIT_CALLED"); + }); + try { + expect(() => parseBatchFile(writeBatch(content))).toThrow(/EXIT_CALLED/); + expect(exitSpy).toHaveBeenCalledWith(1); + } finally { + exitSpy.mockRestore(); + } + } + + it("exits with a clear message on malformed JSON, naming the offending line", () => { + expectExitOne(['{"outputKey":"renders/a.mp4"}', "{not json"].join("\n")); + }); + + it("rejects entries missing outputKey", () => { + expectExitOne('{"variables":{"name":"Alice"}}\n'); + }); + + it("rejects variables that's not a plain object", () => { + expectExitOne('{"outputKey":"renders/a.mp4","variables":[1,2,3]}\n'); + }); +}); diff --git a/packages/cli/src/commands/lambda/render-batch.ts b/packages/cli/src/commands/lambda/render-batch.ts new file mode 100644 index 000000000..b5b99a236 --- /dev/null +++ b/packages/cli/src/commands/lambda/render-batch.ts @@ -0,0 +1,446 @@ +/** + * `hyperframes lambda render-batch --batch ` — + * fan out N personalised renders of the same project, one per JSONL line. + * + * The headline ergonomic for automated template-rendering pipelines on + * Lambda: deploy the site once (or accept a `--site-id` to skip), then + * call `renderToLambda` for each batch entry with per-entry `variables` + * and `outputKey`. Concurrent Step Functions executions are capped at + * `--max-concurrent` (default 50) via a semaphore so a 10 000-entry batch + * file doesn't try to start 10 000 executions simultaneously and trip the + * AWS account's concurrent-execution limit. + * + * Per-entry results land in a manifest: one row per input line with the + * `executionArn` + status. `--dry-run` skips the AWS calls and prints the + * manifest with `status: "would-invoke"` for each entry so callers can + * lint their batch file without paying for any executions. + * + * JSONL format (one JSON object per line): + * + * {"outputKey": "renders/alice.mp4", "variables": {"name": "Alice"}} + * {"outputKey": "renders/bob.mp4", "variables": {"name": "Bob"}} + */ + +import { existsSync, readFileSync } from "node:fs"; +import { join, resolve as resolvePath } from "node:path"; +import type { + DistributedFormat, + SerializableDistributedRenderConfig, + SiteHandle, +} from "@hyperframes/aws-lambda/sdk"; +import { c } from "../../ui/colors.js"; +import { errorBox } from "../../ui/format.js"; +import { + loadProjectVariableSchema, + reportVariableIssues, + validateVariablesAgainstSchema, +} from "../../utils/variables.js"; +import { requireStack } from "./state.js"; + +// Dynamic-import the SDK so tsup keeps it out of the static-import head of +// the CLI bundle. See sites.ts loadSDK() for the full rationale. +async function loadSDK(): Promise { + return import("@hyperframes/aws-lambda/sdk"); +} + +/** Arguments accepted by `hyperframes lambda render-batch`. */ +export interface RenderBatchArgs { + projectDir: string; + stackName: string; + /** Path to the JSONL batch file. Each line is a {@link BatchEntry}. */ + batch: string; + /** + * Skip the project upload and re-use an existing pre-deployed site. The + * batch verb deploys the site once and reuses it across renders by + * default — this flag is for cases where the site was uploaded by a + * separate `sites create` step (CI / cross-machine). + */ + siteId?: string; + /** Composition config — fps/width/height/format required, rest optional. */ + fps: 24 | 30 | 60; + width: number; + height: number; + format: DistributedFormat; + codec?: "h264" | "h265"; + quality?: "draft" | "standard" | "high"; + chunkSize?: number; + maxParallelChunks?: number; + /** + * Maximum in-flight Step Functions starts at any moment. Caps fan-out + * so a 10 000-entry batch doesn't try to spawn 10 000 executions + * simultaneously. Defaults to 50. + * + * Distinct from `maxParallelChunks` (which caps chunks PER render). + * Lambda concurrent-execution limits live one level up at the AWS + * account level and this CLI cannot enforce those; the cap here is + * purely orchestrator-side. + */ + maxConcurrent?: number; + /** + * `--strict-variables` applies to every batch entry's pre-validation. + * Mismatches print as warnings; in strict mode the first failing entry + * aborts the run before any AWS call. + */ + strictVariables?: boolean; + /** + * Don't actually invoke `renderToLambda`. Print the manifest with + * `status: "would-invoke"` for every entry. Used to lint the batch + * file before committing to N billable executions. + */ + dryRun?: boolean; + /** Print machine-readable JSON instead of the human-friendly summary. */ + json: boolean; +} + +/** + * A single line in the JSONL batch file. Each entry produces one Step + * Functions execution with the supplied `variables` injected into the + * composition's `window.__hfVariables`. + */ +export interface BatchEntry { + /** + * Final output S3 key for this entry's render. Per-entry so the caller + * controls the output layout (e.g. `renders/users/alice.mp4`). Without + * an explicit value the SDK falls back to its + * `renders//output.` default, which makes a 100-row + * batch unreadable. + */ + outputKey: string; + /** + * Variable overrides for this entry. Merged over the composition's + * declared defaults inside the chunk worker (via + * `window.__hfVariables`). Optional — pass `{}` if a row needs the + * composition's defaults verbatim. + */ + variables?: Record; + /** + * Optional explicit Step Functions execution name. Defaults to + * `hf-render-` (generated by the SDK). Useful when the caller + * wants to correlate batch rows with downstream systems. + */ + executionName?: string; +} + +/** Single row of the manifest emitted by `render-batch`. */ +interface BatchManifestEntry { + /** 1-based index of the source JSONL line (after blank-line stripping). */ + inputLine: number; + /** Output S3 key the SDK was asked to produce. */ + outputKey: string; + /** + * SFN execution ARN, or `null` for entries that failed-to-start or were + * `--dry-run`-skipped. The latter case has `status: "would-invoke"`. + */ + executionArn: string | null; + /** Stable status discriminator the caller's manifest consumer can switch on. */ + status: "started" | "would-invoke" | "failed-to-start"; + /** Error message when `status === "failed-to-start"`. */ + error?: string; +} + +const DEFAULT_MAX_CONCURRENT = 50; + +/** + * Run the batch render. Throws on usage errors (bad CLI flags, missing + * project dir, malformed batch file); per-entry failures are captured in + * the manifest with `status: "failed-to-start"` rather than aborting the + * whole batch. + */ +// fallow-ignore-next-line complexity +export async function runRenderBatch(args: RenderBatchArgs): Promise { + const projectDir = resolvePath(args.projectDir); + const stack = requireStack(args.stackName); + + const batchPath = resolvePath(args.batch); + if (!existsSync(batchPath)) { + errorBox("Batch file not found", `No such file: ${batchPath}`); + process.exit(1); + } + const entries = parseBatchFile(batchPath); + if (entries.length === 0) { + errorBox("Empty batch", `${batchPath} contains zero entries (every line was blank).`); + process.exit(1); + } + + // Pre-validate every entry's variables against the composition's + // schema. Mismatches print as warnings; strict mode aborts before any + // AWS call. Schema is loaded once and reused across entries — a 10k-row + // batch with the per-entry parser would do 10k readFile + DOM parses. + // + // In strict mode we accumulate every failing entry first, then exit + // once with the full list. The naive "exit on first failure" pattern + // would force the caller to fix-one → re-run × N times on a batch with + // N broken rows. + const schema = loadProjectVariableSchema(join(projectDir, "index.html")); + const strict = args.strictVariables ?? false; + let hadStrictIssue = false; + for (const { entry, lineNumber } of entries) { + if (!entry.variables || Object.keys(entry.variables).length === 0) continue; + const issues = validateVariablesAgainstSchema(entry.variables, schema); + if (issues.length === 0) continue; + if (!args.json) { + console.log(""); + console.log(c.dim(`Batch entry on line ${lineNumber}:`)); + } + // Pass strict: false here so the helper just prints; we own the + // single exit-at-end below. + reportVariableIssues(issues, { strict: false, quiet: args.json }); + if (strict) hadStrictIssue = true; + } + if (hadStrictIssue) { + errorBox( + "Variable validation failed", + "Aborting batch due to variable issues in one or more entries (--strict-variables mode).", + ); + process.exit(1); + } + + const config: SerializableDistributedRenderConfig = { + fps: args.fps, + width: args.width, + height: args.height, + format: args.format, + codec: args.codec, + quality: args.quality, + chunkSize: args.chunkSize, + maxParallelChunks: args.maxParallelChunks, + runtimeCap: "lambda", + }; + + // Deploy the site once and reuse it across every entry. --site-id and + // --dry-run both skip the deploy via a synthesised handle. + let siteHandle: SiteHandle; + if (args.siteId) { + siteHandle = makePlaceholderSiteHandle(args.siteId, stack.bucketName); + } else if (args.dryRun) { + siteHandle = makePlaceholderSiteHandle("dry-run-site", stack.bucketName); + } else { + const { deploySite } = await loadSDK(); + siteHandle = await deploySite({ + projectDir, + bucketName: stack.bucketName, + region: stack.region, + }); + if (!args.json) { + console.log( + c.success( + siteHandle.uploaded + ? `Site uploaded once for the batch: ${siteHandle.siteId}` + : `Site already up to date (skipped upload): ${siteHandle.siteId}`, + ), + ); + console.log(); + } + } + + const maxConcurrent = args.maxConcurrent ?? DEFAULT_MAX_CONCURRENT; + // Skip the SDK import entirely on --dry-run; the startEntry closure + // short-circuits before touching `renderToLambda` so it can stay + // undefined. + const renderToLambda = args.dryRun ? undefined : (await loadSDK()).renderToLambda; + + const startEntry = async (item: { + entry: BatchEntry; + lineNumber: number; + }): Promise => { + const { entry, lineNumber } = item; + if (args.dryRun) { + return { + inputLine: lineNumber, + outputKey: entry.outputKey, + executionArn: null, + status: "would-invoke", + }; + } + if (!renderToLambda) { + // Unreachable: dryRun returns above; this branch is for the TS + // narrower since `renderToLambda` is undefined under dryRun. + throw new Error("[render-batch] renderToLambda is undefined outside --dry-run"); + } + try { + const handle = await renderToLambda({ + siteHandle, + bucketName: stack.bucketName, + stateMachineArn: stack.stateMachineArn, + region: stack.region, + config: { ...config, variables: entry.variables }, + executionName: entry.executionName, + outputKey: entry.outputKey, + }); + return { + inputLine: lineNumber, + outputKey: entry.outputKey, + executionArn: handle.executionArn, + status: "started", + }; + } catch (err) { + return { + inputLine: lineNumber, + outputKey: entry.outputKey, + executionArn: null, + status: "failed-to-start", + error: err instanceof Error ? err.message : String(err), + }; + } + }; + + const manifest = await runWithConcurrencyLimit(entries, maxConcurrent, startEntry); + + if (args.json) { + console.log(JSON.stringify(manifest, null, 2)); + return; + } + + const started = manifest.filter((m) => m.status === "started").length; + const failed = manifest.filter((m) => m.status === "failed-to-start").length; + const wouldInvoke = manifest.filter((m) => m.status === "would-invoke").length; + + if (args.dryRun) { + console.log(c.success(`Dry-run complete: ${wouldInvoke} entries would invoke.`)); + } else { + console.log(c.success(`Batch dispatched: ${started} started, ${failed} failed-to-start.`)); + } + console.log(); + for (const row of manifest) { + const tag = + row.status === "started" + ? c.success("✓") + : row.status === "would-invoke" + ? c.dim("·") + : c.error("✗"); + const detail = + row.status === "failed-to-start" + ? c.error(row.error ?? "unknown error") + : (row.executionArn ?? c.dim("(no execution)")); + console.log(` ${tag} line ${row.inputLine} ${c.dim(row.outputKey)} ${detail}`); + } + if (failed > 0) process.exitCode = 1; +} + +/** + * Synthesise a SiteHandle from a known siteId for paths that skip the + * `deploySite` upload (`--site-id` and `--dry-run`). The SDK reads only + * `siteId` + `projectS3Uri` when `uploaded: false`, so the byte / time + * fields are intentional placeholders. + */ +function makePlaceholderSiteHandle(siteId: string, bucketName: string): SiteHandle { + return { + siteId, + bucketName, + projectS3Uri: `s3://${bucketName}/sites/${siteId}/project.tar.gz`, + bytes: 0, + uploadedAt: "", + uploaded: false, + }; +} + +/** + * Read the JSONL batch file and return one parsed entry per non-blank + * line. Reads the whole file into memory — fine for typical batch sizes, + * an in-memory bound the caller can size around. Calls `errorBox` and + * `process.exit(1)` on the first malformed line so the caller doesn't + * have to sift through thousands of rows looking for the typo. + * + * Exported for unit-test coverage; production callers go through + * {@link runRenderBatch} which handles the file-not-found case. + */ +// fallow-ignore-next-line complexity +export function parseBatchFile(path: string): Array<{ entry: BatchEntry; lineNumber: number }> { + const raw = readFileSync(path, "utf8"); + const lines = raw.split(/\r?\n/); + const out: Array<{ entry: BatchEntry; lineNumber: number }> = []; + for (let i = 0; i < lines.length; i++) { + const line = lines[i]!.trim(); + if (line === "") continue; + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch (err) { + errorBox( + `Invalid JSON in batch file on line ${i + 1}`, + err instanceof Error ? err.message : String(err), + ); + process.exit(1); + } + if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) { + errorBox( + `Invalid batch entry on line ${i + 1}`, + 'Each line must be a JSON object with at least { "outputKey": "..." }.', + ); + process.exit(1); + } + const obj = parsed as Record; + const outputKey = obj.outputKey; + if (typeof outputKey !== "string" || outputKey.length === 0) { + errorBox( + `Missing outputKey on line ${i + 1}`, + 'Each batch entry needs a non-empty "outputKey" string (e.g. "renders/alice.mp4").', + ); + process.exit(1); + } + if (obj.variables !== undefined) { + if ( + obj.variables === null || + typeof obj.variables !== "object" || + Array.isArray(obj.variables) + ) { + errorBox( + `Invalid variables on line ${i + 1}`, + '"variables" must be a JSON object (or omitted).', + ); + process.exit(1); + } + } + if (obj.executionName !== undefined && typeof obj.executionName !== "string") { + errorBox( + `Invalid executionName on line ${i + 1}`, + '"executionName" must be a string (or omitted).', + ); + process.exit(1); + } + out.push({ + entry: { + outputKey, + variables: obj.variables as Record | undefined, + executionName: obj.executionName as string | undefined, + }, + lineNumber: i + 1, + }); + } + return out; +} + +/** + * Run `worker` against every input with at most `limit` concurrent + * invocations. Preserves input order in the returned array; each output + * is positional with its input. + * + * Implementation: index-cursor + N concurrent producers each picking the + * next index until the input is drained. Uses `Promise.all` and + * propagates the first worker rejection — partial-failure isolation is + * the caller's responsibility (in this file, `startEntry` wraps the + * per-entry render in try/catch so a single failure surfaces in the + * manifest rather than aborting the batch). + * + * Exported so unit tests can pin the concurrency cap independently of + * the AWS-dependent fan-out path. + */ +export async function runWithConcurrencyLimit( + inputs: readonly T[], + limit: number, + worker: (input: T, index: number) => Promise, +): Promise { + if (limit < 1) throw new Error(`runWithConcurrencyLimit: limit must be ≥ 1, got ${limit}`); + const results = new Array(inputs.length); + let cursor = 0; + const workerCount = Math.min(limit, inputs.length); + await Promise.all( + Array.from({ length: workerCount }, async () => { + while (cursor < inputs.length) { + const idx = cursor++; + results[idx] = await worker(inputs[idx]!, idx); + } + }), + ); + return results; +}