fix ScheduledAt functionality with InsertManyTx

.github/workflows/ci.yml

@@ -242,7 +242,7 @@ jobs:
       - name: Setup sqlc
         uses: sqlc-dev/setup-sqlc@v4
         with:
-          sqlc-version: "1.22.0"
+          sqlc-version: "1.24.0"
 
       - name: Run sqlc diff
         run: |

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Ensure `ScheduledAt` is respected on `InsertManyTx`. [PR #121](https://github.com/riverqueue/river/pull/121).
+
 ## [0.0.14] - 2023-12-13
 
 ### Fixed

client_test.go

@@ -1016,6 +1016,24 @@ func Test_Client_InsertManyTx(t *testing.T) {
 		require.Len(t, jobs, 2, "Expected to find exactly two jobs of kind: "+(noOpArgs{}).Kind())
 	})
 
+	t.Run("SupportsScheduledJobs", func(t *testing.T) {
+		t.Parallel()
+
+		client, bundle := setup(t)
+
+		startClient(ctx, t, client)
+
+		count, err := client.InsertManyTx(ctx, bundle.tx, []InsertManyParams{{noOpArgs{}, &InsertOpts{ScheduledAt: time.Now().Add(time.Minute)}}})
+		require.NoError(t, err)
+		require.Equal(t, int64(1), count)
+
+		insertedJobs, err := bundle.queries.JobGetByKind(ctx, bundle.tx, noOpArgs{}.Kind())
+		require.NoError(t, err)
+		require.Len(t, insertedJobs, 1)
+		require.Equal(t, dbsqlc.JobStateScheduled, insertedJobs[0].State)
+		require.WithinDuration(t, time.Now().Add(time.Minute), insertedJobs[0].ScheduledAt, 2*time.Second)
+	})
+
 	// A client's allowed to send nil to their driver so they can, for example,
 	// easily use test transactions in their test suite.
 	t.Run("WithDriverWithoutPool", func(t *testing.T) {

internal/dbadapter/db_adapter.go

@@ -305,6 +305,7 @@ func (a *StandardAdapter) JobInsertManyTx(ctx context.Context, tx pgx.Tx, params
 			Priority:    int16(min(params.Priority, math.MaxInt16)),
 			Queue:       params.Queue,
 			State:       params.State,
+			ScheduledAt: params.ScheduledAt,
 			Tags:        tags,
 		}
 	}

internal/dbsqlc/river_job.sql

@@ -184,6 +184,7 @@ INSERT INTO river_job(
   metadata,
   priority,
   queue,
+  scheduled_at,
   state,
   tags
 ) VALUES (
@@ -194,6 +195,7 @@ INSERT INTO river_job(
   @metadata,
   @priority,
   @queue,
+  @scheduled_at,
   @state,
   @tags
 );