respect non-zero ScheduledAt from JobArgsWithInsertOpts

This allows for job arg definitions to utilize custom logic at the args
level for determining when the job should be scheduled.

Fixes #484.

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - Include `pending` state in `JobListParams` by default so pending jobs are included in `JobList` / `JobListTx` results.
+- Respect `ScheduledAt` if set to a non-zero value by `JobArgsWithInsertOpts`. This allows for job arg definitions to utilize custom logic at the args level for determining when the job should be scheduled. [PR #XXX](https://github.com/riverqueue/river/pull/XXX).
 
 ## [0.10.1] - 2024-07-23
 

client.go

@@ -1221,13 +1221,16 @@ func insertParamsFromConfigArgsAndOptions(archetype *baseservice.Archetype, conf
 		Tags:        tags,
 	}
 
-	if insertOpts.ScheduledAt.IsZero() {
+	if !insertOpts.ScheduledAt.IsZero() {
+		insertParams.ScheduledAt = &insertOpts.ScheduledAt
+		insertParams.State = rivertype.JobStateScheduled
+	} else if !jobInsertOpts.ScheduledAt.IsZero() {
+		insertParams.ScheduledAt = &jobInsertOpts.ScheduledAt
+		insertParams.State = rivertype.JobStateScheduled
+	} else {
 		// Use a stubbed time if there was one, but otherwise prefer the value
 		// generated by the database. createdAt is nil unless time is stubbed.
 		insertParams.ScheduledAt = createdAt
-	} else {
-		insertParams.ScheduledAt = &insertOpts.ScheduledAt
-		insertParams.State = rivertype.JobStateScheduled
 	}
 
 	if insertOpts.Pending {

client_test.go

@@ -4617,15 +4617,31 @@ func TestInsertParamsFromJobArgsAndOptions(t *testing.T) {
 	t.Run("WorkerInsertOptsOverrides", func(t *testing.T) {
 		t.Parallel()
 
-		insertParams, _, err := insertParamsFromConfigArgsAndOptions(archetype, config, &customInsertOptsJobArgs{}, nil)
+		nearFuture := time.Now().Add(5 * time.Minute)
+
+		insertParams, _, err := insertParamsFromConfigArgsAndOptions(archetype, config, &customInsertOptsJobArgs{
+			ScheduledAt: nearFuture,
+		}, nil)
 		require.NoError(t, err)
 		// All these come from overrides in customInsertOptsJobArgs's definition:
 		require.Equal(t, 42, insertParams.MaxAttempts)
 		require.Equal(t, 2, insertParams.Priority)
 		require.Equal(t, "other", insertParams.Queue)
+		require.NotNil(t, insertParams.ScheduledAt)
+		require.Equal(t, nearFuture, *insertParams.ScheduledAt)
 		require.Equal(t, []string{"tag1", "tag2"}, insertParams.Tags)
 	})
 
+	t.Run("WorkerInsertOptsScheduledAtNotRespectedIfZero", func(t *testing.T) {
+		t.Parallel()
+
+		insertParams, _, err := insertParamsFromConfigArgsAndOptions(archetype, config, &customInsertOptsJobArgs{
+			ScheduledAt: time.Time{},
+		}, nil)
+		require.NoError(t, err)
+		require.Nil(t, insertParams.ScheduledAt)
+	})
+
 	t.Run("TagFormatValidated", func(t *testing.T) {
 		t.Parallel()
 
@@ -4717,7 +4733,9 @@ func TestID(t *testing.T) {
 	})
 }
 
-type customInsertOptsJobArgs struct{}
+type customInsertOptsJobArgs struct {
+	ScheduledAt time.Time `json:"scheduled_at"`
+}
 
 func (w *customInsertOptsJobArgs) Kind() string { return "customInsertOpts" }
 
@@ -4726,6 +4744,7 @@ func (w *customInsertOptsJobArgs) InsertOpts() InsertOpts {
 		MaxAttempts: 42,
 		Priority:    2,
 		Queue:       "other",
+		ScheduledAt: w.ScheduledAt,
 		Tags:        []string{"tag1", "tag2"},
 	}
 }