Plugins

Most custom plugins should embed river.PluginDefaults, then implement one or more of River’s operation-specific hook or middleware interfaces:

type logPlugin struct {
    river.PluginDefaults
}

func (*logPlugin) InsertBegin(
    ctx context.Context,
    params *rivertype.JobInsertParams,
) error {
    fmt.Printf("inserting job with kind %q\n", params.Kind)
    return nil
}

func (*logPlugin) WorkBegin(ctx context.Context, job *rivertype.JobRow) error {
    fmt.Printf("working job with kind %q\n", job.Kind)
    return nil
}

PluginDefaults provides the marker methods that are required to implement rivertype.Plugin, and guarantees forward compatibility if the interface is extended in the future.

Install plugins globally with Config.Plugins:

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    // Order is significant.
    Plugins: []rivertype.Plugin{
        &logPlugin{},
        &timingPlugin{},
    },
})

River checks each plugin for all supported hook and middleware operation interfaces when the client is initialized. A plugin may implement any number of them, and will only be invoked for the operations it supports.

Order in Config.Plugins is significant. Hooks run in list order. Middleware earlier in the list wrap middleware that appear later, so the first applicable middleware is the first entered and last exited.

Define callback methods on a plugin object and add it to plugins. These callbacks log when insertion and work begin:

class LogPlugin
  def insert_begin(params)
    puts "Inserting #{params.kind}"
  end

  def work_begin(job)
    puts "Working #{job.kind}"
  end
end

config = River::Config.new(plugins: [LogPlugin.new])

River calls the lifecycle methods the object implements. Earlier plugins wrap later ones.

Implement the hook, then install it through a Plugin. The plugin can be added to the client builder as a single extension:

struct LogHook;

impl riverqueue::Hook for LogHook {
    async fn insert_begin(&self, insert: &mut riverqueue::InsertContext) -> Result<(), riverqueue::BoxError> {
        println!("Inserting {}", insert.kind);
        Ok(())
    }
}

struct LogPlugin;

impl riverqueue::Plugin for LogPlugin {
    fn install(&self, extensions: &mut riverqueue::Extensions) {
        extensions.hook(LogHook);
    }
}

let client = riverqueue::Client::builder(pool).plugin(LogPlugin).build()?;

Define a named RiverPlugin with the hooks you need, then add it to the client’s plugins array:

import type { RiverPlugin } from "riverqueue";

const logPlugin: RiverPlugin = {
  name: "log-jobs",
  hooks: {
    beforeInsert: ({ requests }) => {
      for (const request of requests) console.log("Inserting", request.kind);
    },

    beforeWork: ({ job }) => console.log("Working", job.kind),
  },
};

const client = new Client(driver, { plugins: [logPlugin] });

Hooks run at a specific point in River’s lifecycle and return immediately. The logPlugin above implements two hook interfaces: rivertype.HookInsertBegin and rivertype.HookWorkBegin.

An insert hook may modify the provided parameters before the job is inserted. A work hook may inspect the job row, but changes made to its context don’t propagate after the hook returns.

Hooks differ from middleware in that they’re invoked at a specific moment in time instead of keeping a frame on the stack during an operation. Because they keep the stack shallower, their use is recommended over middleware when either can do the job.

Implement callbacks such as insert_begin(params) or work_begin(job) on a plugin object. Insertion parameters may be changed before persistence. A hook runs at a single lifecycle point; use middleware when code must wrap execution.

Implement only the methods you need on Hook; other methods have no-op defaults. Hooks can return errors to fail the operation they observe. Register them directly with .hook(...), or through a plugin’s install method.

Put callbacks in the client’s hooks or a plugin’s hooks object. beforeWork runs before argument decoding and afterWork observes or replaces the result. Event and metric hooks observe runtime activity.

Available hook operation interfaces are:

• rivertype.HookInsertBegin: Invoked before each job is inserted.
• rivertype.HookMetricEmit: Receives metrics emitted by River. Implementations should avoid blocking because metrics are emitted in hot paths.
• rivertype.HookPeriodicJobsStart: Invoked before the periodic job enqueuer starts on a newly elected leader.
• rivertype.HookWorkBegin: Invoked before a job is worked.
• rivertype.HookWorkEnd: Invoked after a job is worked and may modify its returned error.

insert_begin(params), insert_end(result), work_begin(job), work_end(job, error), and job_finalize(job, state) are supported. See the Ruby plugin reference for their lifecycle and return values.

The Hook trait includes insertion, work, metrics, and periodic-job callbacks. Override the relevant method so the compiler checks its signature.

beforeInsert, afterInsert, beforeWork, afterWork, onEvent, onMetric, and onPeriodicJobsStart are available. See the observability guide.

Middleware wrap an operation and receive a doInner function that invokes the rest of the middleware stack and the underlying River operation. This lets middleware run code both before and after an operation, measure its duration, or pass a modified context to inner middleware and workers:

type timingPlugin struct {
    river.PluginDefaults
}

