feat: add ability to filter jobs by tags

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Job list: filter jobs matching any of the selected exact tags. [PR #548](https://github.com/riverqueue/riverui/pull/548).
+
 ### Fixed
 
 - Job args: preserve large numeric JSON values exactly when displaying and copying args, while keeping object keys sorted. [Fixes #593](https://github.com/riverqueue/riverui/issues/593). [PR #594](https://github.com/riverqueue/riverui/pull/594).

go.mod

@@ -8,11 +8,11 @@ require (
 	github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/riverqueue/apiframe v0.0.0-20251229202423-2b52ce1c482e
-	github.com/riverqueue/river v0.41.1
-	github.com/riverqueue/river/riverdriver v0.41.1
-	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.1
-	github.com/riverqueue/river/rivershared v0.41.1
-	github.com/riverqueue/river/rivertype v0.41.1
+	github.com/riverqueue/river v0.42.1-0.20260803004224-dc39f530d6db
+	github.com/riverqueue/river/riverdriver v0.42.1-0.20260803004224-dc39f530d6db
+	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.42.1-0.20260803004224-dc39f530d6db
+	github.com/riverqueue/river/rivershared v0.42.1-0.20260803004224-dc39f530d6db
+	github.com/riverqueue/river/rivertype v0.42.1-0.20260803004224-dc39f530d6db
 	github.com/rs/cors v1.11.1
 	github.com/samber/slog-http v1.12.1
 	github.com/stretchr/testify v1.11.1

go.sum

@@ -35,16 +35,16 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/riverqueue/apiframe v0.0.0-20251229202423-2b52ce1c482e h1:OwOgxT3MRpOj5Mp6DhFdZP43FOQOf2hhywAuT5XZCR4=
 github.com/riverqueue/apiframe v0.0.0-20251229202423-2b52ce1c482e/go.mod h1:O7UmsAMjpMYuToN4au5GNXdmN1gli+5FTldgXqAfaD0=
-github.com/riverqueue/river v0.41.1 h1:Eat4tnIL16PTX1StVTwTAm2iKDowG99t2fDZPfwFPE8=
-github.com/riverqueue/river v0.41.1/go.mod h1:5KXJLKDPwjVC99ISXv2OHDQtNm1YwAeEAx9VYCd1RF0=
-github.com/riverqueue/river/riverdriver v0.41.1 h1:cwmmFCJkJlH6geaUTbOqC8vBUe00vTBm248GVD3HrqY=
-github.com/riverqueue/river/riverdriver v0.41.1/go.mod h1:nAZirY7z2clR4gaZxuGzvae8bwgycl9mrvb3+RWOehQ=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.1 h1:9VCL/tebPcMqrg9dwxQnL80/np4b2HnXwSwaICsszGM=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.1/go.mod h1:Ga+M+38X+cDMk6GsONljP3PrY5ylfBYMBn3MRj2+c9U=
-github.com/riverqueue/river/rivershared v0.41.1 h1:wHYBCXxCocvPhkH90B4fAUyMMWMVGG+/EFubyH15R2E=
-github.com/riverqueue/river/rivershared v0.41.1/go.mod h1:wQJhQ+jP7SILzEF3t7vrUQvuCMRL0sIhCU2DwjFfn/g=
-github.com/riverqueue/river/rivertype v0.41.1 h1:wBO2+nRfmpSohBRkC1/dxEkRyLS96FWlkUjukexOydc=
-github.com/riverqueue/river/rivertype v0.41.1/go.mod h1:D1Ad+EaZiaXbQbJcJcfeicXJMBKno0n6UcfKI5Q7DIQ=
+github.com/riverqueue/river v0.42.1-0.20260803004224-dc39f530d6db h1:a8TDxboPR2Yyf+k9ifyzlEvGOoYBeKQWu3r1HCKAQh8=
+github.com/riverqueue/river v0.42.1-0.20260803004224-dc39f530d6db/go.mod h1:pD+hDP0ZW3SbuTwh0CXDlYQ/M3Q7HDntT3Q4lhfYugY=
+github.com/riverqueue/river/riverdriver v0.42.1-0.20260803004224-dc39f530d6db h1:YmnE4JbUsi+XfvhkJD6IAbuiXmc/GB/0RzM7LEFd3SQ=
+github.com/riverqueue/river/riverdriver v0.42.1-0.20260803004224-dc39f530d6db/go.mod h1:b2IBlA29E3H233XwgbiJlezdoALSWhetZKt1LlBJEQU=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.42.1-0.20260803004224-dc39f530d6db h1:1aDHGtTlxeA9f0oE4/MsdGPXX18YaB5ngTDh/tb2eEQ=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.42.1-0.20260803004224-dc39f530d6db/go.mod h1:x+Yx1dcPLuriu8TqDHpAvZ5YQRJVnh9mxuxlqS0I6HA=
+github.com/riverqueue/river/rivershared v0.42.1-0.20260803004224-dc39f530d6db h1:DK/GXyMms5565iH7SSxSQ4I19tNa35cOdUjiqAUap9M=
+github.com/riverqueue/river/rivershared v0.42.1-0.20260803004224-dc39f530d6db/go.mod h1:EThAIEr49dlUQFhVLJcQGKoMlnPKOq+UxdMi9jevVsk=
+github.com/riverqueue/river/rivertype v0.42.1-0.20260803004224-dc39f530d6db h1:02uJDqbCs41J3IPL9YWdxLabiIpdxWGpt/AfAo+T1d8=
+github.com/riverqueue/river/rivertype v0.42.1-0.20260803004224-dc39f530d6db/go.mod h1:D1Ad+EaZiaXbQbJcJcfeicXJMBKno0n6UcfKI5Q7DIQ=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=

handler_api_endpoint.go

@@ -423,6 +423,7 @@ type jobListRequest struct {
 	Priorities []int16             `json:"-" validate:"omitempty,min=0,max=10"`                                                                      // from ExtractRaw
 	Queues     []string            `json:"-" validate:"omitempty,max=100"`                                                                           // from ExtractRaw
 	State      *rivertype.JobState `json:"-" validate:"omitempty,oneof=available cancelled completed discarded pending retryable running scheduled"` // from ExtractRaw
+	Tags       []string            `json:"-" validate:"omitempty,max=100"`                                                                           // from ExtractRaw
 }
 
 func (req *jobListRequest) ExtractRaw(r *http.Request) error {
@@ -467,6 +468,10 @@ func (req *jobListRequest) ExtractRaw(r *http.Request) error {
 		req.Queues = queues
 	}
 
+	if tags := r.URL.Query()["tags"]; len(tags) > 0 {
+		req.Tags = tags
+	}
+
 	return nil
 }
 
@@ -492,6 +497,10 @@ func (a *jobListEndpoint[TTx]) Execute(ctx context.Context, req *jobListRequest)
 			params = params.Queues(req.Queues...)
 		}
 
+		if len(req.Tags) > 0 {
+			params = params.TagsAny(req.Tags...)
+		}
+
 		if req.State == nil {
 			params = params.States(rivertype.JobStateRunning).OrderBy(river.JobListOrderByTime, river.SortOrderAsc)
 		} else {

handler_api_endpoint_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"log/slog"
 	"net/http"
+	"net/http/httptest"
 	"testing"
 	"time"
 
@@ -671,6 +672,33 @@ func TestAPIHandlerJobList(t *testing.T) {
 		require.Equal(t, job.ID, resp.Data[0].ID)
 	})
 
+	t.Run("FilterByTags", func(t *testing.T) {
+		t.Parallel()
+
+		endpoint, bundle := setupEndpoint(ctx, t, newJobListEndpoint)
+
+		job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{
+			State: ptrutil.Ptr(rivertype.JobStateRunning),
+			Tags:  []string{"alpha-tag", "shared"},
+		})
+		job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{
+			State: ptrutil.Ptr(rivertype.JobStateRunning),
+			Tags:  []string{"beta"},
+		})
+		_ = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{
+			State: ptrutil.Ptr(rivertype.JobStateRunning),
+			Tags:  []string{"ALPHA-TAG"},
+		})
+
+		resp, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &jobListRequest{
+			Tags: []string{"alpha-tag", "beta"},
+		})
+		require.NoError(t, err)
+		require.Len(t, resp.Data, 2)
+		require.Equal(t, job1.ID, resp.Data[0].ID)
+		require.Equal(t, job2.ID, resp.Data[1].ID)
+	})
+
 	t.Run("FilterByState", func(t *testing.T) {
 		t.Parallel()
 
@@ -713,6 +741,42 @@ func TestAPIHandlerJobList(t *testing.T) {
 	})
 }
 
