diff --git a/CHANGELOG.md b/CHANGELOG.md index 36d9d54d..0dd89c0c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- **Breaking change:** JobList/JobListTx now support querying Jobs by a list of Job Kinds and States. Also allows for filtering by specific timestamp values. Thank you Jos Kraaijeveld (@thatjos)! 🙏🏻 [PR #236](https://github.com/riverqueue/river/pull/236). +- **Breaking change:** There are a number of small breaking changes in the job list API using `JobList`/`JobListTx`: + - Now support querying jobs by a list of Job Kinds and States. Also allows for filtering by specific timestamp values. Thank you Jos Kraaijeveld (@thatjos)! 🙏🏻 [PR #236](https://github.com/riverqueue/river/pull/236). + - Job listing now defaults to ordering by job ID (`JobListOrderByID`) instead of a job timestamp dependent on on requested job state. The previous ordering behavior is still available with `NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)`. [PR #XXX](https://github.com/riverqueue/river/pull/XXX). + - The function `JobListCursorFromJob` no longer needs a sort order parameter. Instead, sort order is determined based on the job list parameters that the cursor is subsequently used with. [PR #XXX](https://github.com/riverqueue/river/pull/XXX). - **Breaking change:** Client `Insert` and `InsertTx` functions now return a `JobInsertResult` struct instead of a `JobRow`. This allows the result to include metadata like the new `UniqueSkippedAsDuplicate` property, so callers can tell whether an inserted job was skipped due to unique constraint. [PR #292](https://github.com/riverqueue/river/pull/292). - **Breaking change:** Client `InsertMany` and `InsertManyTx` now return number of jobs inserted as `int` instead of `int64`. This change was made to make the type in use a little more idiomatic. [PR #293](https://github.com/riverqueue/river/pull/293). - **Breaking change:** `river.JobState*` type aliases have been removed. All job state constants should be accessed through `rivertype.JobState*` instead. [PR #300](https://github.com/riverqueue/river/pull/300). diff --git a/client.go b/client.go index acebd845..6382ac17 100644 --- a/client.go +++ b/client.go @@ -1489,7 +1489,7 @@ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobL } res := &JobListResult{Jobs: jobs} if len(jobs) > 0 { - res.LastCursor = JobListCursorFromJob(jobs[len(jobs)-1], params.sortField) + res.LastCursor = jobListCursorFromJobAndParams(jobs[len(jobs)-1], params) } return res, nil } @@ -1519,7 +1519,7 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara } res := &JobListResult{Jobs: jobs} if len(jobs) > 0 { - res.LastCursor = JobListCursorFromJob(jobs[len(jobs)-1], params.sortField) + res.LastCursor = jobListCursorFromJobAndParams(jobs[len(jobs)-1], params) } return res, nil } diff --git a/client_test.go b/client_test.go index c9ddafa4..708866af 100644 --- a/client_test.go +++ b/client_test.go @@ -1500,7 +1500,24 @@ func Test_Client_JobList(t *testing.T) { require.Equal(t, []int64{job3.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) }) - t.Run("SortsAvailableRetryableAndScheduledJobsByScheduledAt", func(t *testing.T) { + t.Run("DefaultsToOrderingByID", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) + require.NoError(t, err) + require.Equal(t, []int64{job1.ID, job2.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderDesc)) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + }) + + t.Run("OrderByTimeSortsAvailableRetryableAndScheduledJobsByScheduledAt", func(t *testing.T) { t.Parallel() client, bundle := setup(t) @@ -1516,7 +1533,7 @@ func Test_Client_JobList(t *testing.T) { job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(state), ScheduledAt: &now}) job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(state), ScheduledAt: ptrutil.Ptr(now.Add(-5 * time.Second))}) - res, err := client.JobList(ctx, NewJobListParams().States(state)) + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(state)) require.NoError(t, err) require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) @@ -1526,7 +1543,7 @@ func Test_Client_JobList(t *testing.T) { } }) - t.Run("SortsCancelledCompletedAndDiscardedJobsByFinalizedAt", func(t *testing.T) { + t.Run("OrderByTimeSortsCancelledCompletedAndDiscardedJobsByFinalizedAt", func(t *testing.T) { t.Parallel() client, bundle := setup(t) @@ -1542,7 +1559,7 @@ func Test_Client_JobList(t *testing.T) { job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(state), FinalizedAt: ptrutil.Ptr(now.Add(-10 * time.Second))}) job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(state), FinalizedAt: ptrutil.Ptr(now.Add(-15 * time.Second))}) - res, err := client.JobList(ctx, NewJobListParams().States(state)) + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(state)) require.NoError(t, err) require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) @@ -1552,7 +1569,7 @@ func Test_Client_JobList(t *testing.T) { } }) - t.Run("SortsRunningJobsByAttemptedAt", func(t *testing.T) { + t.Run("OrderByTimeSortsRunningJobsByAttemptedAt", func(t *testing.T) { t.Parallel() client, bundle := setup(t) @@ -1561,7 +1578,7 @@ func Test_Client_JobList(t *testing.T) { job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateRunning), AttemptedAt: &now}) job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateRunning), AttemptedAt: ptrutil.Ptr(now.Add(-5 * time.Second))}) - res, err := client.JobList(ctx, NewJobListParams().States(rivertype.JobStateRunning)) + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(rivertype.JobStateRunning)) require.NoError(t, err) require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) @@ -1583,11 +1600,70 @@ func Test_Client_JobList(t *testing.T) { res, err := client.JobList(ctx, nil) require.NoError(t, err) - // sort order is switched by ScheduledAt values: - require.Equal(t, []int64{job2.ID, job3.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + // sort order defaults to ID + require.Equal(t, []int64{job1.ID, job2.ID, job3.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + }) + + t.Run("PaginatesWithAfter_JobListOrderByID", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + job3 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + + res, err := client.JobList(ctx, NewJobListParams().After(JobListCursorFromJob(job1))) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job3.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByID, res.LastCursor.sortField) + require.Equal(t, job3.ID, res.LastCursor.id) + + // No more results + res, err = client.JobList(ctx, NewJobListParams().After(JobListCursorFromJob(job3))) + require.NoError(t, err) + require.Equal(t, []int64{}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Nil(t, res.LastCursor) + + // Descending + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByID, SortOrderDesc).After(JobListCursorFromJob(job3))) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByID, res.LastCursor.sortField) + require.Equal(t, job1.ID, res.LastCursor.id) }) - t.Run("PaginatesWithAfter", func(t *testing.T) { + t.Run("PaginatesWithAfter_JobListOrderByScheduledAt", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + now := time.Now().UTC() + job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{ScheduledAt: &now}) + job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{ScheduledAt: ptrutil.Ptr(now.Add(1 * time.Second))}) + job3 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{ScheduledAt: ptrutil.Ptr(now.Add(2 * time.Second))}) + + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderAsc).After(JobListCursorFromJob(job1))) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job3.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByScheduledAt, res.LastCursor.sortField) + require.Equal(t, job3.ID, res.LastCursor.id) + + // No more results + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderAsc).After(JobListCursorFromJob(job3))) + require.NoError(t, err) + require.Equal(t, []int64{}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Nil(t, res.LastCursor) + + // Descending + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderDesc).After(JobListCursorFromJob(job3))) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByScheduledAt, res.LastCursor.sortField) + require.Equal(t, job1.ID, res.LastCursor.id) + }) + + t.Run("PaginatesWithAfter_JobListOrderByTime", func(t *testing.T) { t.Parallel() client, bundle := setup(t) @@ -1600,26 +1676,23 @@ func Test_Client_JobList(t *testing.T) { job5 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateCompleted), ScheduledAt: ptrutil.Ptr(now.Add(-7 * time.Second)), FinalizedAt: ptrutil.Ptr(now.Add(-5 * time.Second))}) job6 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateCompleted), ScheduledAt: ptrutil.Ptr(now.Add(-7 * time.Second)), FinalizedAt: &now}) - res, err := client.JobList(ctx, NewJobListParams().States(rivertype.JobStateAvailable).After(JobListCursorFromJob(job1, JobListOrderByTime))) + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(rivertype.JobStateAvailable).After(JobListCursorFromJob(job1))) require.NoError(t, err) require.Equal(t, []int64{job2.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByTime, res.LastCursor.sortField) require.Equal(t, job2.ID, res.LastCursor.id) - res, err = client.JobList(ctx, NewJobListParams().States(rivertype.JobStateRunning).After(JobListCursorFromJob(job3, JobListOrderByTime))) + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(rivertype.JobStateRunning).After(JobListCursorFromJob(job3))) require.NoError(t, err) require.Equal(t, []int64{job4.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByTime, res.LastCursor.sortField) require.Equal(t, job4.ID, res.LastCursor.id) - res, err = client.JobList(ctx, NewJobListParams().States(rivertype.JobStateCompleted).After(JobListCursorFromJob(job5, JobListOrderByTime))) + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(rivertype.JobStateCompleted).After(JobListCursorFromJob(job5))) require.NoError(t, err) require.Equal(t, []int64{job6.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByTime, res.LastCursor.sortField) require.Equal(t, job6.ID, res.LastCursor.id) - - res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderAsc).After(JobListCursorFromJob(job4, JobListOrderByScheduledAt))) - require.NoError(t, err) - require.Equal(t, []int64{job1.ID, job3.ID, job2.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) - require.Equal(t, JobListOrderByScheduledAt, res.LastCursor.sortField) - require.Equal(t, job2.ID, res.LastCursor.id) }) t.Run("MetadataOnly", func(t *testing.T) { diff --git a/job_list_params.go b/job_list_params.go index 77b08c0a..f8637f44 100644 --- a/job_list_params.go +++ b/job_list_params.go @@ -9,6 +9,7 @@ import ( "time" "github.com/riverqueue/river/internal/dblist" + "github.com/riverqueue/river/internal/util/ptrutil" "github.com/riverqueue/river/rivertype" ) @@ -16,33 +17,48 @@ import ( // job list query. type JobListCursor struct { id int64 + job *rivertype.JobRow // used for JobListCursorFromJob path; not serialized kind string queue string sortField JobListOrderByField - time time.Time + time time.Time // may be empty } // JobListCursorFromJob creates a JobListCursor from a JobRow. -func JobListCursorFromJob(job *rivertype.JobRow, sortField JobListOrderByField) *JobListCursor { - time := job.CreatedAt - switch sortField { +func JobListCursorFromJob(job *rivertype.JobRow) *JobListCursor { + // Other fields are initialized when the cursor is used in After below. + return &JobListCursor{job: job} +} + +func jobListCursorFromJobAndParams(job *rivertype.JobRow, listParams *JobListParams) *JobListCursor { + // A pointer so that we can detect a condition where we accidentally left + // this value unset. + var cursorTime *time.Time + + // Don't include a `default` so `exhaustive` lint can detect omissions. + switch listParams.sortField { + case JobListOrderByID: + cursorTime = ptrutil.Ptr(time.Time{}) case JobListOrderByTime: - time = jobListTimeValue(job) + cursorTime = ptrutil.Ptr(jobListTimeValue(job)) case JobListOrderByFinalizedAt: if job.FinalizedAt != nil { - time = *job.FinalizedAt + cursorTime = job.FinalizedAt } case JobListOrderByScheduledAt: - time = job.ScheduledAt - default: + cursorTime = &job.ScheduledAt + } + + if cursorTime == nil { panic("invalid sort field") } + return &JobListCursor{ id: job.ID, kind: job.Kind, queue: job.Queue, - sortField: sortField, - time: time, + sortField: listParams.sortField, + time: *cursorTime, } } @@ -73,6 +89,10 @@ func (c *JobListCursor) UnmarshalText(text []byte) error { // MarshalText implements encoding.TextMarshaler to encode the cursor as an // opaque string. func (c JobListCursor) MarshalText() ([]byte, error) { + if c.job != nil { + return nil, errors.New("cursor initialized with only a job can't be marshaled; try a cursor from JobListResult instead") + } + wrapperValue := jobListPaginationCursorJSON{ ID: c.id, Kind: c.kind, @@ -103,6 +123,7 @@ type SortOrder int const ( // SortOrderAsc specifies that the sort should in ascending order. SortOrderAsc SortOrder = iota + // SortOrderDesc specifies that the sort should in descending order. SortOrderDesc ) @@ -111,16 +132,30 @@ const ( type JobListOrderByField string const ( + // JobListOrderByID specifies that the sort should be by job ID. + JobListOrderByID JobListOrderByField = "id" + // JobListOrderByFinalizedAt specifies that the sort should be by - // finalized_at. + // `finalized_at`. // // This option must be used in conjunction with filtering by only finalized // job states. JobListOrderByFinalizedAt JobListOrderByField = "finalized_at" - // JobListOrderByScheduledAt specifies that the sort should be by scheduled_at. + + // JobListOrderByScheduledAt specifies that the sort should be by + // `scheduled_at`. JobListOrderByScheduledAt JobListOrderByField = "scheduled_at" - // JobListOrderByTime specifies that the sort should be by time. The specific - // time field used will vary by the first specified job state. + + // JobListOrderByTime specifies that the sort should be by the "best fit" + // time field based on listed state. The best fit is determined by looking + // at the first value given to JobListParams.States. If multiple states are + // specified, the ones after the first will be ignored. + // + // The specific time field used for sorting depends on requested state: + // + // * States `available`, `retryable`, or `scheduled` use `scheduled_at`. + // * State `running` uses `attempted_at`. + // * States `cancelled`, `completed`, or `discarded` use `finalized_at`. JobListOrderByTime JobListOrderByField = "time" ) @@ -146,7 +181,7 @@ type JobListParams struct { func NewJobListParams() *JobListParams { return &JobListParams{ paginationCount: 100, - sortField: JobListOrderByTime, + sortField: JobListOrderByID, sortOrder: SortOrderAsc, states: []rivertype.JobState{ rivertype.JobStateAvailable, @@ -178,7 +213,7 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) { conditionsBuilder := &strings.Builder{} conditions := make([]string, 0, 10) namedArgs := make(map[string]any) - orderBy := []dblist.JobListOrderBy{} + orderBy := make([]dblist.JobListOrderBy, 0, 2) var sortOrder dblist.SortOrder switch p.sortOrder { @@ -208,16 +243,20 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) { } var timeField string - if len(p.states) > 0 && p.sortField == JobListOrderByTime { + switch { + case p.sortField == JobListOrderByID: + // no time field + + case len(p.states) > 0 && p.sortField == JobListOrderByTime: timeField = jobListTimeFieldForState(p.states[0]) - } else { + orderBy = append(orderBy, dblist.JobListOrderBy{Expr: timeField, Order: sortOrder}) + + default: timeField = string(p.sortField) + orderBy = append(orderBy, dblist.JobListOrderBy{Expr: timeField, Order: sortOrder}) } - orderBy = append(orderBy, []dblist.JobListOrderBy{ - {Expr: timeField, Order: sortOrder}, - {Expr: "id", Order: sortOrder}, - }...) + orderBy = append(orderBy, dblist.JobListOrderBy{Expr: "id", Order: sortOrder}) if p.metadataFragment != "" { conditions = append(conditions, `metadata @> @metadata_fragment::jsonb`) @@ -225,12 +264,20 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) { } if p.after != nil { - if sortOrder == dblist.SortOrderAsc { - conditions = append(conditions, fmt.Sprintf(`("%s" > @cursor_time OR ("%s" = @cursor_time AND "id" > @after_id))`, timeField, timeField)) + if p.after.time.IsZero() { // order by ID only + if sortOrder == dblist.SortOrderAsc { + conditions = append(conditions, "(id > @after_id)") + } else { + conditions = append(conditions, "(id < @after_id)") + } } else { - conditions = append(conditions, fmt.Sprintf(`("%s" < @cursor_time OR ("%s" = @cursor_time AND "id" < @after_id))`, timeField, timeField)) + if sortOrder == dblist.SortOrderAsc { + conditions = append(conditions, fmt.Sprintf(`("%s" > @cursor_time OR ("%s" = @cursor_time AND "id" > @after_id))`, timeField, timeField)) + } else { + conditions = append(conditions, fmt.Sprintf(`("%s" < @cursor_time OR ("%s" = @cursor_time AND "id" < @after_id))`, timeField, timeField)) + } + namedArgs["cursor_time"] = p.after.time } - namedArgs["cursor_time"] = p.after.time namedArgs["after_id"] = p.after.id } @@ -259,7 +306,12 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) { // after the given cursor. func (p *JobListParams) After(cursor *JobListCursor) *JobListParams { result := p.copy() - result.after = cursor + + if cursor.job == nil { + result.after = cursor + } else { + result.after = jobListCursorFromJobAndParams(cursor.job, result) + } return result } @@ -311,7 +363,7 @@ func (p *JobListParams) Queues(queues ...string) *JobListParams { func (p *JobListParams) OrderBy(field JobListOrderByField, direction SortOrder) *JobListParams { result := p.copy() switch field { - case JobListOrderByTime, JobListOrderByScheduledAt: + case JobListOrderByID, JobListOrderByTime, JobListOrderByScheduledAt: result.sortField = field case JobListOrderByFinalizedAt: result.sortField = field @@ -341,6 +393,7 @@ func (p *JobListParams) States(states ...rivertype.JobState) *JobListParams { } func jobListTimeFieldForState(state rivertype.JobState) string { + // Don't include a `default` so `exhaustive` lint can detect omissions. switch state { case rivertype.JobStateAvailable, rivertype.JobStateRetryable, rivertype.JobStateScheduled: return "scheduled_at" @@ -348,28 +401,31 @@ func jobListTimeFieldForState(state rivertype.JobState) string { return "attempted_at" case rivertype.JobStateCancelled, rivertype.JobStateCompleted, rivertype.JobStateDiscarded: return "finalized_at" - default: - return "created_at" // should never happen } + + return "created_at" // should never happen } func jobListTimeValue(job *rivertype.JobRow) time.Time { + // Don't include a `default` so `exhaustive` lint can detect omissions. switch job.State { case rivertype.JobStateAvailable, rivertype.JobStateRetryable, rivertype.JobStateScheduled: return job.ScheduledAt + case rivertype.JobStateRunning: if job.AttemptedAt == nil { // This should never happen unless a job has been manually manipulated. return job.CreatedAt } return *job.AttemptedAt + case rivertype.JobStateCancelled, rivertype.JobStateCompleted, rivertype.JobStateDiscarded: if job.FinalizedAt == nil { // This should never happen unless a job has been manually manipulated. return job.CreatedAt } return *job.FinalizedAt - default: - return job.CreatedAt // should never happen } + + return job.CreatedAt // should never happen } diff --git a/job_list_params_test.go b/job_list_params_test.go index f8533167..b3866fa1 100644 --- a/job_list_params_test.go +++ b/job_list_params_test.go @@ -15,6 +15,42 @@ import ( func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { t.Parallel() + jobRow := &rivertype.JobRow{ + ID: 4, + Kind: "test", + Queue: "test", + State: rivertype.JobStateRunning, + } + + cursor := JobListCursorFromJob(jobRow) + require.Zero(t, cursor.id) + require.Equal(t, jobRow, cursor.job) + require.Zero(t, cursor.kind) + require.Zero(t, cursor.queue) + require.Zero(t, cursor.sortField) + require.Zero(t, cursor.time) +} + +func Test_JobListCursor_jobListCursorFromJobAndParams(t *testing.T) { + t.Parallel() + + t.Run("OrderByID", func(t *testing.T) { + t.Parallel() + + jobRow := &rivertype.JobRow{ + ID: 4, + Kind: "test", + Queue: "test", + State: rivertype.JobStateRunning, + } + + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().After(JobListCursorFromJob(jobRow))) + require.Equal(t, jobRow.ID, cursor.id) + require.Equal(t, jobRow.Kind, cursor.kind) + require.Equal(t, jobRow.Queue, cursor.queue) + require.Zero(t, cursor.time) + }) + for i, state := range []rivertype.JobState{ rivertype.JobStateAvailable, rivertype.JobStateRetryable, @@ -22,7 +58,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { } { i, state := i, state - t.Run(fmt.Sprintf("ScheduledAtUsedFor%sJob", state), func(t *testing.T) { + t.Run(fmt.Sprintf("OrderByTimeScheduledAtUsedFor%sJob", state), func(t *testing.T) { t.Parallel() now := time.Now().UTC() @@ -35,7 +71,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { ScheduledAt: now.Add(-10 * time.Second), } - cursor := JobListCursorFromJob(jobRow, JobListOrderByTime) + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) require.Equal(t, jobRow.ID, cursor.id) require.Equal(t, jobRow.Kind, cursor.kind) require.Equal(t, jobRow.Queue, cursor.queue) @@ -50,7 +86,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { } { i, state := i, state - t.Run(fmt.Sprintf("FinalizedAtUsedFor%sJob", state), func(t *testing.T) { + t.Run(fmt.Sprintf("OrderByTimeFinalizedAtUsedFor%sJob", state), func(t *testing.T) { t.Parallel() now := time.Now().UTC() @@ -65,7 +101,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { ScheduledAt: now.Add(-10 * time.Second), } - cursor := JobListCursorFromJob(jobRow, JobListOrderByTime) + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) require.Equal(t, jobRow.ID, cursor.id) require.Equal(t, jobRow.Kind, cursor.kind) require.Equal(t, jobRow.Queue, cursor.queue) @@ -73,7 +109,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { }) } - t.Run("RunningJobUsesAttemptedAt", func(t *testing.T) { + t.Run("OrderByTimeRunningJobUsesAttemptedAt", func(t *testing.T) { t.Parallel() now := time.Now().UTC() @@ -87,14 +123,14 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { ScheduledAt: now.Add(-10 * time.Second), } - cursor := JobListCursorFromJob(jobRow, JobListOrderByTime) + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) require.Equal(t, jobRow.ID, cursor.id) require.Equal(t, jobRow.Kind, cursor.kind) require.Equal(t, jobRow.Queue, cursor.queue) require.Equal(t, *jobRow.AttemptedAt, cursor.time) }) - t.Run("UnknownJobStateUsesCreatedAt", func(t *testing.T) { + t.Run("OrderByTimeUnknownJobStateUsesCreatedAt", func(t *testing.T) { t.Parallel() now := time.Now().UTC() @@ -107,7 +143,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { ScheduledAt: now.Add(-10 * time.Second), } - cursor := JobListCursorFromJob(jobRow, JobListOrderByTime) + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) require.Equal(t, jobRow.ID, cursor.id) require.Equal(t, jobRow.Kind, cursor.kind) require.Equal(t, jobRow.Queue, cursor.queue) @@ -122,20 +158,36 @@ func Test_JobListCursor_MarshalJSON(t *testing.T) { t.Parallel() now := time.Now().UTC() - params := &JobListCursor{ + cursor := &JobListCursor{ id: 4, kind: "test_kind", queue: "test_queue", time: now, } - text, err := json.Marshal(params) + text, err := json.Marshal(cursor) require.NoError(t, err) require.NotEqual(t, "", text) unmarshaledParams := &JobListCursor{} require.NoError(t, json.Unmarshal(text, unmarshaledParams)) - require.Equal(t, params, unmarshaledParams) + require.Equal(t, cursor, unmarshaledParams) + }) + + t.Run("ErrorsOnJobOnlyCursor", func(t *testing.T) { + t.Parallel() + + jobRow := &rivertype.JobRow{ + ID: 4, + Kind: "test", + Queue: "test", + State: rivertype.JobStateRunning, + } + + cursor := JobListCursorFromJob(jobRow) + + _, err := json.Marshal(cursor) + require.EqualError(t, err, "json: error calling MarshalText for type *river.JobListCursor: cursor initialized with only a job can't be marshaled; try a cursor from JobListResult instead") }) }