func (*timingPlugin) Work(
    ctx context.Context,
    job *rivertype.JobRow,
    doInner func(context.Context) error,
) error {
    start := time.Now()
    err := doInner(ctx)

    slog.InfoContext(ctx, "job worked",
        "kind", job.Kind,
        "duration", time.Since(start),
    )

    return err
}

Middleware must call doInner for the underlying operation to run. They may return early intentionally, but doing so prevents River and any inner middleware from handling the operation.

Implement work(job, operation) to wrap job execution. Call operation.call to continue, and use ensure for cleanup or timing even if work raises:

class TimingPlugin
  def work(job, operation)
    started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    operation.call
  ensure
    elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started
    puts "#{job.kind} took #{elapsed} seconds"
  end
end

Call operation.call to invoke the inner hooks and worker. Install the object in Config#plugins.

Implement WorkMiddleware and call next.run(job) to continue execution. This wrapper measures elapsed time and returns the worker’s result:

struct TimingMiddleware;

impl riverqueue::WorkMiddleware for TimingMiddleware {
    async fn work(
        &self,
        _context: &riverqueue::WorkContext,
        job: riverqueue::JobRow,
        next: riverqueue::WorkNext<'_>,
    ) -> Result<riverqueue::WorkOutcome, riverqueue::WorkError> {
        let started = std::time::Instant::now();
        let result = next.run(job).await;

        println!("Job took {:?}", started.elapsed());
        result
    }
}

let client = riverqueue::Client::builder(pool).work_middleware(TimingMiddleware).build()?;

Add a middleware function that awaits next() to run the handler. A finally block lets it record timing for both successful and failed attempts:

const client = new Client(driver, {
  middleware: [async ({ job }, next) => {
    const started = performance.now();

    try {
      return await next();
    } finally {
      console.log(job.kind, performance.now() - started);
    }
  }],
});

Call next() exactly once and return its outcome. Earlier middleware wraps later middleware.

Available middleware operation interfaces are:

• rivertype.JobInsertMiddleware: Invoked around a batch of jobs being inserted.
• rivertype.WorkerMiddleware: Invoked around a job being worked.

insert_many(params, operation) wraps a batch insertion, and work(job, operation) wraps one work attempt.

InsertMiddleware::insert_many wraps insertion; WorkMiddleware::work wraps an attempt. Register them on the client builder or in Plugin::install.

insertMiddleware wraps insertion and middleware wraps work. They can be supplied directly on the client or grouped in a named RiverPlugin.

Plugins installed through Config.Plugins apply globally. To install plugins only for a particular job kind, implement river.JobArgsWithPlugins on its job args:

type EmailArgs struct {
    To string `json:"to"`
}

func (EmailArgs) Kind() string { return "email" }

func (EmailArgs) Plugins() []rivertype.Plugin {
    return []rivertype.Plugin{
        &logPlugin{},
        &timingPlugin{},
    }
}

The plugins may implement hooks, middleware, or both. Their hooks run after globally installed hooks, and their middleware runs inside globally installed middleware.

River resolves Plugins once for each job kind using a zero-valued instance of its args, then caches the result. The returned plugins must depend only on the job type, not on fields like EmailArgs.To.

A job-specific rivertype.JobInsertMiddleware runs once around any insertion batch containing that job kind. It receives the complete batch, which may also contain other kinds of jobs.

Ruby plugins are configured on the client. To limit a callback to a kind, inspect job.kind or the insertion parameters and skip unrelated jobs. Middleware must still invoke operation.call for unrelated kinds.

Rust extensions are installed on the client. Filter by JobRow.kind or InsertContext.kind inside the extension when it should apply only to a particular kind, and continue the inner operation for other kinds.

Plugins are installed on the client. Filter by context.job.kind in work middleware or hooks. For unrelated jobs, middleware should immediately return next().

The base plugin interfaces are intentionally minimal, so a method with a subtly incorrect signature would compile but never be invoked. Add compile-time assertions for every operation interface a plugin is expected to implement:

var (
    _ river.JobArgsWithPlugins = EmailArgs{}

    _ rivertype.Plugin          = (*logPlugin)(nil)
    _ rivertype.HookInsertBegin = (*logPlugin)(nil)
    _ rivertype.HookWorkBegin   = (*logPlugin)(nil)

    _ rivertype.Plugin           = (*timingPlugin)(nil)
    _ rivertype.WorkerMiddleware = (*timingPlugin)(nil)
)

Ruby dispatches plugin methods dynamically. Exercise each callback through a real insertion or River::Testing.perform_job to catch misspelled names and incorrect arity.

Trait implementations are checked by the compiler. Implement the specific Hook, InsertMiddleware, or WorkMiddleware trait and test that the plugin installs the extension on the client.

Annotate plugin objects with RiverPlugin or use satisfies RiverPlugin so TypeScript checks callback signatures. Use a real client when testing execution of hooks and middleware; workOnce runs the handler alone.