The Backend Stack for an AI Video App

Calling a video model is one HTTP request.

Building a product around that request is not.

Once a real user can upload an image, click Generate, close the tab, come back two minutes later, retry a failed job, and expect you not to charge them repeatedly for the same action, you have a backend problem rather than a model-integration problem.

Short answer: An AI video app backend needs durable job state, media storage, asynchronous prediction execution, a reliable completion path, explicit retry rules, and per-job cost tracking. For direct each::api predictions, this example polls the prediction resource; when the job is wrapped in each::workflows, workflow webhooks provide the supported callback path. Add your own queue only when the application needs admission control, preprocessing, scheduling, or backpressure; the execution platform's queue solves a different problem.

You might also need your own job queue. But if the generation platform already executes predictions asynchronously, that queue has a different job: it controls when your application submits more work, not when the model starts running.

Canberk Sinangil, Co-founder & CTO of each::labs, puts the broader distinction this way:

“A catalog gets you into development. Reliability gets you into production.”

Model access gets the demo running. The rest of this article is the infrastructure that keeps the feature working afterward.

The example uses each::api for prediction execution and each::storage for input media, but most of the architecture applies to any long-running generative-media API.

A production AI video job has a lifecycle, not a request/response

A conventional API handler often looks like this:

request
   ↓
do work
   ↓
return result

That is fine when “do work” takes 80 milliseconds.

Video generation can run for tens of seconds or minutes. The browser's original HTTP request should not be responsible for keeping that work alive. each::api creates predictions asynchronously and returns a predictionID that the application can persist and retrieve later.

The product lifecycle looks more like this:

User
  │
  │ POST /video-jobs
  ▼
Application API ─────────────────────► Job database
  │                                      ▲
  │                                      │
  ├──── input media ───► each::storage   │
  │                                      │
  └──── prediction ────► each::api       │
                            │             │
                            ▼             │
                     async execution      │
                            │             │
                            ▼             │
                  polling reconciliation ─┘
                            │
                            └─────────────► Job database

User ◄──────── GET /video-jobs/{id}

Two boundaries matter here.

First, your application job is not the model prediction. Your product needs an identity and state for the user's task before a prediction may even exist.

Second, submission is not execution. Once the prediction has been accepted, the web request can finish while generation continues elsewhere.

The split between your backend and each::labs is easier to reason about when the responsibilities are explicit:

Concern	Your application	each::labs
Product-facing job ID	Own and persist	—
Product status/state	Own and normalize	Exposes prediction state
Input media plumbing	Integrate/upload	each::storage
Prediction execution	Submit and reference	each::api
Application admission queue	Own, if needed	—
Execution queue	—	Managed prediction execution
Terminal-state discovery	Poll and update state	Prediction resource
Retry/product policy	Own	Exposes execution/error behavior
Actual execution cost	Store/use on job	metrics.cost

Everything that follows comes from those boundaries.

Start with a durable application job

Create the application job before you create the model prediction.

A minimal record might look like this:

video_jobs

id
user_id
status
prediction_id
input_url
output_url
attempt_count
cost_usd
error_code
created_at
updated_at
completed_at

Not every product needs exactly these columns. It does need its own job identity.

Suppose the user's task is:

job_123

and each::api returns:

pred_456

pred_456 identifies execution inside the prediction platform. job_123 identifies the thing your product owes the user.

The distinction becomes useful as soon as anything goes wrong between those two systems. An upload can fail before a prediction exists. A prediction can succeed while your application fails to save the result. A webhook can arrive twice. Post-processing can fail after model execution succeeds.

None of those events changes what the prediction API thinks happened. They can still change what your product needs to tell the user.

Normalize provider state into product state

Your frontend should not need to understand the prediction provider's full state vocabulary.

There is another practical reason to keep that vocabulary behind the backend: the current each::labs first-party documentation is in transition. The high-level Get Prediction guide lists:

created
starting
processing
success
error
cancelled

while the newer OpenAPI contract lists:

starting
processing
success
failed
cancelled

Your application does not need to expose that inconsistency.

Normalize both published failure values into one product state:

Application state	Prediction value accepted by our backend
submitted	created, if returned
processing	starting, processing
completed	success
failed	error, failed
cancelled	cancelled

The client now gets a stable contract that you own.

Your application can also have states the prediction system does not:

uploading
waiting_for_capacity
submitting
submitted
processing
saving_result
completed
failed