+func TestAPIHandlerJobListCustomSchema(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	endpoint, bundle := setupEndpointWithCustomSchema(ctx, t, newJobListEndpoint)
+	jobParams := testfactory.Job_Build(t, &testfactory.JobOpts{
+		State: ptrutil.Ptr(rivertype.JobStateRunning),
+		Tags:  []string{"custom-schema-tag"},
+	})
+	jobParams.Schema = bundle.client.Schema()
+	job, err := bundle.exec.JobInsertFull(ctx, jobParams)
+	require.NoError(t, err)
+
+	resp, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &jobListRequest{
+		Tags: []string{"custom-schema-tag"},
+	})
+	require.NoError(t, err)
+	require.Len(t, resp.Data, 1)
+	require.Equal(t, job.ID, resp.Data[0].ID)
+}
+
+func TestJobListRequestExtractRaw(t *testing.T) {
+	t.Parallel()
+
+	req := httptest.NewRequestWithContext(
+		t.Context(),
+		http.MethodGet,
+		"/api/jobs?tags=ALPHA&tags=customer%3A123",
+		nil,
+	)
+	params := &jobListRequest{}
+
+	require.NoError(t, params.ExtractRaw(req))
+	require.Equal(t, []string{"ALPHA", "customer:123"}, params.Tags)
+}
+
 func TestAPIHandlerJobRetry(t *testing.T) {
 	t.Parallel()
 

src/components/job-search/parser.test.ts

@@ -12,12 +12,17 @@ import { JobFilterTypeID } from "./types";
 describe("parser", () => {
   describe("parseFiltersFromText", () => {
     it("parses simple filters", () => {
-      const result = parseFiltersFromText("kind:batch queue:priority");
-      expect(result).toHaveLength(2);
+      const result = parseFiltersFromText(
+        "kind:batch queue:priority tags:customer,urgent",
+      );
+      expect(result).toHaveLength(3);
       expect(result[0].match).toBe("kind:");
       expect(result[0].values).toEqual(["batch"]);
       expect(result[1].match).toBe("queue:");
       expect(result[1].values).toEqual(["priority"]);
+      expect(result[2].match).toBe("tags:");
+      expect(result[2].typeId).toBe(JobFilterTypeID.TAGS);
+      expect(result[2].values).toEqual(["customer", "urgent"]);
     });
 
     it("parses comma-separated values", () => {

src/components/job-search/types.ts

@@ -3,6 +3,7 @@ export enum JobFilterTypeID {
   KIND = "kind",
   PRIORITY = "priority",
   QUEUE = "queue",
+  TAGS = "tags",
 }
 
 export interface FilterType {
@@ -39,4 +40,9 @@ export const AVAILABLE_FILTERS: FilterType[] = [
     label: "queue",
     match: "queue:",
   },
+  {
+    id: JobFilterTypeID.TAGS,
+    label: "tags",
+    match: "tags:",
+  },
 ];

src/routes/jobs/index.schema.ts

@@ -49,4 +49,8 @@ export const jobSearchSchema = z.object({
     .optional()
     .transform((v) => (Array.isArray(v) ? v : v ? [v] : undefined)),
   state: z.nativeEnum(JobState).default(defaultValues.state),
+  tags: z
+    .union([z.string(), z.array(z.string().min(1))])
+    .optional()
+    .transform((v) => (Array.isArray(v) ? v : v ? [v] : undefined)),
 });