Generate CometAPI Videos with Node.js Polling and Webhooks

Last reviewed: 2026-08-03

Direct answer

A CometAPI video request is an asynchronous job. A reliable Node.js service submits the render, stores the returned identifier before acknowledging its own caller, and polls that identifier until the status is completed or failed. A webhook can reduce notification delay, but polling should remain the source of truth for missed callbacks and provider-specific callback bodies.

The CometAPI polling and webhook guide recommends storing task IDs, acknowledging callbacks quickly, handling duplicates idempotently, retaining raw callback data, and reconciling final state through polling. The Sora 2 create reference supplies the concrete request flow: submit POST /v1/videos, save the returned id, poll until completed or failed, and use the supported content route to download the result.

The implementation below includes persistent file-backed jobs and callback events, bounded HTTP work, exact-duplicate callback detection, restart recovery, a task-scoped download lock, and a recovery route for a known remote job whose first local write failed.

Save the code as video-worker.mjs, install Express, and run it with Node.js 18 or newer, as required by the refetched Express 5 API reference .

npm install express
node video-worker.mjs

Set COMETAPI_BASE_URL to the documented API base with a trailing slash. Load the complete authorization header value into COMETAPI_AUTH_HEADER through your deployment secret store. Do not place either value in source, a URL, or logs.

This tutorial server deliberately binds to 127.0.0.1. Do not expose it directly by changing the bind address. For deployment, place a trusted ingress on the same host or private network. Require an authenticated application identity with explicit render permission before forwarding /jobs or its recovery route. Before forwarding the webhook route, apply the callback-verification mechanism documented for the selected video adapter. If the adapter does not document such a mechanism, keep the route behind an equivalent private or allowlisted ingress boundary and retain polling as the recovery path.

import express from 'express';
import {createHash} from 'node:crypto';
import {mkdir, readFile, readdir, rename, writeFile} from 'node:fs/promises';
import {join} from 'node:path';

const baseText = process.env.COMETAPI_BASE_URL;
const authHeader = process.env.COMETAPI_AUTH_HEADER;
const dataRoot = process.env.VIDEO_JOB_DIR ?? './video-job-data';
const outputDir = process.env.VIDEO_OUTPUT_DIR ?? './video-output';
const listenHost = '127.0.0.1';
const listenPort = 3000;

if (!baseText || !baseText.endsWith('/')) {
  throw new Error('COMETAPI_BASE_URL must end with a slash');
}
if (!authHeader) {
  throw new Error('COMETAPI_AUTH_HEADER is required');
}

const apiBase = new URL(baseText);
const jobsDir = join(dataRoot, 'jobs');
const eventsDir = join(dataRoot, 'events');
const terminalStatuses = new Set(['completed', 'failed']);

await Promise.all([
  mkdir(jobsDir, {recursive: true, mode: 0o700}),
  mkdir(eventsDir, {recursive: true, mode: 0o700}),
  mkdir(outputDir, {recursive: true, mode: 0o700})
]);

class AcceptedButUntrackedError extends Error {
  constructor(videoId, cause) {
    super('A remote video was accepted but local tracking failed');
    this.name = 'AcceptedButUntrackedError';
    this.videoId = videoId;
    this.cause = cause;
  }
}

function safeId(value) {
  const text = String(value ?? '');
  const allowed = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_';

  if (!text || [...text].some(character => !allowed.includes(character))) {
    throw new Error('The video identifier contains unsupported characters');
  }

  return text;
}

function endpoint(path) {
  return new URL(path, apiBase);
}

function now() {
  return new Date().toISOString();
}

function wait(milliseconds) {
  return new Promise(resolve => setTimeout(resolve, milliseconds));
}

function logEvent(fields) {
  console.info(JSON.stringify({...fields, recordedAt: now()}));
}

async function readJsonFile(filePath) {
  try {
    return JSON.parse(await readFile(filePath, 'utf8'));
  } catch (error) {
    if (error.code === 'ENOENT') {
      return null;
    }
    throw error;
  }
}

async function atomicWrite(filePath, value) {
  const temporaryPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
  await writeFile(temporaryPath, JSON.stringify(value, null, 2), {
    flag: 'wx',
    mode: 0o600
  });
  await rename(temporaryPath, filePath);
}

class FileStore {
  constructor() {
    this.locks = new Map();
  }