You may never need all of them. The point is to persist enough state that you can reconstruct what happened without depending on one web process still being alive.

Store the input before you create expensive work

An image-to-video job cannot start until the input image lives somewhere the model can reach.

each::storage uses a two-step upload:

POST /v1/upload/presign
        ↓
presigned_url + public_url
        ↓
PUT raw file bytes to presigned_url
        ↓
pass public_url into model input

The presign response contains a presigned_url, a public_url, and any required_headers. Those headers have to be sent unchanged with the upload PUT.

Once the upload succeeds, the public_url can become model input.

There are two expiry clocks to keep separate. The presigned upload URL accepts uploads for 15 minutes. The stored file has its own retention deadline: 180 days by default, configurable from 60 seconds to 365 days. each::storage accepts files up to 100 MB.

That 100 MB figure is a storage limit, not a universal model limit. The image-to-video model used later in this article currently documents an input-image limit of 50 MB, so the application has to enforce the tighter constraint.

The useful sequence is:

job created
   ↓
image uploaded
   ↓
input_url persisted
   ↓
prediction submitted

If prediction creation fails afterward, the uploaded input is still there. Recovery does not have to begin by asking the user for the file again.

Do you actually need your own queue?

AI architecture diagrams have a habit of inserting a queue as soon as a task becomes asynchronous.

Sometimes that is exactly right.

Sometimes it gives you another service without solving another problem.

An application queue and an execution queue operate at different boundaries:

	Application queue	Prediction/execution queue
Controls	When your app submits work	When inference actually runs
Main purpose	Admission and backpressure	Asynchronous execution
Useful for	Quotas, preprocessing, bursts, prioritization	Long-running model work
Owner	Your backend	Execution platform
Always needed in your app?	No	Part of the execution platform

The high-level each::api prediction guide describes created as “queued, not yet started,” although the newer OpenAPI status enum no longer exposes created. Either way, execution itself is asynchronous.

Your own queue solves a problem earlier in the lifecycle.

Add an application queue when submission needs control

Imagine 800 users press Generate during the same minute.

If every click immediately becomes a prediction, your application has already made its admission decision before the execution platform gets involved.

A queue becomes useful when you need to:

• enforce per-user generation limits;
• perform preprocessing before submission;
• prioritize jobs;
• absorb bursts;
• protect an upstream concurrency limit;
• schedule work;
• stop failure recovery from becoming a submission storm.

The current each::labs error reference gives this a concrete shape. A prediction 429 represents a concurrency cap rather than a normal requests-per-second throttle. A rejected request creates no prediction and is not billed. Capacity returns when an in-flight prediction settles.

Sending the same request again immediately cannot create a free slot.

If you are working through high-volume admission and backpressure in more depth, the existing Scaling AI Image Generation in Consumer Apps article goes further into that problem.

Don't add a queue because the diagram looks unfinished without one

For a modest-volume product, this may be enough:

1. create the application job;
2. persist the input;
3. submit an asynchronous prediction;
4. save predictionID;
5. return the application job ID.

The expensive work is already detached from the web request.

That is the architecture in the example below. There is no Redis, RabbitMQ or separate queue worker because the example does not yet have an admission problem.

Add infrastructure when you can name the failure it prevents.

Submit the prediction, then get out of its way

Once the input exists and the job has been admitted, prediction submission should be fairly boring.

The current Create Prediction endpoint is:

POST https://api.eachlabs.ai/v1/prediction

with Bearer authentication:

Authorization: Bearer YOUR_API_KEY

A request requires:

model
input

The machine-readable Create Prediction contract also exposes webhook_url and webhook_secret, but the current webhook overview says webhook support is currently limited to Workflows V2. Because those first-party sources conflict, this runnable direct-prediction backend does not depend on a prediction webhook.

The current OpenAPI contract marks version as deprecated and ignored. A new integration should not make it part of its own backend contract.

Successful creation returns:

predictionID

Conceptually:

response = requests.post(
    "https://api.eachlabs.ai/v1/prediction",
    headers={
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    },
    json={
        "model": MODEL_SLUG,
        "input": model_input,
    },
)

prediction_id = response.json()["predictionID"]

Save that ID immediately, then return your job ID to the browser:

{
  "id": "job_123",
  "status": "submitted"
}

The frontend continues asking about:

GET /video-jobs/job_123

It does not need to know which model API, prediction ID or orchestration layer sits behind that route.

