Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/rivenmedia/riven-ts/llms.txt

Use this file to discover all available pages before exploring further.

Riven processes all media pipeline work asynchronously through BullMQ backed by Redis. Queues are divided into three categories: flow queues that implement the core multi-step pipeline, plugin queues that fan events out to plugin hooks, and sandboxed worker queues that run CPU-intensive tasks in isolated Node.js worker threads. All queues connect to Redis via settings.redisUrl.

​
Queue Topology

​
Flow Queues

Flow queues power the core pipeline. Each queue has a corresponding BullMQ FlowProducer-compatible schema that defines the job’s input, output, and allowed child jobs.
Queue namePurpose
process-item-requestFetches metadata from TMDB/TVDB and creates MediaItem entities
process-media-itemOrchestrates a single media item through scrape → validate → download → complete
request-content-servicesPolls content source plugins (Overseerr, MDBList, …) for new requests
scrape-itemFans the scrape request to all enabled scraper plugins; collects stream candidates
download-itemDrives the debrid cache check and download flow for a single item
download-item.rank-streamsRanks available streams by quality, resolution, and language preferences
download-item.find-valid-torrentQueries debrid provider(s) for cached availability; triggers download if needed
request-stream-linkAcquires a play-ready stream URL and runs health-check loop
request-subtitlesPost-processing step: requests subtitles from subtitle plugins
post-process-media-itemOrchestrates post-download steps (stream link + subtitles)

​
Plugin Queues

Every plugin that registers a hook for a RivenEvent type gets its own dedicated queue named using the pattern: 
${eventType}.plugin[${pluginName}]
For example, a plugin named torrentio that handles riven.media-item.scrape.requested events will consume from the queue riven.media-item.scrape.requested.plugin[torrentio]. This isolation means a slow or failing plugin cannot block other plugins from processing the same event.

​
Sandboxed Worker Queues

Sandboxed workers run in isolated Node.js worker threads (useWorkerThreads: true) for CPU-heavy tasks that would otherwise block the event loop:
Queue nameTaskConcurrency
scrape-item.parse-scrape-resultsParse and normalise raw scraper responsesfloor(parallelism × 0.25)
download-item.map-items-to-filesMap torrent file trees to MediaItem entitiesfloor(parallelism × 0.75)
download-item.validate-torrent-filesValidate torrent file integrity and namingfloor(parallelism × 0.25)

​
Flow Orchestration

Riven uses BullMQ’s FlowProducer to build parent→child job trees. A parent job waits for all its children to complete before its own processor runs. This gives you multi-step pipelines with automatic dependency tracking and a single point of failure handling. 
// Example: process-media-item flow structure
//
// process-media-item (parent)
//   └── scrape-item (child)
//         └── scrape-item.parse-scrape-results (sandboxed child)
//               └── riven.media-item.scrape.requested.plugin[torrentio] (plugin child)
//               └── riven.media-item.scrape.requested.plugin[comet]     (plugin child)

// Each level waits for its children before proceeding.
// The parent processor reads child results via job.getChildrenValues().
The ProcessMediaItemFlow schema illustrates how steps are modelled as an enum on the job input — the processor checks job.data.step and re-enqueues with the next step value, turning a multi-step workflow into a linear state machine backed by the queue: 
// Steps for process-media-item
step: z.enum([
  "scrape",
  "validate-scrape",
  "download",
  "validate-download",
  "complete",
])

​
Deduplication

Jobs are assigned a deduplication.id option so that enqueuing the same logical work twice has no effect if the job is already waiting or active: 
// enqueue-process-media-item.ts
createProcessMediaItemJob(
  mediaItem.fullTitle,
  { step, mediaItem, isRootItem },
  {
    opts: {
      deduplication: {
        id: `process-${mediaItem.type}-${mediaItem.id}`,
      },
    },
  },
);

// enqueue-process-item-request.ts
opts: {
  deduplication: {
    id: `process-item-request-${item.id}`,
  },
},
Deduplication IDs are scoped to the queue. Calling clearDeduplicationJob(queueName, id) (used by the removeItemRequest mutation) removes the pending job and its deduplication record atomically.

​
Retry Strategy

​
Flow Workers

Standard flow workers use BullMQ’s default retry behaviour with 3 attempts. The scrape worker has a custom backoffStrategy tied to settings.scrapeCooldownHours: 
// scrape-item backoff tiers
backoffStrategy: (attemptsMade) => {
  const [after2, after5, after10] = settings.scrapeCooldownHours;

  if (attemptsMade >= 10) return Duration.fromObject({ hours: after10 }).as("milliseconds");
  if (attemptsMade >= 5)  return Duration.fromObject({ hours: after5 }).as("milliseconds");
  if (attemptsMade >= 2)  return Duration.fromObject({ hours: after2 }).as("milliseconds");

  return Duration.fromObject({ minutes: 30 }).as("milliseconds");
}

​
DataSource Workers (Plugin HTTP Clients)

BaseDataSource wraps every HTTP call in a BullMQ job. The retry strategy for HTTP requests uses these defaults:
SettingDefault
Attempts3
Base backoff delay10 000 ms
Non-fatal status codes408, 425, 429, 500, 502, 503, 504
Status codes in the non-fatal set are retried up to the attempt limit. Any other error (e.g. 404, 403) is treated as fatal and the job fails immediately without further retries.

​
Rate Limiting

BaseDataSource supports BullMQ’s built-in queue rate limiter via the optional rateLimiterOptions property. When an upstream API responds with HTTP 429, the DataSource calls queue.rateLimit(), which pauses the worker and re-queues the in-flight job to be retried after the duration specified in the Retry-After response header. 
// Plugin DataSource rate limiter example
class MyDataSource extends BaseDataSource<MySettings> {
  protected override readonly rateLimiterOptions = {
    max: 100,      // max 100 requests
    duration: 60_000, // per 60 seconds
  };
}

​
Concurrency

Flow worker concurrency defaults to floor(availableParallelism() × 1.5) — biased above the CPU count because flow workers are primarily I/O-bound (database queries, HTTP calls). Sandboxed workers use lower multipliers because they perform CPU-bound parsing and validation. 
// create-flow-worker.ts default
concurrency: normaliseConcurrency(os.availableParallelism() * 1.5),

// normaliseConcurrency clamps to at least 1
export function normaliseConcurrency(concurrency: number) {
  return Math.max(1, Math.floor(concurrency));
}
DataSource workers default to a concurrency of 200 because they only perform HTTP I/O. If you observe many timeout errors from a specific plugin, lower the DataSource concurrency property in that plugin’s implementation.

​
Job Retention

All flow workers share the same retention policy to prevent Redis memory growth: 
removeOnComplete: { count: 5000 },
removeOnFail: {
  age: 60 * 60 * 24, // keep failed jobs for 24 hours
  count: 5000,
},
Failed jobs are retained for 24 hours, giving you time to inspect them in the BullMQ dashboard before they are automatically pruned.

Build docs developers (and LLMs) love

Get started for freeTalk to us