  async locked(name, operation) {
    const previous = this.locks.get(name) ?? Promise.resolve();
    let release;
    const gate = new Promise(resolve => {
      release = resolve;
    });

    this.locks.set(name, gate);
    await previous;

    try {
      return await operation();
    } finally {
      release();
      if (this.locks.get(name) === gate) {
        this.locks.delete(name);
      }
    }
  }

  jobPath(videoId) {
    return join(jobsDir, `${safeId(videoId)}.json`);
  }

  eventPath(eventId) {
    return join(eventsDir, `${safeId(eventId)}.json`);
  }

  async getJob(videoId) {
    return readJsonFile(this.jobPath(videoId));
  }

  async createJob(job) {
    return this.locked(`job-${safeId(job.videoId)}`, async () => {
      if (await this.getJob(job.videoId)) {
        throw new Error(`Job ${job.videoId} already exists`);
      }
      await atomicWrite(this.jobPath(job.videoId), job);
      return job;
    });
  }

  async updateJob(videoId, update) {
    return this.locked(`job-${safeId(videoId)}`, async () => {
      const current = await this.getJob(videoId);
      const next = await update(current);
      await atomicWrite(this.jobPath(videoId), next);
      return next;
    });
  }

  async listJobs() {
    const names = (await readdir(jobsDir)).filter(name => name.endsWith('.json'));
    const jobs = await Promise.all(names.map(name => readJsonFile(join(jobsDir, name))));
    return jobs.filter(Boolean);
  }

  async getEvent(eventId) {
    return readJsonFile(this.eventPath(eventId));
  }

  async insertEvent(event) {
    return this.locked(`event-${safeId(event.eventId)}`, async () => {
      const current = await this.getEvent(event.eventId);
      if (current) {
        return {inserted: false, event: current};
      }
      await atomicWrite(this.eventPath(event.eventId), event);
      return {inserted: true, event};
    });
  }

  async updateEvent(eventId, update) {
    return this.locked(`event-${safeId(eventId)}`, async () => {
      const current = await this.getEvent(eventId);
      if (!current) {
        throw new Error(`Event ${eventId} does not exist`);
      }
      const next = await update(current);
      await atomicWrite(this.eventPath(eventId), next);
      return next;
    });
  }

  async listPendingEvents() {
    const names = (await readdir(eventsDir)).filter(name => name.endsWith('.json'));
    const events = await Promise.all(names.map(name => readJsonFile(join(eventsDir, name))));
    return events.filter(event => event && !event.processedAt);
  }
}

const store = new FileStore();

async function withDeadline(deadline, label, operation) {
  const remaining = deadline - Date.now();
  if (remaining <= 0) {
    throw new Error(`${label} deadline reached`);
  }

  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), remaining);

  try {
    return await operation(controller.signal);
  } catch (error) {
    if (controller.signal.aborted) {
      throw new Error(`${label} deadline reached`);
    }
    throw error;
  } finally {
    clearTimeout(timer);
  }
}

async function readJsonResponse(response) {
  const text = await response.text();
  let body = null;

  if (text) {
    try {
      body = JSON.parse(text);
    } catch {
      body = {rawPreview: text.slice(0, 300)};
    }
  }

  if (!response.ok) {
    const error = new Error(`CometAPI request returned ${response.status}`);
    error.status = response.status;
    throw error;
  }

  return body;
}

async function requestJson(path, init, deadline, label) {
  return withDeadline(deadline, label, async signal => {
    const response = await fetch(endpoint(path), {...init, signal});
    const body = await readJsonResponse(response);
    return {body, httpStatus: response.status};
  });
}

function baseJob(videoId) {
  const timestamp = now();
  return {
    videoId,
    model: null,
    status: 'unknown',
    progress: null,
    createdAt: timestamp,
    updatedAt: timestamp,
    lastPolledAt: null,
    lastWebhookAt: null,
    lastWebhookEventId: null,
    webhookStatus: null,
    webhookProgress: null,
    webhookVideoUrlPresent: false,
    downloadedAt: null,
    outputPath: null,
    bytes: null,
    orphaned: true
  };
}

