Validation Utilities for Riven Plugin Development

The @repo/util-plugin-sdk ships a set of validation utilities that plug into the Riven startup sequence and into individual BullMQ job processors. These utilities cover three areas: startup validation (deciding whether a plugin should load at all), error classification (telling BullMQ whether a failed job should be retried), and Zod schema helpers for encoding, decoding, and refining structured data.

Startup validation flow

Every plugin exports a validator function. Riven calls it after parsing the plugin’s settings and before the plugin is considered ready to receive events:

scan env vars → settings.set(prefix, schema) → validator({ dataSources, settings })
  → (success) → settings.lock() → plugin registered
  → (false)   → retry after delay
  → (FatalValidationError) → plugin disabled, error logged

The validator receives a DataSourceMap (all DataSources already instantiated) and the PluginSettings instance. It must return Promise<boolean> (or the synchronous boolean — the schema accepts Promisable<boolean>):

import type { RivenPlugin } from '@repo/util-plugin-sdk';
import { TvdbAPI } from './datasource/tvdb.datasource.ts';

export const plugin: RivenPlugin = {
  // ...
  async validator({ dataSources }) {
    return dataSources.get(TvdbAPI).validate();
  },
};

Delegating to BaseDataSource.validate() is the standard pattern because the DataSource already has the settings object and the HTTP client ready to go.

FatalValidationError

import { FatalValidationError } from '@repo/util-plugin-sdk/errors/fatal-validation-error';

FatalValidationError extends Error and its name is "FatalValidationError". Throw it from either your plugin’s validator function or from BaseDataSource.validate() when the failure is permanent — when retrying will never succeed because the configuration is fundamentally wrong.

import { FatalValidationError } from '@repo/util-plugin-sdk/errors/fatal-validation-error';

public override async validate(): Promise<boolean> {
  try {
    await this.#getAuthToken();
    return true;
  } catch (error) {
    if (error instanceof AuthenticationError) {
      // Wrong API key — retrying won't help
      throw new FatalValidationError(
        'TVDB authentication failed — verify the TVDB_API_KEY setting.',
      );
    }
    // Transient network issue — let Riven retry
    this.logger.error('TVDB validation error', { err: error });
    return false;
  }
}

A FatalValidationError disables the plugin for the lifetime of the process. Only throw it when the misconfiguration cannot be corrected without a restart (e.g. the wrong API key was provided, or the URL points to a non-existent server).

UnrecoverableError

import { UnrecoverableError } from '@repo/util-plugin-sdk/errors/unrecoverable-error';

UnrecoverableError is re-exported from BullMQ via the SDK and used inside job processors (hooks and DataSource workers) to signal that this specific job should not be retried, even if retries are configured. The job moves directly to the failed state. BaseDataSource uses it internally when it cannot decode a request body:

throw new UnrecoverableError('Unable to decode non-string request body.');

Use it in your own hook handlers when the event payload is structurally invalid and re-processing the same data would produce the same failure:

import { UnrecoverableError } from '@repo/util-plugin-sdk/errors/unrecoverable-error';

export const myHook = async ({ event }) => {
  if (!event.item.imdbId && !event.item.tvdbId) {
    throw new UnrecoverableError(
      'Cannot index item without at least one external ID.',
    );
  }
  // ...
};

Zod codecs

`json(schema)`

Encodes/decodes between a raw JSON string and a fully-typed parsed object. The codec validates the decoded value against the provided Zod schema.

import { json } from '@repo/util-plugin-sdk/validation';
import z from 'zod';

const myCodec = json(z.object({ name: z.string(), count: z.number() }));

// Decode (parse JSON string → object)
const decoded = myCodec.decode('{"name":"Riven","count":42}');
// → { name: 'Riven', count: 42 }

// Encode (object → JSON string)
const encoded = myCodec.encode({ name: 'Riven', count: 42 });
// → '{"name":"Riven","count":42}'

json() is built on z.codec() from Zod, so it integrates naturally with other Zod schema compositions. If the input string is not valid JSON, a z.ZodError is thrown with code: "invalid_format" and format: "json".

`urlSearchParamsCodec`

A pre-built codec that encodes/decodes between a query-string (string) and a URLSearchParams instance.

import { urlSearchParamsCodec } from '@repo/util-plugin-sdk/validation';

// Decode (string → URLSearchParams)
const params = urlSearchParamsCodec.decode('page=1&limit=20');
// → URLSearchParams { 'page' => '1', 'limit' => '20' }

// Encode (URLSearchParams → string)
const str = urlSearchParamsCodec.encode(new URLSearchParams({ page: '2' }));
// → 'page=2'

BaseDataSource uses this codec internally to serialise request params for BullMQ job payloads (which must be JSON-serialisable) and to restore them before performing the actual HTTP fetch.

`atLeastOnePropertyRequired(obj, fields?)`

Returns true if at least one property of obj (optionally filtered to fields) is non-null and non-empty. “Empty” means null, undefined, "", 0, or [].

import { atLeastOnePropertyRequired } from '@repo/util-plugin-sdk/validation';

