Skip to content

Commit 9a0cf34

Browse files
committed
Implement JobDeleteMany operations
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.
1 parent cce56b9 commit 9a0cf34

19 files changed

Lines changed: 1194 additions & 72 deletions

File tree

client.go

Lines changed: 79 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2049,6 +2049,71 @@ func validateQueueName(queueName string) error {
20492049
return nil
20502050
}
20512051

2052+
// JobDeleteManyResult is the result of a job list operation. It contains a list of
2053+
// jobs and a cursor for fetching the next page of results.
2054+
type JobDeleteManyResult struct {
2055+
// Jobs is a slice of job returned as part of the list operation.
2056+
Jobs []*rivertype.JobRow
2057+
}
2058+
2059+
// JobDeleteMany deletes many jobs at once based on the conditions defined by
2060+
// JobDeleteManyParams. Running jobs are always ignored.
2061+
//
2062+
// params := river.NewJobDeleteManyParams().First(10).State(rivertype.JobStateCompleted)
2063+
// jobRows, err := client.JobDeleteMany(ctx, params)
2064+
// if err != nil {
2065+
// // handle error
2066+
// }
2067+
func (c *Client[TTx]) JobDeleteMany(ctx context.Context, params *JobDeleteManyParams) (*JobDeleteManyResult, error) {
2068+
if !c.driver.PoolIsSet() {
2069+
return nil, errNoDriverDBPool
2070+
}
2071+
2072+
if params == nil {
2073+
params = NewJobDeleteManyParams()
2074+
}
2075+
params.schema = c.config.Schema
2076+
2077+
listParams, err := dblist.JobMakeDriverParams(ctx, params.toDBParams(), c.driver.SQLFragmentColumnIn)
2078+
if err != nil {
2079+
return nil, err
2080+
}
2081+
2082+
jobs, err := c.driver.GetExecutor().JobDeleteMany(ctx, (*riverdriver.JobDeleteManyParams)(listParams))
2083+
if err != nil {
2084+
return nil, err
2085+
}
2086+
2087+
return &JobDeleteManyResult{Jobs: jobs}, nil
2088+
}
2089+
2090+
// JobDeleteManyTx deletes many jobs at once based on the conditions defined by
2091+
// JobDeleteManyParams. Running jobs are always ignored.
2092+
//
2093+
// params := river.NewJobDeleteManyParams().First(10).States(river.JobStateCompleted)
2094+
// jobRows, err := client.JobDeleteManyTx(ctx, tx, params)
2095+
// if err != nil {
2096+
// // handle error
2097+
// }
2098+
func (c *Client[TTx]) JobDeleteManyTx(ctx context.Context, tx TTx, params *JobDeleteManyParams) (*JobDeleteManyResult, error) {
2099+
if params == nil {
2100+
params = NewJobDeleteManyParams()
2101+
}
2102+
params.schema = c.config.Schema
2103+
2104+
listParams, err := dblist.JobMakeDriverParams(ctx, params.toDBParams(), c.driver.SQLFragmentColumnIn)
2105+
if err != nil {
2106+
return nil, err
2107+
}
2108+
2109+
jobs, err := c.driver.UnwrapExecutor(tx).JobDeleteMany(ctx, (*riverdriver.JobDeleteManyParams)(listParams))
2110+
if err != nil {
2111+
return nil, err
2112+
}
2113+
2114+
return &JobDeleteManyResult{Jobs: jobs}, nil
2115+
}
2116+
20522117
// JobListResult is the result of a job list operation. It contains a list of
20532118
// jobs and a cursor for fetching the next page of results.
20542119
type JobListResult struct {
@@ -2091,10 +2156,16 @@ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobL
20912156
return nil, err
20922157
}
20932158

2094-
jobs, err := dblist.JobList(ctx, c.driver.GetExecutor(), dbParams, c.driver.SQLFragmentColumnIn)
2159+
listParams, err := dblist.JobMakeDriverParams(ctx, dbParams, c.driver.SQLFragmentColumnIn)
2160+
if err != nil {
2161+
return nil, err
2162+
}
2163+
2164+
jobs, err := c.driver.GetExecutor().JobList(ctx, listParams)
20952165
if err != nil {
20962166
return nil, err
20972167
}
2168+
20982169
res := &JobListResult{Jobs: jobs}
20992170
if len(jobs) > 0 {
21002171
res.LastCursor = jobListCursorFromJobAndParams(jobs[len(jobs)-1], params)
@@ -2126,10 +2197,16 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara
21262197
return nil, err
21272198
}
21282199

2129-
jobs, err := dblist.JobList(ctx, c.driver.UnwrapExecutor(tx), dbParams, c.driver.SQLFragmentColumnIn)
2200+
listParams, err := dblist.JobMakeDriverParams(ctx, dbParams, c.driver.SQLFragmentColumnIn)
2201+
if err != nil {
2202+
return nil, err
2203+
}
2204+
2205+
jobs, err := c.driver.UnwrapExecutor(tx).JobList(ctx, listParams)
21302206
if err != nil {
21312207
return nil, err
21322208
}
2209+
21332210
res := &JobListResult{Jobs: jobs}
21342211
if len(jobs) > 0 {
21352212
res.LastCursor = jobListCursorFromJobAndParams(jobs[len(jobs)-1], params)

0 commit comments

Comments
 (0)