Add tag filtering to JobListParams

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Added `JobListParams.TagsAll` and `JobListParams.TagsAny` for filtering jobs that match every or any exact tag, respectively. [PR #1339](https://github.com/riverqueue/river/pull/1339).
+
 ## [0.42.0] - 2026-07-31
 
 ### Added

client.go

@@ -2398,7 +2398,7 @@ func (c *Client[TTx]) jobDeleteMany(ctx context.Context, exec riverdriver.Execut
 		return nil, errors.New("delete with no filters not allowed to prevent accidental deletion of all jobs; either specify a predicate (e.g. JobDeleteManyParams.IDs, JobDeleteManyParams.Kinds, ...) or call JobDeleteManyParams.All")
 	}
 
-	listParams, err := dblist.JobMakeDriverParams(ctx, params.toDBParams(), c.driver.SQLFragmentColumnIn)
+	listParams, err := dblist.JobMakeDriverParams(ctx, params.toDBParams(), c.driver)
 	if err != nil {
 		return nil, err
 	}
@@ -2451,7 +2451,7 @@ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobL
 		return nil, err
 	}
 
-	listParams, err := dblist.JobMakeDriverParams(ctx, dbParams, c.driver.SQLFragmentColumnIn)
+	listParams, err := dblist.JobMakeDriverParams(ctx, dbParams, c.driver)
 	if err != nil {
 		return nil, err
 	}
@@ -2492,7 +2492,7 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara
 		return nil, err
 	}
 
-	listParams, err := dblist.JobMakeDriverParams(ctx, dbParams, c.driver.SQLFragmentColumnIn)
+	listParams, err := dblist.JobMakeDriverParams(ctx, dbParams, c.driver)
 	if err != nil {
 		return nil, err
 	}

insert_opts.go

