Queue API

Overview

The Queue API provides endpoints for managing BullMQ media processing jobs and monitoring queue health. Jobs are enqueued via this API and processed by background workers. Base Path: /api/queue Key Features:

Enqueue media jobs (transcoding, thumbnails, captions)
Real-time queue statistics (depth, waiting, active, completed)
Subscription tier enforcement
Taxonomy cache management

GET /api/queue/stats

Returns BullMQ queue depths and taxonomy cache statistics. Useful for dashboard observability. Authentication: None (public endpoint)

boolean

Request success status

stats

object

BullMQ queue statistics object containing:

waiting - Jobs waiting to be processed
active - Currently processing jobs
completed - Successfully completed jobs
failed - Failed jobs
delayed - Jobs scheduled for future execution

taxonomy_cache

object

Taxonomy cache statistics:

size - Number of cached entries
hits - Cache hit count
misses - Cache miss count

Example Request:

curl http://localhost:3001/api/queue/stats

Response:

{
  "ok": true,
  "stats": {
    "waiting": 3,
    "active": 1,
    "completed": 142,
    "failed": 2,
    "delayed": 0
  },
  "taxonomy_cache": {
    "size": 1847,
    "hits": 9234,
    "misses": 128
  }
}

Use Cases:

Health check dashboards
Queue depth monitoring alerts
Performance analytics
Worker scaling decisions

POST /api/queue/media

Enqueue a media job that already exists in the Directus media_jobs collection. Authentication: Directus JWT (Bearer token)

You must create the Directus media_jobs record before calling this endpoint. This endpoint only pushes the job into BullMQ so workers can pick it up.

Request Parameters

jobId

string

required

UUID of the existing media_jobs Directus record

operation

string

required

Operation type (e.g., transcoding, thumbnail, caption_generation). Used for tier validation.

input_file_id

string

UUID of the input file from Directus directus_files (optional)

input_url

string

URL of remote input file (optional, alternative to input_file_id)

params

object

Job-specific parameters (e.g., {"format": "mp4", "resolution": "1080p"})

creator_profile_id

string

UUID of the associated platform_connections record (optional)

priority

number

Job priority (1-100, default: 10). Lower numbers = higher priority.

Response

boolean

Request success status

bullJobId

string

BullMQ job ID (UUID) assigned by the queue

jobId

string

Directus media_jobs record UUID (echoed from request)

Example Request

curl -X POST http://localhost:3001/api/queue/media \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." \
  -H "Content-Type: application/json" \
  -d '{
    "jobId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
    "operation": "transcoding",
    "input_file_id": "a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d",
    "params": {
      "format": "mp4",
      "resolution": "1080p",
      "codec": "h264"
    },
    "creator_profile_id": "d3f21c8e-4a5b-4c9d-8e7f-1a2b3c4d5e6f",
    "priority": 5
  }'

Success Response:

{
  "ok": true,
  "bullJobId": "bull-job-uuid-12345",
  "jobId": "f47ac10b-58cc-4372-a567-0e02b2c3d479"
}

Subscription Tier Enforcement

The endpoint validates if the user’s subscription tier allows the requested operation: From queue.js:88:

const check = await canPerform(userId, operation, _directusAdminClient);
if (!check.allowed) {
  return res.status(403).json({
    error: "tier_limit_reached",
    reason: check.reason,
    message:
      check.reason === "limit_reached"
        ? "You have reached your plan limit for this feature. Upgrade to continue."
        : "This feature is not available on your current plan.",
  });
}

Tier Limit Error Response:

{
  "error": "tier_limit_reached",
  "reason": "limit_reached",
  "message": "You have reached your plan limit for this feature. Upgrade to continue."
}

Tier Feature Unavailable:

{
  "error": "tier_limit_reached",
  "reason": "feature_unavailable",
  "message": "This feature is not available on your current plan."
}

Supported Operations

Common operation values (validated against subscription tier):

transcoding - Video format conversion
thumbnail - Generate video thumbnail
caption_generation - AI caption generation
download - Download remote media
upload - Upload processing
analysis - Media analysis/metadata extraction

POST /api/queue/taxonomy/invalidate

Bust the taxonomy cache (e.g., after admin updates taxonomy_terms in Directus). Authentication: None (internal loopback only)

This endpoint is designed for internal use only. It should be called via localhost after Directus webhooks update taxonomy data.

Example Request:

curl -X POST http://localhost:3001/api/queue/taxonomy/invalidate

Response:

{
  "ok": true,
  "message": "Taxonomy cache invalidated"
}

Use Case: Directus webhook configuration:

Webhook URL: http://localhost:3001/api/queue/taxonomy/invalidate
Trigger: After Create/Update/Delete on taxonomy_terms collection
Method: POST

Queue Architecture

Job Lifecycle

Create Directus Record - Client creates media_jobs record with status=pending
Enqueue - Client calls POST /api/queue/media with jobId
BullMQ Processing - Worker picks up job from queue
Status Updates - Worker updates Directus record (status=processing, status=completed, status=failed)
Client Polling - Client polls Directus media_jobs record for status changes

