API генерации видео: JavaScript-пример очереди и ожидания в Node.js

Отделите создание ролика от ожидания нескольких заказов

Пусть пользователь поставил в работу три карточки товара. Их создание уже вернуло ID Gate, а ваш backend сохранил запросы и ключи. Теперь нужно обновлять состояние независимо от открытых вкладок. Задача обработчика — прочитать известные ID и получить результат каждого, а не создавать генерацию при каждом запуске программы.

Ниже используется Node.js 22 или новее со встроенным fetch. Дополнительный SDK для одного GET не требуется. Модуль делает ограниченное число проверок, сохраняет снимки и завершает проход. Если часть заказов ещё работает, следующий проход продолжит по тем же ID. Это понятная основа для вашей серверной очереди.

Платный POST /v1/media/generations остаётся отдельным шагом по базовому примеру. Он использует постоянный Idempotency-Key заказа. Этот материал не повторяет отправку сцены, а решает другую задачу: как пережить долгое ожидание, несколько параллельных работ и временный отказ сети.

Задайте границы одному проходу обработчика

У вас должно быть две отдельные границы: сколько заданий проверяется одновременно и сколько GET можно сделать для каждого за один проход. В примере по умолчанию работают два обработчика, а каждый ID проверяется не больше шести раз. Между проверками есть пауза, и один HTTP-вызов ограничен тайм-аутом.

Это настройки вашего приложения, а не заявленные лимиты Gate. Их выбирают по бюджету нагрузки, количеству заказов и ограничениям аккаунта. Нельзя обещать, что шесть проверок всегда означают ровно тридцать секунд: время самих HTTP-запросов тоже входит в ожидание. При медленной сети проход займёт дольше.

Если список большой, выбирайте небольшую порцию из своей базы и сохраняйте время следующей проверки. Не загружайте тысячи ID только ради одного запуска. Пример с локальным JSON удобен для приёмки; постоянная производственная очередь должна учитывать владение задачами и не отдавать один заказ нескольким независимым отправителям.

Подготовьте список существующих ID

Создайте jobs.json с массивом настоящих идентификаторов из HTTP 202. Учебные строки ниже показывают только формат; не выдавайте их за работающие задания. Ключ CELESTRIA_API_KEY задаётся в окружении серверного процесса. Он не включается в jobs.json, который может сохраняться вместе с заказами.

Модуль убирает повторяющиеся ID, чтобы одна локальная порция не опрашивала один и тот же заказ несколькими обработчиками. Это не глобальная блокировка между серверами. Если одновременно запустить два независимых процесса, они оба смогут делать GET; собственного распределённого владельца задания пример не создаёт.

Сохраняйте рядом внутренний номер заказа и исходное тело отдельно от списка для опроса. ID отвечает на вопрос, какую работу обновить, но не заменяет данные для восстановления неизвестной отправки. Если серверный ID ещё не получен, задача должна пройти отдельную процедуру восстановления с прежним ключом и телом.

[
  "media_generation_EXAMPLE_FIRST",
  "media_generation_EXAMPLE_SECOND"
]

Полный пример Node.js: безопасный проход по очереди

Сохраните код как poll-batch.mjs и запустите node poll-batch.mjs jobs.json из серверной папки с переменной ключа. Каждый успешный ответ сохраняется в файл с ID задания, права нового файла ограничены. После завершения программа печатает решение для каждого ID. Она никогда не выполняет POST генерации.

Функция pollExistingJobs экспортируется отдельно, чтобы её можно было подключить к вашему процессу и проверить на имитации fetch. В примере onSnapshot пишет файлы; в рабочем сервисе замените его коротким обновлением базы. Не держите транзакцию открытой между сетевыми проверками и паузами.

Временная сетевая ошибка, 429 или 5xx возвращают решение continue_later. Другие HTTP-ошибки требуют проверки авторизации или данных. Статусы completed, failed и cancelled считаются конечными; прочие продолжают ожидание в рамках заданной границы. Отдельно обрабатывайте неизвестные статусы при развитии своего интерфейса.