@@ -65,9 +65,9 @@ type InsertOpts struct {
 	// JobArgsWithInsertOpts, however, it will work in both cases.
 	ScheduledAt time.Time
 
-	// Tags are an arbitrary list of keywords to add to the job. They have no
-	// functional behavior and are meant entirely as a user-specified construct
-	// to help group and categorize jobs.
+	// Tags are an arbitrary list of keywords to add to the job. They don't
+	// affect job execution, but can be used with JobListParams.TagsAll and
+	// JobListParams.TagsAny to group and filter jobs.
 	//
 	// Tags should conform to the regex `\A[\w][\w\-]+[\w]\z` and be a maximum
 	// of 255 characters long. No special characters are allowed.

internal/dblist/db_list.go

@@ -33,6 +33,8 @@ type JobListParams struct {
 	Queues     []string
 	Schema     string
 	States     []rivertype.JobState
+	TagsAll    []string
+	TagsAny    []string
 	Where      []WherePredicate
 }
 
@@ -41,15 +43,21 @@ type WherePredicate struct {
 	SQL       string
 }
 
+type sqlFragmentBuilder interface {
+	SQLFragmentColumnContainsAll(column, namedArg string, values []string) (string, any, error)
+	SQLFragmentColumnContainsAny(column, namedArg string, values []string) (string, any, error)
+	SQLFragmentColumnIn(column string, values any) (string, any, error)
+}
+
 // JobMakeDriverParams converts client-level parameters for job and delete to
 // driver-level parameters for use with an executor, which generally goes by
 // converting typed fields for IDs, kinds, queues, etc. to lower-level SQL.
 //
 // This was originally implemented for listing jobs, but since the logic is so
 // similar, it also performs the same function for JobDeleteMany. This works
-// because `riverdriver.JobListParams` is identical to `JobDeleteMany` and
-// therefore pointer-level converts to it.
-func JobMakeDriverParams(ctx context.Context, params *JobListParams, sqlFragmentColumnIn func(column string, values any) (string, any, error)) (*riverdriver.JobListParams, error) {
+// because `riverdriver.JobDeleteManyParams` has `JobListParams` as its
+// underlying type and therefore pointer-level converts to it.
+func JobMakeDriverParams(ctx context.Context, params *JobListParams, sqlFragmentBuilder sqlFragmentBuilder) (*riverdriver.JobListParams, error) {
 	var (
 		namedArgs    = make(map[string]any)
 		whereBuilder strings.Builder
@@ -75,7 +83,7 @@ func JobMakeDriverParams(ctx context.Context, params *JobListParams, sqlFragment
 		writeAndAfterFirst()
 
 		const column = "id"
-		sqlFragment, arg, err := sqlFragmentColumnIn(column, params.IDs)
+		sqlFragment, arg, err := sqlFragmentBuilder.SQLFragmentColumnIn(column, params.IDs)
 		if err != nil {
 			return nil, fmt.Errorf("error building SQL fragment for %q: %w", column, err)
 		}
@@ -87,7 +95,7 @@ func JobMakeDriverParams(ctx context.Context, params *JobListParams, sqlFragment
 		writeAndAfterFirst()
 
 		const column = "kind"
-		sqlFragment, arg, err := sqlFragmentColumnIn(column, params.Kinds)
+		sqlFragment, arg, err := sqlFragmentBuilder.SQLFragmentColumnIn(column, params.Kinds)
 		if err != nil {
 			return nil, fmt.Errorf("error building SQL fragment for %q: %w", column, err)
 		}
@@ -99,7 +107,7 @@ func JobMakeDriverParams(ctx context.Context, params *JobListParams, sqlFragment
 		writeAndAfterFirst()
 
 		const column = "priority"
-		sqlFragment, arg, err := sqlFragmentColumnIn(column, params.Priorities)
+		sqlFragment, arg, err := sqlFragmentBuilder.SQLFragmentColumnIn(column, params.Priorities)
 		if err != nil {
 			return nil, fmt.Errorf("error building SQL fragment for %q: %w", column, err)
 		}
@@ -111,7 +119,7 @@ func JobMakeDriverParams(ctx context.Context, params *JobListParams, sqlFragment
 		writeAndAfterFirst()
 
 		const column = "queue"
-		sqlFragment, arg, err := sqlFragmentColumnIn(column, params.Queues)
+		sqlFragment, arg, err := sqlFragmentBuilder.SQLFragmentColumnIn(column, params.Queues)
 		if err != nil {
 			return nil, fmt.Errorf("error building SQL fragment for %q: %w", column, err)
 		}
@@ -123,7 +131,7 @@ func JobMakeDriverParams(ctx context.Context, params *JobListParams, sqlFragment
 		writeAndAfterFirst()
 
 		const column = "state"
-		sqlFragment, arg, err := sqlFragmentColumnIn(column,
+		sqlFragment, arg, err := sqlFragmentBuilder.SQLFragmentColumnIn(column,
 			sliceutil.Map(params.States, func(v rivertype.JobState) string { return string(v) }))
 		if err != nil {
 			return nil, fmt.Errorf("error building SQL fragment for %q: %w", column, err)
@@ -132,6 +140,36 @@ func JobMakeDriverParams(ctx context.Context, params *JobListParams, sqlFragment
 		namedArgs[column] = arg
 	}
 
+	if len(params.TagsAll) > 0 {
+		writeAndAfterFirst()
+
+		const (
+			column   = "tags"
+			namedArg = "tags_all"
+		)
+		sqlFragment, arg, err := sqlFragmentBuilder.SQLFragmentColumnContainsAll(column, namedArg, params.TagsAll)
+		if err != nil {
+			return nil, fmt.Errorf("error building SQL fragment for %q: %w", namedArg, err)
+		}
+		whereBuilder.WriteString(sqlFragment)
+		namedArgs[namedArg] = arg
+	}
+
+	if len(params.TagsAny) > 0 {
+		writeAndAfterFirst()
+
+		const (
+			column   = "tags"
+			namedArg = "tags_any"
+		)
+		sqlFragment, arg, err := sqlFragmentBuilder.SQLFragmentColumnContainsAny(column, namedArg, params.TagsAny)
+		if err != nil {
+			return nil, fmt.Errorf("error building SQL fragment for %q: %w", namedArg, err)
+		}
+		whereBuilder.WriteString(sqlFragment)
+		namedArgs[namedArg] = arg
+	}
+
 	for _, where := range params.Where {
 		writeAndAfterFirst()
 

internal/dblist/db_list_test.go

@@ -45,7 +45,7 @@ func TestJobListNoJobs(t *testing.T) {
 			States:     []rivertype.JobState{rivertype.JobStateCompleted},
 			LimitCount: 1,
 			OrderBy:    []JobListOrderBy{{Expr: "id", Order: SortOrderAsc}},
-		}, bundle.driver.SQLFragmentColumnIn)
+		}, bundle.driver)
 		require.NoError(t, err)
 
 		_, err = bundle.exec.JobList(ctx, listParams)
@@ -64,7 +64,7 @@ func TestJobListNoJobs(t *testing.T) {
 			Where: []WherePredicate{
 				{NamedArgs: map[string]any{"foo": "bar"}, SQL: "queue = 'test' AND priority = 1 AND args->>'foo' = @foo"},
 			},
-		}, bundle.driver.SQLFragmentColumnIn)
+		}, bundle.driver)
 		require.NoError(t, err)
 
 		_, err = bundle.exec.JobList(ctx, listParams)
@@ -112,7 +112,7 @@ func TestJobListWithJobs(t *testing.T) {
 	execTest := func(ctx context.Context, t *testing.T, bundle *testBundle, params *JobListParams, testFunc testListFunc) {
 		t.Helper()
 
-		listParams, err := JobMakeDriverParams(ctx, params, bundle.driver.SQLFragmentColumnIn)
+		listParams, err := JobMakeDriverParams(ctx, params, bundle.driver)
 		require.NoError(t, err)
 
 		t.Logf("testing JobList in Executor")
@@ -326,7 +326,7 @@ func TestJobListWithJobs(t *testing.T) {
 			},
 		}
 
-		_, err := JobMakeDriverParams(ctx, params, bundle.driver.SQLFragmentColumnIn)
+		_, err := JobMakeDriverParams(ctx, params, bundle.driver)
 		require.EqualError(t, err, `expected "1" to contain named arg symbol @not_present`)
 	})
 
@@ -344,7 +344,7 @@ func TestJobListWithJobs(t *testing.T) {
 			},
 		}
 
-		_, err := JobMakeDriverParams(ctx, params, bundle.driver.SQLFragmentColumnIn)
+		_, err := JobMakeDriverParams(ctx, params, bundle.driver)
 		require.EqualError(t, err, "named argument @duplicate already registered")
 	})
 }

job_list_params.go

@@ -177,6 +177,8 @@ type JobListParams struct {
 	sortField      JobListOrderByField
 	sortOrder      SortOrder
 	states         []rivertype.JobState
+	tagsAll        []string
+	tagsAny        []string
 	where          []dblist.WherePredicate
 }
 
@@ -214,6 +216,8 @@ func (p *JobListParams) copy() *JobListParams {
 		sortOrder:      p.sortOrder,
 		schema:         p.schema,
 		states:         append([]rivertype.JobState(nil), p.states...),
+		tagsAll:        append([]string(nil), p.tagsAll...),
+		tagsAny:        append([]string(nil), p.tagsAny...),
 		where:          append([]dblist.WherePredicate(nil), p.where...),
 	}
 }
