perf: reduce job queue db calls (#11846)

Continuation of #11489. This adds a new, optional `updateJobs` db
adapter method that reduces the amount of database calls for the jobs
queue.

## MongoDB

### Previous: running a set of 50 queued jobs
- 1x db.find (= 1x `Model.paginate`)
- 50x db.updateOne (= 50x `Model.findOneAndUpdate`)

### Now: running a set of 50 queued jobs
- 1x db.updateJobs (= 1x `Model.find` and 1x `Model.updateMany`)

**=> 51 db round trips before, 2 db round trips after**


### Previous: upon task completion
- 1x db.find (= 1x `Model.paginate`)
- 1x db.updateOne (= 1x `Model.findOneAndUpdate`)

### Now: upon task completion
- 1x db.updateJobs (= 1x `Model.findOneAndUpdate`)


**=> 2 db round trips before, 1 db round trip after**


## Drizzle (e.g. Postgres)

### running a set of 50 queued jobs
 - 1x db.query[tablename].findMany
 - 50x db.select 
 - 50x upsertRow
 
This is unaffected by this PR and will be addressed in a future PR

Author: AlessioGr

packages/db-mongodb/src/index.ts

@@ -17,7 +17,6 @@ import type {
   TypeWithVersion,
   UpdateGlobalArgs,
   UpdateGlobalVersionArgs,
-  UpdateManyArgs,
   UpdateOneArgs,
   UpdateVersionArgs,
 } from 'payload'
@@ -55,6 +54,7 @@ import { commitTransaction } from './transactions/commitTransaction.js'
 import { rollbackTransaction } from './transactions/rollbackTransaction.js'
 import { updateGlobal } from './updateGlobal.js'
 import { updateGlobalVersion } from './updateGlobalVersion.js'
+import { updateJobs } from './updateJobs.js'
 import { updateMany } from './updateMany.js'
 import { updateOne } from './updateOne.js'
 import { updateVersion } from './updateVersion.js'
@@ -227,6 +227,7 @@ export function mongooseAdapter({
       mongoMemoryServer,
       sessions: {},
       transactionOptions: transactionOptions === false ? undefined : transactionOptions,
+      updateJobs,
       updateMany,
       url,
       versions: {},

packages/db-mongodb/src/updateJobs.ts

@@ -0,0 +1,83 @@
+import type { MongooseUpdateQueryOptions } from 'mongoose'
+import type { BaseJob, UpdateJobs, Where } from 'payload'
+
+import type { MongooseAdapter } from './index.js'
+
+import { buildQuery } from './queries/buildQuery.js'
+import { getCollection } from './utilities/getEntity.js'
+import { getSession } from './utilities/getSession.js'
+import { handleError } from './utilities/handleError.js'
+import { transform } from './utilities/transform.js'
+
+export const updateJobs: UpdateJobs = async function updateMany(
+  this: MongooseAdapter,
+  { id, data, limit, req, returning, where: whereArg },
+) {
+  const where = id ? { id: { equals: id } } : (whereArg as Where)
+
+  const { collectionConfig, Model } = getCollection({
+    adapter: this,
+    collectionSlug: 'payload-jobs',
+  })
+
+  const options: MongooseUpdateQueryOptions = {
+    lean: true,
+    new: true,
+    session: await getSession(this, req),
+  }
+
+  let query = await buildQuery({
+    adapter: this,
+    collectionSlug: collectionConfig.slug,
+    fields: collectionConfig.flattenedFields,
+    where,
+  })
+
+  transform({ adapter: this, data, fields: collectionConfig.fields, operation: 'write' })
+
+  let result: BaseJob[] = []
+
+  try {
+    if (id) {
+      if (returning === false) {
+        await Model.updateOne(query, data, options)
+        return null
+      } else {
+        const doc = await Model.findOneAndUpdate(query, data, options)
+        result = doc ? [doc] : []
+      }
+    } else {
+      if (typeof limit === 'number' && limit > 0) {
+        const documentsToUpdate = await Model.find(
+          query,
+          {},
+          { ...options, limit, projection: { _id: 1 } },
+        )
+        if (documentsToUpdate.length === 0) {
+          return null
+        }
+
+        query = { _id: { $in: documentsToUpdate.map((doc) => doc._id) } }
+      }
+
+      await Model.updateMany(query, data, options)
+
+      if (returning === false) {
+        return null
+      }
+
+      result = await Model.find(query, {}, options)
+    }
+  } catch (error) {
+    handleError({ collection: collectionConfig.slug, error, req })
+  }
+
+  transform({
+    adapter: this,
+    data: result,
+    fields: collectionConfig.fields,
+    operation: 'read',
+  })
+
+  return result
+}

packages/db-postgres/src/index.ts

@@ -33,6 +33,7 @@ import {
   rollbackTransaction,
   updateGlobal,
   updateGlobalVersion,
+  updateJobs,
   updateMany,
   updateOne,
   updateVersion,
@@ -172,6 +173,7 @@ export function postgresAdapter(args: Args): DatabaseAdapterObj<PostgresAdapter>
       find,
       findGlobal,
       findGlobalVersions,
+      updateJobs,
       // @ts-expect-error - vestiges of when tsconfig was not strict. Feel free to improve
       findOne,
       findVersions,

packages/db-sqlite/src/index.ts

@@ -34,6 +34,7 @@ import {
   rollbackTransaction,
   updateGlobal,
   updateGlobalVersion,
+  updateJobs,
   updateMany,
   updateOne,
   updateVersion,
@@ -127,6 +128,7 @@ export function sqliteAdapter(args: Args): DatabaseAdapterObj<SQLiteAdapter> {
       tables: {},
       // @ts-expect-error - vestiges of when tsconfig was not strict. Feel free to improve
       transactionOptions: args.transactionOptions || undefined,
+      updateJobs,
       updateMany,
       versionsSuffix: args.versionsSuffix || '_v',
 

packages/db-vercel-postgres/src/index.ts

@@ -34,6 +34,7 @@ import {
   rollbackTransaction,
   updateGlobal,
   updateGlobalVersion,
+  updateJobs,
   updateMany,
   updateOne,
   updateVersion,
@@ -138,6 +139,7 @@ export function vercelPostgresAdapter(args: Args = {}): DatabaseAdapterObj<Verce
       tables: {},
       tablesFilter: args.tablesFilter,
       transactionOptions: args.transactionOptions || undefined,
+      updateJobs,
       versionsSuffix: args.versionsSuffix || '_v',
 
       // DatabaseAdapter

packages/drizzle/src/index.ts

@@ -33,6 +33,7 @@ export { commitTransaction } from './transactions/commitTransaction.js'
 export { rollbackTransaction } from './transactions/rollbackTransaction.js'
 export { updateGlobal } from './updateGlobal.js'
 export { updateGlobalVersion } from './updateGlobalVersion.js'
+export { updateJobs } from './updateJobs.js'
 export { updateMany } from './updateMany.js'
 export { updateOne } from './updateOne.js'
 export { updateVersion } from './updateVersion.js'

packages/drizzle/src/updateJobs.ts

@@ -0,0 +1,66 @@
+import type { UpdateJobs, Where } from 'payload'
+
+import toSnakeCase from 'to-snake-case'
+
+import type { DrizzleAdapter } from './types.js'
+
+import { findMany } from './find/findMany.js'
+import { upsertRow } from './upsertRow/index.js'
+import { getTransaction } from './utilities/getTransaction.js'
+
+export const updateJobs: UpdateJobs = async function updateMany(
+  this: DrizzleAdapter,
+  { id, data, limit: limitArg, req, returning, sort: sortArg, where: whereArg },
+) {
+  const whereToUse: Where = id ? { id: { equals: id } } : whereArg
+  const limit = id ? 1 : limitArg
+
+  const db = await getTransaction(this, req)
+  const collection = this.payload.collections['payload-jobs'].config
+  const tableName = this.tableNameMap.get(toSnakeCase(collection.slug))
+  const sort = sortArg !== undefined && sortArg !== null ? sortArg : collection.defaultSort
+
+  const jobs = await findMany({
+    adapter: this,
+    collectionSlug: 'payload-jobs',
+    fields: collection.flattenedFields,
+    limit: id ? 1 : limit,
+    pagination: false,
+    req,
+    sort,
+    tableName,
+    where: whereToUse,
+  })
+  if (!jobs.docs.length) {
+    return []
+  }
+
+  const results = []
+
+  // TODO: We need to batch this to reduce the amount of db calls. This can get very slow if we are updating a lot of rows.
+  for (const job of jobs.docs) {
+    const updateData = {
+      ...job,
+      ...data,
+    }
+
+    const result = await upsertRow({
+      id: job.id,
+      adapter: this,
+      data: updateData,
+      db,
+      fields: collection.flattenedFields,
+      ignoreResult: returning === false,
+      operation: 'update',
+      req,
+      tableName,
+    })
+    results.push(result)
+  }
+
+  if (returning === false) {
+    return null
+  }
+
+  return results
+}

packages/payload/src/database/createDatabaseAdapter.ts

@@ -8,6 +8,7 @@ import type {
   RollbackTransaction,
 } from './types.js'
 
+import { defaultUpdateJobs } from './defaultUpdateJobs.js'
 import { createMigration } from './migrations/createMigration.js'
 import { migrate } from './migrations/migrate.js'
 import { migrateDown } from './migrations/migrateDown.js'
@@ -31,6 +32,7 @@ export function createDatabaseAdapter<T extends BaseDatabaseAdapter>(
     | 'migrateReset'
     | 'migrateStatus'
     | 'migrationDir'
+    | 'updateJobs'
   >,
 ): T {
   return {
@@ -45,6 +47,7 @@ export function createDatabaseAdapter<T extends BaseDatabaseAdapter>(
     migrateReset,
     migrateStatus,
     rollbackTransaction,
+    updateJobs: defaultUpdateJobs,
 
     ...args,
     // Ensure migrationDir is set

packages/payload/src/database/defaultUpdateJobs.ts

@@ -0,0 +1,53 @@
+import type { BaseJob, DatabaseAdapter } from '../index.js'
+import type { UpdateJobs } from './types.js'
+
+import { jobsCollectionSlug } from '../queues/config/index.js'
+import { sanitizeUpdateData } from '../queues/utilities/sanitizeUpdateData.js'
+
+export const defaultUpdateJobs: UpdateJobs = async function updateMany(
+  this: DatabaseAdapter,
+  { id, data, limit, req, returning, where },
+) {
+  const updatedJobs: BaseJob[] | null = []
+
+  const jobsToUpdate: BaseJob[] = (
+    id
+      ? [
+          await this.findOne({
+            collection: jobsCollectionSlug,
+            req,
+            where: { id: { equals: id } },
+          }),
+        ]
+      : (
+          await this.find({
+            collection: jobsCollectionSlug,
+            limit,
+            pagination: false,
+            req,
+            where,
+          })
+        ).docs
+  ).filter(Boolean) as BaseJob[]
+
+  if (!jobsToUpdate) {
+    return null
+  }
+
+  for (const job of jobsToUpdate) {
+    const updateData = {
+      ...job,
+      ...data,
+    }
+    const updatedJob = await this.updateOne({
+      id: job.id,
+      collection: jobsCollectionSlug,
+      data: sanitizeUpdateData({ data: updateData }),
+      req,
+      returning,
+    })
+    updatedJobs.push(updatedJob)
+  }
+
+  return updatedJobs
+}

packages/payload/src/database/types.ts

@@ -1,5 +1,5 @@
 import type { TypeWithID } from '../collections/config/types.js'
-import type { CollectionSlug, GlobalSlug } from '../index.js'
+import type { BaseJob, CollectionSlug, GlobalSlug } from '../index.js'
 import type {
   Document,
   JoinQuery,
@@ -147,6 +147,8 @@ export interface BaseDatabaseAdapter {
 
   updateGlobalVersion: UpdateGlobalVersion
 
+  updateJobs: UpdateJobs
+
   updateMany: UpdateMany
 
   updateOne: UpdateOne
@@ -540,6 +542,32 @@ export type UpdateManyArgs = {
 
 export type UpdateMany = (args: UpdateManyArgs) => Promise<Document[] | null>
 
+export type UpdateJobsArgs = {
+  data: Record<string, unknown>
+  req?: Partial<PayloadRequest>
+  /**
+   * If true, returns the updated documents
+   *
+   * @default true
+   */
+  returning?: boolean
+} & (
+  | {
+      id: number | string
+      limit?: never
+      sort?: never
+      where?: never
+    }
+  | {
+      id?: never
+      limit?: number
+      sort?: Sort
+      where: Where
+    }
+)
+
+export type UpdateJobs = (args: UpdateJobsArgs) => Promise<BaseJob[] | null>
+
 export type UpsertArgs = {
   collection: CollectionSlug
   data: Record<string, unknown>