Queue Configuration

From server/utils/actionBus.js:

const mediaQueue = new Queue('media', {
  connection: redisConnection,
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000,
    },
    removeOnComplete: 100, // Keep last 100 completed jobs
    removeOnFail: 1000,    // Keep last 1000 failed jobs
  },
});

Job Priority Levels

1-3 - Critical (real-time user actions)
4-6 - High (scheduled urgent tasks)
7-10 - Normal (default background processing)
11+ - Low (batch operations, cleanup)

Error Handling

Missing required fields:

{
  "error": "jobId and operation are required"
}

Invalid token:

{
  "error": "Invalid or expired Directus token"
}

Queue enqueue failure:

{
  "ok": false,
  "error": "Redis connection refused"
}

Stats retrieval failure:

{
  "ok": false,
  "error": "Failed to connect to Redis"
}

Monitoring & Observability

Queue Metrics Dashboard

Use the /stats endpoint to build real-time dashboards:

// Poll every 5 seconds
setInterval(async () => {
  const res = await fetch('http://localhost:3001/api/queue/stats');
  const { stats } = await res.json();
  
  console.log(`Queue depth: ${stats.waiting}`);
  console.log(`Active jobs: ${stats.active}`);
  console.log(`Failed jobs: ${stats.failed}`);
}, 5000);

Alerting Thresholds

Recommended alert triggers:

High queue depth: waiting > 100 (consider scaling workers)
High failure rate: failed / (completed + failed) > 0.1 (10%)
Stalled jobs: active > 0 for > 30 minutes (worker crash?)

Integration Example

Complete workflow for enqueuing a transcoding job:

// 1. Upload file to Directus
const formData = new FormData();
formData.append('file', videoFile);

const uploadRes = await fetch('http://127.0.0.1:8055/files', {
  method: 'POST',
  headers: { 'Authorization': `Bearer ${directusToken}` },
  body: formData,
});
const { data: fileData } = await uploadRes.json();
const fileId = fileData.id;

// 2. Create media_jobs record
const jobRes = await fetch('http://127.0.0.1:8055/items/media_jobs', {
  method: 'POST',
  headers: {
    'Authorization': `Bearer ${directusToken}`,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    operation: 'transcoding',
    input_file_id: fileId,
    status: 'pending',
    params: { format: 'mp4', resolution: '1080p' },
  }),
});
const { data: jobData } = await jobRes.json();
const jobId = jobData.id;

// 3. Enqueue in BullMQ
const enqueueRes = await fetch('http://localhost:3001/api/queue/media', {
  method: 'POST',
  headers: {
    'Authorization': `Bearer ${directusToken}`,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    jobId,
    operation: 'transcoding',
    input_file_id: fileId,
    params: { format: 'mp4', resolution: '1080p' },
    priority: 5,
  }),
});
const { ok, bullJobId } = await enqueueRes.json();

console.log(`Job enqueued: ${bullJobId}`);

// 4. Poll for completion
const pollInterval = setInterval(async () => {
  const statusRes = await fetch(`http://127.0.0.1:8055/items/media_jobs/${jobId}`, {
    headers: { 'Authorization': `Bearer ${directusToken}` },
  });
  const { data } = await statusRes.json();
  
  if (data.status === 'completed') {
    console.log('Job completed!', data.output_file_id);
    clearInterval(pollInterval);
  } else if (data.status === 'failed') {
    console.error('Job failed:', data.error);
    clearInterval(pollInterval);
  }
}, 2000);

Endpoints

MCP Tools

Collections

Overview

GET /api/queue/stats

POST /api/queue/media

Request Parameters

Response

Example Request

Subscription Tier Enforcement

Supported Operations

POST /api/queue/taxonomy/invalidate

Queue Architecture

Job Lifecycle

Queue Configuration

Job Priority Levels

Error Handling

Monitoring & Observability

Queue Metrics Dashboard

Alerting Thresholds

Integration Example

Next Steps

Captions API

Credentials API

Build docs developers (and LLMs) love

Endpoints

MCP Tools

Collections

​Overview

​GET /api/queue/stats

​POST /api/queue/media

​Request Parameters

​Response

​Example Request

​Subscription Tier Enforcement

​Supported Operations

​POST /api/queue/taxonomy/invalidate

​Queue Architecture

​Job Lifecycle

​Queue Configuration

​Job Priority Levels

​Error Handling

​Monitoring & Observability

​Queue Metrics Dashboard

​Alerting Thresholds

​Integration Example

​Next Steps

Captions API

Credentials API

Build docs developers (and LLMs) love

Overview

GET /api/queue/stats

POST /api/queue/media

Request Parameters

Response

Example Request

Subscription Tier Enforcement

Supported Operations

POST /api/queue/taxonomy/invalidate

Queue Architecture

Job Lifecycle

Queue Configuration

Job Priority Levels

Error Handling

Monitoring & Observability

Queue Metrics Dashboard

Alerting Thresholds

Integration Example

Next Steps