@@ -294,6 +298,8 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) {
 		Queues:     p.queues,
 		Schema:     p.schema,
 		States:     p.states,
+		TagsAll:    p.tagsAll,
+		TagsAny:    p.tagsAny,
 		Where:      p.where,
 	}, nil
 }
@@ -419,6 +425,32 @@ func (p *JobListParams) States(states ...rivertype.JobState) *JobListParams {
 	return paramsCopy
 }
 
+// TagsAll returns an updated filter set that will only return jobs containing
+// all of the given tags. Matching is exact and case-sensitive. TagsAll is
+// combined with TagsAny and all other filters using AND.
+//
+// Calling TagsAll replaces any tags supplied to a previous TagsAll call.
+// Calling it with no tags removes the filter.
+func (p *JobListParams) TagsAll(tags ...string) *JobListParams {
+	paramsCopy := p.copy()
+	paramsCopy.tagsAll = make([]string, len(tags))
+	copy(paramsCopy.tagsAll, tags)
+	return paramsCopy
+}
+
+// TagsAny returns an updated filter set that will only return jobs containing
+// at least one of the given tags. Matching is exact and case-sensitive.
+// TagsAny is combined with TagsAll and all other filters using AND.
+//
+// Calling TagsAny replaces any tags supplied to a previous TagsAny call.
+// Calling it with no tags removes the filter.
+func (p *JobListParams) TagsAny(tags ...string) *JobListParams {
+	paramsCopy := p.copy()
+	paramsCopy.tagsAny = make([]string, len(tags))
+	copy(paramsCopy.tagsAny, tags)
+	return paramsCopy
+}
+
 // NamedArgs are named arguments for use with JobListParams.Where. Keys should
 // look like "my_param", and map to parameters like "@my_param" in SQL queries.
 // "@" are present in the SQL, but not in the keys of this map.

job_list_params_test.go