async function submitVideo(prompt) {
  const form = new FormData();
  form.set('model', 'sora-2');
  form.set('prompt', prompt);
  form.set('seconds', '4');
  form.set('size', '1280x720');

  const startedAt = Date.now();
  const {body, httpStatus} = await requestJson(
    'videos',
    {
      method: 'POST',
      headers: {Authorization: authHeader},
      body: form
    },
    Date.now() + 30000,
    'create request'
  );

  const videoId = safeId(body?.id ?? body?.task_id);
  const timestamp = now();
  const job = {
    ...baseJob(videoId),
    model: body?.model ?? null,
    status: body?.status ?? 'unknown',
    progress: Number.isFinite(body?.progress) ? body.progress : null,
    createdAt: timestamp,
    updatedAt: timestamp,
    orphaned: false
  };

  try {
    await store.createJob(job);
  } catch (error) {
    logEvent({
      event: 'video_persist_failed',
      videoId,
      httpStatus,
      errorName: error.name
    });
    throw new AcceptedButUntrackedError(videoId, error);
  }

  logEvent({
    event: 'video_created',
    videoId,
    model: job.model,
    status: job.status,
    httpStatus,
    latencyMs: Date.now() - startedAt
  });

  return job;
}

async function applyPolledState(videoId, body) {
  return store.updateJob(videoId, current => {
    const job = current ?? baseJob(videoId);
    const timestamp = now();

    return {
      ...job,
      model: body?.model ?? job.model,
      status: body?.status ?? job.status,
      progress: Number.isFinite(body?.progress) ? body.progress : job.progress,
      lastPolledAt: timestamp,
      updatedAt: timestamp
    };
  });
}

async function pollOnce(videoId, deadline) {
  const startedAt = Date.now();
  const {body, httpStatus} = await requestJson(
    `videos/${encodeURIComponent(safeId(videoId))}`,
    {headers: {Authorization: authHeader}},
    deadline,
    'poll request'
  );

  const job = await applyPolledState(videoId, body);
  logEvent({
    event: 'video_status',
    videoId,
    model: job.model,
    status: job.status,
    progress: job.progress,
    httpStatus,
    latencyMs: Date.now() - startedAt
  });

  return job;
}

async function waitForNextPoll(intervalMs, deadline) {
  const remaining = deadline - Date.now();
  if (remaining <= 0) {
    throw new Error('polling deadline reached');
  }

  await wait(Math.min(intervalMs, remaining));

  if (Date.now() >= deadline) {
    throw new Error('polling deadline reached');
  }
}

async function pollVideo(videoId, options = {}) {
  const intervalMs = options.intervalMs ?? 5000;
  const timeoutMs = options.timeoutMs ?? 15 * 60 * 1000;
  const deadline = Date.now() + timeoutMs;

  while (true) {
    const job = await pollOnce(videoId, deadline);
    if (terminalStatuses.has(job.status)) {
      return job;
    }
    await waitForNextPoll(intervalMs, deadline);
  }
}

async function downloadIfNeeded(videoId) {
  const id = safeId(videoId);

  return store.locked(`download-${id}`, async () => {
    const current = await store.getJob(id);
    if (!current || current.downloadedAt || current.status !== 'completed') {
      return current;
    }

    const bytes = await withDeadline(
      Date.now() + 2 * 60 * 1000,
      'content download',
      async signal => {
        const response = await fetch(
          endpoint(`videos/${encodeURIComponent(id)}/content`),
          {headers: {Authorization: authHeader}, signal}
        );

        if (!response.ok) {
          const error = new Error(`Video content request returned ${response.status}`);
          error.status = response.status;
          throw error;
        }

        return Buffer.from(await response.arrayBuffer());
      }
    );

    const outputPath = join(outputDir, `${id}.mp4`);
    const temporaryPath = `${outputPath}.${process.pid}.${Date.now()}.tmp`;
    await writeFile(temporaryPath, bytes, {mode: 0o600});
    await rename(temporaryPath, outputPath);

    const saved = await store.updateJob(id, job => ({
      ...(job ?? baseJob(id)),
      downloadedAt: now(),
      updatedAt: now(),
      outputPath,
      bytes: bytes.length
    }));

    logEvent({
      event: 'video_saved',
      videoId: id,
      outputPath,
      bytes: bytes.length
    });

    return saved;
  });
}

async function finishTerminalJob(job) {
  if (job.status === 'failed') {
    logEvent({event: 'video_failed', videoId: job.videoId, status: job.status});
    return job;
  }

  if (job.status === 'completed') {
    return downloadIfNeeded(job.videoId);
  }

  return job;
}

const activePollers = new Map();

