feat: scheduling jobs

docs/jobs-queue/schedules.mdx

@@ -0,0 +1,156 @@
+---
+title: Job Schedules
+label: Schedules
+order: 60
+desc: Payload allows you to schedule jobs to run periodically
+keywords: jobs queue, application framework, typescript, node, react, nextjs, scheduling, cron, schedule
+---
+
+Payload's `schedule` property lets you enqueue Jobs regularly according to a cron schedule - daily, weekly, hourly, or any custom interval. This is ideal for tasks or workflows that must repeat automatically and without manual intervention.
+
+Scheduling Jobs differs significantly from running them:
+
+- **Queueing**: Scheduling only creates (enqueues) the Job according to your cron expression. It does not immediately execute any business logic.
+- **Running**: Execution happens separately through your Jobs runner - such as autorun, or manual invocation using `payload.jobs.run()` or the `payload-jobs/run` endpoint.
+
+Use the `schedule` property specifically when you have recurring tasks or workflows. To enqueue a single Job to run once in the future, use the `waitUntil` property instead.
+
+## Example use cases
+
+**Regular emails or notifications**
+
+Send nightly digests, weekly newsletters, or hourly updates.
+
+**Batch processing during off-hours**
+
+Process analytics data or rebuild static sites during low-traffic times.
+
+**Periodic data synchronization**
+
+Regularly push or pull updates to or from external APIs.
+
+## Handling schedules
+
+Something needs to actually trigger the scheduling of jobs (execute the scheduling lifecycle seen below). By default, the `jobs.autorun` configuration, as well as the `/api/payload-jobs/run` will also handle scheduling for the queue specified in the `autorun` configuration.
+
+You can disable this behavior by setting `disableScheduling: true` in your `autorun` configuration, or by passing `disableScheduling=true` to the `/api/payload-jobs/run` endpoint. This is useful if you want to handle scheduling manually, for example, by using a cron job or a serverless function that calls the `/api/payload-jobs/handle-schedules` endpoint or the `payload.jobs.handleSchedules()` local API method.
+
+## Defining schedules on Tasks or Workflows
+
+Schedules are defined using the `schedule` property:
+
+```ts
+export type ScheduleConfig = {
+  cron: string // required, supports seconds precision
+  queue: string // required, the queue to push Jobs onto
+  hooks?: {
+    // Optional hooks to customize scheduling behavior
+    beforeSchedule?: BeforeScheduleFn
+    afterSchedule?: AfterScheduleFn
+  }
+}
+```
+
+### Example schedule
+
+The following example demonstrates scheduling a Job to enqueue every day at midnight:
+
+```ts
+import type { TaskConfig } from 'payload'
+
+export const SendDigestEmail: TaskConfig<'SendDigestEmail'> = {
+  slug: 'SendDigestEmail',
+  schedule: [
+    {
+      cron: '0 0 * * *', // Every day at midnight
+      queue: 'nightly',
+    },
+  ],
+  handler: async () => {
+    await sendDigestToAllUsers()
+  },
+}
+```
+
+This configuration only queues the Job - it does not execute it immediately. To actually run the queued Job, you configure autorun in your Payload config (note that autorun should **not** be used on serverless platforms):
+
+```ts
+export default buildConfig({
+  jobs: {
+    scheduler: 'cron',
+    autoRun: [
+      {
+        cron: '* * * * *', // Runs every minute
+        queue: 'nightly',
+      },
+    ],
+    tasks: [SendDigestEmail],
+  },
+})
+```
+
+That way, Payload's scheduler will automatically enqueue the job into the `nightly` queue every day at midnight. The autorun configuration will check the `nightly` queue every minute and execute any Jobs that are due to run.
+
+## Scheduling lifecycle
+
+Here's how the scheduling process operates in detail:
+
+1. **Cron evaluation**: Payload (or your external trigger in `manual` mode) identifies which schedules are due to run. To do that, it will
+   read the `payload-jobs-stats` global which contains information about the last time each scheduled task or workflow was run.
+2. **BeforeSchedule hook**:
+   - The default beforeSchedule hook checks how many active or runnable jobs of the same type that have been queued by the scheduling system currently exist.
+     If such a job exists, it will skip scheduling a new one.
+   - You can provide your own `beforeSchedule` hook to customize this behavior. For example, you might want to allow multiple overlapping Jobs or dynamically set the Job input data.
+3. **Enqueue Job**: Payload queues up a new job. This job will have `waitUntil` set to the next scheduled time based on the cron expression.
+4. **AfterSchedule hook**:
+   - The default afterSchedule hook updates the `payload-jobs-stats` global metadata with the last scheduled time for the Job.
+   - You can provide your own afterSchedule hook to it for custom logging, metrics, or other post-scheduling actions.
+
+## Customizing concurrency and input (Advanced)
+
+You may want more control over concurrency or dynamically set Job inputs at scheduling time. For instance, allowing multiple overlapping Jobs to be scheduled, even if a previously scheduled job has not completed yet, or preparing dynamic data to pass to your Job handler:
+
+```ts
+import { countRunnableOrActiveJobsForQueue } from 'payload'
+
+schedule: [
+  {
+    cron: '* * * * *', // every minute
+    queue: 'reports',
+    hooks: {
+      beforeSchedule: async ({ queueable, req }) => {
+        const runnableOrActiveJobsForQueue =
+          await countRunnableOrActiveJobsForQueue({
+            queue: queueable.scheduleConfig.queue,
+            req,
+            taskSlug: queueable.taskConfig?.slug,
+            workflowSlug: queueable.workflowConfig?.slug,
+            onlyScheduled: true,
+          })
+
+        // Allow up to 3 simultaneous scheduled jobs and set dynamic input
+        return {
+          shouldSchedule: runnableOrActiveJobsForQueue < 3,
+          input: { text: 'Hi there' },
+        }
+      },
+    },
+  },
+]
+```
+
+This allows fine-grained control over how many Jobs can run simultaneously and provides dynamically computed input values each time a Job is scheduled.
+
+## Scheduling in serverless environments
+
+On serverless platforms, scheduling must be triggered externally since Payload does not automatically run cron schedules in ephemeral environments. You have two main ways to trigger scheduling manually:
+
+- **Invoke via Payload's API:** `payload.jobs.handleSchedules()`
+- **Use the REST API endpoint:** `/api/payload-jobs/handle-schedules`
+- **Use the run endpoint, which also handles scheduling by default:** `GET /api/payload-jobs/run`
+
+For example, on Vercel, you can set up a Vercel Cron to regularly trigger scheduling:
+
+- **Vercel Cron Job:** Configure Vercel Cron to periodically call `GET /api/payload-jobs/handle-schedules`. If you would like to auto-run your scheduled jobs as well, you can use the `GET /api/payload-jobs/run` endpoint.
+
+Once Jobs are queued, their execution depends entirely on your configured runner setup (e.g., autorun, or manual invocation).

