Jobs

Now that we have covered Tasks and Workflows, we can tie them together with a concept called a Job.

For example, let's say we have a Workflow or Task that describes the logic to sync information from Payload to a third-party system. This is how you'd declare how to sync that info, but it wouldn't do anything on its own. In order to run that task or workflow, you'd create a Job that references the corresponding Task or Workflow.

Jobs are stored in the Payload database in the payload-jobs collection, and you can decide to keep a running list of all jobs, or configure Payload to delete the job when it has been successfully executed.

Queuing a new job

In order to queue a job, you can use the payload.jobs.queue function.

Here's how you'd queue a new Job, which will run a createPostAndUpdate workflow:

1const createdJob = await payload.jobs.queue({
// Pass the name of the workflow
workflow: 'createPostAndUpdate',
// The input type will be automatically typed
// according to the input you've defined for this workflow
input: {
  title: 'my title',
},
9})

In addition to being able to queue new Jobs based on Workflows, you can also queue a job for a single Task:

1const createdJob = await payload.jobs.queue({
2  task: 'createPost',
3  input: {
4    title: 'my title',
5  },
6})

Where to Queue Jobs

Jobs can be queued from anywhere in your application. Here are the most common scenarios:

From Collection Hooks

The most common place - queue jobs in response to document changes:

1{
slug: 'posts',
hooks: {
  afterChange: [
    async ({ req, doc, operation }) => {
      // Only send notification for published posts
      if (operation === 'update' && doc.status === 'published') {
        await req.payload.jobs.queue({
          task: 'notifySubscribers',
          input: {
            postId: doc.id,
          },
        })
      }
    },
  ],
},
18}

From Field Hooks

Queue jobs based on specific field changes:

1{
name: 'featuredImage',
type: 'upload',
relationTo: 'media',
hooks: {
  afterChange: [
    async ({ req, value, previousValue }) => {
      // Generate image variants when image changes
      if (value !== previousValue) {
        await req.payload.jobs.queue({
          task: 'generateImageVariants',
          input: {
            imageId: value,
          },
        })
      }
    },
  ],
},
20}

From Custom Endpoints

Queue jobs from your API routes:

1export const POST = async (req: PayloadRequest) => {
const job = await req.payload.jobs.queue({
  workflow: 'generateMonthlyReport',
  input: {
    month: new Date().getMonth(),
    year: new Date().getFullYear(),
  },
})
9
return Response.json({
  message: 'Report generation queued',
  jobId: job.id,
})
14}

From Server Actions

Queue jobs from Next.js server actions:

1'use server'
2
3import { getPayload } from 'payload'
4import config from '@payload-config'
5
6export async function scheduleEmail(userId: string) {
7  const payload = await getPayload({ config })
8
9  await payload.jobs.queue({
10    task: 'sendEmail',
11    input: { userId },
12  })
13}

Job Options

When queuing a job, you can pass additional options:

1await payload.jobs.queue({
task: 'sendEmail',
input: { userId: '123' },
4
// Schedule the job to run in the future
waitUntil: new Date('2024-12-25T00:00:00Z'),
7
// Assign to a specific queue
queue: 'high-priority',
10
// Add custom metadata for tracking
log: [
  {
    message: 'Email queued by admin',
    createdAt: new Date().toISOString(),
  },
],
18})

Common options

waitUntil - Schedule the job to run at a specific date/time in the future
queue - Assign the job to a specific queue (defaults to 'default')
log - Add custom log entries for debugging or tracking
req - Pass the request context for access control

Check Job Status

After queuing a job, you can check its status:

1const job = await payload.jobs.queue({
2  task: 'processPayment',
3  input: { orderId: '123' },
4})
5
6// Later, check the job status
7const updatedJob = await payload.findByID({
8  collection: 'payload-jobs',
9  id: job.id,
10})
11
12console.log(updatedJob.completedAt) // When it finished
13console.log(updatedJob.hasError) // If it failed
14console.log(updatedJob.taskStatus) // Details of each task

Job Status Fields

Each job document contains:

1{
id: 'job_123',
taskSlug: 'sendEmail',        // Or workflowSlug for workflows
input: { userId: '123' },     // The input you provided
completedAt: '2024-01-15...',  // When job completed (null if pending)
hasError: false,              // True if job failed
totalTried: 1,                // Number of attempts
processing: false,            // True if currently running
taskStatus: {                 // Status of each task (for workflows)
  sendEmail: {
    '1': {
      complete: true,
      output: { emailSent: true }
    }
  }
},
log: [                        // Execution log
  {
    message: 'Job started',
    createdAt: '...'
  }
]
23}

Access Control

By default, Payload's job operations bypass access control when used from the Local API. You can enable access control by passing overrideAccess: false to any job operation.

To define custom access control for jobs, add an access property to your Jobs Config:

1import type { SanitizedConfig } from 'payload'
2
3const config: SanitizedConfig = {
// ...
jobs: {
  access: {
    // Control who can queue new jobs
    queue: ({ req }) => {
      return req.user?.roles?.includes('admin')
    },
    // Control who can run jobs
    run: ({ req }) => {
      return req.user?.roles?.includes('admin')
    },
    // Control who can cancel jobs
    cancel: ({ req }) => {
      return req.user?.roles?.includes('admin')
    },
  },
},
21}

Each access control function receives the current req object and should return a boolean. If no access control is defined, the default behavior allows any authenticated user to perform the operation.

To use access control in the Local API:

1const req = await createLocalReq({ user }, payload)
2
3await payload.jobs.queue({
4  workflow: 'createPost',
5  input: { title: 'My Post' },
6  overrideAccess: false, // Enable access control
7  req, // Pass the request with user context
8})

Cancelling Jobs

Payload allows you to cancel jobs that are either queued or currently running. When cancelling a running job, the current task will finish executing, but no subsequent tasks will run. This happens because the job checks its cancellation status between tasks.

To cancel a specific job, use the payload.jobs.cancelByID method with the job's ID:

1await payload.jobs.cancelByID({
2  id: createdJob.id,
3})

To cancel multiple jobs at once, use the payload.jobs.cancel method with a Where query:

1await payload.jobs.cancel({
where: {
  workflowSlug: {
    equals: 'createPost',
  },
},
7})

See what others are building with Payload.

"Payload has transformed the way our clients manage content. It's an indispensable tool for any modern agency."

Get the resources you need to start building today

Microsoft chose Payload to tell the world about AI.