That separation buys you room to change the implementation later without changing the product contract.

Poll direct predictions; use workflow webhooks when you orchestrate

After submission, your backend needs to learn how the prediction ended.

For a direct each::api prediction, use the current Get Prediction endpoint:

GET /v1/prediction/{id}

The application can poll that resource from a worker or, as the minimal example below does, reconcile when the client reads the application job. The prediction object provides the terminal state, output, and fields such as metrics.cost.

There is a current documentation conflict around direct-prediction webhooks: the machine-readable prediction contract exposes webhook fields, while the webhook overview says webhooks are currently supported only for Workflows V2. A production article should not hide that discrepancy or make a copy/paste backend depend on the ambiguous path.

If this job later becomes an each::workflow, workflow webhooks are the supported callback mechanism. At that point your handler should still be idempotent because callback delivery can be retried.

Retry policy is also a cost policy

A generic service can sometimes get away with:

failed?
retry
failed again?
retry again

Generative APIs deserve more care.

A second generation attempt can consume more inference and produce a different output. Even with identical input, that is not equivalent to retrying an idempotent database read.

Canberk's rule is:

“The semantics have to come before the pattern. A retry here isn't a retry — it's a second purchase of a different product.”

The useful question is not “do we retry?” It is what failed, and what will another attempt actually do?

Failure	Sensible default
Invalid user input	Fail without retrying unchanged
Concurrency rejection	Wait or queue
Definite pre-submission network failure	Bounded recovery may be reasonable
Ambiguous 500 from prediction creation	Do not blindly create another prediction unless duplicate creation is ruled out
Prediction execution failure	Retry only according to product policy
Duplicate webhook	ACK idempotently; create no work
App failed after model success	Recover the result; don't regenerate

Ambiguous creation failures are awkward for a reason

The general each::labs error guide recommends exponential backoff for transient 500 responses.

But POST /v1/prediction does not currently document an idempotency key.

If your client receives an ambiguous server error, the published contract does not establish that repeating the creation request cannot create another prediction.

The working example below therefore does not automatically retry an ambiguous create request. It keeps the application job and surfaces the failure.

That is less convenient than hiding the problem inside a retry decorator. It is also safer than silently buying another generation.

429 is different

The prediction error documentation is more explicit here.

A rejected 429 creates no prediction and is not billed.

The response is therefore about capacity:

429
 ↓
wait / queue / tell user capacity is busy

Repeated submission still does not help while every execution slot remains occupied.

Keep attempts visible

If your product deliberately starts a second prediction after a failed execution, record it:

attempt_count = 2

A retry that can create another paid artifact should not disappear inside infrastructure boilerplate.

For a deeper treatment of provider failover after this point, see AI Model Fallback Best Practices. That is a separate problem from deciding whether the same generation should be retried.

Record what the job actually cost

If inference cost matters to the product, store it on the product operation.

The Get Prediction response can expose:

{
  "metrics": {
    "predict_time": 12.5,
    "cost": 0.05
  }
}

metrics.cost is the execution cost in USD.

When the application reconciles the terminal prediction, persist that value:

video_jobs.cost_usd

Now you can answer questions an aggregate account balance cannot. Which flow produces the most retries? Which accounts generate the most spend? Did a release change average cost per completed job? Are failed generations becoming expensive enough to deserve engineering time?

Cost controls also sit on both sides of submission:

Before:
quotas
admission
concurrency policy
retry policy

After:
actual cost
attempt count
failure type
completion status

The organization balance endpoint is useful as a coarse account-level guardrail, but its current contract says balance reads are eventually consistent with top-ups.

That makes it useful operational telemetry, not transactional per-job accounting.

Put it together: a minimal working AI video backend

The example below uses:

• Python;
• FastAPI;
• SQLite for durable application jobs;
• requests for each::api and each::storage;
• the current wan-v2-6-image-to-video model;
• direct prediction polling for state, output and cost.

There is intentionally no application queue.

Install the dependencies

pip install fastapi uvicorn requests python-multipart

Set the API key:

export EACHLABS_API_KEY="your-api-key"

app.py

import os
import sqlite3
import uuid
from datetime import datetime, timezone
from typing import Any

import requests
from fastapi import (
    FastAPI,
    File,
    Form,
    HTTPException,
    UploadFile,
)


EACHLABS_API_KEY = os.environ["EACHLABS_API_KEY"]