packages/db-mongodb/src/utilities/transform.ts

@@ -404,6 +404,10 @@ export const transform = ({
   parentIsLocalized = false,
   validateRelationships = true,
 }: Args) => {
+  if (!data) {
+    return null
+  }
+
   if (Array.isArray(data)) {
     for (const item of data) {
       transform({ adapter, data: item, fields, globalSlug, operation, validateRelationships })

packages/payload/package.json

@@ -92,7 +92,7 @@
     "busboy": "^1.6.0",
     "ci-info": "^4.1.0",
     "console-table-printer": "2.12.1",
-    "croner": "9.0.0",
+    "croner": "9.1.0",
     "dataloader": "2.2.3",
     "deepmerge": "4.3.1",
     "file-type": "19.3.0",

packages/payload/src/config/sanitize.ts

@@ -33,7 +33,8 @@ import {
 } from '../locked-documents/config.js'
 import { getPreferencesCollection, preferencesCollectionSlug } from '../preferences/config.js'
 import { getQueryPresetsConfig, queryPresetsCollectionSlug } from '../query-presets/config.js'
-import { getDefaultJobsCollection, jobsCollectionSlug } from '../queues/config/index.js'
+import { getDefaultJobsCollection, jobsCollectionSlug } from '../queues/config/collection.js'
+import { getJobStatsGlobal } from '../queues/config/global.js'
 import { flattenBlock } from '../utilities/flattenAllFields.js'
 import { getSchedulePublishTask } from '../versions/schedule/job.js'
 import { addDefaultsToConfig } from './defaults.js'
@@ -300,7 +301,28 @@ export const sanitizeConfig = async (incomingConfig: Config): Promise<SanitizedC
 
   // Need to add default jobs collection before locked documents collections
   if (config.jobs.enabled) {
-    let defaultJobsCollection = getDefaultJobsCollection(config as unknown as Config)
+    // Check for schedule property in both tasks and workflows
+    const hasScheduleProperty =
+      (config?.jobs?.tasks?.length && config.jobs.tasks.some((task) => task.schedule)) ||
+      (config?.jobs?.workflows?.length &&
+        config.jobs.workflows.some((workflow) => workflow.schedule))
+
+    if (hasScheduleProperty) {
+      config.jobs.scheduling = true
+      // Add payload-jobs-stats global for tracking when a job of a specific slug was last run
+      ;(config.globals ??= []).push(
+        await sanitizeGlobal(
+          config as unknown as Config,
+          getJobStatsGlobal(config as unknown as Config),
+          richTextSanitizationPromises,
+          validRelationships,
+        ),
+      )
+
+      config.jobs.stats = true
+    }
+
+    let defaultJobsCollection = getDefaultJobsCollection(config.jobs)
 
     if (typeof config.jobs.jobsCollectionOverrides === 'function') {
       defaultJobsCollection = config.jobs.jobsCollectionOverrides({
@@ -329,7 +351,7 @@ export const sanitizeConfig = async (incomingConfig: Config): Promise<SanitizedC
       validRelationships,
     )
 
-    configWithDefaults.collections!.push(sanitizedJobsCollection)
+    ;(config.collections ??= []).push(sanitizedJobsCollection)
   }
 
   configWithDefaults.collections!.push(

packages/payload/src/database/defaultUpdateJobs.ts

@@ -1,7 +1,7 @@
 import type { DatabaseAdapter, Job } from '../index.js'
 import type { UpdateJobs } from './types.js'
 
-import { jobsCollectionSlug } from '../queues/config/index.js'
+import { jobsCollectionSlug } from '../queues/config/collection.js'
 
 export const defaultUpdateJobs: UpdateJobs = async function updateMany(
   this: DatabaseAdapter,

packages/payload/src/database/types.ts

@@ -142,6 +142,9 @@ export interface BaseDatabaseAdapter {
     }
   }
 
+  /**
+   * Updates a global that exists. If the global doesn't exist yet, this will not work - you should use `createGlobal` instead.
+   */
   updateGlobal: UpdateGlobal
 
   updateGlobalVersion: UpdateGlobalVersion

packages/payload/src/index.ts

@@ -133,6 +133,7 @@ import { countVersionsLocal } from './collections/operations/local/countVersions
 import { consoleEmailAdapter } from './email/consoleEmailAdapter.js'
 import { fieldAffectsData, type FlattenedBlock } from './fields/config/types.js'
 import { getJobsLocalAPI } from './queues/localAPI.js'
+import { _internal_jobSystemGlobals } from './queues/utilities/getCurrentDate.js'
 import { isNextBuild } from './utilities/isNextBuild.js'
 import { getLogger } from './utilities/logger.js'
 import { serverInit as serverInitTelemetry } from './utilities/telemetry/events/serverInit.js'
@@ -847,24 +848,38 @@ export class BasePayload {
 
       await Promise.all(
         cronJobs.map((cronConfig) => {
-          const job = new Cron(cronConfig.cron ?? DEFAULT_CRON, async () => {
+          const jobAutorunCron = new Cron(cronConfig.cron ?? DEFAULT_CRON, async () => {
+            if (
+              _internal_jobSystemGlobals.shouldAutoSchedule &&
+              !cronConfig.disableScheduling &&
+              this.config.jobs.scheduling
+            ) {
+              await this.jobs.handleSchedules({
+                queue: cronConfig.queue,
+              })
+            }
+
+            if (!_internal_jobSystemGlobals.shouldAutoRun) {
+              return
+            }
+
             if (typeof this.config.jobs.shouldAutoRun === 'function') {
               const shouldAutoRun = await this.config.jobs.shouldAutoRun(this)
 
               if (!shouldAutoRun) {
-                job.stop()
-
-                return false
+                jobAutorunCron.stop()
+                return
               }
             }
 
             await this.jobs.run({
               limit: cronConfig.limit ?? DEFAULT_LIMIT,
               queue: cronConfig.queue,
+              silent: cronConfig.silent,
             })
           })
 
-          this.crons.push(job)
+          this.crons.push(jobAutorunCron)
         }),
       )
     }
@@ -913,8 +928,10 @@ export const reload = async (
   payload: Payload,
   skipImportMapGeneration?: boolean,
 ): Promise<void> => {
-  await payload.destroy()
-
+  if (typeof payload.db.destroy === 'function') {
+    // Only destroy db, as we then later only call payload.db.init and not payload.init
+    await payload.db.destroy()
+  }
   payload.config = config
 
   payload.collections = config.collections.reduce(
@@ -1158,6 +1175,7 @@ export type {
 
 export type { CompoundIndex } from './collections/config/types.js'
 export type { SanitizedCompoundIndex } from './collections/config/types.js'
+
 export { createDataloaderCacheKey, getDataLoader } from './collections/dataloader.js'
 export { countOperation } from './collections/operations/count.js'
 export { createOperation } from './collections/operations/create.js'
@@ -1174,7 +1192,6 @@ export { updateOperation } from './collections/operations/update.js'
 export { updateByIDOperation } from './collections/operations/updateByID.js'
 
 export { buildConfig } from './config/build.js'
-
 export {
   type ClientConfig,
   createClientConfig,
@@ -1183,6 +1200,7 @@ export {
   type UnsanitizedClientConfig,
 } from './config/client.js'
 export { defaults } from './config/defaults.js'
+
 export { type OrderableEndpointBody } from './config/orderable/index.js'
 export { sanitizeConfig } from './config/sanitize.js'
 export type * from './config/types.js'
@@ -1301,6 +1319,7 @@ export {
 
 export type { ValidationFieldError } from './errors/index.js'
 export { baseBlockFields } from './fields/baseFields/baseBlockFields.js'
+
 export { baseIDField } from './fields/baseFields/baseIDField.js'
 
 export {
@@ -1424,13 +1443,15 @@ export type {
 
 export { getDefaultValue } from './fields/getDefaultValue.js'
 export { traverseFields as afterChangeTraverseFields } from './fields/hooks/afterChange/traverseFields.js'
+
 export { promise as afterReadPromise } from './fields/hooks/afterRead/promise.js'
 export { traverseFields as afterReadTraverseFields } from './fields/hooks/afterRead/traverseFields.js'
 export { traverseFields as beforeChangeTraverseFields } from './fields/hooks/beforeChange/traverseFields.js'
 export { traverseFields as beforeValidateTraverseFields } from './fields/hooks/beforeValidate/traverseFields.js'
 
 export { sortableFieldTypes } from './fields/sortableFieldTypes.js'
 export { validations } from './fields/validations.js'
+
 export type {
   ArrayFieldValidation,
   BlocksFieldValidation,
@@ -1485,6 +1506,7 @@ export type {
 
 export { docAccessOperation as docAccessOperationGlobal } from './globals/operations/docAccess.js'
 export { findOneOperation } from './globals/operations/findOne.js'
+
 export { findVersionByIDOperation as findVersionByIDOperationGlobal } from './globals/operations/findVersionByID.js'
 export { findVersionsOperation as findVersionsOperationGlobal } from './globals/operations/findVersions.js'
 export { restoreVersionOperation as restoreVersionOperationGlobal } from './globals/operations/restoreVersion.js'
@@ -1505,8 +1527,7 @@ export type {
   TabsPreferences,
 } from './preferences/types.js'
 export type { QueryPreset } from './query-presets/types.js'
-export { jobAfterRead } from './queues/config/index.js'
-
+export { jobAfterRead } from './queues/config/collection.js'
 export type { JobsConfig, RunJobAccess, RunJobAccessArgs } from './queues/config/types/index.js'
 export type {
   RunInlineTaskFunction,
@@ -1521,6 +1542,7 @@ export type {
   TaskOutput,
   TaskType,
 } from './queues/config/types/taskTypes.js'
+
 export type {
   BaseJob,
   JobLog,
@@ -1531,8 +1553,14 @@ export type {
   WorkflowHandler,
   WorkflowTypes,
 } from './queues/config/types/workflowTypes.js'
-
+export { countRunnableOrActiveJobsForQueue } from './queues/operations/handleSchedules/countRunnableOrActiveJobsForQueue.js'
 export { importHandlerPath } from './queues/operations/runJobs/runJob/importHandlerPath.js'
+
+export {
+  _internal_jobSystemGlobals,
+  _internal_resetJobSystemGlobals,
+  getCurrentDate,
+} from './queues/utilities/getCurrentDate.js'
 export { getLocalI18n } from './translations/getLocalI18n.js'
 export * from './types/index.js'
 export { getFileByPath } from './uploads/getFileByPath.js'