function ensurePolling(videoId) {
  const id = safeId(videoId);
  if (activePollers.has(id)) {
    return activePollers.get(id);
  }

  const task = pollVideo(id).then(finishTerminalJob);
  activePollers.set(id, task);
  void task.then(
    () => activePollers.delete(id),
    () => activePollers.delete(id)
  );
  return task;
}

function scheduleTracking(job) {
  setImmediate(() => {
    const task = terminalStatuses.has(job.status)
      ? finishTerminalJob(job)
      : ensurePolling(job.videoId);

    void task.catch(error => {
      logEvent({
        event: 'video_tracking_error',
        videoId: job.videoId,
        errorName: error.name,
        httpStatus: error.status ?? null
      });
    });
  });
}

function normalizeWebhook(payload) {
  const taskId = typeof payload?.task_id === 'string'
    ? payload.task_id
    : typeof payload?.id === 'string'
      ? payload.id
      : null;

  return {
    taskId: taskId ? safeId(taskId) : null,
    status: typeof payload?.status === 'string' ? payload.status : 'unknown',
    progress: Number.isFinite(payload?.progress) ? payload.progress : null,
    videoUrlPresent: Boolean(payload?.video_url ?? payload?.result?.video_url)
  };
}

async function recordWebhook(request) {
  const normalized = normalizeWebhook(request.body);
  if (!normalized.taskId) {
    const error = new Error('The callback body has no task identifier');
    error.status = 400;
    throw error;
  }

  const rawBody = request.rawBody ?? JSON.stringify(request.body ?? {});
  const eventId = createHash('sha256').update(rawBody).digest('hex');
  const event = {
    eventId,
    taskId: normalized.taskId,
    receivedAt: now(),
    normalized,
    rawBody,
    processedAt: null,
    attempts: 0,
    lastAttemptAt: null,
    lastError: null
  };

  return store.insertEvent(event);
}

async function reconcileEvent(event) {
  await store.updateEvent(event.eventId, current => ({
    ...current,
    attempts: current.attempts + 1,
    lastAttemptAt: now(),
    lastError: null
  }));

  await store.updateJob(event.taskId, current => {
    const job = current ?? baseJob(event.taskId);
    return {
      ...job,
      lastWebhookAt: event.receivedAt,
      lastWebhookEventId: event.eventId,
      webhookStatus: event.normalized.status,
      webhookProgress: event.normalized.progress,
      webhookVideoUrlPresent: event.normalized.videoUrlPresent,
      updatedAt: now()
    };
  });

  try {
    const polledJob = await pollOnce(event.taskId, Date.now() + 30000);
    await store.updateEvent(event.eventId, current => ({
      ...current,
      processedAt: now(),
      lastError: null
    }));

    logEvent({
      event: 'video_webhook_reconciled',
      eventId: event.eventId,
      videoId: event.taskId,
      webhookStatus: event.normalized.status,
      polledStatus: polledJob.status,
      videoUrlPresent: event.normalized.videoUrlPresent
    });

    if (terminalStatuses.has(polledJob.status)) {
      await finishTerminalJob(polledJob);
    } else {
      void ensurePolling(event.taskId).catch(error => {
        logEvent({
          event: 'video_polling_error',
          videoId: event.taskId,
          errorName: error.name,
          httpStatus: error.status ?? null
        });
      });
    }
  } catch (error) {
    await store.updateEvent(event.eventId, current => ({
      ...current,
      lastError: {
        name: error.name,
        httpStatus: error.status ?? null,
        recordedAt: now()
      }
    }));
    throw error;
  }
}

const activeEvents = new Map();

function ensureEventReconciliation(event) {
  if (activeEvents.has(event.eventId)) {
    return activeEvents.get(event.eventId);
  }

  const task = reconcileEvent(event);
  activeEvents.set(event.eventId, task);
  void task.then(
    () => activeEvents.delete(event.eventId),
    () => activeEvents.delete(event.eventId)
  );
  return task;
}

async function recoverWork() {
  for (const event of await store.listPendingEvents()) {
    try {
      await ensureEventReconciliation(event);
    } catch (error) {
      logEvent({
        event: 'video_webhook_recovery_error',
        eventId: event.eventId,
        videoId: event.taskId,
        errorName: error.name,
        httpStatus: error.status ?? null
      });
    }
  }

  for (const job of await store.listJobs()) {
    if (job.status === 'completed' && !job.downloadedAt) {
      try {
        await finishTerminalJob(job);
      } catch (error) {
        logEvent({
          event: 'video_download_recovery_error',
          videoId: job.videoId,
          errorName: error.name,
          httpStatus: error.status ?? null
        });
      }
    } else if (!terminalStatuses.has(job.status)) {
      void ensurePolling(job.videoId).catch(error => {
        logEvent({
          event: 'video_polling_recovery_error',
          videoId: job.videoId,
          errorName: error.name,
          httpStatus: error.status ?? null
        });
      });
    }
  }
}

