diff --git a/CHANGELOG.md b/CHANGELOG.md index 08615905..ccf6a658 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Add a Pro-only dead letter queue panel with ID, kind, queue, priority, and tag filters, dead letter autocomplete, bounded database queries, and More/Fewer controls to inspect discarded jobs and retry them as new jobs. [PR #711](https://github.com/riverqueue/riverui/pull/711). +- Add a Pro-only dead letter queue panel that lists discarded jobs newest first, with ID, kind, queue, priority, and tag filters, dead letter autocomplete, and More/Fewer controls to inspect jobs and retry them as new jobs. [PR #711](https://github.com/riverqueue/riverui/pull/711). ### Changed @@ -17,6 +17,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Show a "Page not found" page with a link back to Jobs for unknown URLs, instead of a blank page. [PR #711](https://github.com/riverqueue/riverui/pull/711). +- Keep a long job kind on one line in the job detail heading, with a tooltip and a copy button for the full kind. [PR #711](https://github.com/riverqueue/riverui/pull/711). +- Cancel job search suggestion requests that a newer keystroke has replaced. [PR #711](https://github.com/riverqueue/riverui/pull/711). - Clear filters and pagination when switching between Jobs and Dead Letters, while preserving them when switching job states within Jobs. [PR #711](https://github.com/riverqueue/riverui/pull/711). - Show queues whose concurrency metadata is missing, `null`, or `{}` as having no concurrency override, matching how River Pro interprets them, and stop showing a zero concurrency config for producers that report none. [PR #704](https://github.com/riverqueue/riverui/pull/704). - Reject negative queue concurrency `global_limit` and `local_limit` values with a 400 error instead of storing them. [PR #704](https://github.com/riverqueue/riverui/pull/704). diff --git a/internal/riveruidriver/river_ui_driver_interface.go b/internal/riveruidriver/river_ui_driver_interface.go index 97826751..0f694c62 100644 --- a/internal/riveruidriver/river_ui_driver_interface.go +++ b/internal/riveruidriver/river_ui_driver_interface.go @@ -5,7 +5,11 @@ package riveruidriver import ( + "cmp" "context" + "errors" + "fmt" + "slices" "time" "github.com/riverqueue/river/riverdriver" @@ -28,12 +32,41 @@ type Executor interface { // JobCountEstimate returns inexpensive database estimates where supported. // Drivers without an estimate strategy return an empty map. JobCountEstimate(ctx context.Context, params *JobCountEstimateParams) (map[rivertype.JobState]JobCountEstimateResult, error) - // JobDeadLetterList returns at most Max jobs, ordered by ID ascending. - // Only Pro endpoints call this query; it requires the dead letter table. + // JobDeadLetterList returns at most Max jobs, ordered by ID descending so + // the newest dead letter jobs come first and stay reachable when more jobs + // match than Max. Only Pro endpoints call this query; it requires the dead + // letter table. JobDeadLetterList(ctx context.Context, params *JobDeadLetterListParams) ([]*JobDeadLetterListResult, error) + // JobDeadLetterAutocomplete returns at most Max distinct values of the + // facet that start with Match and aren't in Exclude. Every driver compares + // and orders values the same way regardless of database collation: Match + // ignores the case of ASCII letters only, and values sort by their UTF-8 + // bytes. Only Pro endpoints call this query; it requires the dead letter + // table. JobDeadLetterAutocomplete(ctx context.Context, params *JobDeadLetterAutocompleteParams) ([]string, error) } +const ( + // JobDeadLetterListMaxDefault is the number of jobs to list when the caller + // has no preference. + JobDeadLetterListMaxDefault = 20 + // JobDeadLetterListMaxLimit is the largest Max that JobDeadLetterList + // accepts. + JobDeadLetterListMaxLimit = 200 + // JobDeadLetterAutocompleteMaxLimit is the largest Max that + // JobDeadLetterAutocomplete accepts. + JobDeadLetterAutocompleteMaxLimit = 100 +) + +// JobDeadLetterFacet is a job attribute that JobDeadLetterAutocomplete can +// suggest values for. The queries compare against these values literally. +type JobDeadLetterFacet string + +const ( + JobDeadLetterFacetJobKind JobDeadLetterFacet = "job_kind" + JobDeadLetterFacetQueueName JobDeadLetterFacet = "queue_name" +) + // JobDeadLetterFilters matches any value within each populated filter and // requires all populated filters to match. Tags match any exact tag. type JobDeadLetterFilters struct { @@ -46,20 +79,46 @@ type JobDeadLetterFilters struct { type JobDeadLetterAutocompleteParams struct { Exclude []string - Facet string + Facet JobDeadLetterFacet Match string Max int Schema string } +// Validate reports whether the params are ones that every driver accepts. +// Drivers call it before querying so they all reject the same inputs. +func (p *JobDeadLetterAutocompleteParams) Validate() error { + switch p.Facet { + case JobDeadLetterFacetJobKind, JobDeadLetterFacetQueueName: + default: + return errors.New("invalid dead letter autocomplete facet") + } + if p.Max < 1 || p.Max > JobDeadLetterAutocompleteMaxLimit { + return fmt.Errorf("dead letter autocomplete max must be between 1 and %d", JobDeadLetterAutocompleteMaxLimit) + } + return nil +} + type JobDeadLetterListParams struct { Filters JobDeadLetterFilters Max int Schema string } +// Validate reports whether the params are ones that every driver accepts. +// Drivers call it before querying so they all reject the same inputs. +func (p *JobDeadLetterListParams) Validate() error { + if p.Max < 1 || p.Max > JobDeadLetterListMaxLimit { + return fmt.Errorf("dead letter list max must be between 1 and %d", JobDeadLetterListMaxLimit) + } + return nil +} + // JobDeadLetterListResult contains only fields displayed in the list. Args, // metadata, and attempt history are loaded separately when opening a job. +// +// FinalizedAt is in UTC. Its precision is the one difference between drivers: +// PostgreSQL returns microseconds while SQLite returns milliseconds. type JobDeadLetterListResult struct { ID int64 `json:"id"` Attempt int `json:"attempt"` @@ -69,6 +128,20 @@ type JobDeadLetterListResult struct { Queue string `json:"queue"` } +// NormalizeJobDeadLetterListResults puts decoded jobs into the form that every +// driver returns. Jobs are ordered by ID descending, because not every database +// guarantees the order of elements within an aggregated JSON array. Timestamps +// are in UTC, because PostgreSQL renders them in the session's time zone while +// SQLite renders them in UTC. +func NormalizeJobDeadLetterListResults(jobs []*JobDeadLetterListResult) { + for _, job := range jobs { + if job.FinalizedAt != nil { + *job.FinalizedAt = job.FinalizedAt.UTC() + } + } + slices.SortFunc(jobs, func(a, b *JobDeadLetterListResult) int { return cmp.Compare(b.ID, a.ID) }) +} + type JobCountByAllStatesCappedParams struct { // Max must be positive and less than math.MaxInt32 so callers get the same // behavior from every database implementation. diff --git a/internal/riveruidriver/riveruidrivertest/job_dead_letter.go b/internal/riveruidriver/riveruidrivertest/job_dead_letter.go index 302cc9e3..2051906b 100644 --- a/internal/riveruidriver/riveruidrivertest/job_dead_letter.go +++ b/internal/riveruidriver/riveruidrivertest/job_dead_letter.go @@ -3,6 +3,7 @@ package riveruidrivertest import ( "context" "fmt" + "slices" "strings" "testing" "time" @@ -16,6 +17,21 @@ import ( "riverqueue.com/riverui/internal/riveruidriver" ) +// createJobDeadLetterTable creates the dead letter table in the test schema and +// returns the prefix that qualifies table names in it. OSS test schemas don't +// include Pro tables, so reproduce only the columns these queries need, +// preserving each database's timestamp column type. +func createJobDeadLetterTable(ctx context.Context, t *testing.T, executor riveruidriver.Executor, schema string) string { + t.Helper() + + prefix := "" + if schema != "" { + prefix = dbutil.SafeIdentifier(schema) + "." + } + require.NoError(t, executor.Exec(ctx, "CREATE TABLE "+prefix+"river_job_dead_letter AS SELECT id, attempt, finalized_at, kind, max_attempts, queue, priority, tags FROM "+prefix+"river_job WHERE 1 = 0")) + return prefix +} + func exerciseJobDeadLetterList( ctx context.Context, t *testing.T, @@ -31,43 +47,37 @@ func exerciseJobDeadLetterList( originalKind = "send_email" customerTag = "customer" emailKind = "archive_email" - kindFacet = "job_kind" + kindFacet = riveruidriver.JobDeadLetterFacetJobKind ) t.Run("JobDeadLetterList", func(t *testing.T) { t.Parallel() executor, schema := executorWithSchema(ctx, t, driver, executorWithTx) - prefix := "" - if schema != "" { - prefix = dbutil.SafeIdentifier(schema) + "." - } - // OSS test schemas don't include Pro tables. Reproduce only the columns - // this query needs, preserving each database's timestamp column type. - require.NoError(t, executor.Exec(ctx, "CREATE TABLE "+prefix+"river_job_dead_letter AS SELECT id, attempt, finalized_at, kind, max_attempts, queue, priority, tags FROM "+prefix+"river_job WHERE 1 = 0")) + prefix := createJobDeadLetterTable(ctx, t, executor, schema) testfactory.Job(ctx, t, executor, &testfactory.JobOpts{Schema: schema}) jobs, err := executor.JobDeadLetterList(ctx, &riveruidriver.JobDeadLetterListParams{Max: 20, Schema: schema}) require.NoError(t, err) require.Empty(t, jobs, "live jobs must not appear in the dead letter list") require.NotNil(t, jobs) - // Deliberately insert in reverse order, with IDs beyond JavaScript's + // Deliberately insert out of result order, with IDs beyond JavaScript's // exact integer range, and enough rows to exceed the maximum UI limit. const firstID int64 = 1970670598291982290 rows := make([]string, 0, 205) - for i := 204; i >= 0; i-- { + for i := range 205 { finalizedAt := "'2026-10-21 10:02:03.456+00:00'" - if i == 0 { + if i == 204 { finalizedAt = "NULL" } rows = append(rows, fmt.Sprintf("(%d, 3, %s, 'send_email', 5, 'mail')", firstID+int64(i), finalizedAt)) } require.NoError(t, executor.Exec(ctx, "INSERT INTO "+prefix+"river_job_dead_letter (id, attempt, finalized_at, kind, max_attempts, queue) VALUES "+strings.Join(rows, ","))) - for _, limit := range []int{1, 20, 40, 200} { + for _, limit := range []int{1, riveruidriver.JobDeadLetterListMaxDefault, 40, riveruidriver.JobDeadLetterListMaxLimit} { jobs, err := executor.JobDeadLetterList(ctx, &riveruidriver.JobDeadLetterListParams{Max: limit, Schema: schema}) require.NoError(t, err) require.Len(t, jobs, limit) for i, job := range jobs { - require.Equal(t, firstID+int64(i), job.ID) + require.Equal(t, firstID+204-int64(i), job.ID, "newest jobs must come first") require.Equal(t, 3, job.Attempt) require.Equal(t, 5, job.MaxAttempts) require.Equal(t, originalKind, job.Kind) @@ -77,21 +87,22 @@ func exerciseJobDeadLetterList( } else { require.NotNil(t, job.FinalizedAt) require.True(t, job.FinalizedAt.Equal(time.Date(2026, time.October, 21, 10, 2, 3, 456_000_000, time.UTC))) + require.Equal(t, time.UTC, job.FinalizedAt.Location(), "timestamps must be in UTC on every driver") } } } - for _, limit := range []int{-1, 0, 201} { + for _, limit := range []int{-1, 0, riveruidriver.JobDeadLetterListMaxLimit + 1} { _, err := executor.JobDeadLetterList(ctx, &riveruidriver.JobDeadLetterListParams{Max: limit, Schema: schema}) require.EqualError(t, err, "dead letter list max must be between 1 and 200") } // Filtering must happen before limiting, without rounding large IDs. jobs, err = executor.JobDeadLetterList(ctx, &riveruidriver.JobDeadLetterListParams{ - Filters: riveruidriver.JobDeadLetterFilters{IDs: []int64{firstID + 204}}, Max: 1, Schema: schema, + Filters: riveruidriver.JobDeadLetterFilters{IDs: []int64{firstID}}, Max: 1, Schema: schema, }) require.NoError(t, err) require.Len(t, jobs, 1) - require.Equal(t, firstID+204, jobs[0].ID) + require.Equal(t, firstID, jobs[0].ID) archive := func(kind, queue string, priority int, tags []string) int64 { job := testfactory.Job(ctx, t, executor, &testfactory.JobOpts{ @@ -108,16 +119,18 @@ func exerciseJobDeadLetterList( filters riveruidriver.JobDeadLetterFilters want []int64 }{ - {name: "IDs", filters: riveruidriver.JobDeadLetterFilters{IDs: []int64{second, first}}, want: []int64{first, second}}, - {name: "Kinds", filters: riveruidriver.JobDeadLetterFilters{Kinds: []string{emailKind, reportKind}}, want: []int64{first, second, third}}, + {name: "IDs", filters: riveruidriver.JobDeadLetterFilters{IDs: []int64{first, second}}, want: []int64{second, first}}, + {name: "IDsWithOtherFilters", filters: riveruidriver.JobDeadLetterFilters{IDs: []int64{first, second, third}, Queues: []string{mailQueue}}, want: []int64{third, first}}, + {name: "IDsNoMatches", filters: riveruidriver.JobDeadLetterFilters{IDs: []int64{firstID - 1}}, want: []int64{}}, + {name: "Kinds", filters: riveruidriver.JobDeadLetterFilters{Kinds: []string{emailKind, reportKind}}, want: []int64{third, second, first}}, {name: "Queues", filters: riveruidriver.JobDeadLetterFilters{Queues: []string{reportsQueue}}, want: []int64{second}}, - {name: "Priorities", filters: riveruidriver.JobDeadLetterFilters{Priorities: []int16{1, 3}}, want: []int64{first, second}}, - {name: "TagsMatchAny", filters: riveruidriver.JobDeadLetterFilters{Tags: []string{urgentTag, "batch"}}, want: []int64{first, second}}, + {name: "Priorities", filters: riveruidriver.JobDeadLetterFilters{Priorities: []int16{1, 3}}, want: []int64{second, first}}, + {name: "TagsMatchAny", filters: riveruidriver.JobDeadLetterFilters{Tags: []string{urgentTag, "batch"}}, want: []int64{second, first}}, {name: "TagsAreExact", filters: riveruidriver.JobDeadLetterFilters{Tags: []string{urgentTag}}, want: []int64{first}}, {name: "Combined", filters: riveruidriver.JobDeadLetterFilters{Kinds: []string{emailKind, reportKind}, Queues: []string{mailQueue}, Priorities: []int16{1, 2}, Tags: []string{customerTag}}, want: []int64{first}}, {name: "NoMatches", filters: riveruidriver.JobDeadLetterFilters{Kinds: []string{"missing"}}, want: []int64{}}, {name: "LiteralValues", filters: riveruidriver.JobDeadLetterFilters{Tags: []string{"urgent' OR 1=1 --"}}, want: []int64{}}, - {name: "EmptyFilters", filters: riveruidriver.JobDeadLetterFilters{IDs: []int64{}, Kinds: []string{}, Queues: []string{}, Tags: []string{}, Priorities: []int16{}}, want: []int64{first, second, third}}, + {name: "EmptyFilters", filters: riveruidriver.JobDeadLetterFilters{IDs: []int64{}, Kinds: []string{}, Queues: []string{}, Tags: []string{}, Priorities: []int16{}}, want: []int64{firstID + 204, firstID + 203, firstID + 202}}, } { t.Run(tt.name, func(t *testing.T) { jobs, err := executor.JobDeadLetterList(ctx, &riveruidriver.JobDeadLetterListParams{Filters: tt.filters, Max: 3, Schema: schema}) @@ -132,7 +145,7 @@ func exerciseJobDeadLetterList( for _, tt := range []struct { name string - facet string + facet riveruidriver.JobDeadLetterFacet match string exclude []string max int @@ -142,7 +155,7 @@ func exerciseJobDeadLetterList( {name: "PrefixCaseInsensitive", facet: kindFacet, match: "ARCHIVE_", max: 100, want: []string{emailKind, reportKind}}, {name: "Exclude", facet: kindFacet, match: "archive_", exclude: []string{reportKind}, max: 100, want: []string{emailKind}}, {name: "Limit", facet: kindFacet, max: 1, want: []string{emailKind}}, - {name: "Queues", facet: "queue_name", max: 100, want: []string{mailQueue, reportsQueue}}, + {name: "Queues", facet: riveruidriver.JobDeadLetterFacetQueueName, max: 100, want: []string{mailQueue, reportsQueue}}, {name: "NoLiveJobs", facet: kindFacet, match: "fake_job", max: 100, want: []string{}}, {name: "LiteralPrefix", facet: kindFacet, match: "%", max: 100, want: []string{}}, } { @@ -154,3 +167,96 @@ func exerciseJobDeadLetterList( } }) } + +// exerciseJobDeadLetterAutocomplete proves that every driver matches and orders +// suggestions the same way for text where database collations disagree: +// mixed-case ASCII and non-ASCII letters. +func exerciseJobDeadLetterAutocomplete( + ctx context.Context, + t *testing.T, + databaseName string, + driver riveruidriver.Driver, + executorWithTx func(ctx context.Context, t *testing.T) (riverdriver.Executor, string), +) { + t.Helper() + const ( + kindFacet = riveruidriver.JobDeadLetterFacetJobKind + queueFacet = riveruidriver.JobDeadLetterFacetQueueName + ) + t.Run("JobDeadLetterAutocomplete", func(t *testing.T) { + t.Parallel() + executor, schema := executorWithSchema(ctx, t, driver, executorWithTx) + prefix := createJobDeadLetterTable(ctx, t, executor, schema) + + if databaseName == riverdriver.DatabaseNamePostgres { + // Give the columns a linguistic collation where one is available so + // the query is shown not to depend on the collation of the database + // or its columns. Unlike "C", this collation folds non-ASCII letters + // in lower() and sorts letters of both cases together. + var hasICU bool + require.NoError(t, executor.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM pg_collation WHERE collname = 'und-x-icu')").Scan(&hasICU)) + if hasICU { + require.NoError(t, executor.Exec(ctx, "ALTER TABLE "+prefix+`river_job_dead_letter ALTER COLUMN kind TYPE text COLLATE "und-x-icu", ALTER COLUMN queue TYPE text COLLATE "und-x-icu"`)) + } + } + + // Listed in the byte order that every driver must return. A linguistic + // collation would instead return alpha, école, École, zebra, Zeta. + const ( + zeta = "Zeta" + alpha = "alpha" + zebra = "zebra" + ecoleUpper = "École" + ecoleLower = "école" + ) + kinds := []string{zeta, alpha, zebra, ecoleUpper, ecoleLower} + rows := make([]string, 0, len(kinds)) + for i, kind := range slices.Backward(kinds) { + rows = append(rows, fmt.Sprintf("(%d, 3, NULL, '%s', 5, '%s')", i+1, kind, strings.ToUpper(kind))) + } + require.NoError(t, executor.Exec(ctx, "INSERT INTO "+prefix+"river_job_dead_letter (id, attempt, finalized_at, kind, max_attempts, queue) VALUES "+strings.Join(rows, ","))) + + for _, tt := range []struct { + name string + facet riveruidriver.JobDeadLetterFacet + match string + exclude []string + max int + want []string + }{ + {name: "ByteOrder", facet: kindFacet, max: 100, want: kinds}, + {name: "ByteOrderQueues", facet: queueFacet, max: 100, want: []string{"ALPHA", "ZEBRA", "ZETA", "ÉCOLE"}}, + {name: "LimitFollowsByteOrder", facet: kindFacet, max: 2, want: []string{zeta, alpha}}, + {name: "MixedCaseMatch", facet: kindFacet, match: "zE", max: 100, want: []string{zeta, zebra}}, + {name: "MixedCaseMatchUpper", facet: kindFacet, match: "ALPHA", max: 100, want: []string{alpha}}, + {name: "MixedCaseMatchQueues", facet: queueFacet, match: "zeb", max: 100, want: []string{"ZEBRA"}}, + {name: "NonASCIILower", facet: kindFacet, match: "é", max: 100, want: []string{ecoleLower}}, + {name: "NonASCIIUpper", facet: kindFacet, match: "É", max: 100, want: []string{ecoleUpper}}, + {name: "NonASCIIFoldsOnlyASCII", facet: kindFacet, match: "éCOLE", max: 100, want: []string{ecoleLower}}, + {name: "NonASCIIFoldsOnlyASCIIQueues", facet: queueFacet, match: "éc", max: 100, want: []string{}}, + {name: "NonASCIIExclude", facet: kindFacet, exclude: []string{ecoleUpper, "ZETA"}, max: 100, want: []string{zeta, alpha, zebra, ecoleLower}}, + } { + t.Run(tt.name, func(t *testing.T) { + values, err := executor.JobDeadLetterAutocomplete(ctx, &riveruidriver.JobDeadLetterAutocompleteParams{Facet: tt.facet, Match: tt.match, Exclude: tt.exclude, Max: tt.max, Schema: schema}) + require.NoError(t, err) + require.Equal(t, tt.want, values) + }) + } + + // Drivers guard their own inputs rather than relying on the caller's + // validation, and reject them before running a query. + for _, facet := range []riveruidriver.JobDeadLetterFacet{"", "tags", "kind", "JOB_KIND"} { + _, err := executor.JobDeadLetterAutocomplete(ctx, &riveruidriver.JobDeadLetterAutocompleteParams{Facet: facet, Max: 100, Schema: schema}) + require.EqualError(t, err, "invalid dead letter autocomplete facet", "facet %q", facet) + } + for _, limit := range []int{-1, 0, riveruidriver.JobDeadLetterAutocompleteMaxLimit + 1} { + _, err := executor.JobDeadLetterAutocomplete(ctx, &riveruidriver.JobDeadLetterAutocompleteParams{Facet: kindFacet, Max: limit, Schema: schema}) + require.EqualError(t, err, "dead letter autocomplete max must be between 1 and 100", "max %d", limit) + } + for _, limit := range []int{1, riveruidriver.JobDeadLetterAutocompleteMaxLimit} { + values, err := executor.JobDeadLetterAutocomplete(ctx, &riveruidriver.JobDeadLetterAutocompleteParams{Facet: kindFacet, Max: limit, Schema: schema}) + require.NoError(t, err) + require.Len(t, values, min(limit, len(kinds))) + } + }) +} diff --git a/internal/riveruidriver/riveruidrivertest/riveruidrivertest.go b/internal/riveruidriver/riveruidrivertest/riveruidrivertest.go index 1fbb6632..e58b7386 100644 --- a/internal/riveruidriver/riveruidrivertest/riveruidrivertest.go +++ b/internal/riveruidriver/riveruidrivertest/riveruidrivertest.go @@ -29,6 +29,7 @@ func Exercise( exerciseJobCountByAllStatesCapped(ctx, t, driver, executorWithTx) exerciseJobCountEstimate(ctx, t, databaseName, driver, executorWithTx) exerciseJobDeadLetterList(ctx, t, driver, executorWithTx) + exerciseJobDeadLetterAutocomplete(ctx, t, databaseName, driver, executorWithTx) } func exerciseGetExecutor( diff --git a/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_job_dead_letter.sql b/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_job_dead_letter.sql index f9f2bf5b..5c43ab08 100644 --- a/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_job_dead_letter.sql +++ b/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_job_dead_letter.sql @@ -12,9 +12,12 @@ CREATE TABLE river_job_dead_letter ( ); -- name: JobDeadLetterList :one --- Apply the limit before aggregation so neither the database nor the browser --- materializes the entire queue. River's common executor exposes QueryRow, --- so return the bounded set as one JSON value. +-- Return the newest jobs first. Apply the limit before aggregation so neither +-- the database nor the browser materializes the entire queue. River's common +-- executor exposes QueryRow, so return the bounded set as one JSON value. +-- +-- This query ignores the filter's IDs; the driver sends ID lookups to +-- JobDeadLetterListByIDs instead. SELECT coalesce(json_agg(json_build_object( 'id', limited.id, 'attempt', limited.attempt, @@ -22,27 +25,60 @@ SELECT coalesce(json_agg(json_build_object( 'kind', limited.kind, 'max_attempts', limited.max_attempts, 'queue', limited.queue -) ORDER BY limited.id), '[]'::json)::text AS jobs +) ORDER BY limited.id DESC), '[]'::json)::text AS jobs FROM ( SELECT id, attempt, finalized_at, kind, max_attempts, queue FROM /* TEMPLATE: schema */river_job_dead_letter - WHERE (NOT jsonb_exists(@filters::jsonb, 'ids') OR id IN (SELECT value::bigint FROM jsonb_array_elements_text(@filters::jsonb -> 'ids'))) + WHERE (NOT jsonb_exists(@filters::jsonb, 'kinds') OR kind IN (SELECT value::text FROM jsonb_array_elements_text(@filters::jsonb -> 'kinds'))) + AND (NOT jsonb_exists(@filters::jsonb, 'priorities') OR priority IN (SELECT value::smallint FROM jsonb_array_elements_text(@filters::jsonb -> 'priorities'))) + AND (NOT jsonb_exists(@filters::jsonb, 'queues') OR queue IN (SELECT value::text FROM jsonb_array_elements_text(@filters::jsonb -> 'queues'))) + AND (NOT jsonb_exists(@filters::jsonb, 'tags') OR tags && ARRAY(SELECT jsonb_array_elements_text(@filters::jsonb -> 'tags'))::varchar[]) + ORDER BY id DESC + LIMIT @max::int +) AS limited; + +-- name: JobDeadLetterListByIDs :one +-- JobDeadLetterList for filters that include IDs. The ID predicate must stand +-- alone as a top-level condition: written as "no IDs requested OR id IN (...)" +-- the planner can't turn it into an index condition and reads the whole +-- table. Collapsing the requested IDs to an array lets it probe the primary +-- key once per ID, whether the statement runs with a custom or generic plan. +SELECT coalesce(json_agg(json_build_object( + 'id', limited.id, + 'attempt', limited.attempt, + 'finalized_at', limited.finalized_at, + 'kind', limited.kind, + 'max_attempts', limited.max_attempts, + 'queue', limited.queue +) ORDER BY limited.id DESC), '[]'::json)::text AS jobs +FROM ( + SELECT id, attempt, finalized_at, kind, max_attempts, queue + FROM /* TEMPLATE: schema */river_job_dead_letter + WHERE id = ANY (ARRAY(SELECT value::bigint FROM jsonb_array_elements_text(@filters::jsonb -> 'ids'))) AND (NOT jsonb_exists(@filters::jsonb, 'kinds') OR kind IN (SELECT value::text FROM jsonb_array_elements_text(@filters::jsonb -> 'kinds'))) AND (NOT jsonb_exists(@filters::jsonb, 'priorities') OR priority IN (SELECT value::smallint FROM jsonb_array_elements_text(@filters::jsonb -> 'priorities'))) AND (NOT jsonb_exists(@filters::jsonb, 'queues') OR queue IN (SELECT value::text FROM jsonb_array_elements_text(@filters::jsonb -> 'queues'))) AND (NOT jsonb_exists(@filters::jsonb, 'tags') OR tags && ARRAY(SELECT jsonb_array_elements_text(@filters::jsonb -> 'tags'))::varchar[]) - ORDER BY id + ORDER BY id DESC LIMIT @max::int ) AS limited; -- name: JobDeadLetterAutocomplete :one -- Suggest values from the archive, including kinds and queues with no live jobs. -SELECT coalesce(json_agg(suggestions.value ORDER BY suggestions.value), '[]'::json)::text AS values +-- +-- The "C" collation makes matching and ordering independent of the database's +-- collation and identical to SQLite: lower() folds only ASCII letters, and +-- values sort by their bytes. The ordering matters before the limit as well, +-- because it decides which suggestions are returned. +SELECT coalesce(json_agg(suggestions.value ORDER BY suggestions.value COLLATE "C"), '[]'::json)::text AS values FROM ( - SELECT DISTINCT CASE WHEN @facet::text = 'job_kind' THEN kind ELSE queue END AS value - FROM /* TEMPLATE: schema */river_job_dead_letter - WHERE left(lower(CASE WHEN @facet::text = 'job_kind' THEN kind ELSE queue END), length(@match::text)) = lower(@match::text) - AND (CASE WHEN @facet::text = 'job_kind' THEN kind ELSE queue END) NOT IN (SELECT jsonb_array_elements_text(@exclude::jsonb)) - ORDER BY value + SELECT DISTINCT candidates.value + FROM ( + SELECT (CASE WHEN @facet::text = 'job_kind' THEN kind ELSE queue END) COLLATE "C" AS value + FROM /* TEMPLATE: schema */river_job_dead_letter + ) AS candidates + WHERE left(lower(candidates.value COLLATE "C"), length(@match::text)) = lower((@match::text) COLLATE "C") + AND candidates.value NOT IN (SELECT excluded.value COLLATE "C" FROM jsonb_array_elements_text(@exclude::jsonb) AS excluded(value)) + ORDER BY candidates.value LIMIT @max::int ) AS suggestions; diff --git a/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_job_dead_letter.sql.go b/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_job_dead_letter.sql.go index add724a0..7451eb97 100644 --- a/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_job_dead_letter.sql.go +++ b/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_job_dead_letter.sql.go @@ -10,13 +10,16 @@ import ( ) const jobDeadLetterAutocomplete = `-- name: JobDeadLetterAutocomplete :one -SELECT coalesce(json_agg(suggestions.value ORDER BY suggestions.value), '[]'::json)::text AS values +SELECT coalesce(json_agg(suggestions.value ORDER BY suggestions.value COLLATE "C"), '[]'::json)::text AS values FROM ( - SELECT DISTINCT CASE WHEN $1::text = 'job_kind' THEN kind ELSE queue END AS value - FROM /* TEMPLATE: schema */river_job_dead_letter - WHERE left(lower(CASE WHEN $1::text = 'job_kind' THEN kind ELSE queue END), length($2::text)) = lower($2::text) - AND (CASE WHEN $1::text = 'job_kind' THEN kind ELSE queue END) NOT IN (SELECT jsonb_array_elements_text($3::jsonb)) - ORDER BY value + SELECT DISTINCT candidates.value + FROM ( + SELECT (CASE WHEN $1::text = 'job_kind' THEN kind ELSE queue END) COLLATE "C" AS value + FROM /* TEMPLATE: schema */river_job_dead_letter + ) AS candidates + WHERE left(lower(candidates.value COLLATE "C"), length($2::text)) = lower(($2::text) COLLATE "C") + AND candidates.value NOT IN (SELECT excluded.value COLLATE "C" FROM jsonb_array_elements_text($3::jsonb) AS excluded(value)) + ORDER BY candidates.value LIMIT $4::int ) AS suggestions ` @@ -29,6 +32,11 @@ type JobDeadLetterAutocompleteParams struct { } // Suggest values from the archive, including kinds and queues with no live jobs. +// +// The "C" collation makes matching and ordering independent of the database's +// collation and identical to SQLite: lower() folds only ASCII letters, and +// values sort by their bytes. The ordering matters before the limit as well, +// because it decides which suggestions are returned. func (q *Queries) JobDeadLetterAutocomplete(ctx context.Context, db DBTX, arg *JobDeadLetterAutocompleteParams) (string, error) { row := db.QueryRow(ctx, jobDeadLetterAutocomplete, arg.Facet, @@ -49,16 +57,15 @@ SELECT coalesce(json_agg(json_build_object( 'kind', limited.kind, 'max_attempts', limited.max_attempts, 'queue', limited.queue -) ORDER BY limited.id), '[]'::json)::text AS jobs +) ORDER BY limited.id DESC), '[]'::json)::text AS jobs FROM ( SELECT id, attempt, finalized_at, kind, max_attempts, queue FROM /* TEMPLATE: schema */river_job_dead_letter - WHERE (NOT jsonb_exists($1::jsonb, 'ids') OR id IN (SELECT value::bigint FROM jsonb_array_elements_text($1::jsonb -> 'ids'))) - AND (NOT jsonb_exists($1::jsonb, 'kinds') OR kind IN (SELECT value::text FROM jsonb_array_elements_text($1::jsonb -> 'kinds'))) + WHERE (NOT jsonb_exists($1::jsonb, 'kinds') OR kind IN (SELECT value::text FROM jsonb_array_elements_text($1::jsonb -> 'kinds'))) AND (NOT jsonb_exists($1::jsonb, 'priorities') OR priority IN (SELECT value::smallint FROM jsonb_array_elements_text($1::jsonb -> 'priorities'))) AND (NOT jsonb_exists($1::jsonb, 'queues') OR queue IN (SELECT value::text FROM jsonb_array_elements_text($1::jsonb -> 'queues'))) AND (NOT jsonb_exists($1::jsonb, 'tags') OR tags && ARRAY(SELECT jsonb_array_elements_text($1::jsonb -> 'tags'))::varchar[]) - ORDER BY id + ORDER BY id DESC LIMIT $2::int ) AS limited ` @@ -68,12 +75,54 @@ type JobDeadLetterListParams struct { Max int32 } -// Apply the limit before aggregation so neither the database nor the browser -// materializes the entire queue. River's common executor exposes QueryRow, -// so return the bounded set as one JSON value. +// Return the newest jobs first. Apply the limit before aggregation so neither +// the database nor the browser materializes the entire queue. River's common +// executor exposes QueryRow, so return the bounded set as one JSON value. +// +// This query ignores the filter's IDs; the driver sends ID lookups to +// JobDeadLetterListByIDs instead. func (q *Queries) JobDeadLetterList(ctx context.Context, db DBTX, arg *JobDeadLetterListParams) (string, error) { row := db.QueryRow(ctx, jobDeadLetterList, arg.Filters, arg.Max) var jobs string err := row.Scan(&jobs) return jobs, err } + +const jobDeadLetterListByIDs = `-- name: JobDeadLetterListByIDs :one +SELECT coalesce(json_agg(json_build_object( + 'id', limited.id, + 'attempt', limited.attempt, + 'finalized_at', limited.finalized_at, + 'kind', limited.kind, + 'max_attempts', limited.max_attempts, + 'queue', limited.queue +) ORDER BY limited.id DESC), '[]'::json)::text AS jobs +FROM ( + SELECT id, attempt, finalized_at, kind, max_attempts, queue + FROM /* TEMPLATE: schema */river_job_dead_letter + WHERE id = ANY (ARRAY(SELECT value::bigint FROM jsonb_array_elements_text($1::jsonb -> 'ids'))) + AND (NOT jsonb_exists($1::jsonb, 'kinds') OR kind IN (SELECT value::text FROM jsonb_array_elements_text($1::jsonb -> 'kinds'))) + AND (NOT jsonb_exists($1::jsonb, 'priorities') OR priority IN (SELECT value::smallint FROM jsonb_array_elements_text($1::jsonb -> 'priorities'))) + AND (NOT jsonb_exists($1::jsonb, 'queues') OR queue IN (SELECT value::text FROM jsonb_array_elements_text($1::jsonb -> 'queues'))) + AND (NOT jsonb_exists($1::jsonb, 'tags') OR tags && ARRAY(SELECT jsonb_array_elements_text($1::jsonb -> 'tags'))::varchar[]) + ORDER BY id DESC + LIMIT $2::int +) AS limited +` + +type JobDeadLetterListByIDsParams struct { + Filters []byte + Max int32 +} + +// JobDeadLetterList for filters that include IDs. The ID predicate must stand +// alone as a top-level condition: written as "no IDs requested OR id IN (...)" +// the planner can't turn it into an index condition and reads the whole +// table. Collapsing the requested IDs to an array lets it probe the primary +// key once per ID, whether the statement runs with a custom or generic plan. +func (q *Queries) JobDeadLetterListByIDs(ctx context.Context, db DBTX, arg *JobDeadLetterListByIDsParams) (string, error) { + row := db.QueryRow(ctx, jobDeadLetterListByIDs, arg.Filters, arg.Max) + var jobs string + err := row.Scan(&jobs) + return jobs, err +} diff --git a/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_queries.go b/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_queries.go index 6021c0d2..f28f55f1 100644 --- a/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_queries.go +++ b/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_queries.go @@ -95,6 +95,14 @@ func JobDeadLetterListRiver(ctx context.Context, db RowQuerier, maxRows int, fil return jobs, nil } +func JobDeadLetterListByIDsRiver(ctx context.Context, db RowQuerier, maxRows int, filters string) (string, error) { + var jobs string + if err := db.QueryRow(ctx, jobDeadLetterListByIDs, filters, maxRows).Scan(&jobs); err != nil { + return "", err + } + return jobs, nil +} + func JobDeadLetterAutocompleteRiver(ctx context.Context, db RowQuerier, facet, match, exclude string, maxRows int) (string, error) { var values string if err := db.QueryRow(ctx, jobDeadLetterAutocomplete, facet, match, exclude, maxRows).Scan(&values); err != nil { diff --git a/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_queries_test.go b/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_queries_test.go index b7e94119..fdb8c45b 100644 --- a/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_queries_test.go +++ b/internal/riveruidriver/riveruipostgres/internal/dbsqlc/river_queries_test.go @@ -6,6 +6,8 @@ import ( "github.com/stretchr/testify/require" "github.com/riverqueue/river/rivertype" + + "riverqueue.com/riverui/internal/riveruidriver" ) func TestJobCountEstimateQuery(t *testing.T) { @@ -29,3 +31,12 @@ func TestJobCountByAllStatesCappedQueryStates(t *testing.T) { require.Contains(t, jobCountByAllStatesCapped, "state = '"+string(state)+"'") } } + +func TestJobDeadLetterAutocompleteQueryFacets(t *testing.T) { + t.Parallel() + + // The query selects kinds for the job kind facet and queues for anything + // else, so its literal must stay in step with the driver's facet values. + require.Contains(t, jobDeadLetterAutocomplete, "= '"+string(riveruidriver.JobDeadLetterFacetJobKind)+"' THEN kind ELSE queue END") + require.NotContains(t, jobDeadLetterAutocomplete, string(riveruidriver.JobDeadLetterFacetQueueName)) +} diff --git a/internal/riveruidriver/riveruipostgres/river_ui_postgres_driver.go b/internal/riveruidriver/riveruipostgres/river_ui_postgres_driver.go index d62d5ada..04157ecf 100644 --- a/internal/riveruidriver/riveruipostgres/river_ui_postgres_driver.go +++ b/internal/riveruidriver/riveruipostgres/river_ui_postgres_driver.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "math" + "slices" "time" "github.com/riverqueue/river/riverdriver" @@ -141,14 +142,20 @@ func (w riverExecutorWrapper) QueryRow(ctx context.Context, query string, args . } func (e *Executor) JobDeadLetterList(ctx context.Context, params *riveruidriver.JobDeadLetterListParams) ([]*riveruidriver.JobDeadLetterListResult, error) { - if params.Max < 1 || params.Max > 200 { - return nil, errors.New("dead letter list max must be between 1 and 200") + if err := params.Validate(); err != nil { + return nil, err } filters, err := json.Marshal(params.Filters) if err != nil { return nil, fmt.Errorf("error encoding dead letter filters: %w", err) } - rawJobs, err := dbsqlc.JobDeadLetterListRiver(schemaTemplateParam(ctx, params.Schema), e.dbtx, params.Max, string(filters)) + // ID lookups use their own query so the database can search the primary + // key instead of reading the whole table. + listQuery := dbsqlc.JobDeadLetterListRiver + if len(params.Filters.IDs) > 0 { + listQuery = dbsqlc.JobDeadLetterListByIDsRiver + } + rawJobs, err := listQuery(schemaTemplateParam(ctx, params.Schema), e.dbtx, params.Max, string(filters)) if err != nil { return nil, fmt.Errorf("error listing dead letter jobs: %w", err) } @@ -156,21 +163,19 @@ func (e *Executor) JobDeadLetterList(ctx context.Context, params *riveruidriver. if err := json.Unmarshal([]byte(rawJobs), &jobs); err != nil { return nil, fmt.Errorf("error decoding dead letter jobs: %w", err) } + riveruidriver.NormalizeJobDeadLetterListResults(jobs) return jobs, nil } func (e *Executor) JobDeadLetterAutocomplete(ctx context.Context, params *riveruidriver.JobDeadLetterAutocompleteParams) ([]string, error) { - if params.Facet != "job_kind" && params.Facet != "queue_name" { - return nil, errors.New("invalid dead letter autocomplete facet") - } - if params.Max < 1 || params.Max > 100 { - return nil, errors.New("dead letter autocomplete max must be between 1 and 100") + if err := params.Validate(); err != nil { + return nil, err } exclude, err := json.Marshal(append([]string{}, params.Exclude...)) if err != nil { return nil, fmt.Errorf("error encoding dead letter exclusions: %w", err) } - rawValues, err := dbsqlc.JobDeadLetterAutocompleteRiver(schemaTemplateParam(ctx, params.Schema), e.dbtx, params.Facet, params.Match, string(exclude), params.Max) + rawValues, err := dbsqlc.JobDeadLetterAutocompleteRiver(schemaTemplateParam(ctx, params.Schema), e.dbtx, string(params.Facet), params.Match, string(exclude), params.Max) if err != nil { return nil, fmt.Errorf("error autocompleting dead letter jobs: %w", err) } @@ -178,5 +183,8 @@ func (e *Executor) JobDeadLetterAutocomplete(ctx context.Context, params *riveru if err := json.Unmarshal([]byte(rawValues), &values); err != nil { return nil, fmt.Errorf("error decoding dead letter suggestions: %w", err) } + // Byte order matches the queries' ordering; sorting here keeps the result + // deterministic where the database doesn't guarantee aggregate order. + slices.Sort(values) return values, nil } diff --git a/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_job_dead_letter.sql b/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_job_dead_letter.sql index 6295e003..e837a095 100644 --- a/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_job_dead_letter.sql +++ b/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_job_dead_letter.sql @@ -12,9 +12,15 @@ CREATE TABLE river_job_dead_letter ( ); -- name: JobDeadLetterList :one --- Apply the limit before aggregation so neither the database nor the browser --- materializes the entire queue. Normalize SQLite timestamps to RFC 3339 for --- the shared result decoder; strftime preserves NULL timestamps. +-- Return the newest jobs first. Apply the limit before aggregation so neither +-- the database nor the browser materializes the entire queue. SQLite doesn't +-- guarantee that json_group_array preserves the subquery's order, so the +-- driver sorts the decoded rows. Normalize SQLite timestamps to RFC 3339 for +-- the shared result decoder; strftime preserves NULL timestamps. Its %f keeps +-- milliseconds, where PostgreSQL returns microseconds. +-- +-- This query ignores the filter's IDs; the driver sends ID lookups to +-- JobDeadLetterListByIDs instead. SELECT CAST(json_group_array(json_object( 'id', limited.id, 'attempt', limited.attempt, @@ -26,17 +32,46 @@ SELECT CAST(json_group_array(json_object( FROM ( SELECT id, attempt, finalized_at, kind, max_attempts, queue FROM /* TEMPLATE: schema */river_job_dead_letter - WHERE (json_type(CAST(@filters AS text), '$.ids') IS NULL OR id IN (SELECT CAST(value AS integer) FROM json_each(CAST(@filters AS text), '$.ids'))) + WHERE (json_type(CAST(@filters AS text), '$.kinds') IS NULL OR kind IN (SELECT CAST(value AS text) FROM json_each(CAST(@filters AS text), '$.kinds'))) + AND (json_type(CAST(@filters AS text), '$.priorities') IS NULL OR priority IN (SELECT CAST(value AS integer) FROM json_each(CAST(@filters AS text), '$.priorities'))) + AND (json_type(CAST(@filters AS text), '$.queues') IS NULL OR queue IN (SELECT CAST(value AS text) FROM json_each(CAST(@filters AS text), '$.queues'))) + AND (json_type(CAST(@filters AS text), '$.tags') IS NULL OR EXISTS (SELECT 1 FROM json_each(river_job_dead_letter.tags) AS job_tag JOIN json_each(CAST(@filters AS text), '$.tags') AS filter_tag ON job_tag.value = filter_tag.value)) + ORDER BY id DESC + LIMIT @max +) AS limited; + +-- name: JobDeadLetterListByIDs :one +-- JobDeadLetterList for filters that include IDs. The ID predicate must stand +-- alone as a top-level condition: written as "no IDs requested OR id IN (...)" +-- SQLite scans the whole table instead of searching the primary key. +SELECT CAST(json_group_array(json_object( + 'id', limited.id, + 'attempt', limited.attempt, + 'finalized_at', strftime('%Y-%m-%dT%H:%M:%fZ', limited.finalized_at), + 'kind', limited.kind, + 'max_attempts', limited.max_attempts, + 'queue', limited.queue +)) AS text) AS jobs +FROM ( + SELECT id, attempt, finalized_at, kind, max_attempts, queue + FROM /* TEMPLATE: schema */river_job_dead_letter + WHERE id IN (SELECT CAST(value AS integer) FROM json_each(CAST(@filters AS text), '$.ids')) AND (json_type(CAST(@filters AS text), '$.kinds') IS NULL OR kind IN (SELECT CAST(value AS text) FROM json_each(CAST(@filters AS text), '$.kinds'))) AND (json_type(CAST(@filters AS text), '$.priorities') IS NULL OR priority IN (SELECT CAST(value AS integer) FROM json_each(CAST(@filters AS text), '$.priorities'))) AND (json_type(CAST(@filters AS text), '$.queues') IS NULL OR queue IN (SELECT CAST(value AS text) FROM json_each(CAST(@filters AS text), '$.queues'))) AND (json_type(CAST(@filters AS text), '$.tags') IS NULL OR EXISTS (SELECT 1 FROM json_each(river_job_dead_letter.tags) AS job_tag JOIN json_each(CAST(@filters AS text), '$.tags') AS filter_tag ON job_tag.value = filter_tag.value)) - ORDER BY id + ORDER BY id DESC LIMIT @max ) AS limited; -- name: JobDeadLetterAutocomplete :one -- Suggest values from the archive, including kinds and queues with no live jobs. +-- +-- SQLite's built-in lower() folds only ASCII letters and its default BINARY +-- collation sorts values by their bytes. The PostgreSQL query pins the "C" +-- collation to behave the same way. A SQLite build that loads the ICU +-- extension replaces lower() with a Unicode-aware version and would also fold +-- non-ASCII letters. SELECT CAST(json_group_array(suggestions.value) AS text) AS values_json FROM ( SELECT DISTINCT candidates.value diff --git a/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_job_dead_letter.sql.go b/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_job_dead_letter.sql.go index 84d6d0a3..3dadb5a5 100644 --- a/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_job_dead_letter.sql.go +++ b/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_job_dead_letter.sql.go @@ -32,6 +32,12 @@ type JobDeadLetterAutocompleteParams struct { } // Suggest values from the archive, including kinds and queues with no live jobs. +// +// SQLite's built-in lower() folds only ASCII letters and its default BINARY +// collation sorts values by their bytes. The PostgreSQL query pins the "C" +// collation to behave the same way. A SQLite build that loads the ICU +// extension replaces lower() with a Unicode-aware version and would also fold +// non-ASCII letters. func (q *Queries) JobDeadLetterAutocomplete(ctx context.Context, db DBTX, arg *JobDeadLetterAutocompleteParams) (string, error) { row := db.QueryRowContext(ctx, jobDeadLetterAutocomplete, arg.Facet, @@ -56,12 +62,11 @@ SELECT CAST(json_group_array(json_object( FROM ( SELECT id, attempt, finalized_at, kind, max_attempts, queue FROM /* TEMPLATE: schema */river_job_dead_letter - WHERE (json_type(CAST(?1 AS text), '$.ids') IS NULL OR id IN (SELECT CAST(value AS integer) FROM json_each(CAST(?1 AS text), '$.ids'))) - AND (json_type(CAST(?1 AS text), '$.kinds') IS NULL OR kind IN (SELECT CAST(value AS text) FROM json_each(CAST(?1 AS text), '$.kinds'))) + WHERE (json_type(CAST(?1 AS text), '$.kinds') IS NULL OR kind IN (SELECT CAST(value AS text) FROM json_each(CAST(?1 AS text), '$.kinds'))) AND (json_type(CAST(?1 AS text), '$.priorities') IS NULL OR priority IN (SELECT CAST(value AS integer) FROM json_each(CAST(?1 AS text), '$.priorities'))) AND (json_type(CAST(?1 AS text), '$.queues') IS NULL OR queue IN (SELECT CAST(value AS text) FROM json_each(CAST(?1 AS text), '$.queues'))) AND (json_type(CAST(?1 AS text), '$.tags') IS NULL OR EXISTS (SELECT 1 FROM json_each(river_job_dead_letter.tags) AS job_tag JOIN json_each(CAST(?1 AS text), '$.tags') AS filter_tag ON job_tag.value = filter_tag.value)) - ORDER BY id + ORDER BY id DESC LIMIT ?2 ) AS limited ` @@ -71,12 +76,55 @@ type JobDeadLetterListParams struct { Max int64 } -// Apply the limit before aggregation so neither the database nor the browser -// materializes the entire queue. Normalize SQLite timestamps to RFC 3339 for -// the shared result decoder; strftime preserves NULL timestamps. +// Return the newest jobs first. Apply the limit before aggregation so neither +// the database nor the browser materializes the entire queue. SQLite doesn't +// guarantee that json_group_array preserves the subquery's order, so the +// driver sorts the decoded rows. Normalize SQLite timestamps to RFC 3339 for +// the shared result decoder; strftime preserves NULL timestamps. Its %f keeps +// milliseconds, where PostgreSQL returns microseconds. +// +// This query ignores the filter's IDs; the driver sends ID lookups to +// JobDeadLetterListByIDs instead. func (q *Queries) JobDeadLetterList(ctx context.Context, db DBTX, arg *JobDeadLetterListParams) (string, error) { row := db.QueryRowContext(ctx, jobDeadLetterList, arg.Filters, arg.Max) var jobs string err := row.Scan(&jobs) return jobs, err } + +const jobDeadLetterListByIDs = `-- name: JobDeadLetterListByIDs :one +SELECT CAST(json_group_array(json_object( + 'id', limited.id, + 'attempt', limited.attempt, + 'finalized_at', strftime('%Y-%m-%dT%H:%M:%fZ', limited.finalized_at), + 'kind', limited.kind, + 'max_attempts', limited.max_attempts, + 'queue', limited.queue +)) AS text) AS jobs +FROM ( + SELECT id, attempt, finalized_at, kind, max_attempts, queue + FROM /* TEMPLATE: schema */river_job_dead_letter + WHERE id IN (SELECT CAST(value AS integer) FROM json_each(CAST(?1 AS text), '$.ids')) + AND (json_type(CAST(?1 AS text), '$.kinds') IS NULL OR kind IN (SELECT CAST(value AS text) FROM json_each(CAST(?1 AS text), '$.kinds'))) + AND (json_type(CAST(?1 AS text), '$.priorities') IS NULL OR priority IN (SELECT CAST(value AS integer) FROM json_each(CAST(?1 AS text), '$.priorities'))) + AND (json_type(CAST(?1 AS text), '$.queues') IS NULL OR queue IN (SELECT CAST(value AS text) FROM json_each(CAST(?1 AS text), '$.queues'))) + AND (json_type(CAST(?1 AS text), '$.tags') IS NULL OR EXISTS (SELECT 1 FROM json_each(river_job_dead_letter.tags) AS job_tag JOIN json_each(CAST(?1 AS text), '$.tags') AS filter_tag ON job_tag.value = filter_tag.value)) + ORDER BY id DESC + LIMIT ?2 +) AS limited +` + +type JobDeadLetterListByIDsParams struct { + Filters string + Max int64 +} + +// JobDeadLetterList for filters that include IDs. The ID predicate must stand +// alone as a top-level condition: written as "no IDs requested OR id IN (...)" +// SQLite scans the whole table instead of searching the primary key. +func (q *Queries) JobDeadLetterListByIDs(ctx context.Context, db DBTX, arg *JobDeadLetterListByIDsParams) (string, error) { + row := db.QueryRowContext(ctx, jobDeadLetterListByIDs, arg.Filters, arg.Max) + var jobs string + err := row.Scan(&jobs) + return jobs, err +} diff --git a/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_queries.go b/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_queries.go index 7376e0b9..71023799 100644 --- a/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_queries.go +++ b/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_queries.go @@ -45,6 +45,14 @@ func JobDeadLetterListRiver(ctx context.Context, db RowQuerier, maxRows int, fil return jobs, nil } +func JobDeadLetterListByIDsRiver(ctx context.Context, db RowQuerier, maxRows int, filters string) (string, error) { + var jobs string + if err := db.QueryRow(ctx, jobDeadLetterListByIDs, filters, maxRows).Scan(&jobs); err != nil { + return "", err + } + return jobs, nil +} + func JobDeadLetterAutocompleteRiver(ctx context.Context, db RowQuerier, facet, match, exclude string, maxRows int) (string, error) { var values string if err := db.QueryRow(ctx, jobDeadLetterAutocomplete, facet, match, exclude, maxRows).Scan(&values); err != nil { diff --git a/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_queries_test.go b/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_queries_test.go index 9baae87d..e7a92900 100644 --- a/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_queries_test.go +++ b/internal/riveruidriver/riveruisqlite/internal/dbsqlc/river_queries_test.go @@ -6,6 +6,8 @@ import ( "github.com/stretchr/testify/require" "github.com/riverqueue/river/rivertype" + + "riverqueue.com/riverui/internal/riveruidriver" ) func TestJobCountByAllStatesCappedQueryStates(t *testing.T) { @@ -23,3 +25,12 @@ func TestJobCountEstimateStat4Query(t *testing.T) { require.Contains(t, jobCountEstimateStat4, "river_job_prioritized_fetching_index") require.Contains(t, jobCountEstimateStat4, "group_concat") } + +func TestJobDeadLetterAutocompleteQueryFacets(t *testing.T) { + t.Parallel() + + // The query selects kinds for the job kind facet and queues for anything + // else, so its literal must stay in step with the driver's facet values. + require.Contains(t, jobDeadLetterAutocomplete, "= '"+string(riveruidriver.JobDeadLetterFacetJobKind)+"' THEN kind ELSE queue END") + require.NotContains(t, jobDeadLetterAutocomplete, string(riveruidriver.JobDeadLetterFacetQueueName)) +} diff --git a/internal/riveruidriver/riveruisqlite/river_ui_sqlite_driver.go b/internal/riveruidriver/riveruisqlite/river_ui_sqlite_driver.go index bb4ec784..8e4a6096 100644 --- a/internal/riveruidriver/riveruisqlite/river_ui_sqlite_driver.go +++ b/internal/riveruidriver/riveruisqlite/river_ui_sqlite_driver.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "math" + "slices" "strconv" "strings" @@ -243,14 +244,20 @@ func (w riverExecutorWrapper) QueryRow(ctx context.Context, query string, args . } func (e *Executor) JobDeadLetterList(ctx context.Context, params *riveruidriver.JobDeadLetterListParams) ([]*riveruidriver.JobDeadLetterListResult, error) { - if params.Max < 1 || params.Max > 200 { - return nil, errors.New("dead letter list max must be between 1 and 200") + if err := params.Validate(); err != nil { + return nil, err } filters, err := json.Marshal(params.Filters) if err != nil { return nil, fmt.Errorf("error encoding dead letter filters: %w", err) } - rawJobs, err := dbsqlc.JobDeadLetterListRiver(schemaTemplateParam(ctx, params.Schema), e.dbtx, params.Max, string(filters)) + // ID lookups use their own query so the database can search the primary + // key instead of reading the whole table. + listQuery := dbsqlc.JobDeadLetterListRiver + if len(params.Filters.IDs) > 0 { + listQuery = dbsqlc.JobDeadLetterListByIDsRiver + } + rawJobs, err := listQuery(schemaTemplateParam(ctx, params.Schema), e.dbtx, params.Max, string(filters)) if err != nil { return nil, fmt.Errorf("error listing dead letter jobs: %w", err) } @@ -258,21 +265,19 @@ func (e *Executor) JobDeadLetterList(ctx context.Context, params *riveruidriver. if err := json.Unmarshal([]byte(rawJobs), &jobs); err != nil { return nil, fmt.Errorf("error decoding dead letter jobs: %w", err) } + riveruidriver.NormalizeJobDeadLetterListResults(jobs) return jobs, nil } func (e *Executor) JobDeadLetterAutocomplete(ctx context.Context, params *riveruidriver.JobDeadLetterAutocompleteParams) ([]string, error) { - if params.Facet != "job_kind" && params.Facet != "queue_name" { - return nil, errors.New("invalid dead letter autocomplete facet") - } - if params.Max < 1 || params.Max > 100 { - return nil, errors.New("dead letter autocomplete max must be between 1 and 100") + if err := params.Validate(); err != nil { + return nil, err } exclude, err := json.Marshal(append([]string{}, params.Exclude...)) if err != nil { return nil, fmt.Errorf("error encoding dead letter exclusions: %w", err) } - rawValues, err := dbsqlc.JobDeadLetterAutocompleteRiver(schemaTemplateParam(ctx, params.Schema), e.dbtx, params.Facet, params.Match, string(exclude), params.Max) + rawValues, err := dbsqlc.JobDeadLetterAutocompleteRiver(schemaTemplateParam(ctx, params.Schema), e.dbtx, string(params.Facet), params.Match, string(exclude), params.Max) if err != nil { return nil, fmt.Errorf("error autocompleting dead letter jobs: %w", err) } @@ -280,5 +285,8 @@ func (e *Executor) JobDeadLetterAutocomplete(ctx context.Context, params *riveru if err := json.Unmarshal([]byte(rawValues), &values); err != nil { return nil, fmt.Errorf("error decoding dead letter suggestions: %w", err) } + // Byte order matches the queries' ordering; sorting here keeps the result + // deterministic where the database doesn't guarantee aggregate order. + slices.Sort(values) return values, nil } diff --git a/riverproui/endpoints.go b/riverproui/endpoints.go index 220714f9..00cf7f4a 100644 --- a/riverproui/endpoints.go +++ b/riverproui/endpoints.go @@ -90,10 +90,7 @@ func (e *endpoints[TTx]) Extensions(ctx context.Context) (map[string]bool, error } defer execTx.Rollback(ctx) - hasDeadLetterTable, err := execTx.TableExists(ctx, &riverdriver.TableExistsParams{ - Schema: schema, - Table: "river_job_dead_letter", - }) + hasDeadLetterTable, err := prohandler.HasDeadLetterTable(ctx, execTx, schema) if err != nil { return nil, err } diff --git a/riverproui/internal/prohandler/dead_letter_endpoints.go b/riverproui/internal/prohandler/dead_letter_endpoints.go index 9676e0f3..7de38e7d 100644 --- a/riverproui/internal/prohandler/dead_letter_endpoints.go +++ b/riverproui/internal/prohandler/dead_letter_endpoints.go @@ -6,6 +6,9 @@ import ( "fmt" "net/http" "strconv" + "strings" + "time" + "unicode/utf8" "github.com/riverqueue/apiframe/apiendpoint" "github.com/riverqueue/apiframe/apierror" @@ -22,11 +25,21 @@ import ( "riverqueue.com/riverui/internal/riveruidriver/riveruisqlite" ) -func (a *ProAPIBundle[TTx]) requireDeadLetterJobs(ctx context.Context) error { - exists, err := a.DB.TableExists(ctx, &riverdriver.TableExistsParams{ - Schema: a.Client.Schema(), - Table: "river_job_dead_letter", +// deadLetterTableName is the table created by River Pro's dead letter queue +// migration. +const deadLetterTableName = "river_job_dead_letter" + +// HasDeadLetterTable reports whether the schema has the dead letter table, and +// therefore whether the dead letter endpoints can serve requests. +func HasDeadLetterTable(ctx context.Context, exec riverdriver.Executor, schema string) (bool, error) { + return exec.TableExists(ctx, &riverdriver.TableExistsParams{ + Schema: schema, + Table: deadLetterTableName, }) +} + +func (a *ProAPIBundle[TTx]) requireDeadLetterJobs(ctx context.Context) error { + exists, err := HasDeadLetterTable(ctx, a.DB, a.Client.Schema()) if err != nil { return fmt.Errorf("error checking dead letter schema support: %w", err) } @@ -36,11 +49,28 @@ func (a *ProAPIBundle[TTx]) requireDeadLetterJobs(ctx context.Context) error { return nil } -// IDs are strings so JavaScript can preserve the full int64 range. +// deadLetterJobListItem is the public shape of a job in the list response. It +// is declared here rather than built on the driver's result, whose JSON tags +// exist to decode a query and shouldn't define the API. IDs are strings so +// JavaScript can preserve the full int64 range. type deadLetterJobListItem struct { - *riveruidriver.JobDeadLetterListResult + ID int64 `json:"id,string"` + Attempt int `json:"attempt"` + FinalizedAt *time.Time `json:"finalized_at"` + Kind string `json:"kind"` + MaxAttempts int `json:"max_attempts"` + Queue string `json:"queue"` +} - ID int64 `json:"id,string"` +func deadLetterJobListItemFromDriver(job *riveruidriver.JobDeadLetterListResult) *deadLetterJobListItem { + return &deadLetterJobListItem{ + ID: job.ID, + Attempt: job.Attempt, + FinalizedAt: job.FinalizedAt, + Kind: job.Kind, + MaxAttempts: job.MaxAttempts, + Queue: job.Queue, + } } type deadLetterJob struct { @@ -85,6 +115,8 @@ func (*deadLetterJobListEndpoint[TTx]) Meta() *apiendpoint.EndpointMeta { return &apiendpoint.EndpointMeta{Pattern: "GET /api/pro/dead-letter-jobs", StatusCode: http.StatusOK} } +// The validate tags must be literals. Limit's maximum mirrors +// riveruidriver.JobDeadLetterListMaxLimit. type deadLetterJobListRequest struct { IDs []int64 `json:"-" validate:"omitempty,max=1000,dive,min=1"` Kinds []string `json:"-" validate:"omitempty,max=100,dive,required,max=1024"` @@ -94,9 +126,31 @@ type deadLetterJobListRequest struct { Limit *int `json:"-" validate:"omitempty,min=1,max=200"` } +// validateDeadLetterText rejects text that the databases can't handle +// consistently. PostgreSQL refuses NUL bytes and invalid UTF-8 in both text and +// jsonb parameters, which would otherwise surface as an internal server error, +// while SQLite accepts them. +func validateDeadLetterText(param string, values ...string) error { + for _, value := range values { + if strings.ContainsRune(value, 0) || !utf8.ValidString(value) { + return apierror.NewBadRequestf("Invalid `%s` value: must be valid UTF-8 without NUL bytes.", param) + } + } + return nil +} + func (req *deadLetterJobListRequest) ExtractRaw(r *http.Request) error { query := r.URL.Query() req.Kinds, req.Queues, req.Tags = query["kinds"], query["queues"], query["tags"] + if err := validateDeadLetterText("kinds", req.Kinds...); err != nil { + return err + } + if err := validateDeadLetterText("queues", req.Queues...); err != nil { + return err + } + if err := validateDeadLetterText("tags", req.Tags...); err != nil { + return err + } for _, value := range query["ids"] { id, err := strconv.ParseInt(value, 10, 64) if err != nil { @@ -127,15 +181,13 @@ func (a *deadLetterJobListEndpoint[TTx]) Execute(ctx context.Context, req *deadL } jobs, err := a.uiDriver.GetExecutor(a.DB).JobDeadLetterList(ctx, &riveruidriver.JobDeadLetterListParams{ Filters: riveruidriver.JobDeadLetterFilters{IDs: req.IDs, Kinds: req.Kinds, Priorities: req.Priorities, Queues: req.Queues, Tags: req.Tags}, - Max: ptrutil.ValOrDefault(req.Limit, 20), + Max: ptrutil.ValOrDefault(req.Limit, riveruidriver.JobDeadLetterListMaxDefault), Schema: a.Client.Schema(), }) if err != nil { return nil, fmt.Errorf("error listing dead letter jobs: %w", err) } - return listResponseFrom(sliceutil.Map(jobs, func(job *riveruidriver.JobDeadLetterListResult) *deadLetterJobListItem { - return &deadLetterJobListItem{JobDeadLetterListResult: job, ID: job.ID} - })), nil + return listResponseFrom(sliceutil.Map(jobs, deadLetterJobListItemFromDriver)), nil } type deadLetterJobRequest struct { @@ -229,16 +281,21 @@ func (*deadLetterAutocompleteEndpoint[TTx]) Meta() *apiendpoint.EndpointMeta { return &apiendpoint.EndpointMeta{Pattern: "GET /api/pro/dead-letter-jobs/autocomplete", StatusCode: http.StatusOK} } +// The validate tags must be literals. Facet's values mirror the +// riveruidriver.JobDeadLetterFacet constants. type deadLetterAutocompleteRequest struct { - Facet string `json:"-" validate:"required,oneof=job_kind queue_name"` - Match string `json:"-" validate:"max=1024"` - Exclude []string `json:"-" validate:"omitempty,max=100,dive,max=1024"` + Facet riveruidriver.JobDeadLetterFacet `json:"-" validate:"required,oneof=job_kind queue_name"` + Match string `json:"-" validate:"max=1024"` + Exclude []string `json:"-" validate:"omitempty,max=100,dive,max=1024"` } func (req *deadLetterAutocompleteRequest) ExtractRaw(r *http.Request) error { query := r.URL.Query() - req.Facet, req.Match, req.Exclude = query.Get("facet"), query.Get("match"), query["exclude"] - return nil + req.Facet, req.Match, req.Exclude = riveruidriver.JobDeadLetterFacet(query.Get("facet")), query.Get("match"), query["exclude"] + if err := validateDeadLetterText("match", req.Match); err != nil { + return err + } + return validateDeadLetterText("exclude", req.Exclude...) } func (a *deadLetterAutocompleteEndpoint[TTx]) Execute(ctx context.Context, req *deadLetterAutocompleteRequest) (*listResponse[string], error) { @@ -246,7 +303,7 @@ func (a *deadLetterAutocompleteEndpoint[TTx]) Execute(ctx context.Context, req * return nil, err } values, err := a.uiDriver.GetExecutor(a.DB).JobDeadLetterAutocomplete(ctx, &riveruidriver.JobDeadLetterAutocompleteParams{ - Facet: req.Facet, Match: req.Match, Exclude: req.Exclude, Max: 100, Schema: a.Client.Schema(), + Facet: req.Facet, Match: req.Match, Exclude: req.Exclude, Max: riveruidriver.JobDeadLetterAutocompleteMaxLimit, Schema: a.Client.Schema(), }) if err != nil { return nil, err diff --git a/riverproui/internal/prohandler/dead_letter_endpoints_test.go b/riverproui/internal/prohandler/dead_letter_endpoints_test.go index f149d285..001cfb80 100644 --- a/riverproui/internal/prohandler/dead_letter_endpoints_test.go +++ b/riverproui/internal/prohandler/dead_letter_endpoints_test.go @@ -6,19 +6,24 @@ import ( "fmt" "net/http" "net/http/httptest" + "reflect" "strings" "testing" "time" "github.com/stretchr/testify/require" + "github.com/riverqueue/apiframe/apiendpoint" + "github.com/riverqueue/apiframe/apierror" "github.com/riverqueue/apiframe/apitest" "github.com/riverqueue/river/riverdriver" + "github.com/riverqueue/river/rivershared/util/dbutil" "github.com/riverqueue/river/rivertype" "riverqueue.com/riverpro/driver" "riverqueue.com/riverui/internal/riverinternaltest/testfactory" + "riverqueue.com/riverui/internal/riveruidriver" ) func insertDeadLetterJob(ctx context.Context, t *testing.T, bundle *setupEndpointTestBundle) *rivertype.JobRow { @@ -68,17 +73,18 @@ func TestDeadLetterJobList(t *testing.T) { encoded, err = json.Marshal(resp) require.NoError(t, err) require.Contains(t, string(encoded), fmt.Sprintf(`"id":"%d"`, job.ID)) + var newest *rivertype.JobRow for range 24 { - insertDeadLetterJob(ctx, t, bundle) + newest = insertDeadLetterJob(ctx, t, bundle) } resp, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobListRequest{}) require.NoError(t, err) - require.Len(t, resp.Data, 20) - require.Equal(t, job.ID, resp.Data[0].ID) + require.Len(t, resp.Data, riveruidriver.JobDeadLetterListMaxDefault) + require.Equal(t, newest.ID, resp.Data[0].ID, "newest jobs must come first") resp, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobListRequest{Limit: new(2)}) require.NoError(t, err) require.Len(t, resp.Data, 2) - require.Less(t, resp.Data[0].ID, resp.Data[1].ID) + require.Greater(t, resp.Data[0].ID, resp.Data[1].ID) resp, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobListRequest{Limit: new(40)}) require.NoError(t, err) require.Len(t, resp.Data, 25) @@ -91,25 +97,200 @@ func TestDeadLetterJobList(t *testing.T) { resp, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobListRequest{Queues: []string{"missing"}}) require.NoError(t, err) require.Empty(t, resp.Data) - for _, invalid := range []*deadLetterJobListRequest{ - {IDs: []int64{0}}, - {IDs: []int64{-1}}, - {IDs: make([]int64, 1001)}, - {Priorities: []int16{0}}, - {Priorities: []int16{5}}, - {Kinds: []string{""}}, - {Queues: []string{strings.Repeat("x", 1025)}}, - {Tags: make([]string, 101)}, - } { - _, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), invalid) - require.Error(t, err) - } - for _, limit := range []int{-1, 0, 201} { + for _, limit := range []int{-1, 0, riveruidriver.JobDeadLetterListMaxLimit + 1} { _, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobListRequest{Limit: &limit}) require.Error(t, err) } } +// The validate tags can't refer to constants, so check that they agree with the +// limits and facets that the drivers enforce. +func TestDeadLetterRequestTagsMatchDriver(t *testing.T) { + t.Parallel() + + validateTag := func(request any, fieldName string) string { + field, ok := reflect.TypeOf(request).FieldByName(fieldName) + require.True(t, ok) + return field.Tag.Get("validate") + } + require.Equal(t, + fmt.Sprintf("omitempty,min=1,max=%d", riveruidriver.JobDeadLetterListMaxLimit), + validateTag(deadLetterJobListRequest{}, "Limit")) + require.Equal(t, + fmt.Sprintf("required,oneof=%s %s", riveruidriver.JobDeadLetterFacetJobKind, riveruidriver.JobDeadLetterFacetQueueName), + validateTag(deadLetterAutocompleteRequest{}, "Facet")) +} + +// validDeadLetterIDs and validDeadLetterValues return slices whose elements +// pass validation, so that only the size of the slice is under test. +func validDeadLetterIDs(count int) []int64 { + ids := make([]int64, count) + for i := range ids { + ids[i] = int64(i) + 1 + } + return ids +} + +func validDeadLetterValues(count int) []string { + values := make([]string, count) + for i := range values { + values[i] = fmt.Sprintf("value_%d", i) + } + return values +} + +func TestDeadLetterJobListValidation(t *testing.T) { + t.Parallel() + ctx := t.Context() + endpoint, _ := setupEndpoint(ctx, t, NewDeadLetterJobListEndpoint) + + const ( + tooManyElements = "must contain at most %d element(s)" + tooLong = "must be at most 1024 character(s) long" + required = "is required" + ) + for _, tt := range []struct { + name string + request *deadLetterJobListRequest + wantErr string // empty when the request is valid + }{ + {name: "IDsAtLimit", request: &deadLetterJobListRequest{IDs: validDeadLetterIDs(1000)}}, + {name: "IDsOverLimit", request: &deadLetterJobListRequest{IDs: validDeadLetterIDs(1001)}, wantErr: fmt.Sprintf(tooManyElements, 1000)}, + {name: "IDZero", request: &deadLetterJobListRequest{IDs: []int64{0}}, wantErr: "must be greater or equal to 1"}, + {name: "IDNegative", request: &deadLetterJobListRequest{IDs: []int64{-1}}, wantErr: "must be greater or equal to 1"}, + {name: "KindsAtLimit", request: &deadLetterJobListRequest{Kinds: validDeadLetterValues(100)}}, + {name: "KindsOverLimit", request: &deadLetterJobListRequest{Kinds: validDeadLetterValues(101)}, wantErr: fmt.Sprintf(tooManyElements, 100)}, + {name: "KindAtLength", request: &deadLetterJobListRequest{Kinds: []string{strings.Repeat("x", 1024)}}}, + {name: "KindOverLength", request: &deadLetterJobListRequest{Kinds: []string{strings.Repeat("x", 1025)}}, wantErr: tooLong}, + {name: "KindEmpty", request: &deadLetterJobListRequest{Kinds: []string{""}}, wantErr: required}, + {name: "QueuesAtLimit", request: &deadLetterJobListRequest{Queues: validDeadLetterValues(100)}}, + {name: "QueuesOverLimit", request: &deadLetterJobListRequest{Queues: validDeadLetterValues(101)}, wantErr: fmt.Sprintf(tooManyElements, 100)}, + {name: "QueueAtLength", request: &deadLetterJobListRequest{Queues: []string{strings.Repeat("x", 1024)}}}, + {name: "QueueOverLength", request: &deadLetterJobListRequest{Queues: []string{strings.Repeat("x", 1025)}}, wantErr: tooLong}, + {name: "QueueEmpty", request: &deadLetterJobListRequest{Queues: []string{""}}, wantErr: required}, + {name: "TagsAtLimit", request: &deadLetterJobListRequest{Tags: validDeadLetterValues(100)}}, + {name: "TagsOverLimit", request: &deadLetterJobListRequest{Tags: validDeadLetterValues(101)}, wantErr: fmt.Sprintf(tooManyElements, 100)}, + {name: "TagAtLength", request: &deadLetterJobListRequest{Tags: []string{strings.Repeat("x", 1024)}}}, + {name: "TagOverLength", request: &deadLetterJobListRequest{Tags: []string{strings.Repeat("x", 1025)}}, wantErr: tooLong}, + {name: "TagEmpty", request: &deadLetterJobListRequest{Tags: []string{""}}, wantErr: required}, + {name: "PrioritiesAll", request: &deadLetterJobListRequest{Priorities: []int16{1, 2, 3, 4}}}, + {name: "PrioritiesOverLimit", request: &deadLetterJobListRequest{Priorities: []int16{1, 2, 3, 4, 1}}, wantErr: fmt.Sprintf(tooManyElements, 4)}, + {name: "PriorityZero", request: &deadLetterJobListRequest{Priorities: []int16{0}}, wantErr: "min"}, + {name: "PriorityTooHigh", request: &deadLetterJobListRequest{Priorities: []int16{5}}, wantErr: "max"}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + _, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), tt.request) + if tt.wantErr == "" { + require.NoError(t, err) + return + } + var badRequest *apierror.BadRequest + require.ErrorAs(t, err, &badRequest) + require.Equal(t, http.StatusBadRequest, badRequest.StatusCode) + require.Contains(t, badRequest.Message, tt.wantErr) + }) + } +} + +func TestDeadLetterAutocompleteValidation(t *testing.T) { + t.Parallel() + ctx := t.Context() + endpoint, _ := setupEndpoint(ctx, t, NewDeadLetterAutocompleteEndpoint) + + const tooLong = "must be at most 1024 character(s) long" + for _, tt := range []struct { + name string + request *deadLetterAutocompleteRequest + wantErr string // empty when the request is valid + }{ + {name: "MatchAtLength", request: &deadLetterAutocompleteRequest{Facet: "job_kind", Match: strings.Repeat("x", 1024)}}, + {name: "MatchOverLength", request: &deadLetterAutocompleteRequest{Facet: "job_kind", Match: strings.Repeat("x", 1025)}, wantErr: tooLong}, + {name: "ExcludeAtLimit", request: &deadLetterAutocompleteRequest{Facet: "job_kind", Exclude: validDeadLetterValues(100)}}, + {name: "ExcludeOverLimit", request: &deadLetterAutocompleteRequest{Facet: "job_kind", Exclude: validDeadLetterValues(101)}, wantErr: "must contain at most 100 element(s)"}, + {name: "ExcludeValueAtLength", request: &deadLetterAutocompleteRequest{Facet: "queue_name", Exclude: []string{strings.Repeat("x", 1024)}}}, + {name: "ExcludeValueOverLength", request: &deadLetterAutocompleteRequest{Facet: "queue_name", Exclude: []string{strings.Repeat("x", 1025)}}, wantErr: tooLong}, + {name: "FacetMissing", request: &deadLetterAutocompleteRequest{}, wantErr: "is required"}, + {name: "FacetUnknown", request: &deadLetterAutocompleteRequest{Facet: "tags"}, wantErr: "should be one of the following values: job_kind queue_name"}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + _, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), tt.request) + if tt.wantErr == "" { + require.NoError(t, err) + return + } + var badRequest *apierror.BadRequest + require.ErrorAs(t, err, &badRequest) + require.Equal(t, http.StatusBadRequest, badRequest.StatusCode) + require.Contains(t, badRequest.Message, tt.wantErr) + }) + } +} + +// Every dead letter endpoint must report a missing table as a bad request that +// names the migration, rather than failing in a query. +func TestDeadLetterJobsRequireTable(t *testing.T) { + t.Parallel() + ctx := t.Context() + + dropTable := func(t *testing.T, bundle *setupEndpointTestBundle) { + t.Helper() + require.NoError(t, bundle.exec.Exec(ctx, "DROP TABLE "+dbutil.SafeIdentifier(bundle.schema)+"."+deadLetterTableName)) + + exists, err := HasDeadLetterTable(ctx, bundle.exec, bundle.schema) + require.NoError(t, err) + require.False(t, exists) + } + requireMissingTableError := func(t *testing.T, err error) { + t.Helper() + var badRequest *apierror.BadRequest + require.ErrorAs(t, err, &badRequest) + require.Equal(t, http.StatusBadRequest, badRequest.StatusCode) + require.Equal(t, "Dead letter jobs require River Pro's dead letter queue migration.", badRequest.Message) + } + + t.Run("List", func(t *testing.T) { + t.Parallel() + endpoint, bundle := setupEndpoint(ctx, t, NewDeadLetterJobListEndpoint) + exists, err := HasDeadLetterTable(ctx, bundle.exec, bundle.schema) + require.NoError(t, err) + require.True(t, exists) + dropTable(t, bundle) + _, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobListRequest{}) + requireMissingTableError(t, err) + }) + t.Run("Autocomplete", func(t *testing.T) { + t.Parallel() + endpoint, bundle := setupEndpoint(ctx, t, NewDeadLetterAutocompleteEndpoint) + dropTable(t, bundle) + _, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterAutocompleteRequest{Facet: "job_kind"}) + requireMissingTableError(t, err) + }) + t.Run("Get", func(t *testing.T) { + t.Parallel() + endpoint, bundle := setupEndpoint(ctx, t, NewDeadLetterJobGetEndpoint) + dropTable(t, bundle) + _, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobRequest{JobID: 1}) + requireMissingTableError(t, err) + }) + t.Run("Retry", func(t *testing.T) { + t.Parallel() + endpoint, bundle := setupEndpoint(ctx, t, NewDeadLetterJobRetryEndpoint) + dropTable(t, bundle) + _, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobRequest{JobID: 1}) + requireMissingTableError(t, err) + }) +} + +func requireDeadLetterNotFound(t *testing.T, err error, jobID int64) { + t.Helper() + var notFound *apierror.NotFound + require.ErrorAs(t, err, ¬Found) + require.Equal(t, http.StatusNotFound, notFound.StatusCode) + require.Equal(t, fmt.Sprintf("Dead letter job with ID %d not found. It may already have been retried.", jobID), notFound.Message) +} + func TestDeadLetterJobGet(t *testing.T) { t.Parallel() ctx := t.Context() @@ -123,7 +304,7 @@ func TestDeadLetterJobGet(t *testing.T) { require.Equal(t, "delivery failed", resp.Errors[0].Error) _, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobRequest{JobID: job.ID + 1}) - require.EqualError(t, err, fmt.Sprintf("Dead letter job with ID %d not found. It may already have been retried.", job.ID+1)) + requireDeadLetterNotFound(t, err, job.ID+1) } func TestDeadLetterJobRetry(t *testing.T) { @@ -148,9 +329,9 @@ func TestDeadLetterJobRetry(t *testing.T) { require.Equal(t, rivertype.JobStateAvailable, live.State) _, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobRequest{JobID: job.ID}) - require.EqualError(t, err, fmt.Sprintf("Dead letter job with ID %d not found. It may already have been retried.", job.ID)) + requireDeadLetterNotFound(t, err, job.ID) _, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobRequest{JobID: live.ID}) - require.EqualError(t, err, fmt.Sprintf("Dead letter job with ID %d not found. It may already have been retried.", live.ID)) + requireDeadLetterNotFound(t, err, live.ID) } func TestDeadLetterJobRetryUniqueConflict(t *testing.T) { @@ -170,7 +351,10 @@ func TestDeadLetterJobRetryUniqueConflict(t *testing.T) { require.NoError(t, err) _, err = apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterJobRequest{JobID: job.ID}) - require.ErrorContains(t, err, "another active job has the same unique properties") + var apiErr *apierror.APIError + require.ErrorAs(t, err, &apiErr) + require.Equal(t, http.StatusConflict, apiErr.StatusCode) + require.Equal(t, "Job can't be retried because another active job has the same unique properties. Wait for it to finish before retrying.", apiErr.Message) _, err = bundle.exec.JobDeadLetterGetByID(ctx, &driver.JobDeadLetterGetByIDParams{ID: job.ID, Schema: bundle.schema}) require.NoError(t, err, "a conflict must retain the dead letter job") } @@ -195,6 +379,21 @@ func TestDeadLetterJobSerialization(t *testing.T) { require.Contains(t, string(encoded), `"errors":[]`) } +func TestDeadLetterJobListItemSerialization(t *testing.T) { + t.Parallel() + + finalizedAt := time.Date(2026, time.October, 21, 10, 2, 3, 456_000_000, time.UTC) + encoded, err := json.Marshal(deadLetterJobListItemFromDriver(&riveruidriver.JobDeadLetterListResult{ + ID: 1970670598291982290, Attempt: 3, FinalizedAt: &finalizedAt, Kind: "send_email", MaxAttempts: 5, Queue: "mail", + })) + require.NoError(t, err) + require.JSONEq(t, `{"id":"1970670598291982290","attempt":3,"finalized_at":"2026-10-21T10:02:03.456Z","kind":"send_email","max_attempts":5,"queue":"mail"}`, string(encoded)) + + encoded, err = json.Marshal(deadLetterJobListItemFromDriver(&riveruidriver.JobDeadLetterListResult{ID: 1})) + require.NoError(t, err) + require.JSONEq(t, `{"id":"1","attempt":0,"finalized_at":null,"kind":"","max_attempts":0,"queue":""}`, string(encoded)) +} + func TestDeadLetterJobListRequest(t *testing.T) { t.Parallel() for _, query := range []string{"limit=invalid", "limit=1.5", "limit=9223372036854775808", "ids=bad", "ids=1.5", "ids=9223372036854775808", "priorities=bad", "priorities=32768"} { @@ -212,6 +411,75 @@ func TestDeadLetterJobListRequest(t *testing.T) { require.Equal(t, []string{"urgent", "batch"}, parsed.Tags) } +func TestDeadLetterRequestRejectsInvalidText(t *testing.T) { + t.Parallel() + + extractList := func(r *http.Request) error { return (&deadLetterJobListRequest{}).ExtractRaw(r) } + extractAutocomplete := func(r *http.Request) error { return (&deadLetterAutocompleteRequest{}).ExtractRaw(r) } + for _, tt := range []struct { + param string + extract func(r *http.Request) error + }{ + {param: "kinds", extract: extractList}, + {param: "queues", extract: extractList}, + {param: "tags", extract: extractList}, + {param: "match", extract: extractAutocomplete}, + {param: "exclude", extract: extractAutocomplete}, + } { + // A NUL byte alone and within text, then invalid UTF-8 sequences. + for _, value := range []string{"%00", "send%00email", "%ff", "caf%c3"} { + t.Run(tt.param+"="+value, func(t *testing.T) { + t.Parallel() + req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/?facet=job_kind&kinds=valid&"+tt.param+"="+value, nil) + err := tt.extract(req) + var badRequest *apierror.BadRequest + require.ErrorAs(t, err, &badRequest) + require.Equal(t, http.StatusBadRequest, badRequest.StatusCode) + require.Equal(t, "Invalid `"+tt.param+"` value: must be valid UTF-8 without NUL bytes.", badRequest.Message) + }) + } + } + + // Other control characters and non-ASCII text are valid in both databases. + req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/?kinds=caf%C3%A9&queues=tab%09bed&tags=%C3%BCber", nil) + var parsed deadLetterJobListRequest + require.NoError(t, parsed.ExtractRaw(req)) + require.Equal(t, []string{"café"}, parsed.Kinds) + require.Equal(t, []string{"tab\tbed"}, parsed.Queues) + require.Equal(t, []string{"über"}, parsed.Tags) +} + +// Requests go through the mounted endpoint so that a value PostgreSQL rejects +// is shown to produce a bad request rather than an internal server error. +func TestDeadLetterInvalidTextStatus(t *testing.T) { + t.Parallel() + ctx := t.Context() + + listEndpoint, _ := setupEndpoint(ctx, t, NewDeadLetterJobListEndpoint) + autocompleteEndpoint, _ := setupEndpoint(ctx, t, NewDeadLetterAutocompleteEndpoint) + mux := http.NewServeMux() + apiendpoint.Mount(mux, listEndpoint, testMountOpts(t)) + apiendpoint.Mount(mux, autocompleteEndpoint, testMountOpts(t)) + + for _, tt := range []struct { + path string + want int + }{ + {path: "/api/pro/dead-letter-jobs?kinds=%00", want: http.StatusBadRequest}, + {path: "/api/pro/dead-letter-jobs?queues=%00", want: http.StatusBadRequest}, + {path: "/api/pro/dead-letter-jobs?tags=%ff", want: http.StatusBadRequest}, + {path: "/api/pro/dead-letter-jobs?kinds=caf%C3%A9&tags=tab%09bed", want: http.StatusOK}, + {path: "/api/pro/dead-letter-jobs/autocomplete?facet=job_kind&match=%00", want: http.StatusBadRequest}, + {path: "/api/pro/dead-letter-jobs/autocomplete?facet=job_kind&match=%ff", want: http.StatusBadRequest}, + {path: "/api/pro/dead-letter-jobs/autocomplete?facet=job_kind&exclude=%00", want: http.StatusBadRequest}, + {path: "/api/pro/dead-letter-jobs/autocomplete?facet=job_kind&match=caf%C3%A9", want: http.StatusOK}, + } { + recorder := httptest.NewRecorder() + mux.ServeHTTP(recorder, httptest.NewRequestWithContext(ctx, http.MethodGet, tt.path, nil)) + require.Equal(t, tt.want, recorder.Code, "%s: %s", tt.path, recorder.Body.String()) + } +} + func TestDeadLetterAutocomplete(t *testing.T) { t.Parallel() ctx := t.Context() @@ -233,8 +501,9 @@ func TestDeadLetterAutocomplete(t *testing.T) { } require.Equal(t, tt.want, got) } - for _, facet := range []string{"", "tags", "id"} { + for _, facet := range []riveruidriver.JobDeadLetterFacet{"", "tags", "id"} { _, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &deadLetterAutocompleteRequest{Facet: facet}) - require.Error(t, err) + var badRequest *apierror.BadRequest + require.ErrorAs(t, err, &badRequest) } } diff --git a/src/components/CopyableHeadingText.tsx b/src/components/CopyableHeadingText.tsx new file mode 100644 index 00000000..abb0dc44 --- /dev/null +++ b/src/components/CopyableHeadingText.tsx @@ -0,0 +1,62 @@ +import { CheckIcon } from "@heroicons/react/16/solid"; +import { ClipboardIcon } from "@heroicons/react/24/outline"; +import { toastSuccess } from "@services/toast"; +import { capitalize } from "@utils/string"; +import { useState } from "react"; + +type CopyableHeadingTextProps = { + // Names what is copied, in lowercase, such as "job kind". + label: string; + text: string; +}; + +// CopyableHeadingText keeps a heading on one line. Text that doesn't fit +// is cut with an ellipsis, and the copy button and tooltip give the full +// text. +export default function CopyableHeadingText({ + label, + text, +}: CopyableHeadingTextProps) { + const [copied, setCopied] = useState(false); + + const copyToClipboard = () => { + if (!navigator.clipboard?.writeText) { + console.error(`Failed to copy ${label}: Clipboard API unavailable`); + return; + } + + navigator.clipboard.writeText(text).then( + () => { + setCopied(true); + toastSuccess({ message: `${capitalize(label)} copied to clipboard` }); + setTimeout(() => setCopied(false), 2000); + }, + (error: unknown) => { + console.error(`Failed to copy ${label} to clipboard:`, error); + }, + ); + }; + + return ( + + + {text} + + + + ); +} diff --git a/src/components/DeadLetterJobDetail.stories.ts b/src/components/DeadLetterJobDetail.stories.ts new file mode 100644 index 00000000..65ace602 --- /dev/null +++ b/src/components/DeadLetterJobDetail.stories.ts @@ -0,0 +1,29 @@ +import type { Meta, StoryObj } from "@storybook/react-vite"; + +import { jobFactory } from "@test/factories/job"; + +import DeadLetterJobDetail from "./DeadLetterJobDetail"; + +const meta: Meta = { + component: DeadLetterJobDetail, + title: "Pages/DeadLetterJobDetail", +}; + +export default meta; + +type Story = StoryObj; + +export const Default: Story = { + args: { + job: jobFactory.discarded().build(), + retry: () => {}, + retryPending: false, + }, +}; + +export const Retrying: Story = { + args: { + ...Default.args, + retryPending: true, + }, +}; diff --git a/src/components/DeadLetterJobDetail.tsx b/src/components/DeadLetterJobDetail.tsx index 825df79a..a6955e3f 100644 --- a/src/components/DeadLetterJobDetail.tsx +++ b/src/components/DeadLetterJobDetail.tsx @@ -1,12 +1,8 @@ import { ArrowUturnLeftIcon } from "@heroicons/react/24/outline"; import { Job } from "@services/jobs"; -import { Button } from "./Button"; -import JobAttempts from "./JobAttempts"; -import JSONTextView from "./JSONTextView"; -import JSONView from "./JSONView"; -import RelativeTimeFormatter from "./RelativeTimeFormatter"; -import TopNavTitleOnly from "./TopNavTitleOnly"; +import ButtonForGroup from "./ButtonForGroup"; +import { JobDetailView } from "./JobDetail"; export default function DeadLetterJobDetail({ job, @@ -18,72 +14,20 @@ export default function DeadLetterJobDetail({ retryPending: boolean; }) { return ( - <> - -
- -
-
-

{job.kind}

-

- ID: {job.id.toString()} -

-
- -
-

- Retrying removes this job from the dead letter queue and enqueues a - new job with the same args and a fresh attempt history. -

-
-
-
Queue
-
{job.queue}
-
-
-
Attempts
-
- {job.attempt} / {job.maxAttempts} -
-
-
-
Created
-
- -
-
-
-
Discarded
-
- {job.finalizedAt ? ( - - ) : ( - "–" - )} -
-
-
-
-
-
Args
-
- -
-
-
-
Metadata
-
- -
-
- -
-
- + + + + } + job={job} + notice="Retrying removes this job from the dead letter queue and enqueues a new job with the same args and a fresh attempt history." + title="Dead Letter Job" + /> ); } diff --git a/src/components/DeadLetterJobList.stories.ts b/src/components/DeadLetterJobList.stories.ts new file mode 100644 index 00000000..3ad5c1f8 --- /dev/null +++ b/src/components/DeadLetterJobList.stories.ts @@ -0,0 +1,57 @@ +import type { Meta, StoryObj } from "@storybook/react-vite"; + +import { FilterTypeId } from "@components/job-search/JobSearch"; +import { jobMinimalFactory } from "@test/factories/job"; + +import DeadLetterJobList from "./DeadLetterJobList"; + +const meta: Meta = { + component: DeadLetterJobList, + title: "Pages/DeadLetterJobList", +}; + +export default meta; + +type Story = StoryObj; + +export const Default: Story = { + args: { + canShowFewer: false, + canShowMore: true, + initialFilters: [], + jobs: jobMinimalFactory.discarded().buildList(10), + loading: false, + onFiltersChange: () => {}, + showFewer: () => {}, + showMore: () => {}, + }, +}; + +export const Empty: Story = { + args: { + ...Default.args, + canShowMore: false, + jobs: [], + }, +}; + +export const NoFilterMatches: Story = { + args: { + ...Empty.args, + initialFilters: [ + { + id: FilterTypeId.KIND, + match: "kind:", + typeId: FilterTypeId.KIND, + values: ["missing"], + }, + ], + }, +}; + +export const RefreshFailed: Story = { + args: { + ...Default.args, + loadError: "Couldn't refresh dead letter jobs: Failed to fetch", + }, +}; diff --git a/src/components/DeadLetterJobList.tsx b/src/components/DeadLetterJobList.tsx index 61312a92..f8e0059b 100644 --- a/src/components/DeadLetterJobList.tsx +++ b/src/components/DeadLetterJobList.tsx @@ -1,18 +1,78 @@ +import { Badge } from "@components/Badge"; import { fetchSuggestions } from "@components/job-search/api"; import { Filter, FilterTypeId, JobSearch, } from "@components/job-search/JobSearch"; +import ListEmptyState from "@components/ListEmptyState"; +import ListPagination from "@components/ListPagination"; +import RelativeTimeFormatter from "@components/RelativeTimeFormatter"; +import TopNavTitleOnly from "@components/TopNavTitleOnly"; import { DeadLetterJobListItem, fetchDeadLetterSuggestions, } from "@services/deadLetterJobs"; import { Link } from "@tanstack/react-router"; -import { Button } from "./Button"; -import RelativeTimeFormatter from "./RelativeTimeFormatter"; -import TopNavTitleOnly from "./TopNavTitleOnly"; +type DeadLetterJobListProps = { + canShowFewer: boolean; + canShowMore: boolean; + filterError?: string; + initialFilters: Filter[]; + jobs: DeadLetterJobListItem[]; + // Set when a refresh fails. The last loaded jobs stay visible. + loadError?: string; + loading: boolean; + onFiltersChange: (filters: Filter[]) => void; + showFewer: () => void; + showMore: () => void; +}; + +const DeadLetterJobRow = ({ job }: { job: DeadLetterJobListItem }) => ( +
  • +
    +
    +
    +
    +
    +

    + + {job.kind}{" "} + + #{job.id.toString()} + + +

    +
    + {job.finalizedAt ? ( + + ) : ( + "–" + )} +
    +
    +
    +
    + {job.attempt.toString()} + / + {job.maxAttempts.toString()} +
    + + {job.queue} + +
    +
    +
  • +); export default function DeadLetterJobList({ canShowFewer, @@ -20,121 +80,76 @@ export default function DeadLetterJobList({ filterError, initialFilters, jobs, + loadError, loading, onFiltersChange, showFewer, showMore, -}: { - canShowFewer: boolean; - canShowMore: boolean; - filterError?: string; - initialFilters: Filter[]; - jobs: DeadLetterJobListItem[]; - loading: boolean; - onFiltersChange: (filters: Filter[]) => void; - showFewer: () => void; - showMore: () => void; -}) { +}: DeadLetterJobListProps) { + const alert = filterError ?? loadError; + return ( <> -
    -

    - Discarded jobs retained in the dead letter queue. Select a job to - inspect its failures and retry it. -

    - - {filterError && ( -

    - {filterError} -

    - )} +
    +
    +
    + +
    + {alert && ( +

    + {alert} +

    + )} +
    {loading ? ( -

    Loading…

    - ) : jobs.length === 0 ? ( -

    - {initialFilters.length - ? "No dead letter jobs match these filters." - : "No dead letter jobs."} +

    + Loading…

    + ) : jobs.length === 0 ? ( +
    + {initialFilters.length ? ( + + ) : ( + + Discarded jobs appear here when the dead letter queue is + enabled with Config.DeadLetter.Enabled in River + Pro. + + } + title="No dead letter jobs." + /> + )} +
    ) : ( <> -
    - - - - - - - - - - - {jobs.map((job) => ( - - - - - - - ))} - -
    - Job - - Queue - - Attempts - - Discarded -
    - - {job.kind}{" "} - #{job.id.toString()} - - {job.queue} - {job.attempt} / {job.maxAttempts} - - {job.finalizedAt ? ( - - ) : ( - "–" - )} -
    -
    - + {jobs.map((job) => ( + + ))} + + )}
    @@ -146,13 +161,15 @@ async function fetchSuggestionsForDeadLetters( type: FilterTypeId, query: string, selectedValues: string[], + signal?: AbortSignal, ): Promise { if (type === FilterTypeId.KIND || type === FilterTypeId.QUEUE) { return fetchDeadLetterSuggestions( type === FilterTypeId.KIND ? "job_kind" : "queue_name", query, selectedValues, + signal, ); } - return fetchSuggestions(type, query, selectedValues); + return fetchSuggestions(type, query, selectedValues, signal); } diff --git a/src/components/JobDetail.tsx b/src/components/JobDetail.tsx index 3040348e..ce26a29e 100644 --- a/src/components/JobDetail.tsx +++ b/src/components/JobDetail.tsx @@ -1,6 +1,7 @@ import { Badge } from "@components/Badge"; import ButtonForGroup from "@components/ButtonForGroup"; import ConfirmationDialog from "@components/ConfirmationDialog"; +import CopyableHeadingText from "@components/CopyableHeadingText"; import JobAttempts from "@components/JobAttempts"; import JobTimeline from "@components/JobTimeline"; import JSONTextView from "@components/JSONTextView"; @@ -17,7 +18,7 @@ import { Job, JobWithKnownMetadata } from "@services/jobs"; import { JobState } from "@services/types"; import { Link } from "@tanstack/react-router"; import { capitalize } from "@utils/string"; -import { FormEvent, useState } from "react"; +import { FormEvent, ReactNode, useState } from "react"; type JobDetailProps = { cancel: () => void; @@ -26,12 +27,44 @@ type JobDetailProps = { retry: () => void; }; +type JobDetailViewProps = { + actions: ReactNode; + job: Job; + // Shown under the heading, for context that applies to the whole job. + notice?: ReactNode; + title: string; +}; + export default function JobDetail({ cancel, deleteFn, job, retry, }: JobDetailProps) { + return ( + + } + job={job} + title="Job Details" + /> + ); +} + +// JobDetailView is the job detail page without its actions, so that other +// panes showing a job (like dead letters) share one layout. +export function JobDetailView({ + actions, + job, + notice, + title, +}: JobDetailViewProps) { const featureEnabledWorkflows = useFeature("ENABLE_WORKFLOWS", true); let jobWithMetadata: JobWithKnownMetadata | undefined; @@ -41,28 +74,26 @@ export default function JobDetail({ return ( <> - +
    {/* Heading */}
    -
    -

    - - {job.kind} - +
    +

    +

    ID: {job.id.toString()}

    + {notice && ( +

    + {notice} +

    + )}
    - + {actions}

    diff --git a/src/components/JobList.tsx b/src/components/JobList.tsx index d8a347fd..024ef1da 100644 --- a/src/components/JobList.tsx +++ b/src/components/JobList.tsx @@ -1,11 +1,12 @@ import { Badge } from "@components/Badge"; -import { Button } from "@components/Button"; import ButtonForGroup from "@components/ButtonForGroup"; import ConfirmationDialog from "@components/ConfirmationDialog"; import { CustomCheckbox } from "@components/CustomCheckbox"; import { Dropdown, DropdownItem, DropdownMenu } from "@components/Dropdown"; import { Filter, JobSearch } from "@components/job-search/JobSearch"; import { JobStateFilters } from "@components/JobStateFilters"; +import ListEmptyState from "@components/ListEmptyState"; +import ListPagination from "@components/ListPagination"; import RelativeTimeFormatter from "@components/RelativeTimeFormatter"; import TopNav from "@components/TopNav"; import { useFeatures } from "@contexts/Features.hook"; @@ -152,35 +153,6 @@ const JobListItem = ({ ); }; -type EmptySetIconProps = React.ComponentProps<"svg">; - -const EmptySetIcon = (props: EmptySetIconProps) => ( - - - - -); - -const EmptyState = () => ( -
    - -

    - No jobs found -

    -

    - Try adjusting your filters to find what you're looking for. -

    -
    -); - export type JobRowsProps = { cancelJobs: (jobIDs: bigint[]) => void; canShowFewer: boolean; @@ -408,7 +380,10 @@ const JobRows = ({
    {jobs.length === 0 ? (
    - +
    ) : ( <> @@ -426,27 +401,12 @@ const JobRows = ({ /> ))} - + )} diff --git a/src/components/JobNotFound.tsx b/src/components/JobNotFound.tsx index 8716d857..6fbd0a63 100644 --- a/src/components/JobNotFound.tsx +++ b/src/components/JobNotFound.tsx @@ -1,7 +1,7 @@ import { JobState } from "@services/types"; import { Button } from "./Button"; -import TopNavTitleOnly from "./TopNavTitleOnly"; +import NotFoundPage from "./NotFoundPage"; type JobNotFoundProps = { jobId: bigint; @@ -9,27 +9,19 @@ type JobNotFoundProps = { export default function JobNotFound({ jobId }: JobNotFoundProps) { return ( - <> - - -
    -
    -

    - 404 -

    -

    - Job not found -

    -

    - Sorry, there is no job with ID {jobId.toString()}. -

    -
    - -
    -
    -
    - + + Return to job list + + } + description={ + <> + Sorry, there is no job with ID {jobId.toString()}. + + } + heading="Job not found" + title="Job Details" + /> ); } diff --git a/src/components/Layout.tsx b/src/components/Layout.tsx index 57c7e9c0..a5634788 100644 --- a/src/components/Layout.tsx +++ b/src/components/Layout.tsx @@ -16,11 +16,16 @@ import { RectangleGroupIcon, XMarkIcon, } from "@heroicons/react/24/outline"; +import { retainedJobSearchKeys } from "@routes/jobs/index.schema"; import { Link, useRouterState } from "@tanstack/react-router"; import { Fragment, PropsWithChildren, useMemo } from "react"; type LayoutProps = PropsWithChildren; +const clearedJobSearch = Object.fromEntries( + retainedJobSearchKeys.map((key) => [key, undefined]), +); + const Layout = ({ children }: LayoutProps) => { const { open: sidebarOpen, setOpen: setSidebarOpen } = useSidebarSetting(); @@ -37,18 +42,15 @@ const Layout = ({ children }: LayoutProps) => { icon: QueueListIcon, name: "Jobs", // Explicitly clear retained Jobs filters when entering from another pane. - search: inJobsPane - ? {} - : { - id: undefined, - kind: undefined, - limit: undefined, - priority: undefined, - queue: undefined, - tags: undefined, - }, + search: inJobsPane ? {} : clearedJobSearch, }, { href: "/queues", icon: InboxStackIcon, name: "Queues" }, + { + hidden: !features.workflowQueries, + href: "/workflows", + icon: RectangleGroupIcon, + name: "Workflows", + }, { hidden: !features.deadLetterJobs, href: "/dead-letter-jobs", @@ -56,12 +58,6 @@ const Layout = ({ children }: LayoutProps) => { name: "Dead Letters", search: {}, }, - { - hidden: !features.workflowQueries, - href: "/workflows", - icon: RectangleGroupIcon, - name: "Workflows", - }, { hidden: !features.durablePeriodicJobs, href: "/periodic-jobs", diff --git a/src/components/ListEmptyState.tsx b/src/components/ListEmptyState.tsx new file mode 100644 index 00000000..8e1e8bc3 --- /dev/null +++ b/src/components/ListEmptyState.tsx @@ -0,0 +1,37 @@ +import { ReactNode } from "react"; + +type EmptySetIconProps = React.ComponentProps<"svg">; + +const EmptySetIcon = (props: EmptySetIconProps) => ( + + + + +); + +type ListEmptyStateProps = { + description: ReactNode; + title: string; +}; + +const ListEmptyState = ({ description, title }: ListEmptyStateProps) => ( +
    + +

    + {title} +

    +

    + {description} +

    +
    +); + +export default ListEmptyState; diff --git a/src/components/ListPagination.tsx b/src/components/ListPagination.tsx new file mode 100644 index 00000000..9843ce4c --- /dev/null +++ b/src/components/ListPagination.tsx @@ -0,0 +1,39 @@ +import { Button } from "@components/Button"; + +type ListPaginationProps = { + canShowFewer: boolean; + canShowMore: boolean; + showFewer: () => void; + showMore: () => void; +}; + +const ListPagination = ({ + canShowFewer, + canShowMore, + showFewer, + showMore, +}: ListPaginationProps) => ( + +); + +export default ListPagination; diff --git a/src/components/NotFoundPage.tsx b/src/components/NotFoundPage.tsx new file mode 100644 index 00000000..a1b4cfbb --- /dev/null +++ b/src/components/NotFoundPage.tsx @@ -0,0 +1,40 @@ +import { ReactNode } from "react"; + +import TopNavTitleOnly from "./TopNavTitleOnly"; + +type NotFoundPageProps = { + action: ReactNode; + description: ReactNode; + heading: string; + title: string; +}; + +export default function NotFoundPage({ + action, + description, + heading, + title, +}: NotFoundPageProps) { + return ( + <> + + +
    +
    +

    + 404 +

    +

    + {heading} +

    +

    + {description} +

    +
    + {action} +
    +
    +
    + + ); +} diff --git a/src/components/WorkflowDetail.tsx b/src/components/WorkflowDetail.tsx index 3dfeaf72..fd3ad668 100644 --- a/src/components/WorkflowDetail.tsx +++ b/src/components/WorkflowDetail.tsx @@ -1,4 +1,5 @@ import ButtonForGroup from "@components/ButtonForGroup"; +import CopyableHeadingText from "@components/CopyableHeadingText"; import { DurationCompact } from "@components/DurationCompact"; import { Subheading } from "@components/Heading"; import { RunningSpinnerIcon } from "@components/icons/jobStateIcons"; @@ -18,12 +19,7 @@ import WorkflowGateInspector, { } from "@components/WorkflowGateInspector"; import { useFeatures } from "@contexts/Features.hook"; // (Dialog is now encapsulated in RetryWorkflowDialog) -import { CheckIcon } from "@heroicons/react/16/solid"; -import { - ArrowPathIcon, - ClipboardIcon, - LinkIcon, -} from "@heroicons/react/24/outline"; +import { ArrowPathIcon, LinkIcon } from "@heroicons/react/24/outline"; import { CheckCircleIcon, PlayCircleIcon, @@ -31,7 +27,6 @@ import { TrashIcon, XCircleIcon, } from "@heroicons/react/24/solid"; -import { toastSuccess } from "@services/toast"; import { type Heroicon, JobState } from "@services/types"; import { Workflow, @@ -122,7 +117,6 @@ export default function WorkflowDetail({ // Modal state for retry const [retryOpen, setRetryOpen] = useState(false); const [retryMode, setRetryMode] = useState(); - const [workflowNameCopied, setWorkflowNameCopied] = useState(false); if (!features.workflowQueries) { return ( @@ -156,52 +150,7 @@ export default function WorkflowDetail({

    - - - {workflowName} - - - +

    ID: {workflow.id} diff --git a/src/components/job-search/JobSearch.test.tsx b/src/components/job-search/JobSearch.test.tsx index 684081b0..64b346e7 100644 --- a/src/components/job-search/JobSearch.test.tsx +++ b/src/components/job-search/JobSearch.test.tsx @@ -170,6 +170,29 @@ describe("JobSearch", () => { expect(input).toHaveValue("kind:batch"); }); + it("aborts a suggestion request when a newer one replaces it", async () => { + const signals: AbortSignal[] = []; + const fetchSuggestions = vi.fn( + ( + _type: FilterTypeId, + _query: string, + _selected: string[], + signal?: AbortSignal, + ) => { + if (signal) signals.push(signal); + return new Promise(() => {}); + }, + ); + render(); + const input = screen.getByTestId("job-search-input"); + fireEvent.change(input, { target: { value: "kind:b" } }); + await waitFor(() => expect(signals).toHaveLength(1)); + fireEvent.change(input, { target: { value: "kind:ba" } }); + await waitFor(() => expect(signals).toHaveLength(2)); + expect(signals[0].aborted).toBe(true); + expect(signals[1].aborted).toBe(false); + }); + it("notifies parent of filter changes", async () => { const onFiltersChange = vi.fn(); render(); diff --git a/src/components/job-search/JobSearch.tsx b/src/components/job-search/JobSearch.tsx index fc7b2155..887ab2f5 100644 --- a/src/components/job-search/JobSearch.tsx +++ b/src/components/job-search/JobSearch.tsx @@ -22,12 +22,14 @@ export interface JobSearchProps { * @param filterTypeId - The ID of the filter type * @param query - The search query * @param selectedValues - Values already selected for this filter + * @param signal - Aborted when a newer request replaces this one * @returns Promise resolving to an array of suggestion strings */ fetchSuggestions?: ( filterTypeId: JobFilterTypeID, query: string, selectedValues: string[], + signal?: AbortSignal, ) => Promise; /** * Initial filters to display diff --git a/src/components/job-search/api.ts b/src/components/job-search/api.ts index 93b4d107..fe4f53ad 100644 --- a/src/components/job-search/api.ts +++ b/src/components/job-search/api.ts @@ -8,6 +8,7 @@ export async function fetchSuggestions( filterTypeId: JobFilterTypeID, query: string, selectedValues: string[], + signal?: AbortSignal, ): Promise { if (filterTypeId === JobFilterTypeID.PRIORITY) { // Priority is a hardcoded list of values—we just need to filter out @@ -31,7 +32,13 @@ export async function fetchSuggestions( } if (fetchType) { - return await fetchAutocomplete(fetchType, query, undefined, selectedValues); + return await fetchAutocomplete( + fetchType, + query, + undefined, + selectedValues, + signal, + ); } return []; diff --git a/src/components/job-search/hooks.ts b/src/components/job-search/hooks.ts index b19c6625..60b1d752 100644 --- a/src/components/job-search/hooks.ts +++ b/src/components/job-search/hooks.ts @@ -21,6 +21,7 @@ interface UseFilterInputProps { filterTypeId: JobFilterTypeID, query: string, selectedValues: string[], + signal?: AbortSignal, ) => Promise; initialFilters: JobFilter[]; onFiltersChange?: (filters: JobFilter[]) => void; @@ -32,6 +33,7 @@ interface UseSuggestionsProps { filterTypeId: JobFilterTypeID, query: string, selectedValues: string[], + signal?: AbortSignal, ) => Promise; suppressSuggestions: boolean; } @@ -237,10 +239,15 @@ export function useSuggestions({ }); const latestSuggestionRequestRef = useRef(0); + const suggestionAbortRef = useRef(undefined); const updateSuggestions = useCallback( async (text: string, cursor: number) => { const requestId = ++latestSuggestionRequestRef.current; + // A newer request replaces this one, so stop the old fetch instead of + // leaving its query running on the server. + suggestionAbortRef.current?.abort(); + suggestionAbortRef.current = undefined; const context = analyzeAutocompleteContext( text, cursor, @@ -289,10 +296,13 @@ export function useSuggestions({ let suggestions: string[] = []; if (context.type === "filter-value" && context.filterTypeId) { + const abortController = new AbortController(); + suggestionAbortRef.current = abortController; suggestions = await fetchSuggestions( context.filterTypeId, context.currentToken || "", context.existingValues || [], + abortController.signal, ); } diff --git a/src/routes/__root.tsx b/src/routes/__root.tsx index 8e12b44a..03f56222 100644 --- a/src/routes/__root.tsx +++ b/src/routes/__root.tsx @@ -1,3 +1,5 @@ +import { Button } from "@components/Button"; +import NotFoundPage from "@components/NotFoundPage"; import { Root } from "@components/Root"; import { type Features, featuresKey, getFeatures } from "@services/features"; import { QueryClient } from "@tanstack/react-query"; @@ -18,6 +20,20 @@ export const Route = createRootRouteWithContext<{ }; }, component: RootComponent, + // Also shown for a pane this server doesn't support, such as a Pro-only + // route opened on an OSS server. + notFoundComponent: () => ( + + Return to job list + + } + description="This page doesn't exist, or this River UI server doesn't support it." + heading="Page not found" + title="Not Found" + /> + ), }); function RootComponent() { diff --git a/src/routes/dead-letter-jobs.test.tsx b/src/routes/dead-letter-jobs.test.tsx index 78c25b77..18de4556 100644 --- a/src/routes/dead-letter-jobs.test.tsx +++ b/src/routes/dead-letter-jobs.test.tsx @@ -1,7 +1,10 @@ import Layout from "@components/Layout"; import { FeaturesContext } from "@contexts/Features"; import { SidebarSettingProvider } from "@contexts/SidebarSetting.provider"; -import { deadLetterJobSearchSchema } from "@routes/dead-letter-jobs/index.schema"; +import { + deadLetterJobRouteSearchSchema, + deadLetterJobSearchSchema, +} from "@routes/dead-letter-jobs/index.schema"; import { listDeadLetterJobsKey } from "@services/deadLetterJobs"; import { featuresKey } from "@services/features"; import { Job, listJobs, listJobsKey } from "@services/jobs"; @@ -30,7 +33,7 @@ import { } from "@testing-library/react"; import { APIResponseError, NotFoundError } from "@utils/api"; import qs from "query-string"; -import { beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import "../global-type-overrides"; import { routeTree } from "../routeTree.gen"; @@ -126,6 +129,10 @@ async function renderRoute(path: string, pro = true) { return { queryClient, router }; } +afterEach(() => { + vi.useRealTimers(); +}); + beforeEach(() => { vi.clearAllMocks(); fetchDeadLetterSuggestions.mockResolvedValue([]); @@ -228,9 +235,46 @@ describe("Dead letter jobs", () => { ).not.toBeInTheDocument(); expect(listDeadLetterJobs).not.toHaveBeenCalled(); expect(getDeadLetterJob).not.toHaveBeenCalled(); + expect( + screen.getByRole("heading", { name: "Page not found" }), + ).toBeInTheDocument(); + expect( + screen.getByRole("link", { name: "Return to job list" }), + ).toHaveAttribute("href", "/jobs"); }, ); + it("opens the list when a bookmarked URL has filters it can't parse", async () => { + const { router } = await renderRoute( + "/dead-letter-jobs?id=abc&priority=5&queue=mail", + ); + expect( + await screen.findByRole("link", { name: `send_email #${job.id}` }), + ).toBeInTheDocument(); + expect(screen.getByTestId("job-search-input")).toHaveValue("queue:mail"); + expect(router.state.location.search).toEqual({ queue: "mail" }); + expect(listDeadLetterJobs).toHaveBeenCalledWith( + expect.objectContaining({ + queryKey: listDeadLetterJobsKey(20, { queues: ["mail"] }), + }), + ); + }); + + it("keeps the loaded jobs visible when a refresh fails", async () => { + const { queryClient } = await renderRoute("/dead-letter-jobs"); + listDeadLetterJobs.mockRejectedValue(new Error("Failed to fetch")); + await act(async () => { + await queryClient.refetchQueries({ queryKey: listDeadLetterJobsKey() }); + }); + expect(await screen.findByRole("alert")).toHaveTextContent( + "Couldn't refresh dead letter jobs: Failed to fetch", + ); + expect( + screen.getByRole("link", { name: `send_email #${job.id}` }), + ).toBeInTheDocument(); + expect(screen.getByTestId("job-search-input")).toBeInTheDocument(); + }); + it("shows the Pro navigation and opens the selected dead letter job", async () => { const { router } = await renderRoute("/dead-letter-jobs"); expect( @@ -397,6 +441,7 @@ describe("Dead letter jobs", () => { }); it("restores the search field on browser Back", async () => { + vi.useFakeTimers({ shouldAdvanceTime: true }); const { router } = await renderRoute("/dead-letter-jobs?queue=mail"); await act(async () => { await router.navigate({ @@ -413,8 +458,9 @@ describe("Dead letter jobs", () => { await waitFor(() => expect(screen.getByTestId("job-search-input")).toHaveValue("queue:mail"), ); + // Run the filter debounce to confirm it doesn't rewrite the restored URL. await act(async () => { - await new Promise((resolve) => setTimeout(resolve, 300)); + await vi.advanceTimersByTimeAsync(300); }); expect(router.state.location.search).toEqual({ queue: "mail" }); }); @@ -442,6 +488,7 @@ describe("Dead letter jobs", () => { "job_kind", "archive_", [], + expect.any(AbortSignal), ), ); fireEvent.change(input, { target: { value: "queue:mail,re" } }); @@ -450,6 +497,7 @@ describe("Dead letter jobs", () => { "queue_name", "re", ["mail"], + expect.any(AbortSignal), ), ); }); @@ -473,6 +521,14 @@ describe("Dead letter jobs", () => { expect(deadLetterJobSearchSchema.safeParse({ priority }).success).toBe( false, ); + // The route schema drops only the filters it can't parse. + expect( + deadLetterJobRouteSearchSchema.parse({ + id: "bad", + kind: "report", + priority: "5", + }), + ).toEqual({ limit: 20, kind: ["report"] }); }); it("retries through the Pro API and opens the newly created job", async () => { diff --git a/src/routes/dead-letter-jobs/$jobId.tsx b/src/routes/dead-letter-jobs/$jobId.tsx index fea226cd..b91311aa 100644 --- a/src/routes/dead-letter-jobs/$jobId.tsx +++ b/src/routes/dead-letter-jobs/$jobId.tsx @@ -1,6 +1,6 @@ import { Button } from "@components/Button"; import DeadLetterJobDetail from "@components/DeadLetterJobDetail"; -import TopNavTitleOnly from "@components/TopNavTitleOnly"; +import NotFoundPage from "@components/NotFoundPage"; import { useRefreshSetting } from "@contexts/RefreshSettings.hook"; import { refreshQueryOptions } from "@contexts/RefreshSettings.query"; import { @@ -9,7 +9,7 @@ import { listDeadLetterJobsKey, retryDeadLetterJob, } from "@services/deadLetterJobs"; -import { getJobKey } from "@services/jobs"; +import { getJobKey, listJobsKeyPrefix } from "@services/jobs"; import { countsByStateKey } from "@services/states"; import { toastError, toastSuccess } from "@services/toast"; import { @@ -34,15 +34,16 @@ export const Route = createFileRoute("/dead-letter-jobs/$jobId")({ queryClient.ensureQueryData(jobQueryOptions), errorComponent: ({ error }) => error instanceof NotFoundError ? ( - <> - -

    -

    {error.message}

    - -
    - + } + description={error.message} + heading="Dead letter job not found" + title="Dead Letter Job" + /> ) : ( ), @@ -68,7 +69,7 @@ export function DeadLetterJobComponent() { toastSuccess({ message: "Job enqueued for retry", duration: 2000 }); await Promise.all([ queryClient.invalidateQueries({ queryKey: listDeadLetterJobsKey() }), - queryClient.invalidateQueries({ queryKey: ["listJobs"] }), + queryClient.invalidateQueries({ queryKey: listJobsKeyPrefix }), queryClient.invalidateQueries({ queryKey: countsByStateKey() }), ]); await navigate({ to: "/jobs/$jobId", params: { jobId: job.id } }); @@ -83,7 +84,11 @@ export function DeadLetterJobComponent() { throwOnError: true, }); - if (!query.data) return

    Loading…

    ; + if (!query.data) { + return ( +

    Loading…

    + ); + } return ( ; +// The route drops a filter it can't parse, as the Jobs list does, so that a +// hand-edited or stale URL still opens the list. Edits in the search field +// use the strict schema above and show the problem instead. +export const deadLetterJobRouteSearchSchema = deadLetterJobSearchSchema.extend({ + id: ids.catch(undefined), + kind: strings.catch(undefined), + priority: priorities.catch(undefined), + queue: strings.catch(undefined), + tags: strings.catch(undefined), +}); + export const searchToDeadLetterFilters = ( search: DeadLetterJobSearch, ): DeadLetterJobFilters => ({ diff --git a/src/routes/dead-letter-jobs/index.tsx b/src/routes/dead-letter-jobs/index.tsx index e392d753..0212ecf5 100644 --- a/src/routes/dead-letter-jobs/index.tsx +++ b/src/routes/dead-letter-jobs/index.tsx @@ -13,6 +13,7 @@ import { createFileRoute, stripSearchParams } from "@tanstack/react-router"; import { useCallback, useMemo, useState } from "react"; import { + deadLetterJobRouteSearchSchema, DeadLetterJobSearch, deadLetterJobSearchSchema, inputFiltersToSearch, @@ -30,7 +31,7 @@ const jobsQueryOptions = (search: DeadLetterJobSearch) => }); export const Route = createFileRoute("/dead-letter-jobs/")({ - validateSearch: deadLetterJobSearchSchema, + validateSearch: deadLetterJobRouteSearchSchema, search: { middlewares: [stripSearchParams({ limit: defaultDeadLetterJobsLimit })], }, @@ -50,7 +51,6 @@ function DeadLetterJobsIndexComponent() { const query = useQuery({ ...jobsQueryOptions(search), ...refreshQueryOptions(intervalMs), - throwOnError: true, }); const onFiltersChange = useCallback( (filters: Filter[]) => { @@ -78,6 +78,11 @@ function DeadLetterJobsIndexComponent() { filterError={filterError} initialFilters={initialFilters} jobs={query.data ?? []} + loadError={ + query.isError + ? `Couldn't refresh dead letter jobs: ${query.error.message}` + : undefined + } loading={query.isLoading} onFiltersChange={onFiltersChange} showFewer={() => diff --git a/src/routes/jobs/index.schema.ts b/src/routes/jobs/index.schema.ts index ab1b4064..f4978cf7 100644 --- a/src/routes/jobs/index.schema.ts +++ b/src/routes/jobs/index.schema.ts @@ -5,6 +5,17 @@ export const minimumLimit = 20; export const defaultLimit = 20; export const maximumLimit = 200; +// Search params kept when moving between job states. Other panes clear +// them when they link to Jobs. +export const retainedJobSearchKeys = [ + "id", + "kind", + "limit", + "priority", + "queue", + "tags", +] as const; + // Define default values export const defaultValues = { limit: defaultLimit, diff --git a/src/routes/jobs/index.tsx b/src/routes/jobs/index.tsx index 69c4fe0f..268afbce 100644 --- a/src/routes/jobs/index.tsx +++ b/src/routes/jobs/index.tsx @@ -6,7 +6,11 @@ import { refreshQueryOptions, } from "@contexts/RefreshSettings.query"; import { useRetryJobs } from "@hooks/use-retry-jobs"; -import { defaultValues, jobSearchSchema } from "@routes/jobs/index.schema"; +import { + defaultValues, + jobSearchSchema, + retainedJobSearchKeys, +} from "@routes/jobs/index.schema"; import { cancelJobs, deleteJobs, @@ -50,7 +54,7 @@ export const Route = createFileRoute("/jobs/")({ search: { middlewares: [ stripSearchParams(defaultValues), - retainSearchParams(["id", "kind", "limit", "priority", "queue", "tags"]), + retainSearchParams([...retainedJobSearchKeys]), ], }, beforeLoad: async ({ context }) => { diff --git a/src/services/deadLetterJobs.ts b/src/services/deadLetterJobs.ts index ea1bcfa8..7a8d5b2e 100644 --- a/src/services/deadLetterJobs.ts +++ b/src/services/deadLetterJobs.ts @@ -95,12 +95,13 @@ export async function fetchDeadLetterSuggestions( facet: "job_kind" | "queue_name", match: string, exclude: string[], + signal?: AbortSignal, ): Promise { const query = new URLSearchParams({ facet, match }); exclude.forEach((value) => query.append("exclude", value)); - const response = await API.get>({ - path: "/pro/dead-letter-jobs/autocomplete", - query, - }); + const response = await API.get>( + { path: "/pro/dead-letter-jobs/autocomplete", query }, + { signal }, + ); return response.data; } diff --git a/src/services/jobs.ts b/src/services/jobs.ts index 487620e1..26d6396b 100644 --- a/src/services/jobs.ts +++ b/src/services/jobs.ts @@ -205,6 +205,9 @@ type ListJobsFilters = { tags?: string[]; }; +// Matches every job list query, for invalidation. +export const listJobsKeyPrefix = ["listJobs"] as const; + export const listJobsKey = (args: ListJobsFilters): ListJobsKey => { return [ "listJobs",