Add missing integration-level job snooze test (#31)

I noticed while trying to refactor something the other day that I had a
problem around snoozing, but only the example snooze test was failing.
Looking into it further, it turned out that it was because the snooze
example was the only snooze test we had.

Example tests aren't great because they do obnoxious things like swallow
all output (making it harder to debug), but they also don't support some
feature like `-count`, making it harder to debug intermittent problems.

Here, add another client test for snooze. We use the new client test
line that makes use of `JobArgsReflectKind` and `WorkFunc`, keeping the
implementation clean and all local to the specific test case that needs
it.

And since snooze and cancel are quite similar, I added a test case for
cancel as well. Cancel did have another test elsewhere, but I figure I
might as well add one in this new line too to line up with the snooze
test case. It should be a pretty fast test, so doesn't hurt to have a
duplicate.

client_test.go

@@ -168,15 +168,27 @@ func Test_Client(t *testing.T) {
 
 	ctx := context.Background()
 
-	type testBundle struct{}
+	type testBundle struct {
+		queries       *dbsqlc.Queries
+		subscribeChan <-chan *Event
+	}
 
 	setup := func(t *testing.T) (*Client[pgx.Tx], *testBundle) {
 		t.Helper()
 
 		config := newTestConfig(t, nil)
 		client := newTestClient(ctx, t, config)
 
-		return client, &testBundle{}
+		subscribeChan, _ := client.Subscribe(
+			EventKindJobCancelled,
+			EventKindJobCompleted,
+			EventKindJobFailed,
+			EventKindJobSnoozed,
+		)
+
+		return client, &testBundle{
+			subscribeChan: subscribeChan,
+		}
 	}
 
 	t.Run("StartInsertAndWork", func(t *testing.T) {
@@ -203,6 +215,64 @@ func Test_Client(t *testing.T) {
 		rivertest.WaitOrTimeout(t, workedChan)
 	})
 
+	t.Run("JobCancel", func(t *testing.T) {
+		t.Parallel()
+
+		client, bundle := setup(t)
+
+		type JobArgs struct {
+			JobArgsReflectKind[JobArgs]
+		}
+
+		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			return JobCancel(fmt.Errorf("a persisted internal error"))
+		}))
+
+		startClient(ctx, t, client)
+
+		insertedJob, err := client.Insert(ctx, &JobArgs{}, nil)
+		require.NoError(t, err)
+
+		event := rivertest.WaitOrTimeout(t, bundle.subscribeChan)
+		require.Equal(t, EventKindJobCancelled, event.Kind)
+		require.Equal(t, JobStateCancelled, event.Job.State)
+		require.WithinDuration(t, time.Now(), *event.Job.FinalizedAt, 2*time.Second)
+
+		updatedJob, err := bundle.queries.JobGetByID(ctx, client.driver.GetDBPool(), insertedJob.ID)
+		require.NoError(t, err)
+		require.Equal(t, dbsqlc.JobStateCancelled, updatedJob.State)
+		require.WithinDuration(t, time.Now(), *updatedJob.FinalizedAt, 2*time.Second)
+	})
+
+	t.Run("JobSnooze", func(t *testing.T) {
+		t.Parallel()
+
+		client, bundle := setup(t)
+
+		type JobArgs struct {
+			JobArgsReflectKind[JobArgs]
+		}
+
+		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			return JobSnooze(15 * time.Minute)
+		}))
+
+		startClient(ctx, t, client)
+
+		insertedJob, err := client.Insert(ctx, &JobArgs{}, nil)
+		require.NoError(t, err)
+
+		event := rivertest.WaitOrTimeout(t, bundle.subscribeChan)
+		require.Equal(t, EventKindJobSnoozed, event.Kind)
+		require.Equal(t, JobStateScheduled, event.Job.State)
+		require.WithinDuration(t, time.Now().Add(15*time.Minute), event.Job.ScheduledAt, 2*time.Second)
+
+		updatedJob, err := bundle.queries.JobGetByID(ctx, client.driver.GetDBPool(), insertedJob.ID)
+		require.NoError(t, err)
+		require.Equal(t, dbsqlc.JobStateScheduled, updatedJob.State)
+		require.WithinDuration(t, time.Now().Add(15*time.Minute), updatedJob.ScheduledAt, 2*time.Second)
+	})
+
 	t.Run("AlternateSchema", func(t *testing.T) {
 		t.Parallel()