const app = express();
app.use(express.json({
  limit: '2mb',
  verify(request, response, buffer) {
    request.rawBody = buffer.toString('utf8');
  }
}));

app.post('/jobs', async (request, response, next) => {
  try {
    const prompt = request.body?.prompt;
    if (typeof prompt !== 'string' || !prompt.trim() || prompt.length > 2000) {
      response.status(400).json({error: 'invalid_prompt'});
      return;
    }

    const job = await submitVideo(prompt.trim());
    response.status(202).json({
      videoId: job.videoId,
      remoteStatus: job.status,
      trackingState: 'persisted'
    });
    scheduleTracking(job);
  } catch (error) {
    if (error instanceof AcceptedButUntrackedError) {
      response.status(202).json({
        videoId: error.videoId,
        remoteState: 'accepted',
        trackingState: 'persistence_failed',
        automaticResubmissionSafe: false,
        recoveryRequired: true
      });
      return;
    }
    next(error);
  }
});

app.post('/jobs/:videoId/recover', async (request, response, next) => {
  try {
    const videoId = safeId(request.params.videoId);
    const job = await pollOnce(videoId, Date.now() + 30000);
    response.status(202).json({
      videoId,
      remoteStatus: job.status,
      trackingState: 'recovered'
    });
    scheduleTracking(job);
  } catch (error) {
    next(error);
  }
});

app.post('/cometapi/video-webhook', async (request, response, next) => {
  try {
    const recorded = await recordWebhook(request);
    response.status(200).json({
      received: true,
      duplicate: !recorded.inserted
    });

    if (!recorded.event.processedAt) {
      setImmediate(() => {
        void ensureEventReconciliation(recorded.event).catch(error => {
          logEvent({
            event: 'video_webhook_error',
            eventId: recorded.event.eventId,
            videoId: recorded.event.taskId,
            errorName: error.name,
            httpStatus: error.status ?? null
          });
        });
      });
    }
  } catch (error) {
    next(error);
  }
});

app.use((error, request, response, next) => {
  logEvent({
    event: 'http_error',
    errorName: error.name,
    httpStatus: error.status ?? 500
  });

  if (!response.headersSent) {
    response.status(error.status ?? 500).json({error: 'request_failed'});
  }
});

app.listen(listenPort, listenHost, () => {
  logEvent({event: 'video_worker_started', host: listenHost, port: listenPort});
  void recoverWork().catch(error => {
    logEvent({event: 'video_recovery_error', errorName: error.name});
  });
});

The file store uses atomic replacement and process-local locks. It is suitable for one tutorial worker on a persistent local volume and survives a normal process restart. Multiple replicas require transactional shared storage with unique job and event identifiers plus an atomic download claim.

The callback handler stores the raw body in a restricted event file before returning 200. A digest of that exact body becomes its event ID, so an exact repeated delivery finds the existing event. Reconciliation stores normalized callback fields, polls CometAPI, and marks the event processed only after that poll succeeds. Startup recovery retries any event that remained unprocessed.

The canonical job status is changed only by polling. Callback state is retained separately as webhookStatus, preventing a provider-specific callback from silently becoming authoritative. The download function holds a download-{videoId} lock from its downloadedAt check through the content request, file rename, and job update. An active poller and webhook reconciliation therefore cannot start concurrent downloads in this single-process implementation.

Logs contain operational fields only: event name, event ID, video ID, model, canonical status, callback status, numeric progress, HTTP status, latency, error class, output path, byte count, host, port, and timestamp. They exclude prompts, authorization values, raw callback bodies, response previews, and generated media URLs.

Happy path

An authorized application caller reaches the loopback worker through the trusted ingress and submits a short prompt.
CometAPI accepts the job and returns a video ID with a queued status.
The worker writes that ID and initial state before returning local 202 with trackingState: 'persisted'.
A bounded poller stores canonical state updates.
If a verified callback arrives, the worker stores its raw body, acknowledges it, normalizes its fields, and polls the same ID for reconciliation.
When polling reports completed, the task-scoped lock permits one content download and records the output path and byte count.
An exact duplicate callback reuses its event record, while any concurrent terminal path observes the same download lock and durable completion record.