@@ -243,4 +243,44 @@ func Test_JobListParams_toDBParams(t *testing.T) {
 			toDBParams()
 		require.EqualError(t, err, "cannot order by finalized_at without finalized state filters")
 	})
+
+	t.Run("TagsAll", func(t *testing.T) {
+		t.Parallel()
+
+		tags := []string{"alpha", "beta"}
+		params := NewJobListParams().TagsAll(tags...)
+		tags[0] = "modified"
+
+		dbParams, err := params.toDBParams()
+		require.NoError(t, err)
+		require.Equal(t, []string{"alpha", "beta"}, dbParams.TagsAll)
+
+		dbParams, err = params.TagsAll("gamma").toDBParams()
+		require.NoError(t, err)
+		require.Equal(t, []string{"gamma"}, dbParams.TagsAll)
+
+		dbParams, err = params.TagsAll().toDBParams()
+		require.NoError(t, err)
+		require.Empty(t, dbParams.TagsAll)
+	})
+
+	t.Run("TagsAny", func(t *testing.T) {
+		t.Parallel()
+
+		tags := []string{"alpha", "beta"}
+		params := NewJobListParams().TagsAny(tags...)
+		tags[0] = "modified"
+
+		dbParams, err := params.toDBParams()
+		require.NoError(t, err)
+		require.Equal(t, []string{"alpha", "beta"}, dbParams.TagsAny)
+
+		dbParams, err = params.TagsAny("gamma").toDBParams()
+		require.NoError(t, err)
+		require.Equal(t, []string{"gamma"}, dbParams.TagsAny)
+
+		dbParams, err = params.TagsAny().toDBParams()
+		require.NoError(t, err)
+		require.Empty(t, dbParams.TagsAny)
+	})
 }

riverdriver/river_driver_interface.go

@@ -128,6 +128,22 @@ type Driver[TTx any] interface {
 	// API is not stable. DO NOT USE.
 	PoolSet(dbPool any) error
 
+	// SQLFragmentColumnContainsAll generates an SQL fragment to be included as
+	// a predicate in a `WHERE` query for a collection column containing all of
+	// the given values. PostgreSQL uses array containment while SQLite compares
+	// values from a JSON array.
+	//
+	// API is not stable. DO NOT USE.
+	SQLFragmentColumnContainsAll(column, namedArg string, values []string) (string, any, error)
+
+	// SQLFragmentColumnContainsAny generates an SQL fragment to be included as
+	// a predicate in a `WHERE` query for a collection column containing at least
+	// one of the given values. PostgreSQL uses array overlap while SQLite
+	// compares values from a JSON array.
+	//
+	// API is not stable. DO NOT USE.
+	SQLFragmentColumnContainsAny(column, namedArg string, values []string) (string, any, error)
+
 	// SQLFragmentColumnIn generates an SQL fragment to be included as a
 	// predicate in a `WHERE` query for the existence of a set of values in a
 	// column like `id IN (...)`. The actual implementation depends on support
@@ -404,13 +420,7 @@ type JobDeleteBeforeParams struct {
 	Schema                      string
 }
 
-type JobDeleteManyParams struct {
-	Max           int32
-	NamedArgs     map[string]any
-	OrderByClause string
-	Schema        string
-	WhereClause   string
-}
+type JobDeleteManyParams JobListParams
 
 type JobGetAvailableParams struct {
 	ClientID       string

riverdriver/riverdatabasesql/river_database_sql_driver.go

@@ -82,6 +82,14 @@ func (d *Driver) GetMigrationTruncateTables(line string, version int) []string {
 func (d *Driver) PoolIsSet() bool          { return d.dbPool != nil }
 func (d *Driver) PoolSet(dbPool any) error { return riverdriver.ErrNotImplemented }
 
+func (d *Driver) SQLFragmentColumnContainsAll(column, namedArg string, values []string) (string, any, error) {
+	return fmt.Sprintf("%s @> @%s", column, namedArg), pq.Array(values), nil
+}
+
+func (d *Driver) SQLFragmentColumnContainsAny(column, namedArg string, values []string) (string, any, error) {
+	return fmt.Sprintf("%s && @%s", column, namedArg), pq.Array(values), nil
+}
+
 func (d *Driver) SQLFragmentColumnIn(column string, values any) (string, any, error) {
 	// Identical to the Pgx implementation except for use of `pg.Array`.
 	return fmt.Sprintf("%s = any(@%s)", column, column), pq.Array(values), nil