fix: job queue error when running tasks in parallel on sqlite

packages/db-mongodb/src/updateOne.ts

@@ -50,12 +50,18 @@ export const updateOne: UpdateOne = async function updateOne(
 
   let result
 
-  const $inc: Record<string, number> = {}
   let updateData: UpdateQuery<any> = data
-  transform({ $inc, adapter: this, data, fields, operation: 'write' })
+
+  const $inc: Record<string, number> = {}
+  const $push: Record<string, { $each: any[] } | any> = {}
+
+  transform({ $inc, $push, adapter: this, data, fields, operation: 'write' })
   if (Object.keys($inc).length) {
     updateData = { $inc, $set: updateData }
   }
+  if (Object.keys($push).length) {
+    updateData = { $push, $set: updateData }
+  }
 
   try {
     if (returning === false) {

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

@@ -209,6 +209,7 @@ const sanitizeDate = ({
 
 type Args = {
   $inc?: Record<string, number>
+  $push?: Record<string, { $each: any[] } | any>
   /** instance of the adapter */
   adapter: MongooseAdapter
   /** data to transform, can be an array of documents or a single document */
@@ -398,6 +399,7 @@ const stripFields = ({
 
 export const transform = ({
   $inc,
+  $push,
   adapter,
   data,
   fields,
@@ -412,7 +414,16 @@ export const transform = ({
 
   if (Array.isArray(data)) {
     for (const item of data) {
-      transform({ $inc, adapter, data: item, fields, globalSlug, operation, validateRelationships })
+      transform({
+        $inc,
+        $push,
+        adapter,
+        data: item,
+        fields,
+        globalSlug,
+        operation,
+        validateRelationships,
+      })
     }
     return
   }
@@ -470,6 +481,26 @@ export const transform = ({
       }
     }
 
+    if (
+      $push &&
+      field.type === 'array' &&
+      operation === 'write' &&
+      field.name in ref &&
+      ref[field.name]
+    ) {
+      const value = ref[field.name]
+      if (value && typeof value === 'object' && '$push' in value) {
+        const push = value.$push
+
+        if (Array.isArray(push)) {
+          $push[`${parentPath}${field.name}`] = { $each: push }
+        } else if (typeof push === 'object') {
+          $push[`${parentPath}${field.name}`] = push
+        }
+        delete ref[field.name]
+      }
+    }
+
     if (field.type === 'date' && operation === 'read' && field.name in ref && ref[field.name]) {
       if (config.localization && fieldShouldBeLocalized({ field, parentIsLocalized })) {
         const fieldRef = ref[field.name] as Record<string, unknown>

packages/drizzle/src/transform/write/array.ts

@@ -71,6 +71,7 @@ export const transformArray = ({
     data.forEach((arrayRow, i) => {
       const newRow: ArrayRowToInsert = {
         arrays: {},
+        arraysToPush: {},
         locales: {},
         row: {
           _order: i + 1,
@@ -104,6 +105,7 @@ export const transformArray = ({
       traverseFields({
         adapter,
         arrays: newRow.arrays,
+        arraysToPush: newRow.arraysToPush,
         baseTableName,
         blocks,
         blocksToDelete,

packages/drizzle/src/transform/write/blocks.ts

@@ -78,6 +78,7 @@ export const transformBlocks = ({
 
     const newRow: BlockRowToInsert = {
       arrays: {},
+      arraysToPush: {},
       locales: {},
       row: {
         _order: i + 1,
@@ -116,6 +117,7 @@ export const transformBlocks = ({
     traverseFields({
       adapter,
       arrays: newRow.arrays,
+      arraysToPush: newRow.arraysToPush,
       baseTableName,
       blocks,
       blocksToDelete,

packages/drizzle/src/transform/write/index.ts

@@ -27,6 +27,7 @@ export const transformForWrite = ({
   // Split out the incoming data into rows to insert / delete
   const rowToInsert: RowToInsert = {
     arrays: {},
+    arraysToPush: {},
     blocks: {},
     blocksToDelete: new Set(),
     locales: {},
@@ -45,6 +46,7 @@ export const transformForWrite = ({
   traverseFields({
     adapter,
     arrays: rowToInsert.arrays,
+    arraysToPush: rowToInsert.arraysToPush,
     baseTableName: tableName,
     blocks: rowToInsert.blocks,
     blocksToDelete: rowToInsert.blocksToDelete,

packages/drizzle/src/transform/write/traverseFields.ts

@@ -4,13 +4,7 @@ import { fieldIsVirtual, fieldShouldBeLocalized } from 'payload/shared'
 import toSnakeCase from 'to-snake-case'
 
 import type { DrizzleAdapter } from '../../types.js'
-import type {
-  ArrayRowToInsert,
-  BlockRowToInsert,
-  NumberToDelete,
-  RelationshipToDelete,
-  TextToDelete,
-} from './types.js'
+import type { NumberToDelete, RelationshipToDelete, RowToInsert, TextToDelete } from './types.js'
 
 import { isArrayOfRows } from '../../utilities/isArrayOfRows.js'
 import { resolveBlockTableName } from '../../utilities/validateExistingBlockIsIdentical.js'
@@ -23,16 +17,20 @@ import { transformTexts } from './texts.js'
 
 type Args = {
   adapter: DrizzleAdapter
-  arrays: {
-    [tableName: string]: ArrayRowToInsert[]
-  }
+  /**
+   * This will delete the array table and then re-insert all the new array rows.
+   */
+  arrays: RowToInsert['arrays']
+  /**
+   * Array rows to push to the existing array. This will simply create
+   * a new row in the array table.
+   */
+  arraysToPush: RowToInsert['arraysToPush']
   /**
    * This is the name of the base table
    */
   baseTableName: string
-  blocks: {
-    [blockType: string]: BlockRowToInsert[]
-  }
+  blocks: RowToInsert['blocks']
   blocksToDelete: Set<string>
   /**
    * A snake-case field prefix, representing prior fields
@@ -82,6 +80,7 @@ type Args = {
 export const traverseFields = ({
   adapter,
   arrays,
+  arraysToPush,
   baseTableName,
   blocks,
   blocksToDelete,
@@ -129,10 +128,6 @@ export const traverseFields = ({
     if (field.type === 'array') {
       const arrayTableName = adapter.tableNameMap.get(`${parentTableName}_${columnName}`)
 
-      if (!arrays[arrayTableName]) {
-        arrays[arrayTableName] = []
-      }
-
       if (isLocalized) {
         if (typeof data[field.name] === 'object' && data[field.name] !== null) {
           Object.entries(data[field.name]).forEach(([localeKey, localeData]) => {
@@ -157,19 +152,33 @@ export const traverseFields = ({
                 textsToDelete,
                 withinArrayOrBlockLocale: localeKey,
               })
-
+              if (!arrays[arrayTableName]) {
+                arrays[arrayTableName] = []
+              }
               arrays[arrayTableName] = arrays[arrayTableName].concat(newRows)
             }
           })
         }
       } else {
+        let value = data[field.name]
+        let push = false
+        if (
+          // TODO do this for localized as well in DRY way
+
+          typeof value === 'object' &&
+          '$push' in value
+        ) {
+          value = Array.isArray(value.$push) ? value.$push : [value.$push]
+          push = true
+        }
+
         const newRows = transformArray({
           adapter,
           arrayTableName,
           baseTableName,
           blocks,
           blocksToDelete,
-          data: data[field.name],
+          data: value,
           field,
           numbers,
           numbersToDelete,
@@ -183,7 +192,17 @@ export const traverseFields = ({
           withinArrayOrBlockLocale,
         })
 
-        arrays[arrayTableName] = arrays[arrayTableName].concat(newRows)
+        if (push) {
+          if (!arraysToPush[arrayTableName]) {
+            arraysToPush[arrayTableName] = []
+          }
+          arraysToPush[arrayTableName] = arraysToPush[arrayTableName].concat(newRows)
+        } else {
+          if (!arrays[arrayTableName]) {
+            arrays[arrayTableName] = []
+          }
+          arrays[arrayTableName] = arrays[arrayTableName].concat(newRows)
+        }
       }
 
       return
@@ -264,6 +283,7 @@ export const traverseFields = ({
             traverseFields({
               adapter,
               arrays,
+              arraysToPush,
               baseTableName,
               blocks,
               blocksToDelete,
@@ -298,6 +318,7 @@ export const traverseFields = ({
           traverseFields({
             adapter,
             arrays,
+            arraysToPush,
             baseTableName,
             blocks,
             blocksToDelete,

packages/drizzle/src/transform/write/types.ts

@@ -2,6 +2,9 @@ export type ArrayRowToInsert = {
   arrays: {
     [tableName: string]: ArrayRowToInsert[]
   }
+  arraysToPush: {
+    [tableName: string]: ArrayRowToInsert[]
+  }
   locales: {
     [locale: string]: Record<string, unknown>
   }
@@ -12,6 +15,9 @@ export type BlockRowToInsert = {
   arrays: {
     [tableName: string]: ArrayRowToInsert[]
   }
+  arraysToPush: {
+    [tableName: string]: ArrayRowToInsert[]
+  }
   locales: {
     [locale: string]: Record<string, unknown>
   }
@@ -37,6 +43,9 @@ export type RowToInsert = {
   arrays: {
     [tableName: string]: ArrayRowToInsert[]
   }
+  arraysToPush: {
+    [tableName: string]: ArrayRowToInsert[]
+  }
   blocks: {
     [tableName: string]: BlockRowToInsert[]
   }

packages/drizzle/src/updateJobs.ts

@@ -13,9 +13,13 @@ export const updateJobs: UpdateJobs = async function updateMany(
   this: DrizzleAdapter,
   { id, data, limit: limitArg, req, returning, sort: sortArg, where: whereArg },
 ) {
-  if (!(data?.log as object[])?.length) {
+  if (
+    !(data?.log as object[])?.length &&
+    !(data.log && typeof data.log === 'object' && '$push' in data.log)
+  ) {
     delete data.log
   }
+
   const whereToUse: Where = id ? { id: { equals: id } } : whereArg
   const limit = id ? 1 : limitArg
 

packages/drizzle/src/upsertRow/index.ts

@@ -44,7 +44,7 @@ export const upsertRow = async <T extends Record<string, unknown> | TypeWithID>(
 }: Args): Promise<T> => {
   let insertedRow: Record<string, unknown> = { id }
   if (id && shouldUseOptimizedUpsertRow({ data, fields })) {
-    const { row } = transformForWrite({
+    const { arraysToPush, row } = transformForWrite({
       adapter,
       data,
       enableAtomicWrites: true,
@@ -54,11 +54,27 @@ export const upsertRow = async <T extends Record<string, unknown> | TypeWithID>(
 
     const drizzle = db as LibSQLDatabase
 
+    // First, handle $push arrays
+
+    if (arraysToPush && Object.keys(arraysToPush)?.length) {
+      await insertArrays({
+        adapter,
+        arrays: [arraysToPush],
+        db,
+        parentRows: [insertedRow],
+        uuidMap: {},
+      })
+    }
+
+    // Then, handle regular row update
+
     if (ignoreResult) {
-      await drizzle
-        .update(adapter.tables[tableName])
-        .set(row)
-        .where(eq(adapter.tables[tableName].id, id))
+      if (row && Object.keys(row).length) {
+        await drizzle
+          .update(adapter.tables[tableName])
+          .set(row)
+          .where(eq(adapter.tables[tableName].id, id))
+      }
       return ignoreResult === 'idOnly' ? ({ id } as T) : null
     }
 
@@ -74,6 +90,22 @@ export const upsertRow = async <T extends Record<string, unknown> | TypeWithID>(
     const findManyKeysLength = Object.keys(findManyArgs).length
     const hasOnlyColumns = Object.keys(findManyArgs.columns || {}).length > 0
 
+    if (!row || !Object.keys(row).length) {
+      // Nothing to update => just fetch current row and return
+      findManyArgs.where = eq(adapter.tables[tableName].id, insertedRow.id)
+
+      const doc = await db.query[tableName].findFirst(findManyArgs)
+
+      return transform<T>({
+        adapter,
+        config: adapter.payload.config,
+        data: doc,
+        fields,
+        joinQuery: false,
+        tableName,
+      })
+    }
+
     if (findManyKeysLength === 0 || hasOnlyColumns) {
       // Optimization - No need for joins => can simply use returning(). This is optimal for very simple collections
       // without complex fields that live in separate tables like blocks, arrays, relationships, etc.
@@ -429,9 +461,9 @@ export const upsertRow = async <T extends Record<string, unknown> | TypeWithID>(
 
     await insertArrays({
       adapter,
-      arrays: [rowToInsert.arrays],
+      arrays: [rowToInsert.arrays, rowToInsert.arraysToPush],
       db,
-      parentRows: [insertedRow],
+      parentRows: [insertedRow, insertedRow],
       uuidMap: arraysBlocksUUIDMap,
     })
 

packages/drizzle/src/upsertRow/insertArrays.ts

@@ -32,6 +32,9 @@ export const insertArrays = async ({
   const rowsByTable: RowsByTable = {}
 
   arrays.forEach((arraysByTable, parentRowIndex) => {
+    if (!arraysByTable || Object.keys(arraysByTable).length === 0) {
+      return
+    }
     Object.entries(arraysByTable).forEach(([tableName, arrayRows]) => {
       // If the table doesn't exist in map, initialize it
       if (!rowsByTable[tableName]) {