Error path

If CometAPI accepts a render but the initial job write fails, the route returns non-error 202 with the known video ID, trackingState: 'persistence_failed', and automaticResubmissionSafe: false.
The caller preserves that ID instead of retrying create. After storage is repaired, an operator uses the recovery route, which polls the existing ID and never submits another render.
Every create, poll, reconciliation, and content request has an abort deadline covering the fetch and response-body read.
Poll sleeps are capped by the remaining global deadline.
Failed callback reconciliation remains unprocessed on disk and is retried during startup recovery.
A missed callback is covered by persisted polling.
A failed canonical status stops downloading and emits a sanitized failure event.
An ingress that cannot authenticate render callers or verify callbacks must not forward those routes from an untrusted network.

Who this is for

This tutorial is for Node.js developers building a service, queue worker, or internal tool around asynchronous CometAPI video generation. It is particularly useful when a render can outlive its initiating HTTP request, a worker may restart during rendering, or a provider callback can arrive more than once.

It is not a prompt-writing or video-editing guide. The goal is job orchestration: create, persist, observe, reconcile, recover, and download. The pattern can be adapted to another CometAPI video model, but that model’s current documentation must remain authoritative for supported request fields and callback behavior.

If your application already handles image output, the related Node.js image saving tutorial is a useful adjacent implementation. Video creation adds asynchronous state, callback reconciliation, and terminal download coordination.

Key takeaways

Treat the create response as an accepted job, not a finished video.
Persist the returned identifier before reporting normal tracked acceptance.
If persistence fails after remote acceptance, return the identifier and explicitly prohibit automatic resubmission.
Poll until completed or failed, even when webhooks are enabled.
Record a callback before returning a quick 2xx response.
Keep callback state separate from canonical polled state.
Replay unprocessed events and nonterminal jobs after restart.
Place the worker behind an ingress that authorizes render callers and applies documented callback verification.
Hold a task-scoped lock across the download check, request, file move, and completion update.
Bound both HTTP work and polling sleeps with the overall deadline.
Treat video_url as optional and use a supported content route when available.
Log identifiers and transitions, not prompts, raw callback data, authorization values, or media URLs.

The CometAPI key boundary guide covers the repository and deployment boundary for the authorization configuration used by this worker.

Sources checked

The article uses four refetched public sources:

Use polling and webhooks for video generation explains task IDs, polling as the baseline, provider-specific callbacks, quick acknowledgments, duplicate handling, raw-body retention, and optional result URLs.
Create a Sora 2 video defines the multipart create request, supported duration and size values, queued response, retrieval flow, terminal statuses, and content download.
HTTP in Node.js documents the Node HTTP client and server primitives, including request, response, timeout, abort, and completion behavior.
The Express 5.x API reference documents Express application setup, routing, middleware, and its Node.js runtime baseline.

The CometAPI pages are authoritative for request fields, task status, callback variability, and result handling. The Node.js and Express documentation support the surrounding service. Poll intervals, storage technology, ingress policy, retention, and deployment topology remain application decisions.

Contract details to verify

Before using a model in production, verify its current CometAPI page instead of assuming every video adapter shares one request or callback shape.

First, confirm the model identifier. The Sora reference documents sora-2 and sora-2-pro, while the general guide uses another model in its minimal example. Store the returned model beside the video ID so operators can identify the adapter used for each render.

Next, validate duration and size. The Sora reference lists four, eight, twelve, sixteen, and twenty seconds. Standard Sora output supports 1280x720 and 720x1280; the larger documented dimensions apply to the Pro model. The size field uses exact width-by-height form rather than ratio labels or shorthand resolution tokens.

Then verify result retrieval. A completed response can contain video_url, but the general guide says that field is optional. Depending on the adapter, the application may need model-specific result fields or the supported /v1/videos/{id}/content route. This worker downloads through the documented Sora content path and does not treat an optional URL as durable storage.

Finally, verify callback support and verification. CometAPI does not define one universal callback payload for every video model. Do not add a callback field unless the selected endpoint documents it, and do not invent a callback-verification rule absent from that adapter’s documentation. Retain the raw payload, normalize only fields your application understands, and confirm final state by polling.

Failure modes

