# Bella OpenAPI Video Generation Workflow: How Video Jobs Are Processed

> Discover the Bella OpenAPI video generation workflow. Learn how video jobs are processed via a two-stage asynchronous pipeline using Redis and AI providers.

- Repository: [Ke Technologies/bella-openapi](https://github.com/lianjiatech/bella-openapi)
- Tags: how-to-guide
- Published: 2026-03-06

---

**Bella OpenAPI processes video generation as a two-stage asynchronous pipeline: jobs are first created and enqueued to Redis, then submitted to AI providers via scheduled executors, with continuous polling until finalization.**

The [lianjiatech/bella-openapi](https://github.com/lianjiatech/bella-openapi) repository implements a robust, queue-driven architecture for handling video generation requests. This article breaks down the complete video generation workflow—from REST API ingestion through Redis-backed job distribution to provider-specific adaptor integration—using actual source file paths and implementation details from the codebase.

## Stage 1: Job Creation and Enqueuing

### REST API Entry Point

Client requests enter through [`VideoController.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoController.java) at [`api/server/src/main/java/com/ke/bella/openapi/endpoints/VideoController.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/endpoints/VideoController.java). The controller exposes a `POST /v1/video` endpoint that accepts a `VideoCreateRequest` payload and delegates to the service layer.

### Service Layer and Persistence

The `VideoService.createVideoJob(...)` method in [`api/server/src/main/java/com/ke/bella/openapi/service/VideoService.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/service/VideoService.java) orchestrates initial job setup:

1. Generates a global `videoId` using `VideoIdGenerator.VIDEO_ID_GENERATOR.generate(spaceCode)`
2. Persists a `VideoJobDB` row with status **`queued`**
3. Enqueues the ID into the Redis submit list

```java
// Inside VideoService.createVideoJob
String videoId = VideoIdGenerator.VIDEO_ID_GENERATOR.generate(spaceCode);
videoJobDB.setStatus(Status.queued.name());
videoRepo.addVideoJob(videoJobDB);
queueManager.enqueueForSubmit(model, videoId);   // Redis LPUSH

```

### Redis Queue Integration

The `VideoJobQueues` class at [`api/server/src/main/java/com/ke/bella/openapi/queue/VideoJobQueues.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/queue/VideoJobQueues.java) manages the **submit queue** (`bella:video:submit:<model>`) and **sync queue** (`bella:video:syncing`). Jobs remain in the submit queue until assigned to a provider channel.

## Stage 2: Submit Queue Processing

### The VideoJobExecutor Scheduler

[`VideoJobExecutor.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoJobExecutor.java) at [`api/server/src/main/java/com/ke/bella/openapi/executor/VideoJobExecutor.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/executor/VideoJobExecutor.java) starts after Spring Boot initialization (`@PostConstruct`). It runs two fixed-rate schedulers:

- **Submit scheduler**: `scheduleIntervalSeconds` (default 5s) → `processVideoJobs()`
- **Sync scheduler**: `syncIntervalSeconds` (default 5s) → `processSyncQueue()`

### RPM Throttling and Channel Selection

For each model, the executor:

1. Acquires a distributed lock (`bella:video:model-lock:<model>`)
2. Loads active channels via `ChannelService`
3. Filters channels with available RPM quota using `ChannelRpmLimiter` at [`api/server/src/main/java/com/ke/bella/openapi/protocol/limiter/ChannelRpmLimiter.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/protocol/limiter/ChannelRpmLimiter.java)
4. Calculates safe batch size via `calculateSafeBatchSizeByRpm`

### Batch Dequeue and Task Submission

The executor dequeues up to the calculated batch size from the submit queue:

```java
List<String> videoIds = queueManager.dequeueForSubmit(model, batchSize);
submitBatchTasksToChannels(videoIds, availableChannels);

```

Each job is assigned to a channel (round-robin), the RPM quota is consumed, and a `VideoJobSubmitTask` is submitted to the `TaskExecutor` worker pool.

## Stage 3: Provider Submission

### VideoJobSubmitTask Execution

[`VideoJobSubmitTask.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoJobSubmitTask.java) at [`api/server/src/main/java/com/ke/bella/openapi/executor/VideoJobSubmitTask.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/executor/VideoJobSubmitTask.java) executes the following:

1. Loads the job from `videoRepo.queryVideoJob`
2. Invokes the **adaptor** via `VideoAdaptor.submitVideoTask(...)`
3. Receives a channel-specific video ID from the provider
4. Performs a CAS (compare-and-swap) update from `queued` → `submitting` → `processing`
5. Enqueues the job ID to the **sync queue** (`VideoJobQueues.enqueueForSync`)

### Adaptor Pattern and Provider Integration

The `VideoAdaptor` interface at [`api/server/src/main/java/com/ke/bella/openapi/protocol/video/VideoAdaptor.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/protocol/video/VideoAdaptor.java) abstracts provider-specific implementations. The `HuoshanAdaptor` at [`api/server/src/main/java/com/ke/bella/openapi/protocol/video/HuoshanAdaptor.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/protocol/video/HuoshanAdaptor.java) converts OpenAPI requests to Huoshan format using `HuoshanVideoConverter` and manages authentication.

```java
String channelVideoId = ((VideoAdaptor<VideoProperty>) adaptor)
        .submitVideoTask(request, baseUrl, (VideoProperty) property, job.getVideoId());

```

## Stage 4: Synchronization and Finalization

### Sync Queue Polling

The sync scheduler in `VideoJobExecutor.processSyncQueue()` dequeues IDs from `bella:video:syncing` and submits `VideoJobSyncTask` instances to the worker pool.

### VideoJobSyncTask and Provider Polling

[`VideoJobSyncTask.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoJobSyncTask.java) at [`api/server/src/main/java/com/ke/bella/openapi/executor/VideoJobSyncTask.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/executor/VideoJobSyncTask.java) handles the polling logic:

1. Loads the job and verifies status is `processing`
2. Checks `minProcessingSeconds` threshold; if not elapsed, re-enqueues
3. Queries provider status via `VideoAdaptor.queryVideoTask(...)`
4. For terminal states (`completed`, `failed`, `cancelled`):
   - **Completed**: Downloads file via `VideoAdaptor.transferVideoToFile(...)` to OpenAI file service
   - Updates DB with size, duration, file ID, error JSON, progress
   - Logs usage and cost via `EndpointLogger`

```java
ChannelVideoResult result = queryChannelVideoStatus(job, channel);
if (isTerminalState(result.getStatus())) {
    handleTerminalState(result, channel);   // download + processSyncResult
} else {
    queueManager.enqueueForSync(videoId);   // poll later
}

```

### Cost Logging and Usage Tracking

`VideoService.processSyncResult(...)` in [`api/server/src/main/java/com/ke/bella/openapi/service/VideoService.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/service/VideoService.java) finalizes the job and invokes `logVideoCost(...)` to record provider usage data into `EndpointProcessData`, which is then forwarded to `EndpointLogger` at [`api/server/src/main/java/com/ke/bella/openapi/protocol/log/EndpointLogger.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/protocol/log/EndpointLogger.java) for centralized billing analytics.

## Job Lifecycle Management

### Status States and Transitions

The `VideoJobDB` entity tracks jobs through the following states:

- **`queued`**: Initial state after creation, waiting in submit queue
- **`submitting`**: Actively being sent to provider
- **`processing`**: Provider accepted job, generation in progress
- **`completed`**: Video generated successfully, file stored
- **`failed`**: Generation error occurred
- **`cancelled`**: Job aborted
- **`deleted`**: Soft-deleted by user

Transitions use CAS (compare-and-swap) logic to prevent race conditions during concurrent executor threads.

### Deletion Constraints

Only jobs in terminal states (`queued`, `completed`, `failed`, `cancelled`) can be deleted. The `VideoService.deleteVideoJob(...)` method performs a soft delete by updating the status to `deleted` rather than removing the database row.

## Code Examples

### Creating a Video Job via cURL

```bash
curl -X POST https://api.example.com/v1/video \
  -H "Authorization: Bearer <access_token>" \
  -H "Content-Type: application/json" \
  -d '{
        "model":"huoshan-video-1.0",
        "prompt":"A futuristic city skyline at sunrise",
        "size":"720p",
        "seconds":"10"
      }'

```

The controller maps this JSON to `VideoCreateRequest`, then `VideoService` persists and enqueues the job.

### Using the Java SDK

```java
BellaOpenApiClient client = BellaOpenApiClient.builder()
        .baseUrl("https://api.example.com")
        .accessToken("YOUR_TOKEN")
        .build();

VideoCreateRequest request = VideoCreateRequest.builder()
        .model("huoshan-video-1.0")
        .prompt("A kitten playing with a ball of yarn")
        .size("720p")
        .seconds("5")
        .build();

VideoJobResponse response = client.video().create(request);
System.out.println("Video job created, id = " + response.getVideoId());

```

Behind the scenes, the SDK sends the same request that ends up in `VideoService.createVideoJob`.

### Querying Job Status

```bash
curl -X GET https://api.example.com/v1/video/{videoId} \
  -H "Authorization: Bearer <access_token>"

```

This endpoint reads `VideoJobDB` via `VideoService.queryVideoJob` and returns the current status.

### Deleting a Finished Job

```bash
curl -X DELETE https://api.example.com/v1/video/{videoId} \
  -H "Authorization: Bearer <access_token>"

```

This calls `VideoService.deleteVideoJob`, which updates the DB status to `deleted`.

## Key Implementation Files

| Component | File | Purpose |
|-----------|------|---------|
| **Controller** | [[`VideoController.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoController.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/endpoints/VideoController.java) | HTTP entry point for video CRUD |
| **Service** | [[`VideoService.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoService.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/service/VideoService.java) | Job creation, status update, sync result handling |
| **Redis Queue Facade** | [[`VideoJobQueues.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoJobQueues.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/queue/VideoJobQueues.java) | Submit & sync queue operations |
| **Executor (submit)** | [[`VideoJobExecutor.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoJobExecutor.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/executor/VideoJobExecutor.java) | Scheduler, RPM throttling, batch dequeue |
| **Submit Task** | [[`VideoJobSubmitTask.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoJobSubmitTask.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/executor/VideoJobSubmitTask.java) | Calls provider adaptor, updates status, pushes to sync queue |
| **Sync Task** | [[`VideoJobSyncTask.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoJobSyncTask.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/executor/VideoJobSyncTask.java) | Polls provider, downloads result, final DB update |
| **Adaptor Interface** | [[`VideoAdaptor.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoAdaptor.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/protocol/video/VideoAdaptor.java) | Abstracts provider-specific calls |
| **Huoshan Provider** | [[`HuoshanAdaptor.java`](https://github.com/lianjiatech/bella-openapi/blob/main/HuoshanAdaptor.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/protocol/video/HuoshanAdaptor.java) | Concrete implementation for Huoshan video API |
| **Request DTO** | [[`VideoCreateRequest.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoCreateRequest.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/sdk/src/main/java/com/ke/bella/openapi/protocol/video/VideoCreateRequest.java) | JSON model for job creation |
| **Database Entity** | [[`VideoJobDB.java`](https://github.com/lianjiatech/bella-openapi/blob/main/VideoJobDB.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/tables/pojos/VideoJobDB.java) | Persistent row (status, model, timestamps, etc.) |
| **RPM Limiter** | [[`ChannelRpmLimiter.java`](https://github.com/lianjiatech/bella-openapi/blob/main/ChannelRpmLimiter.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/protocol/limiter/ChannelRpmLimiter.java) | Enforces per-channel request-per-minute caps |
| **Endpoint Logger** | [[`EndpointLogger.java`](https://github.com/lianjiatech/bella-openapi/blob/main/EndpointLogger.java)](https://github.com/lianjiatech/bella-openapi/blob/develop/api/server/src/main/java/com/ke/bella/openapi/protocol/log/EndpointLogger.java) | Centralised usage/cost logging for video jobs |

## Summary

- **Bella OpenAPI** implements an asynchronous two-stage pipeline for video generation, separating job submission from result synchronization.
- **Redis queues** (`bella:video:submit:<model>` and `bella:video:syncing`) decouple the REST API from provider communication, enabling horizontal scaling.
- **VideoJobExecutor** manages both stages via scheduled threads, enforcing **RPM limits** through `ChannelRpmLimiter` and distributed locking.
- **Adaptor pattern** (`VideoAdaptor` / `HuoshanAdaptor`) abstracts provider-specific protocols, allowing new video providers without changing core orchestration logic.
- **CAS state transitions** ensure thread-safe updates from `queued` → `processing` → `completed`/`failed`, with final cost logging via `EndpointLogger`.

## Frequently Asked Questions

### How does Bella OpenAPI handle high concurrency for video generation?

Bella OpenAPI uses **distributed Redis queues** and **per-model locking** (`bella:video:model-lock:<model>`) to prevent race conditions. The `VideoJobExecutor` calculates safe batch sizes based on remaining **RPM (requests per minute)** quotas via `ChannelRpmLimiter`, ensuring channels never exceed provider rate limits. Jobs exceeding capacity remain in the Redis queue until capacity frees up.

### What is the purpose of the VideoAdaptor interface?

The `VideoAdaptor` interface (defined in [`api/server/src/main/java/com/ke/bella/openapi/protocol/video/VideoAdaptor.java`](https://github.com/lianjiatech/bella-openapi/blob/main/api/server/src/main/java/com/ke/bella/openapi/protocol/video/VideoAdaptor.java)) abstracts provider-specific video APIs. Implementations like `HuoshanAdaptor` handle protocol conversion, authentication, and error mapping without modifying the core job orchestration. This pattern allows Bella OpenAPI to support multiple video providers (e.g., Huoshan, OpenAI, or custom endpoints) by simply adding new adaptor implementations.

### How does the system ensure video jobs don't get lost during processing?

Jobs persist in **PostgreSQL** via `VideoJobDB` with strict status state machines (`queued` → `submitting` → `processing` → terminal). The `VideoJobSubmitTask` uses **CAS (compare-and-swap)** updates to ensure only valid state transitions occur. If a worker crashes, jobs remain in Redis queues (`bella:video:submit:<model>` or `bella:video:syncing`) until the `VideoJobExecutor` recovers them on the next scheduling interval (default 5 seconds).

### Can I delete a video job while it's processing?

No. Bella OpenAPI restricts deletion to **terminal states** only: `queued`, `completed`, `failed`, or `cancelled`. The `VideoService.deleteVideoJob(...)` method performs a soft delete by updating the status to `deleted` rather than removing the record. This preserves audit trails and prevents accidental termination of active provider tasks that might incur costs.