feat: allow where in payload.jobs.run (#9877)

Example:

```ts
  await payload.jobs.queue({
        task: 'MyTask',
        input: {
          message: `secret`,
        },
 })

await payload.jobs.run({ where: { 'input.message':  { equals: 'secret' } } })
```

Author: AlessioGr

docs/jobs-queue/queues.mdx

@@ -105,6 +105,9 @@ const results = await payload.jobs.run()
 
 // You can customize the queue name and limit by passing them as arguments:
 await payload.jobs.run({ queue: 'nightly', limit: 100 })
+
+// You can provide a where clause to filter the jobs that should be run:
+await payload.jobs.run({ where: { 'input.message': { equals: 'secret' } } })
 ```
 
 **Run a single job:**

packages/payload/src/queues/localAPI.ts

@@ -6,6 +6,7 @@ import {
   type PayloadRequest,
   type RunningJob,
   type TypedJobs,
+  type Where,
 } from '../index.js'
 import { runJobs } from './operations/runJobs/index.js'
 
@@ -70,13 +71,15 @@ export const getJobsLocalAPI = (payload: Payload) => ({
     overrideAccess?: boolean
     queue?: string
     req?: PayloadRequest
+    where?: Where
   }): Promise<ReturnType<typeof runJobs>> => {
     const newReq: PayloadRequest = args?.req ?? (await createLocalReq({}, payload))
     const result = await runJobs({
       limit: args?.limit,
       overrideAccess: args?.overrideAccess !== false,
       queue: args?.queue,
       req: newReq,
+      where: args?.where,
     })
     return result
   },

packages/payload/src/queues/operations/runJobs/index.ts

@@ -25,6 +25,7 @@ export type RunJobsArgs = {
   overrideAccess?: boolean
   queue?: string
   req: PayloadRequest
+  where?: Where
 }
 
 export type RunJobsResult = {
@@ -45,6 +46,7 @@ export const runJobs = async ({
   overrideAccess,
   queue,
   req,
+  where: whereFromProps,
 }: RunJobsArgs): Promise<RunJobsResult> => {
   if (!overrideAccess) {
     const hasAccess = await req.payload.config.jobs.access.run({ req })
@@ -94,6 +96,10 @@ export const runJobs = async ({
     })
   }
 
+  if (whereFromProps) {
+    where.and.push(whereFromProps)
+  }
+
   // Find all jobs and ensure we set job to processing: true as early as possible to reduce the chance of
   // the same job being picked up by another worker
   const jobsQuery: {

test/queues/int.spec.ts

@@ -965,4 +965,90 @@ describe('Queues', () => {
     expect(allCompletedJobs.totalDocs).toBe(1)
     expect(allCompletedJobs.docs[0].id).toBe(lastJobID)
   })
+
+  it('ensure where query for id in payload.jobs.run works and only runs the specified job', async () => {
+    payload.config.jobs.deleteJobOnComplete = false
+
+    let lastJobID: string = null
+    for (let i = 0; i < 3; i++) {
+      const job = await payload.jobs.queue({
+        task: 'CreateSimple',
+        input: {
+          message: 'from single task',
+        },
+      })
+      lastJobID = job.id
+    }
+
+    await payload.jobs.run({
+      where: {
+        id: {
+          equals: lastJobID,
+        },
+      },
+    })
+
+    const allSimples = await payload.find({
+      collection: 'simple',
+      limit: 100,
+    })
+
+    expect(allSimples.totalDocs).toBe(1)
+    expect(allSimples.docs[0].title).toBe('from single task')
+
+    const allCompletedJobs = await payload.find({
+      collection: 'payload-jobs',
+      limit: 100,
+      where: {
+        completedAt: {
+          exists: true,
+        },
+      },
+    })
+
+    expect(allCompletedJobs.totalDocs).toBe(1)
+    expect(allCompletedJobs.docs[0].id).toBe(lastJobID)
+  })
+
+  it('ensure where query for input data in payload.jobs.run works and only runs the specified job', async () => {
+    payload.config.jobs.deleteJobOnComplete = false
+
+    for (let i = 0; i < 3; i++) {
+      await payload.jobs.queue({
+        task: 'CreateSimple',
+        input: {
+          message: `from single task ${i}`,
+        },
+      })
+    }
+
+    await payload.jobs.run({
+      where: {
+        'input.message': {
+          equals: 'from single task 2',
+        },
+      },
+    })
+
+    const allSimples = await payload.find({
+      collection: 'simple',
+      limit: 100,
+    })
+
+    expect(allSimples.totalDocs).toBe(1)
+    expect(allSimples.docs[0].title).toBe('from single task 2')
+
+    const allCompletedJobs = await payload.find({
+      collection: 'payload-jobs',
+      limit: 100,
+      where: {
+        completedAt: {
+          exists: true,
+        },
+      },
+    })
+
+    expect(allCompletedJobs.totalDocs).toBe(1)
+    expect((allCompletedJobs.docs[0].input as any).message).toBe('from single task 2')
+  })
 })