The create call succeeds but local storage fails. The remote job may already consume work. This worker returns 202, the known ID, and an explicit unsafe-resubmission flag instead of a generic 5xx. Preserve the ID, repair storage, and recover that job by polling it through the recovery route.

The service is exposed without an ingress boundary. An untrusted caller could create billable renders, fill callback storage, or cause polling of arbitrary identifiers. Keep the worker bound to loopback. Authorize application callers, rate-limit creation, enforce body limits, and verify callbacks using the selected adapter’s documented mechanism before forwarding them.

A poller and callback observe completion together. Both can enter the terminal path, but the task-scoped download lock serializes the check and download. The second path observes downloadedAt after the first finishes. Multiple replicas need a shared atomic claim because process-local locks cannot coordinate across hosts.

A callback never arrives. The persisted polling loop continues independently. Startup recovery scans nonterminal jobs, so a restart does not turn a missed callback into an abandoned render.

The same callback arrives twice. The worker hashes the exact raw body and atomically stores one local event file for that digest. An exact repeat receives a successful duplicate acknowledgment. Shared transactional uniqueness is required when multiple replicas accept callbacks.

The callback body has unfamiliar nesting. The normalizer accepts task_id or id, a string status, numeric progress, and the presence of a result URL. The complete raw body remains in restricted storage for adapter-specific analysis but never enters logs.

The process exits after acknowledging a callback. The raw event is already stored and remains unprocessed until reconciliation succeeds. Startup recovery finds it and retries the poll.

A poll request stalls. The global deadline drives an abort controller around both the fetch and response-body parsing. One stalled request cannot run beyond the worker’s advertised polling deadline.

A callback reports completion but polling disagrees. Callback state remains in webhookStatus, while polling owns canonical status. Downloading begins only after polling reports completed.

The final media URL expires. The Sora reference describes finished download URLs as temporary. Download completed content promptly and store the bytes under application control.

FAQ

Should polling or webhooks be primary?

Polling should remain the recovery path and source of truth. Webhooks can lower notification latency, but their payloads vary by model and deliveries can be missed or repeated.

Why does a persistence failure return 202?

CometAPI has already accepted the remote render. Returning a generic server error can invite automatic create retries and duplicate work. The special 202 response identifies the existing video, marks local tracking as failed, and says automatic resubmission is unsafe.

How should the routes be protected?

Keep the Node.js listener on loopback. A trusted ingress should authenticate and authorize application callers before forwarding render or recovery requests. It should verify callbacks using the selected adapter’s documented mechanism before forwarding them; otherwise use a private or allowlisted boundary and rely on polling.

Does this example survive a restart?

Yes, for one worker process using a persistent local volume. Startup recovery replays pending callback events, resumes nonterminal polling, and downloads completed jobs without a download record. Multiple replicas require shared transactional storage.

How is callback idempotency enforced?

An event ID is derived from the exact raw body. Atomic insertion creates one local record, processedAt prevents completed reconciliation from running again, and the active-event map prevents concurrent processing within one process.

What should the worker log?

Log event and video identifiers, model, canonical and callback status, numeric progress, HTTP status, latency, error class, output path, byte count, host, port, and timestamp. Exclude prompts, authorization values, raw callbacks, response previews, and media URLs.

Is `video_url` always present on completion?

No. The guide says it can be absent. Use documented model-specific fields or the content route when available.

Which first render is a useful smoke test?

Use a four-second, 1280x720 Sora request with a short, non-sensitive prompt. Confirm persistence before local acknowledgment, terminal polling, one saved content file, one event for an exact duplicate callback, and restart recovery for an interrupted job.

Reader next step

Run the worker on a persistent local directory and keep its listener on loopback. Submit one small render through the protected ingress, confirm trackingState: 'persisted', and verify that the job file appears before polling starts. Stop the process while the job is nonterminal, restart it, and confirm recovery continues with the same video ID.

Next, send the same sanitized callback test body twice through the trusted callback boundary. Confirm that only one event is stored and that reconciliation marks it processed only after a successful poll. Arrange for polling and callback reconciliation to observe a completed test job concurrently, then verify that the download lock produces one content request and one final file.

Finally, simulate local persistence failure after a remote acceptance. Confirm that the caller receives the video ID, a non-error 202, and automaticResubmissionSafe: false; repair storage and recover that exact ID without another create call. Before enabling real callbacks, recheck the CometAPI polling and webhook guide and the selected adapter documentation. For Sora request fields, verify the Sora 2 create reference .