EACHLABS_BASE_URL = "https://api.eachlabs.ai/v1"

MODEL_SLUG = "wan-v2-6-image-to-video"

DB_PATH = "video_jobs.db"


app = FastAPI(title="AI Video Backend")


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def get_db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db() -> None:
    with get_db() as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS video_jobs (
                id TEXT PRIMARY KEY,
                status TEXT NOT NULL,
                prediction_id TEXT UNIQUE,
                input_url TEXT,
                output_url TEXT,
                prompt TEXT NOT NULL,
                attempt_count INTEGER NOT NULL DEFAULT 0,
                cost_usd REAL,
                error_code TEXT,
                error_message TEXT,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                completed_at TEXT
            )
            """
        )


init_db()


def eachlabs_headers() -> dict[str, str]:
    return {
        "Authorization": f"Bearer {EACHLABS_API_KEY}",
    }


def get_job(job_id: str) -> dict[str, Any]:
    with get_db() as conn:
        row = conn.execute(
            "SELECT * FROM video_jobs WHERE id = ?",
            (job_id,),
        ).fetchone()

    if row is None:
        raise HTTPException(
            status_code=404,
            detail="Video job not found",
        )

    return dict(row)


def get_job_by_prediction_id(
    prediction_id: str,
) -> dict[str, Any] | None:
    with get_db() as conn:
        row = conn.execute(
            """
            SELECT *
            FROM video_jobs
            WHERE prediction_id = ?
            """,
            (prediction_id,),
        ).fetchone()

    return dict(row) if row else None


def update_job(job_id: str, **values: Any) -> None:
    if not values:
        return

    values["updated_at"] = now_iso()

    assignments = ", ".join(
        f"{column} = ?"
        for column in values
    )

    params = list(values.values()) + [job_id]

    with get_db() as conn:
        conn.execute(
            f"""
            UPDATE video_jobs
            SET {assignments}
            WHERE id = ?
            """,
            params,
        )


def upload_image_to_eachlabs(
    contents: bytes,
    content_type: str,
) -> str:
    presign = requests.post(
        f"{EACHLABS_BASE_URL}/upload/presign",
        headers={
            **eachlabs_headers(),
            "Content-Type": "application/json",
        },
        json={
            "content_type": content_type,
            "file_type": "image",
        },
        timeout=30,
    )

    presign.raise_for_status()
    upload = presign.json()

    put_response = requests.put(
        upload["presigned_url"],
        data=contents,
        headers={
            "Content-Type": content_type,
            **(upload.get("required_headers") or {}),
        },
        timeout=120,
    )

    put_response.raise_for_status()

    return upload["public_url"]


def create_prediction(
    image_url: str,
    prompt: str,
) -> str:
    response = requests.post(
        f"{EACHLABS_BASE_URL}/prediction",
        headers={
            **eachlabs_headers(),
            "Content-Type": "application/json",
        },
        json={
            "model": MODEL_SLUG,
            "input": {
                "image_url": image_url,
                "prompt": prompt,
                "resolution": "720p",
                "duration": "5",
            },
        },
        timeout=30,
    )

    # Deliberately no blind retry here.
    # The current creation contract does not document
    # an idempotency key for ambiguous failures.
    response.raise_for_status()

    return response.json()["predictionID"]


def get_prediction(
    prediction_id: str,
) -> dict[str, Any]:
    response = requests.get(
        f"{EACHLABS_BASE_URL}/prediction/{prediction_id}",
        headers=eachlabs_headers(),
        timeout=30,
    )

    response.raise_for_status()
    return response.json()


def extract_output_url(output: Any) -> str | None:
    if isinstance(output, str):
        return output

    if isinstance(output, dict):
        url = output.get("url")
        return url if isinstance(url, str) else None

    if isinstance(output, list) and output:
        first = output[0]

        if isinstance(first, str):
            return first

        if isinstance(first, dict):
            url = first.get("url")
            return url if isinstance(url, str) else None

    return None



def reconcile_job(
    job: dict[str, Any],
) -> dict[str, Any]:
    prediction_id = job.get("prediction_id")

    if not prediction_id:
        return job

    if (
        job["status"] in {
            "completed",
            "failed",
            "cancelled",
        }
        and job.get("cost_usd") is not None
    ):
        return job

    prediction = get_prediction(prediction_id)

    prediction_status = prediction.get("status")
    metrics = prediction.get("metrics") or {}

    # The current first-party docs expose two status
    # vocabularies. Normalize both into our app state.
    if prediction_status == "created":
        update_job(
            job["id"],
            status="submitted",
        )

    elif prediction_status in {
        "starting",
        "processing",
    }:
        update_job(
            job["id"],
            status="processing",
        )

    elif prediction_status == "success":
        update_job(
            job["id"],
            status="completed",
            output_url=extract_output_url(
                prediction.get("output")
            ),
            cost_usd=metrics.get("cost"),
            completed_at=(
                job.get("completed_at")
                or now_iso()
            ),
            error_code=None,
            error_message=None,
        )

    elif prediction_status in {
        "error",
        "failed",
    }:
        update_job(
            job["id"],
            status="failed",
            cost_usd=metrics.get("cost"),
            error_code=(
                job.get("error_code")
                or "prediction_failed"
            ),
            error_message=(
                job.get("error_message")
                or str(
                    prediction.get("logs")
                    or "Prediction failed"
                )
            ),
            completed_at=(
                job.get("completed_at")
                or now_iso()
            ),
        )

    elif prediction_status == "cancelled":
        update_job(
            job["id"],
            status="cancelled",
            cost_usd=metrics.get("cost"),
            completed_at=(
                job.get("completed_at")
                or now_iso()
            ),
        )

    return get_job(job["id"])


@app.post("/video-jobs")
async def create_video_job(
    prompt: str = Form(...),
    image: UploadFile = File(...),
):
    job_id = f"job_{uuid.uuid4().hex}"

    created_at = now_iso()

    with get_db() as conn:
        conn.execute(
            """
            INSERT INTO video_jobs (
                id,
                status,
                prompt,
                created_at,
                updated_at
            )
            VALUES (?, ?, ?, ?, ?)
            """,
            (
                job_id,
                "uploading",
                prompt,
                created_at,
                created_at,
            ),
        )

    try:
        contents = await image.read()

        content_type = (
            image.content_type
            or "application/octet-stream"
        )

        input_url = upload_image_to_eachlabs(
            contents,
            content_type,
        )

        update_job(
            job_id,
            input_url=input_url,
            status="submitting",
        )

        prediction_id = create_prediction(
            input_url,
            prompt,
        )

        update_job(
            job_id,
            prediction_id=prediction_id,
            status="submitted",
            attempt_count=1,
        )

    except requests.HTTPError as exc:
        response = exc.response

        status_code = (
            response.status_code
            if response is not None
            else None
        )

        error_body = (
            response.text
            if response is not None
            else str(exc)
        )

        update_job(
            job_id,
            status="failed",
            error_code=(
                "upstream_concurrency"
                if status_code == 429
                else f"upstream_http_{status_code}"
                if status_code
                else "upstream_request_failed"
            ),
            error_message=error_body,
            completed_at=now_iso(),
        )

        raise HTTPException(
            status_code=(
                503
                if status_code == 429
                else 502
            ),
            detail={
                "job_id": job_id,
                "message": (
                    "Could not submit video generation"
                ),
            },
        )

    return {
        "id": job_id,
        "status": "submitted",
    }


    webhook_status = payload.get("status")

    if webhook_status == "succeeded":
        update_job(
            job["id"],
            status="completed",
            output_url=extract_output_url(
                payload.get("output")
            ),
            completed_at=now_iso(),
            error_code=None,
            error_message=None,
        )

    elif webhook_status == "failed":
        update_job(
            job["id"],
            status="failed",
            error_code="prediction_failed",
            error_message=webhook_error_detail(
                payload.get("output")
            ),
            completed_at=now_iso(),
        )

    else:
        raise HTTPException(
            status_code=400,
            detail="Unknown webhook status",
        )

    return {"received": True}


@app.get("/video-jobs/{job_id}")
def read_video_job(job_id: str):
    job = get_job(job_id)

    try:
        # Reconcile the durable app job against the
        # current direct-prediction state.
        job = reconcile_job(job)
    except requests.HTTPError:
        # A temporary reconciliation failure should not
        # overwrite valid application state.
        pass

    return {
        "id": job["id"],
        "status": job["status"],
        "output_url": job["output_url"],
        "cost_usd": job["cost_usd"],
        "error": (
            {
                "code": job["error_code"],
                "message": job["error_message"],
            }
            if job["error_code"]
            else None
        ),
        "created_at": job["created_at"],
        "completed_at": job["completed_at"],
    }

Run it:

uvicorn app:app --reload

Create a job:

curl -X POST http://localhost:8000/video-jobs \
  -F 'prompt=Slow cinematic camera push toward the product, soft studio light' \
  -F 'image=@product.png'

The route uploads the input, creates the asynchronous prediction, persists its predictionID, and returns:

{
  "id": "job_d5ca0f...",
  "status": "submitted"
}

It does not wait for the video.

Each read reconciles the durable application job against the current prediction. In a higher-volume service, move that polling into a scheduled worker so completion does not depend on the user refreshing the page.

The frontend still talks to your API:

curl http://localhost:8000/video-jobs/job_d5ca0f...

While work is running:

{
  "id": "job_d5ca0f...",
  "status": "processing",
  "output_url": null,
  "cost_usd": null,
  "error": null,
  "created_at": "2026-09-03T10:00:00+00:00",
  "completed_at": null
}

After completion and reconciliation:

{
  "id": "job_d5ca0f...",
  "status": "completed",
  "output_url": "https://...",
  "cost_usd": 0.42,
  "error": null,
  "created_at": "2026-09-03T10:00:00+00:00",
  "completed_at": "2026-09-03T10:01:24+00:00"
}

0.42 is illustrative. The application stores whatever metrics.cost the actual prediction reports.

The model call uses the current wan-v2-6-image-to-video slug with image_url, prompt, resolution and duration. Its current model page documents 720p/1080p output and video durations up to 15 seconds.

What the example deliberately leaves out

This is not a complete platform.

There is no authentication system, quota service, billing product, application queue, CDN configuration or multi-provider routing layer.

What it does prove is the lifecycle:

1. Create your own durable job
2. Persist the input
3. Submit asynchronous work
4. Store the prediction ID
5. Return your job ID
6. Poll and reconcile the prediction
7. Persist output and cost
8. Let the client read product state

Where an application queue fits later

If submission volume grows:

POST /video-jobs
      ↓
job database
      ↓
application queue
      ↓
submission worker
      ↓
each::api

The client-facing API does not have to change.

It still knows only:

job_123

The queue changes when the backend submits work, not the identity of the user's task.

If this eventually becomes a multi-stage media chain rather than a single generation, How to Chain Image, Upscale, and Video in One API Call covers the handoff problem between generative steps.

When this backend is more architecture than you need

Not every prototype needs a queue, six database tables, a workflow engine and an event bus.

With one model, modest traffic, a handful of users and no serious admission-control problem, this may be perfectly reasonable:

application job
      ↓
persisted input
      ↓
async prediction
      ↓
poll direct prediction
      ↓
persist result

Canberk's rule for adding orchestration is:

“If all we add is another network hop, you shouldn't use us.”

An extra layer should remove complexity you actually have, not justify itself with hypothetical scale.

Add an application queue when you have an admission problem. Move polling into workers when read-time reconciliation no longer fits the traffic pattern. Add workflow webhooks when the product becomes a workflow. Add fallback when provider failure becomes a product requirement.

Otherwise you have built another service to deploy, monitor and debug.

FAQ

Do AI video apps need a job queue?

They need asynchronous execution, but not always an additional application queue. Add your own when you need admission control, preprocessing, quotas, prioritization, burst absorption or protection from upstream concurrency limits.

Should I use polling or webhooks for video generation?

For direct each::api predictions, polling GET /v1/prediction/{id} is the defensible current path because first-party pages conflict about direct-prediction webhook availability. For each::workflows, webhooks are explicitly supported and avoid continuous polling; polling still remains useful for reconciliation.

Where should generated AI videos be stored?

Persist the output reference on your own application job. If your product needs retention, permissions or delivery behavior beyond the output URL's guarantees, copy the completed asset into storage you control.

For input media, each::storage provides the presigned-upload flow used in the example.

What should I store in an AI video job record?

At minimum, keep your own job ID and status, the external prediction ID, input and output references, attempt count, timestamps, failure information and actual execution cost when the prediction system exposes it.

How many times should a failed AI video generation be retried?

There is no useful universal number. Invalid input should not be retried unchanged. Capacity errors should wait for capacity. An ambiguous create-prediction failure should not be blindly repeated unless duplicate creation can be ruled out. Retrying a prediction that already executed is a new potentially billable generation, not ordinary HTTP recovery.