Implement JobDeleteMany operations (#962)

Here, add functions for `Client.JobDeleteMany` which lets jobs be
deleted in batches. The main impetus for this is to give us a way of
implementing a "purge queue" function, which has been previously
requested by users, and which is generally just a good feature.

We copy the `JobList` API very closely and use almost all the same
implementation, with the caveat that I've removed some of the more
complex features that I don't think will be as necessary for deletion,
with the sorting features being the biggest one, but also arbitrary SQL
in `WHERE` queries. These can always be added in later because they're
purely additive to the API, but I'm hoping that we won't need them.

Author: brandur

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - The project now tests against [libSQL](https://github.com/tursodatabase/libsql), a popular SQLite fork. It's used through the same `riversqlite` driver that SQLite uses. [PR #957](https://github.com/riverqueue/river/pull/957)
+- Added `JobDeleteMany` operations that remove many jobs in a single operation according to input criteria. [PR #962](https://github.com/riverqueue/river/pull/962)
 
 ### Changed
 

client.go

@@ -2052,6 +2052,71 @@ func validateQueueName(queueName string) error {
 	return nil
 }
 
+// JobDeleteManyResult is the result of a job list operation. It contains a list of
+// jobs and a cursor for fetching the next page of results.
+type JobDeleteManyResult struct {
+	// Jobs is a slice of job returned as part of the list operation.
+	Jobs []*rivertype.JobRow
+}
+
+// JobDeleteMany deletes many jobs at once based on the conditions defined by
+// JobDeleteManyParams. Running jobs are always ignored.
+//
+//	params := river.NewJobDeleteManyParams().First(10).State(rivertype.JobStateCompleted)
+//	jobRows, err := client.JobDeleteMany(ctx, params)
+//	if err != nil {
+//		// handle error
+//	}
+func (c *Client[TTx]) JobDeleteMany(ctx context.Context, params *JobDeleteManyParams) (*JobDeleteManyResult, error) {
+	if !c.driver.PoolIsSet() {
+		return nil, errNoDriverDBPool
+	}
+
+	if params == nil {
+		params = NewJobDeleteManyParams()
+	}
+	params.schema = c.config.Schema
+
+	listParams, err := dblist.JobMakeDriverParams(ctx, params.toDBParams(), c.driver.SQLFragmentColumnIn)
+	if err != nil {
+		return nil, err
+	}
+
+	jobs, err := c.driver.GetExecutor().JobDeleteMany(ctx, (*riverdriver.JobDeleteManyParams)(listParams))
+	if err != nil {
+		return nil, err
+	}
+
+	return &JobDeleteManyResult{Jobs: jobs}, nil
+}
+
+// JobDeleteManyTx deletes many jobs at once based on the conditions defined by
+// JobDeleteManyParams. Running jobs are always ignored.
+//
+//	params := river.NewJobDeleteManyParams().First(10).States(river.JobStateCompleted)
+//	jobRows, err := client.JobDeleteManyTx(ctx, tx, params)
+//	if err != nil {
+//		// handle error
+//	}
+func (c *Client[TTx]) JobDeleteManyTx(ctx context.Context, tx TTx, params *JobDeleteManyParams) (*JobDeleteManyResult, error) {
+	if params == nil {
+		params = NewJobDeleteManyParams()
+	}
+	params.schema = c.config.Schema
+
+	listParams, err := dblist.JobMakeDriverParams(ctx, params.toDBParams(), c.driver.SQLFragmentColumnIn)
+	if err != nil {
+		return nil, err
+	}
+
+	jobs, err := c.driver.UnwrapExecutor(tx).JobDeleteMany(ctx, (*riverdriver.JobDeleteManyParams)(listParams))
+	if err != nil {
+		return nil, err
+	}
+
+	return &JobDeleteManyResult{Jobs: jobs}, nil
+}
+
 // JobListResult is the result of a job list operation. It contains a list of
 // jobs and a cursor for fetching the next page of results.
 type JobListResult struct {
@@ -2094,10 +2159,16 @@ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobL
 		return nil, err
 	}
 
-	jobs, err := dblist.JobList(ctx, c.driver.GetExecutor(), dbParams, c.driver.SQLFragmentColumnIn)
+	listParams, err := dblist.JobMakeDriverParams(ctx, dbParams, c.driver.SQLFragmentColumnIn)
+	if err != nil {
+		return nil, err
+	}
+
+	jobs, err := c.driver.GetExecutor().JobList(ctx, listParams)
 	if err != nil {
 		return nil, err
 	}
+
 	res := &JobListResult{Jobs: jobs}
 	if len(jobs) > 0 {
 		res.LastCursor = jobListCursorFromJobAndParams(jobs[len(jobs)-1], params)
@@ -2129,10 +2200,16 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara
 		return nil, err
 	}
 
-	jobs, err := dblist.JobList(ctx, c.driver.UnwrapExecutor(tx), dbParams, c.driver.SQLFragmentColumnIn)
+	listParams, err := dblist.JobMakeDriverParams(ctx, dbParams, c.driver.SQLFragmentColumnIn)
+	if err != nil {
+		return nil, err
+	}
+
+	jobs, err := c.driver.UnwrapExecutor(tx).JobList(ctx, listParams)
 	if err != nil {
 		return nil, err
 	}
+
 	res := &JobListResult{Jobs: jobs}
 	if len(jobs) > 0 {
 		res.LastCursor = jobListCursorFromJobAndParams(jobs[len(jobs)-1], params)