import { readFileSync, writeFileSync } from "node:fs";
import { resolve } from "node:path";
import { pathToFileURL } from "node:url";
import { setTimeout as sleep } from "node:timers/promises";

const terminal = new Set(["completed", "failed", "cancelled"]);

export async function pollExistingJobs(ids, apiKey, options = {}) {
  const fetchImpl = options.fetchImpl ?? fetch;
  const sleepImpl = options.sleepImpl ?? sleep;
  const save = options.onSnapshot ?? (() => {});
  const limit = options.concurrency ?? 2;
  const attempts = options.attempts ?? 6;
  const unique = [...new Set(ids)];
  if (!Number.isInteger(limit) || limit < 1 || limit > 4) throw new Error("Параллельность: 1–4");
  if (!Number.isInteger(attempts) || attempts < 1 || attempts > 12) throw new Error("Проверок за проход: 1–12");
  if (!unique.every((id) => typeof id === "string" && /^media_generation_[A-Za-z0-9_-]+$/.test(id))) {
    throw new Error("Нужен список настоящих ID общих заданий Gate");
  }
  let next = 0;
  const outcomes = new Array(unique.length);
  async function worker() {
    while (next < unique.length) {
      const index = next++;
      const id = unique[index];
      let outcome = { id, action: "continue_later" };
      for (let count = 0; count < attempts; count++) {
        let response;
        try {
          response = await fetchImpl(
            `https://api.celestriagate.com/v1/media/generations/${encodeURIComponent(id)}`,
            { headers: { Authorization: `Bearer ${apiKey}` }, signal: AbortSignal.timeout(30_000) },
          );
        } catch {
          outcome = { id, action: "continue_later", reason: "network_error" };
          break;
        }
        if (!response.ok) {
          outcome = { id, action: response.status === 429 || response.status >= 500 ? "continue_later" : "check_error", http_status: response.status };
          await response.arrayBuffer();
          break;
        }
        const job = await response.json();
        if (job.id !== id) throw new Error("ID ответа не совпал с заказом");
        await save(job);
        outcome = { id, action: terminal.has(job.status) ? "terminal" : "continue_later", status: job.status };
        if (terminal.has(job.status)) break;
        if (count + 1 < attempts) await sleepImpl(5_000);
      }
      outcomes[index] = outcome;
    }
  }
  await Promise.all(Array.from({ length: Math.min(limit, unique.length) }, worker));
  return outcomes;
}

async function main() {
  const apiKey = process.env.CELESTRIA_API_KEY;
  if (!apiKey) throw new Error("Нужен серверный CELESTRIA_API_KEY");
  const file = process.argv[2];
  if (!file) throw new Error("Запуск: node poll-batch.mjs jobs.json");
  const ids = JSON.parse(readFileSync(file, "utf8"));
  if (!Array.isArray(ids)) throw new Error("jobs.json должен содержать массив ID");
  const results = await pollExistingJobs(ids, apiKey, {
    onSnapshot(job) {
      writeFileSync(`${job.id}.json`, JSON.stringify(job, null, 2), { mode: 0o600 });
      console.log(job.id, job.status);
    },
  });
  console.log(JSON.stringify(results, null, 2));
}

if (process.argv[1] && import.meta.url === pathToFileURL(resolve(process.argv[1])).href) {
  main().catch((error) => { console.error(error.message); process.exitCode = 1; });
}

Превратите статусы API в понятные действия

queued, submitting и pending означают принятый или передаваемый заказ, in_progress — выполнение. awaiting_billing требует дождаться подтверждения окончательного расхода. completed позволяет показать файлы и сумму; failed и cancelled останавливают обычный цикл ожидания, но сохраняют ошибку и финансовые данные для разбора.

Решение continue_later в программе не является статусом Gate. Это действие вашего обработчика: закончить текущий проход и вернуться к прежнему ID позже. Оно не означает отмену, не требует нового ключа и не доказывает, что провайдер остановился. Подпишите такую разницу в коде и журнале, чтобы дежурный разработчик не перепутал уровни.

