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.

Every Riven plugin that communicates with an external HTTP API does so through a class that extends BaseDataSource. Under the hood it layers a BullMQ job queue on top of Apollo’s RESTDataSource, giving every outgoing request automatic retry logic, exponential backoff, HTTP response caching via KeyvAdapter, and rate-limit awareness — without any extra boilerplate in your plugin code.

​
How BaseDataSource works

When your code calls this.get(...) or this.post(...), the base class:
  1. Checks the Keyv HTTP cache. If a valid cached response exists the request is served immediately, bypassing the queue entirely.
  2. Otherwise, enqueues a FetchJobInput job on the plugin’s dedicated BullMQ queue (<pluginSymbol>-<ClassName>-fetch-queue).
  3. The worker picks up the job (respecting concurrency and any rateLimiterOptions) and performs the real HTTP fetch via Apollo’s RESTDataSource.
  4. On a 429 response the worker calls queue.rateLimit(waitMs) and throws Worker.RateLimitError(), which pauses the queue until the Retry-After interval elapses.
  5. On any other retryable status code the job is retried up to requestAttempts times with exponential backoff (jitter ±50 %).
  6. Fatal status codes cause the job to be returned immediately with success: false instead of retrying.
The BullMQ worker is I/O-bound, which is why concurrency defaults to 200. If your API is latency-sensitive or frequently times out, lower this value.

​
Extending BaseDataSource

import { BaseDataSource } from '@repo/util-plugin-sdk';
import type { RateLimiterOptions } from '@repo/util-plugin-sdk';
import type { AugmentedRequest } from '@apollo/datasource-rest';

import type { MySettings } from '../my-settings.schema.ts';

export class MyAPI extends BaseDataSource<MySettings> {
  public override baseURL = this.settings.url;
  public override serviceName = 'MyService';

  // Optional: cap this datasource to 50 requests per second
  protected override rateLimiterOptions: RateLimiterOptions = {
    max: 50,
    duration: 1000,
  };

  // Attach auth headers to every outgoing request
  protected override async willSendRequest(
    _path: string,
    requestOpts: AugmentedRequest,
  ) {
    requestOpts.headers['x-api-key'] = this.settings.apiKey;
  }

  // Called at plugin startup — must return true to allow the plugin to load
  public override async validate(): Promise<boolean> {
    try {
      await this.get('health');
      return true;
    } catch {
      return false;
    }
  }

  // Business methods call this.get / this.post / etc.
  public async getItem(id: string): Promise<unknown> {
    return this.get(`items/${id}`);
  }
}
Keep DataSource classes thin — they are API wrappers, not business logic. Transform and validate responses with Zod schemas in the calling code or a dedicated transformer module.

​
The validate() method

validate() is one of two abstract members on BaseDataSource — along with baseURL. Riven calls it at plugin startup (and on explicit health-check queries) to confirm the plugin can reach its API. Return true to signal success, false to signal a transient failure that may be retried, or throw a FatalValidationError when the credentials are permanently wrong. The return type is Promisable<boolean> (from type-fest), so you may return a plain boolean or a Promise<boolean>. 
import { FatalValidationError } from '@repo/util-plugin-sdk/errors/fatal-validation-error';

public override async validate(): Promise<boolean> {
  try {
    await this.#getAuthToken(); // throws TvdbAPIError on bad key
    return true;
  } catch (error) {
    // Distinguish credential failures from transient network issues
    if (error instanceof TvdbAPIError) {
      throw new FatalValidationError(
        'TVDB authentication failed — check TVDB_API_KEY',
      );
    }
    this.logger.error('TVDB validation error', { err: error });
    return false;
  }
}

​
Configuration reference

BaseDataSource is instantiated by the Riven core automatically when your plugin is registered. You never call new MyAPI(config) yourself in production code. The config object type is:
​
settings
T extends Record<string, unknown>
required
The parsed, frozen settings object produced by PluginSettings.get(YourSchema). Available as this.settings inside the class.
​
pluginSymbol
symbol
required
The unique symbol from your pluginConfig.name. Used to namespace the BullMQ queue.
​
concurrency
number
default:"200"
Maximum number of in-flight HTTP requests the worker processes simultaneously. Override the protected readonly concurrency field on your subclass to change it.
​
requestAttempts
number
default:"3"
How many times a failing job is retried before being marked as failed.
​
requestBackoffDelay
number
default:"10000"
Base delay in milliseconds for exponential backoff between retries. Actual delay is delay * 2^attempt ± 50 % jitter.
​
connection
ConnectionOptions
required
BullMQ Redis connection options, injected by the framework.
​
logger
Logger
required
Winston logger instance, injected by the framework. Available as this.logger.

​
Status code behaviour

Retryable (non-fatal)

408, 425, 429, 500, 502, 503, 504 — the job is retried up to requestAttempts times using exponential backoff.

Fatal (no retry)

Any other 4xx or 5xx — the job returns success: false immediately without consuming any retry attempts.

​
Rate limiting

Set rateLimiterOptions on your subclass to engage BullMQ’s built-in rate limiter: 
protected override rateLimiterOptions: RateLimiterOptions = {
  max: 150,      // maximum 150 jobs …
  duration: 60_000, // … per 60 seconds
};
When the upstream API responds with 429 Too Many Requests, BaseDataSource automatically reads the Retry-After response header (as either seconds or an HTTP date), calls this.queue.rateLimit(waitMs), and throws Worker.RateLimitError(). The queue pauses and all pending jobs wait the correct duration before resuming.

​
HTTP caching

HTTP responses that include standard cache headers (Cache-Control, ETag, Last-Modified) are stored in Keyv automatically by the Apollo layer. On the next matching request BaseDataSource detects the cached entry and short-circuits the queue, returning the cached response without creating a BullMQ job at all.

​
BullMQ job context

Inside any event hook or DataSource method you can access the current BullMQ job and token via the dataSourceContext AsyncLocalStorage store: 
import { dataSourceContext } from '@repo/util-plugin-sdk/datasource/context';

// Inside a hook handler or datasource method:
const ctx = dataSourceContext.getStore(); // { job, token } | undefined
The Riven core populates this store before running each hook worker, so child jobs created by DataSource requests are automatically linked as flow children of the parent hook job.

​
Real-world example: CometAPI

The Comet scraper plugin shows a complete BaseDataSource subclass in production: 
import { BaseDataSource } from '@repo/util-plugin-sdk';
import type { RateLimiterOptions } from '@repo/util-plugin-sdk';
import type { CometSettings } from '../comet-settings.schema.ts';

export class CometAPI extends BaseDataSource<CometSettings> {
  public override baseURL = this.settings.url;
  public override serviceName = 'Comet';

  protected override rateLimiterOptions: RateLimiterOptions = {
    max: 150,
    duration: 60 * 1000,
  };

  public override async validate() {
    try {
      await this.get('validate');
      return true;
    } catch {
      return false;
    }
  }

  public async scrape({ item }: ParamsFor<MediaItemScrapeRequestedEvent>) {
    const { identifier, imdbId, scrapeType } =
      await this.#getCometScrapeConfig(item);

    const response = await this.get<unknown>(
      `/stream/${scrapeType}/${imdbId}${identifier ?? ''}.json`,
    );

    return CometScrapeResponse.parse(response);
  }
}
The CometSettings schema supplies url (the instance base URL), which becomes this.settings.url — set via the RIVEN_PLUGIN_SETTING__REPO_PLUGIN_COMET__url environment variable.

Build docs developers (and LLMs) love

Get started for freeTalk to us