Skip to content

Cross-compatible hook/middleware + "plugins" which are allowed to be hook and middleware #1284

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
brandur wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Cross-compatible hook/middleware + "plugins" which are allowed to be hook and middleware #1284

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
6d60ff7
Cross-compatible hook/middleware + "plugins" which are allowed to be …
brandur Jun 12, 2026
000c152
Make `Plugin` a marker interface
brandur Jul 7, 2026
0e83736
Consolidate hook and middleware lookup into `pluginlookup`
brandur Jul 8, 2026
50adaf0
Bring `rivertest` more in line with unified hook/middleware path from…
brandur Jul 8, 2026
4c5579e
Code review feedback to allow legacy hooks/middlewares to work unchanged
brandur Jul 9, 2026
9873cfd
Cleaner use of internal plugins + plugin initialization
brandur Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Changed

- Both hooks and middleware should now be registered under a general `Config.Plugins` instead, a change that simpifies things somewhat, and better handles cases where a hook or middleware might be _both_ a hook and a middleware as opposed to only one or the other. `Config.Hooks` and `Config.Middleware` are still available for use, but considered deprecated in favor of the more general `Config.Plugins`. [PR #1284](https://github.com/riverqueue/river/pull/1284).
- Hooks passed to `Config.Hooks` that also implement middleware now have their middleware functionality activate automatically. The converse is also true for middleware sent to `Config.Middleware` that implement hooks. [PR #1284](https://github.com/riverqueue/river/pull/1284).

## [0.40.0] - 2026-07-02

⚠️ Version 0.40.0 contains a new database migration, version 7, that rolls up some database cleanups and a few SQLite features:
Expand Down
164 changes: 93 additions & 71 deletions client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,16 +17,15 @@ import (

"github.com/riverqueue/river/internal/dblist"
"github.com/riverqueue/river/internal/dbunique"
"github.com/riverqueue/river/internal/hooklookup"
"github.com/riverqueue/river/internal/jobcompleter"
"github.com/riverqueue/river/internal/jobexecutor"
"github.com/riverqueue/river/internal/leadership"
"github.com/riverqueue/river/internal/maintenance"
"github.com/riverqueue/river/internal/middlewarelookup"
"github.com/riverqueue/river/internal/notifier"
"github.com/riverqueue/river/internal/notifylimiter"
"github.com/riverqueue/river/internal/pluginlookup"
"github.com/riverqueue/river/internal/rivercommon"
"github.com/riverqueue/river/internal/rivermiddleware"
"github.com/riverqueue/river/internal/riverplugin"
"github.com/riverqueue/river/internal/workunit"
"github.com/riverqueue/river/riverdriver"
"github.com/riverqueue/river/rivershared/baseservice"
Expand Down Expand Up @@ -241,6 +240,11 @@ type Config struct {
// work hook runs and the insertion hooks on either side of it are skipped.
//
// Jobs may have their own specific hooks by implementing JobArgsWithHooks.
//
// If a type in Hooks also implements rivertype.Middleware, it will be
// installed as middleware too.
//
// Deprecated: Use Plugins instead.
Hooks []rivertype.Hook

// Logger is the structured logger to use for logging purposes. If none is
Expand Down Expand Up @@ -274,8 +278,27 @@ type Config struct {
// middlewares will run one after another, and the work middleware between
// them will not run. When a job is worked, the work middleware runs and the
// insertion middlewares on either side of it are skipped.
//
// If a type in Middleware also implements rivertype.Hook, it will be
// installed as a hook too.
//
// Deprecated: Use Plugins instead.
Middleware []rivertype.Middleware

// Plugins contains extensions installed globally as hooks, middleware, or
// both.
//
// A type qualifies as a plugin by implementing [rivertype.Plugin]. Most
// existing hook and middleware implementations already do this by embedding
// HookDefaults or MiddlewareDefaults. If a type participates on both sides,
// it may embed PluginDefaults, or embed both HookDefaults and
// MiddlewareDefaults directly and define its own IsPlugin method, then
// implement any operation-specific hook or middleware interfaces it needs.
//
// Hooks and Middleware are still supported for backward compatibility, but
// Plugins is the preferred place to register new extensions.
Plugins []rivertype.Plugin

// PeriodicJobs are a set of periodic jobs to run at the specified intervals
// in the client.
PeriodicJobs []*PeriodicJob
Expand Down Expand Up @@ -500,6 +523,7 @@ func (c *Config) WithDefaults() *Config {
MaxAttempts: cmp.Or(c.MaxAttempts, MaxAttemptsDefault),
Middleware: c.Middleware,
PeriodicJobs: c.PeriodicJobs,
Plugins: c.Plugins,
PollOnly: c.PollOnly,
Queues: c.Queues,
ReindexerIndexNames: reindexerIndexNames,
Expand All @@ -520,6 +544,35 @@ func (c *Config) WithDefaults() *Config {
}
}

func middlewareFromConfig(config *Config) []rivertype.Middleware {
middleware := make([]rivertype.Middleware, 0,
len(config.Middleware)+len(config.JobInsertMiddleware)+len(config.WorkerMiddleware))
middleware = append(middleware, config.Middleware...)

for _, jobInsertMiddleware := range config.JobInsertMiddleware {
middleware = append(middleware, jobInsertMiddleware)
}

outerLoop:
for _, workerMiddleware := range config.WorkerMiddleware {
// Don't add the middleware if it also implements JobInsertMiddleware
// and the instance has been added to config.JobInsertMiddleware. This
// is a hedge to make sure we don't accidentally double add middleware
// as we've converted over to the unified config.Middleware setting.
if workerMiddlewareAsJobInsertMiddleware, ok := workerMiddleware.(rivertype.JobInsertMiddleware); ok {
for _, jobInsertMiddleware := range config.JobInsertMiddleware {
if workerMiddlewareAsJobInsertMiddleware == jobInsertMiddleware {
continue outerLoop
}
}
}

middleware = append(middleware, workerMiddleware)
}

return middleware
}

func (c *Config) validate() error {
if c.CancelledJobRetentionPeriod < -1 {
return errors.New("CancelledJobRetentionPeriod time cannot be less than zero, except for -1 (infinite)")
Expand Down Expand Up @@ -678,28 +731,27 @@ type Client[TTx any] struct {
baseService baseservice.BaseService
baseStartStop startstop.BaseStartStop

clientNotifyBundle *ClientNotifyBundle[TTx]
completer jobcompleter.JobCompleter
config *Config
driver riverdriver.Driver[TTx]
elector *leadership.Elector
hookLookupByJob *hooklookup.JobHookLookup
hookLookupGlobal hooklookup.HookLookupInterface
insertNotifyLimiter *notifylimiter.Limiter
middlewareLookupGlobal middlewarelookup.MiddlewareLookupInterface
notifier *notifier.Notifier // may be nil in poll-only mode
periodicJobs *PeriodicJobBundle
pilot riverpilot.Pilot
producersByQueueName map[string]*producer
producersMu sync.RWMutex
queueMaintainer *maintenance.QueueMaintainer
queueMaintainerLeader *maintenance.QueueMaintainerLeader
queues *QueueBundle
services []startstop.Service
stopped <-chan struct{}
stuckJobCount atomic.Int32
subscriptionManager *subscriptionManager
testSignals clientTestSignals
clientNotifyBundle *ClientNotifyBundle[TTx]
completer jobcompleter.JobCompleter
config *Config
driver riverdriver.Driver[TTx]
elector *leadership.Elector
pluginLookupByJob *pluginlookup.JobPluginLookup
pluginLookupGlobal pluginlookup.PluginLookupInterface
insertNotifyLimiter *notifylimiter.Limiter
notifier *notifier.Notifier // may be nil in poll-only mode
periodicJobs *PeriodicJobBundle
pilot riverpilot.Pilot
producersByQueueName map[string]*producer
producersMu sync.RWMutex
queueMaintainer *maintenance.QueueMaintainer
queueMaintainerLeader *maintenance.QueueMaintainerLeader
queues *QueueBundle
services []startstop.Service
stopped <-chan struct{}
stuckJobCount atomic.Int32
subscriptionManager *subscriptionManager
testSignals clientTestSignals

// workCancel cancels the context used for all work goroutines. Normal Stop
// does not cancel that context.
Expand Down Expand Up @@ -802,8 +854,14 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
}
}

for _, hook := range config.Hooks {
if withBaseService, ok := hook.(baseservice.WithBaseService); ok {
allPlugins := pluginlookup.NormalizePlugins(
config.Hooks,
middlewareFromConfig(config),
append(riverplugin.DefaultPlugins(), config.Plugins...),
)

for _, plugin := range allPlugins {
if withBaseService, ok := plugin.(baseservice.WithBaseService); ok {
baseservice.Init(archetype, withBaseService)
}
}
Expand All @@ -815,8 +873,8 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
},
config: config,
driver: driver,
hookLookupByJob: hooklookup.NewJobHookLookup(),
hookLookupGlobal: hooklookup.NewHookLookup(config.Hooks),
pluginLookupByJob: pluginlookup.NewJobPluginLookup(),
pluginLookupGlobal: pluginlookup.NewPluginLookup(allPlugins),
producersByQueueName: make(map[string]*producer),
testSignals: clientTestSignals{},
workCancel: func(cause error) {}, // replaced on start, but here in case StopAndCancel is called before start up
Expand All @@ -834,41 +892,6 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
client.baseService.Name = "Client" // Have to correct the name because base service isn't embedded like it usually is
client.insertNotifyLimiter = notifylimiter.NewLimiter(archetype, config.FetchCooldown)

// Validation ensures that config.JobInsertMiddleware/WorkerMiddleware or
// the more abstract config.Middleware for middleware are set, but not both,
// so in practice we never append all three of these to each other.
{
middleware := rivermiddleware.DefaultMiddleware()
middleware = append(middleware, config.Middleware...)
for _, jobInsertMiddleware := range config.JobInsertMiddleware {
middleware = append(middleware, jobInsertMiddleware)
}
outerLoop:
for _, workerMiddleware := range config.WorkerMiddleware {
// Don't add the middleware if it also implements JobInsertMiddleware
// and the instance has been added to config.JobInsertMiddleware. This
// is a hedge to make sure we don't accidentally double add middleware
// as we've converted over to the unified config.Middleware setting.
if workerMiddlewareAsJobInsertMiddleware, ok := workerMiddleware.(rivertype.JobInsertMiddleware); ok {
for _, jobInsertMiddleware := range config.JobInsertMiddleware {
if workerMiddlewareAsJobInsertMiddleware == jobInsertMiddleware {
continue outerLoop
}
}
}

middleware = append(middleware, workerMiddleware)
}

for _, middleware := range middleware {
if withBaseService, ok := middleware.(baseservice.WithBaseService); ok {
baseservice.Init(archetype, withBaseService)
}
}

client.middlewareLookupGlobal = middlewarelookup.NewMiddlewareLookup(middleware)
}

pluginDriver, _ := driver.(driverPlugin[TTx])
if pluginDriver != nil {
pluginDriver.PluginInit(archetype)
Expand Down Expand Up @@ -996,7 +1019,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
{
periodicJobEnqueuer, err := maintenance.NewPeriodicJobEnqueuer(archetype, &maintenance.PeriodicJobEnqueuerConfig{
AdvisoryLockPrefix: config.AdvisoryLockPrefix,
HookLookupGlobal: client.hookLookupGlobal,
PluginLookupGlobal: client.pluginLookupGlobal,
Insert: client.insertMany,
Pilot: client.pilot,
Schema: config.Schema,
Expand Down Expand Up @@ -1976,8 +1999,8 @@ func (c *Client[TTx]) insertManyShared(
doInner := func(ctx context.Context) ([]*rivertype.JobInsertResult, error) {
for _, params := range insertParams {
for _, hook := range append(
c.hookLookupGlobal.ByHookKind(hooklookup.HookKindInsertBegin),
c.hookLookupByJob.ByJobArgs(params.Args).ByHookKind(hooklookup.HookKindInsertBegin)...,
c.pluginLookupGlobal.ByKind(pluginlookup.HookKindInsertBegin),
c.pluginLookupByJob.ByJobArgs(params.Args).ByKind(pluginlookup.HookKindInsertBegin)...,
) {
if err := hook.(rivertype.HookInsertBegin).InsertBegin(ctx, params); err != nil { //nolint:forcetypeassert
return nil, err
Expand Down Expand Up @@ -2008,7 +2031,7 @@ func (c *Client[TTx]) insertManyShared(
return insertResults, nil
}

jobInsertMiddleware := c.middlewareLookupGlobal.ByMiddlewareKind(middlewarelookup.MiddlewareKindJobInsert)
jobInsertMiddleware := c.pluginLookupGlobal.ByKind(pluginlookup.MiddlewareKindJobInsert)
if len(jobInsertMiddleware) > 0 {
// Wrap middlewares in reverse order so the one defined first is wrapped
// as the outermost function and is first to receive the operation.
Expand Down Expand Up @@ -2286,14 +2309,13 @@ func (c *Client[TTx]) producerAdd(queueName string, queueConfig QueueConfig) (*p
ErrorHandler: c.config.ErrorHandler,
FetchCooldown: cmp.Or(queueConfig.FetchCooldown, c.config.FetchCooldown),
FetchPollInterval: cmp.Or(queueConfig.FetchPollInterval, c.config.FetchPollInterval),
HookLookupByJob: c.hookLookupByJob,
HookLookupGlobal: c.hookLookupGlobal,
PluginLookupByJob: c.pluginLookupByJob,
PluginLookupGlobal: c.pluginLookupGlobal,
JobStuckHandler: c.config.JobStuckHandler,
JobStuckCount: &c.stuckJobCount,
JobStuckThreshold: c.config.JobStuckThreshold,
JobTimeout: c.config.JobTimeout,
MaxWorkers: queueConfig.MaxWorkers,
MiddlewareLookupGlobal: c.middlewareLookupGlobal,
Notifier: c.notifier,
Queue: queueName,
QueueEventCallback: c.subscriptionManager.distributeQueueEvent,
Expand Down
Loading
Loading