84 lines
3.5 KiB
TypeScript
84 lines
3.5 KiB
TypeScript
/**
|
|
* Produce worker entrypoint: the compose/mix job runner as its own process.
|
|
*
|
|
* Kept separate from src/main.ts (the live-channel supervisor) so a produce
|
|
* worker runs where its lane needs it without dragging the supervisor's
|
|
* dependencies (Redis, the fence, signed control) along: a remote-lane box needs
|
|
* only Postgres + object storage + ffmpeg.
|
|
*
|
|
* MEDIA_WORKER_LANE=remote → claim only `input_json->>'lane' = 'remote'` rows
|
|
* MEDIA_WORKER_LANE unset / 'local' (the default worker) → claim lane local or NULL
|
|
*
|
|
* Everything else is env parity with the API's storage (S3_ENDPOINT/S3_ACCESS_KEY/
|
|
* S3_SECRET_KEY/S3_BUCKET/S3_KEK) and DATABASE_URL for the job tables.
|
|
*/
|
|
import { appendFileSync, mkdirSync } from "node:fs";
|
|
import pino from "pino";
|
|
import { Pool as PgPool } from "pg";
|
|
import { laneFromEnv, resolveProduceFont } from "./env";
|
|
import { PostgresProduceJobStore } from "./producer/pg-store";
|
|
import { ProduceRunner } from "./producer/runner";
|
|
import { WorkerS3Storage, workerStorageConfigFromEnv } from "./producer/s3-storage";
|
|
|
|
const logger = pino({ level: process.env.LOG_LEVEL ?? "info" });
|
|
|
|
export async function startProduceWorker(): Promise<{ runner: ProduceRunner; pool: PgPool }> {
|
|
const lane = laneFromEnv(process.env.MEDIA_WORKER_LANE);
|
|
const fontFile = resolveProduceFont(process.env.MEDIA_PRODUCE_FONT);
|
|
const scratchRoot = process.env.MEDIA_SCRATCH_DIR ?? "/var/tmp/media-worker/produce";
|
|
mkdirSync(scratchRoot, { recursive: true });
|
|
const pool = new PgPool({ connectionString: process.env.DATABASE_URL });
|
|
const runner = new ProduceRunner({
|
|
jobs: new PostgresProduceJobStore(pool),
|
|
storage: new WorkerS3Storage(workerStorageConfigFromEnv()),
|
|
fontFile,
|
|
scratchRoot,
|
|
lane,
|
|
redactor: {
|
|
redact: (line) => {
|
|
if (process.env.MEDIA_DEBUG_STDERR === "1") {
|
|
appendFileSync("/tmp/ffmpeg-debug.log", line + "\n");
|
|
return line;
|
|
}
|
|
return "[produce stderr redacted]";
|
|
},
|
|
},
|
|
onStderr: (message) => logger.warn({ message }, "ffmpeg stderr"),
|
|
onEvent: (event) => {
|
|
switch (event.type) {
|
|
case "idle": return;
|
|
case "claimed": return logger.info({ jobId: event.jobId, lane }, "produce job claimed");
|
|
case "complete": return logger.info({ jobId: event.jobId, detail: event.detail }, "produce job complete");
|
|
case "failed": return logger.warn({ jobId: event.jobId, detail: event.detail }, "produce job failed");
|
|
case "store-error": return logger.warn({ detail: event.detail }, "produce store error");
|
|
}
|
|
},
|
|
});
|
|
logger.info({ lane, fontFile }, "produce worker starting");
|
|
runner.start();
|
|
return { runner, pool };
|
|
}
|
|
|
|
/**
|
|
* Entry when run directly (`tsx src/produce.ts` / `node dist/produce.js`). No
|
|
* top-level await: this package compiles CommonJS, so the start promise is
|
|
* driven by main() and a boot failure still sets the process exit code.
|
|
*/
|
|
async function main(): Promise<void> {
|
|
const entry = await startProduceWorker();
|
|
for (const signal of ["SIGINT", "SIGTERM"] as const) {
|
|
process.on(signal, () => {
|
|
logger.info({ signal }, "produce worker stopping");
|
|
// Stop polling first; an in-flight job finishes (its DB row is its lease),
|
|
// then the pool closes and the process exits cleanly.
|
|
void entry.runner.stop().finally(() => {
|
|
void entry.pool.end().finally(() => process.exit(0));
|
|
});
|
|
});
|
|
}
|
|
}
|
|
|
|
void main().catch((error: unknown) => {
|
|
logger.error({ err: String(error) }, "produce worker startup failed");
|
|
process.exitCode = 1;
|
|
});
|