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

CHANGELOG.md

@@ -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:

client.go

@@ -17,14 +17,13 @@ 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/workunit"
@@ -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
@@ -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
@@ -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,
@@ -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)")
@@ -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.
@@ -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 {
+	var (
+		configuredMiddleware = middlewareFromConfig(config)
+		allMiddleware        = append(rivermiddleware.DefaultMiddleware(), configuredMiddleware...)
+		allPlugins           = pluginlookup.NormalizePlugins(config.Hooks, allMiddleware, config.Plugins)
+	)
+
+	for _, plugin := range allPlugins {
+		if withBaseService, ok := plugin.(baseservice.WithBaseService); ok {
 			baseservice.Init(archetype, withBaseService)
 		}
 	}
@@ -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
@@ -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)
@@ -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,
@@ -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
@@ -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.
@@ -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,