Если HTTP-ответ повреждён или ID не совпал с заказом, не перезаписывайте запись чужими данными. Пример прекращает выполнение при несоответствии ID. В промышленном обработчике фиксируйте диагностическое событие и продолжайте независимые заказы по своей политике. Исключение в локальной программе тоже не отменяет серверные генерации.

Сохраняйте статус и billing вместе с заказом

После каждого успешного GET onSnapshot получает целый объект. Благодаря этому закрытие программы не уничтожает последний подтверждённый статус. Не сохраняйте только ссылку: она может истечь. ID, модель, provider, error и billing нужны для повторного получения результата и объяснения расхода.

В своей базе обновляйте одну запись существующей попытки, а не добавляйте новый расход после каждого GET. billing.cost_rub относится к окончательной стоимости, reserved_rub — к текущему резерву. Awaiting_billing не следует превращать в завершённый финансовый отчёт по предварительной оценке.

Файлы снимков могут содержать приватные исходники или подписанные ссылки, поэтому папка обработки не должна раздаваться веб-сервером как публичная. API-ключ в снимке не хранится. В журнал печатайте ID и статус, а подробный JSON сохраняйте только там, где это нужно вашему приложению и разрешено его политикой данных.

Как пережить остановку процесса без повторной покупки

Остановка Node.js завершает локальный опрос, но не работу Gate. После перезапуска прочитайте ту же порцию ID или выберите незавершённые записи из базы. Получение completed повторно полезно для свежей ссылки и не должно снова добавлять итоговую стоимость в ваш отчёт.

Не связывайте сигнал завершения процесса с автоматическим cancel всех задач. Отмена доступна только на безопасных этапах и не гарантирует бесплатной остановки уже переданной провайдеру работы. Поведение «оператор перезапустил worker — клиент потерял видео» обычно противоречит намерению пользователя.

Если отдельная задача долго находится awaiting_billing, сохраните её состояние и вернитесь позже. Не создавайте повторную генерацию ради обхода ожидания расхода. Работа с неизвестным исходом оплачиваемого POST относится к слою отправки; этот GET-обработчик должен оставаться простым и не подменять её новой покупкой.

Локальная приёмка очереди до подключения настоящих ID

Проверьте функцию на имитации: два задания отвечают in_progress, одно затем completed, другое awaiting_billing. Убедитесь, что одновременно выполняется не больше разрешённого числа fetch и что каждое проверяется не больше attempts. Повтор ID во входном массиве должен остаться одним элементом обработки.

Отдельно верните 429, 500 и сетевое исключение: решения должны приглашать продолжить позже, без POST и смены ID. Для failed и cancelled проверьте сохранение snapshot. Измените ID ответа и убедитесь, что чужой результат не попадает в запись заказа. Такие тесты не требуют платной генерации.

После этого подключите несколько реальных существующих заказов и проверьте только чтение результатов. Для художественной оценки запускается отдельный согласованный тест нужной модели. Так пример JavaScript превращается в устойчивую часть приложения: очередь ждёт результат, а не расходует баланс случайными повторными отправками.

Частые вопросы

Почему здесь нет POST генерации?

Обработчик получает существующие ID и занимается только ожиданием. Платная отправка выделяется отдельно, где сохраняются тело и Idempotency-Key заказа.

Шесть проверок гарантируют завершение видео?

Нет. Это предел одного локального прохода. Неготовое задание сохраняется и проверяется позже по тому же ID.

Можно ли запускать несколько таких worker одновременно?

GET сам по себе не создаёт платных задач, но процессы могут дублировать нагрузку. Для производственной очереди нужен собственный механизм владения порциями заказов.

Закрытие Node.js отменяет видео?

Нет. Оно прекращает локальное ожидание. Генерация продолжает существовать на сервере; после перезапуска используйте сохранённый ID.

Источники

• Документация Celestria Gate
• Подтверждённые расходы fal
• Node.js: встроенный fetch
• Node.js: promise-таймеры