const ExternalIdsSchema = z.object({
  imdbId: z.string().nullish(),
  tvdbId: z.string().nullish(),
  tmdbId: z.string().nullish(),
}).refine(
  (val) => atLeastOnePropertyRequired(val, ['imdbId', 'tvdbId', 'tmdbId']),
  'At least one of imdbId, tvdbId, or tmdbId is required',
);

The optional fields array restricts which keys are checked. If fields is omitted, all keys in the object are evaluated.

// Passes — imdbId is present
ExternalIdsSchema.parse({ imdbId: 'tt1234567', tvdbId: null, tmdbId: null });

// Fails — all IDs are null/empty
ExternalIdsSchema.parse({ imdbId: null, tvdbId: null, tmdbId: null });
// → ZodError: At least one of imdbId, tvdbId, or tmdbId is required

`recordIsNotEmpty(obj)`

Returns true if the plain object has at least one key. Useful as a Zod .refine() predicate to reject empty result maps.

import { recordIsNotEmpty } from '@repo/util-plugin-sdk/validation';

const NonEmptyResultsSchema = z.record(z.string(), z.string()).refine(
  recordIsNotEmpty,
  'At least one result is required',
);

// Passes
NonEmptyResultsSchema.parse({ abc123: 'My Movie (2024)' });

// Fails
NonEmptyResultsSchema.parse({});
// → ZodError: At least one result is required

BaseDataSource.validate() best practices

validate() is the abstract method every BaseDataSource subclass must implement. It is called at startup and can be re-triggered via a GraphQL query. Follow these patterns:

Use the cheapest endpoint

Prefer a /health, /ping, or /status endpoint over a data endpoint. If no health endpoint exists, a lightweight read (e.g. fetching your own user profile) works well.

Return false for transient errors

Wrap the call in try/catch. Return false for network timeouts or 5xx responses — Riven will retry. This lets the plugin eventually come online after a brief API outage.

Throw FatalValidationError for auth failures

When the response clearly indicates bad credentials (401, 403, or a specific error body) throw FatalValidationError so the operator gets an immediate, actionable error message.

Cache auth tokens

If validation involves obtaining an auth token (like TVDB’s /login), cache the token as a private field. The validate() method can then double as the token-refresh path used by all subsequent requests.

Complete example: TVDB validate()

import { BaseDataSource } from '@repo/util-plugin-sdk';
import { FatalValidationError } from '@repo/util-plugin-sdk/errors/fatal-validation-error';
import { DateTime } from 'luxon';

interface TvdbToken {
  value: string;
  expiresAt: DateTime;
}

export class TvdbAPI extends BaseDataSource<TvdbSettings> {
  public override baseURL = 'https://api4.thetvdb.com/v4/';

  #token: TvdbToken | null = null;
  #inFlightLoginRequest: Promise<unknown> | null = null;

  async #getAuthToken(): Promise<TvdbToken> {
    const now = DateTime.utc();

    if (this.#token && this.#token.expiresAt > now) {
      return this.#token;
    }

    this.#inFlightLoginRequest ??= this.post<unknown>('login', {
      body: { apikey: this.settings.apiKey },
    });

    const response = await this.#inFlightLoginRequest.finally(() => {
      this.#inFlightLoginRequest = null;
    });

    const { data } = postLogin200Schema.parse(response);

    if (!data?.token) {
      throw new Error('No token in TVDB login response');
    }

    this.#token = {
      value: data.token,
      expiresAt: now.plus({ days: 25 }),
    };

    return this.#token;
  }

  public override async validate(): Promise<boolean> {
    try {
      await this.#getAuthToken();
      return true;
    } catch (error) {
      this.logger.error('TVDB validation error', { err: error });
      return false;
    }
  }
}

The token is cached for 25 days; all subsequent willSendRequest calls read this.#token.value without hitting /login again.

Overview

Building a Plugin

Reference

Validation Utilities for Riven Plugin Development

Startup validation flow

FatalValidationError

UnrecoverableError

Zod codecs

`json(schema)`

`urlSearchParamsCodec`

Zod refinement helpers

`atLeastOnePropertyRequired(obj, fields?)`

`recordIsNotEmpty(obj)`

BaseDataSource.validate() best practices

Use the cheapest endpoint

Return false for transient errors

Throw FatalValidationError for auth failures

Cache auth tokens

Complete example: TVDB validate()

Build docs developers (and LLMs) love

Overview

Building a Plugin

Reference

Documentation Index

​Startup validation flow

​FatalValidationError

​UnrecoverableError

​Zod codecs

​json(schema)

​urlSearchParamsCodec

​Zod refinement helpers

​atLeastOnePropertyRequired(obj, fields?)

​recordIsNotEmpty(obj)

​BaseDataSource.validate() best practices

Use the cheapest endpoint

Return false for transient errors

Throw FatalValidationError for auth failures

Cache auth tokens

​Complete example: TVDB validate()

Build docs developers (and LLMs) love

Startup validation flow

FatalValidationError

UnrecoverableError

Zod codecs

`json(schema)`

`urlSearchParamsCodec`

Zod refinement helpers

`atLeastOnePropertyRequired(obj, fields?)`

`recordIsNotEmpty(obj)`

BaseDataSource.validate() best practices

Complete example: TVDB validate()