Plugins

Plugins run custom code at predefined points in River's lifecycle. They're useful for shared behavior like logging, telemetry, modifying job parameters, and integrating River with other systems.

A plugin can implement hooks, middleware, or both. Hooks run at a specific point and return immediately, while middleware wrap an operation and stay on the stack for its duration.

Defining and installing a plugin

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.

Hooks

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.

Hook operations

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.

Middleware

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.

Middleware operations

Available middleware operation interfaces are:

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

Job-specific plugins

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.

Choosing between hooks and middleware

Hooks should generally be preferred when either type can perform the work. They're more granular and don't add a frame to the stack for the duration of an operation. Use middleware when code needs to wrap an operation, measure its duration, or propagate a modified context.

Testing interface compliance

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