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.

Use withPlugins to activate every hook and middleware interface a value implements. These interfaces all extend the common Plugin marker. A reusable plugin can group callbacks in one class:

import com.riverqueue.HookWorkEnd;
import com.riverqueue.WorkContext;
import com.riverqueue.WorkerMiddleware;

final class JobInstrumentation implements HookWorkEnd, WorkerMiddleware {
    @Override
    public void work(WorkContext<?> context, WorkContext.Step next) throws Exception {
        System.out.println("Working " + context.job().kind());
        next.run();
    }

    @Override
    public Throwable workEnd(WorkContext<?> context, Throwable failure) {
        if (failure != null) {
            System.getLogger("jobs").log(System.Logger.Level.ERROR,
                "Job " + context.job().id() + " failed", failure);
        }
        return failure;
    }
}

var client = new Client(database).withPlugins(new JobInstrumentation());

Individual hook and middleware interfaces also accept lambdas or method references, which can be passed together to withPlugins. Use withHooks or withMiddleware to activate only that side of an implementation.

All three methods return a new client with registrations appended. Use the returned client for insertion and worker configuration. Plugins run before separately registered hooks and middleware, regardless of which registration method was called first; each list preserves its own order. Repeated registrations invoke callbacks repeatedly. Instances may run concurrently, so protect shared mutable state.

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.

Register hooks with withPlugins, or use withHooks to activate only hook interfaces. Plugin hooks run first, followed by separately registered hooks, in each list’s registration order. Insertion and work hooks execute inside the innermost middleware. Work hooks receive decoded arguments; if HookWorkBegin throws, the handler and work-end hooks are skipped.

HookWorkEnd receives the handler’s failure, or null on success. Return the failure to preserve it, another failure to replace it, or null to report success. Each work-end hook receives the previous hook’s result; throwing skips the remaining work-end hooks.

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.

Java exposes four hook interfaces:

Use worker subscriptions for job events.

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.

Register a WorkerMiddleware through withPlugins to wrap an attempt, including its work hooks. Call next.run() once and let failures propagate:

import com.riverqueue.WorkerMiddleware;

WorkerMiddleware timeWork = (context, next) -> {
    long started = System.nanoTime();
    try {
        next.run();
    } finally {
        System.out.println(context.job().kind() + ": "
            + (System.nanoTime() - started) + " ns");
    }
};

var client = new Client(database).withPlugins(timeWork);

Plugin middleware wraps separately registered middleware. Within each list, the first registered middleware is outermost; code after next unwinds in reverse order. Use withMiddleware when only middleware interfaces should be activated.

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.

WorkerMiddleware wraps work. JobInsertMiddleware wraps a whole insertion batch; a single insertion is a one-element batch:

import com.riverqueue.JobInsertMiddleware;

JobInsertMiddleware logBatch = (connection, submissions, next) -> {
    var results = next.call();
    System.out.println("Inserted " + results.size() + " jobs");
    return results;
};

var client = new Client(database).withPlugins(logBatch);

submissions is an immutable list of the caller’s requests, before merging job-type defaults. Call next.call() exactly once and return its results in input order. Insert hooks and middleware run in the same transaction; throwing, even after next returns, rolls back the insertion and writes made through that connection during the operation.

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.

Plugins, hooks, and middleware are installed on the client. Filter work callbacks by context.job().kind() and insert hooks by type.kind(). In insertion middleware, inspect each submission’s type().kind(). Middleware must still invoke next for unrelated kinds.

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)
)

Assign lambdas to a specific hook or middleware interface so the compiler checks their signatures. For classes, implement the relevant interfaces and annotate methods with @Override. Use a real client and worker runtime to test execution of hooks and middleware.

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.