     1	import { RestApiValidationError, RestApiResourceError } from '../../lib/rest-api-errors.js'
     2	
     3	export const BulkOperationsPlugin = {
     4	  name: 'bulk-operations',
     5	  dependencies: ['rest-api'],
     6	
     7	  async install ({ api, log, addHook, addScopeMethod, helpers, pluginOptions }) {
     8	    const bulkOptions = pluginOptions || {}
     9	    const {
    10	      maxBulkOperations = 100,
    11	      defaultAtomic = true,
    12	      batchSize = 100,
    13	      enableOptimizations = true
    14	    } = bulkOptions
    15	
    16	    log.info('Installing Bulk Operations plugin', { maxBulkOperations, defaultAtomic })
    17	
    18	    // Add bulk methods to each scope
    19	    addScopeMethod('bulkPost', async ({ scope, scopeName, params, context, runHooks }) => {
    20	      const { inputRecords, atomic = defaultAtomic } = params
    21	
    22	      // Validate bulk size
    23	      if (!Array.isArray(inputRecords) || inputRecords.length === 0) {
    24	        throw new RestApiValidationError('Bulk operations require an array of records', {
    25	          fields: ['data'],
    26	          violations: [{ field: 'data', rule: 'required_array', message: 'Must be a non-empty array' }]
    27	        })
    28	      }
    29	
    30	      if (inputRecords.length > maxBulkOperations) {
    31	        throw new RestApiValidationError(`Bulk operations limited to ${maxBulkOperations} records`, {
    32	          fields: ['data'],
    33	          violations: [{
    34	            field: 'data',
    35	            rule: 'max_items',
    36	            message: `Cannot process more than ${maxBulkOperations} records at once`
    37	          }]
    38	        })
    39	      }
    40	
    41	      const results = []
    42	      const errors = []
    43	      let transaction = null
    44	
    45	      try {
    46	        // Start transaction if atomic mode
    47	        if (atomic && helpers.newTransaction) {
    48	          transaction = await helpers.newTransaction()
    49	        }
    50	
    51	        // Process records in batches
    52	        for (let i = 0; i < inputRecords.length; i += batchSize) {
    53	          const batch = inputRecords.slice(i, i + batchSize)
    54	
    55	          for (let j = 0; j < batch.length; j++) {
    56	            const recordIndex = i + j
    57	            const inputRecord = batch[j]
    58	
    59	            try {
    60	              // Create individual context for each record
    61	              const recordContext = {
    62	                ...context,
    63	                bulkOperation: true,
    64	                bulkIndex: recordIndex
    65	              }
    66	
    67	              // Use the existing post method with transaction
    68	              const result = await scope.post({
    69	                inputRecord: inputRecord.data ? inputRecord : { data: inputRecord },
    70	                transaction
    71	              }, recordContext)
    72	
    73	              results.push({
    74	                index: recordIndex,
    75	                status: 'success',
    76	                data: result.data
    77	              })
    78	            } catch (error) {
    79	              if (atomic) {
    80	                // In atomic mode, rollback and throw
    81	                if (transaction) await transaction.rollback()
    82	                throw error
    83	              } else {
    84	                // In non-atomic mode, collect errors
    85	                errors.push({
    86	                  index: recordIndex,
    87	                  status: 'error',
    88	                  error: {
    89	                    code: error.code || 'UNKNOWN_ERROR',
    90	                    message: error.message,
    91	                    details: error.details
    92	                  }
    93	                })
    94	              }
    95	            }
    96	          }
    97	        }
    98	
    99	        // Commit transaction if atomic
   100	        if (transaction) {
   101	          await transaction.commit()
   102	        }
   103	
   104	        // Build response
   105	        return {
   106	          data: results.filter(r => r.status === 'success').map(r => r.data),
   107	          errors: errors.length > 0 ? errors : undefined,
   108	          meta: {
   109	            total: inputRecords.length,
   110	            succeeded: results.length,
   111	            failed: errors.length,
   112	            atomic
   113	          }
   114	        }
   115	      } catch (error) {
   116	        // Ensure rollback on error
   117	        if (transaction && !transaction.isCompleted()) {
   118	          await transaction.rollback()
   119	        }
   120	        throw error
   121	      }
   122	    })
   123	
   124	    addScopeMethod('bulkPatch', async ({ scope, scopeName, params, context, runHooks }) => {
   125	      const { operations, atomic = defaultAtomic } = params
   126	
   127	      // Validate operations
   128	      if (!Array.isArray(operations) || operations.length === 0) {
   129	        throw new RestApiValidationError('Bulk patch requires an array of operations', {
   130	          fields: ['operations'],
   131	          violations: [{ field: 'operations', rule: 'required_array', message: 'Must be a non-empty array' }]
   132	        })
   133	      }
   134	
   135	      if (operations.length > maxBulkOperations) {
   136	        throw new RestApiValidationError(`Bulk operations limited to ${maxBulkOperations} operations`, {
   137	          fields: ['operations'],
   138	          violations: [{
   139	            field: 'operations',
   140	            rule: 'max_items',
   141	            message: `Cannot process more than ${maxBulkOperations} operations at once`
   142	          }]
   143	        })
   144	      }
   145	
   146	      const results = []
   147	      const errors = []
   148	      let transaction = null
   149	
   150	      try {
   151	        if (atomic && helpers.newTransaction) {
   152	          transaction = await helpers.newTransaction()
   153	        }
   154	
   155	        for (let i = 0; i < operations.length; i++) {
   156	          const operation = operations[i]
   157	
   158	          // Validate operation structure
   159	          if (!operation.id || !operation.data) {
   160	            errors.push({
   161	              index: i,
   162	              status: 'error',
   163	              error: {
   164	                code: 'INVALID_OPERATION',
   165	                message: 'Operation must include id and data',
   166	                details: { operation }
   167	              }
   168	            })
   169	            if (atomic) {
   170	              if (transaction) await transaction.rollback()
   171	              throw new RestApiValidationError('Invalid operation structure', {
   172	                fields: [`operations[${i}]`],
   173	                violations: [{
   174	                  field: `operations[${i}]`,
   175	                  rule: 'required_fields',
   176	                  message: 'Operation must include id and data'
   177	                }]
   178	              })
   179	            }
   180	            continue
   181	          }
   182	
   183	          try {
   184	            const recordContext = {
   185	              ...context,
   186	              bulkOperation: true,
   187	              bulkIndex: i
   188	            }
   189	
   190	            const result = await scope.patch({
   191	              id: operation.id,
   192	              inputRecord: { data: operation.data },
   193	              transaction
   194	            }, recordContext)
   195	
   196	            // If patch doesn't return full record, fetch it
   197	            let resultData = result.data
   198	            if (!resultData && result.id) {
   199	              const fetchedRecord = await scope.get({
   200	                id: result.id,
   201	                transaction
   202	              }, recordContext)
   203	              resultData = fetchedRecord.data
   204	            }
   205	
   206	            results.push({
   207	              index: i,
   208	              id: operation.id,
   209	              status: 'success',
   210	              data: resultData
   211	            })
   212	          } catch (error) {
   213	            if (atomic) {
   214	              if (transaction) await transaction.rollback()
   215	              throw error
   216	            } else {
   217	              errors.push({
   218	                index: i,
   219	                id: operation.id,
   220	                status: 'error',
   221	                error: {
   222	                  code: error.code || 'UNKNOWN_ERROR',
   223	                  message: error.message,
   224	                  details: error.details
   225	                }
   226	              })
   227	            }
   228	          }
   229	        }
   230	
   231	        if (transaction) {
   232	          await transaction.commit()
   233	        }
   234	
   235	        return {
   236	          data: results.filter(r => r.status === 'success').map(r => r.data),
   237	          errors: errors.length > 0 ? errors : undefined,
   238	          meta: {
   239	            total: operations.length,
   240	            succeeded: results.length,
   241	            failed: errors.length,
   242	            atomic
   243	          }
   244	        }
   245	      } catch (error) {
   246	        if (transaction && !transaction.isCompleted()) {
   247	          await transaction.rollback()
   248	        }
   249	        throw error
   250	      }
   251	    })
   252	
   253	    addScopeMethod('bulkDelete', async ({ scope, scopeName, params, context, runHooks }) => {
   254	      const { ids, atomic = defaultAtomic } = params
   255	
   256	      // Validate IDs
   257	      if (!Array.isArray(ids) || ids.length === 0) {
   258	        throw new RestApiValidationError('Bulk delete requires an array of IDs', {
   259	          fields: ['ids'],
   260	          violations: [{ field: 'ids', rule: 'required_array', message: 'Must be a non-empty array' }]
   261	        })
   262	      }
   263	
   264	      if (ids.length > maxBulkOperations) {
   265	        throw new RestApiValidationError(`Bulk operations limited to ${maxBulkOperations} IDs`, {
   266	          fields: ['ids'],
   267	          violations: [{
   268	            field: 'ids',
   269	            rule: 'max_items',
   270	            message: `Cannot process more than ${maxBulkOperations} IDs at once`
   271	          }]
   272	        })
   273	      }
   274	
   275	      const results = []
   276	      const errors = []
   277	      let transaction = null
   278	
   279	      try {
   280	        if (atomic && helpers.newTransaction) {
   281	          transaction = await helpers.newTransaction()
   282	        }
   283	
   284	        // Process deletes
   285	        for (let i = 0; i < ids.length; i++) {
   286	          const id = ids[i]
   287	
   288	          try {
   289	            const recordContext = {
   290	              ...context,
   291	              bulkOperation: true,
   292	              bulkIndex: i
   293	            }
   294	
   295	            await scope.delete({
   296	              id,
   297	              transaction
   298	            }, recordContext)
   299	
   300	            results.push({
   301	              index: i,
   302	              id,
   303	              status: 'success'
   304	            })
   305	          } catch (error) {
   306	            if (atomic) {
   307	              if (transaction) await transaction.rollback()
   308	              throw error
   309	            } else {
   310	              errors.push({
   311	                index: i,
   312	                id,
   313	                status: 'error',
   314	                error: {
   315	                  code: error.code || 'UNKNOWN_ERROR',
   316	                  message: error.message,
   317	                  details: error.details
   318	                }
   319	              })
   320	            }
   321	          }
   322	        }
   323	
   324	        if (transaction) {
   325	          await transaction.commit()
   326	        }
   327	
   328	        return {
   329	          meta: {
   330	            total: ids.length,
   331	            succeeded: results.length,
   332	            failed: errors.length,
   333	            deleted: results.filter(r => r.status === 'success').map(r => r.id),
   334	            atomic
   335	          },
   336	          errors: errors.length > 0 ? errors : undefined
   337	        }
   338	      } catch (error) {
   339	        if (transaction && !transaction.isCompleted()) {
   340	          await transaction.rollback()
   341	        }
   342	        throw error
   343	      }
   344	    })
   345	
   346	    // Hook into scope creation to add bulk routes
   347	    addHook('afterAddScope', 'bulkOperationsRoutes', {}, async ({ scopeName }) => {
   348	      // Note: context is not available here during route registration
   349	      // The urlPrefix will be calculated per-request in the handler
   350	      const urlPrefix = api.vars.transport?.mountPath || ''
   351	      const scopePath = `${urlPrefix}/${scopeName}`
   352	
   353	      // Create route handlers
   354	      const createBulkRouteHandler = (method) => {
   355	        return async ({ context, body, query }) => {
   356	          try {
   357	            // Parse query params for atomic mode override
   358	            const atomic = query?.atomic !== undefined
   359	              ? query.atomic === 'true'
   360	              : defaultAtomic
   361	
   362	            let params
   363	            if (method === 'bulkPost') {
   364	              // For bulk create, expect array in data field (JSON:API style)
   365	              params = {
   366	                inputRecords: body?.data || body,
   367	                atomic
   368	              }
   369	            } else if (method === 'bulkPatch') {
   370	              // For bulk update, expect operations array
   371	              params = {
   372	                operations: body?.operations || body,
   373	                atomic
   374	              }
   375	            } else if (method === 'bulkDelete') {
   376	              // For bulk delete, expect IDs array
   377	              params = {
   378	                ids: body?.data || body?.ids || body,
   379	                atomic
   380	              }
   381	            }
   382	
   383	            // Call the scope method
   384	            const result = await api.scopes[scopeName][method](params)
   385	
   386	            return result
   387	          } catch (error) {
   388	            // Let transport plugin handle error mapping
   389	            throw error
   390	          }
   391	        }
   392	      }
   393	
   394	      // Register bulk routes
   395	      await api.addRoute({
   396	        method: 'POST',
   397	        path: `${scopePath}/bulk`,
   398	        handler: createBulkRouteHandler('bulkPost')
   399	      })
   400	
   401	      await api.addRoute({
   402	        method: 'PATCH',
   403	        path: `${scopePath}/bulk`,
   404	        handler: createBulkRouteHandler('bulkPatch')
   405	      })
   406	
   407	      await api.addRoute({
   408	        method: 'DELETE',
   409	        path: `${scopePath}/bulk`,
   410	        handler: createBulkRouteHandler('bulkDelete')
   411	      })
   412	
   413	      log.debug(`Added bulk operation routes for scope: ${scopeName}`)
   414	    })
   415	
   416	    // Add optimized bulk insert for Knex if available
   417	    if (enableOptimizations) {
   418	      addHook('beforeBulkPost', 'optimizedBulkInsert', {}, async ({ scope, params, context }) => {
   419	        // Check if we can use optimized path
   420	        if (params.atomic && api.knex?.instance && context.bulkOperation === undefined) {
   421	          // This is the main bulk operation, not individual records
   422	          const { inputRecords } = params
   423	
   424	          // Transform records for direct insertion
   425	          const recordsToInsert = []
   426	          for (const inputRecord of inputRecords) {
   427	            // Run validation
   428	            const validated = await scope.validateInput({
   429	              inputRecord,
   430	              operation: 'create'
   431	            })
   432	
   433	            // Transform to database format
   434	            const dbRecord = await scope.transformForDatabase({
   435	              record: validated,
   436	              operation: 'create'
   437	            })
   438	
   439	            recordsToInsert.push(dbRecord)
   440	          }
   441	
   442	          // Perform bulk insert
   443	          const knex = api.knex.instance
   444	          const tableName = scope.vars.schemaInfo.tableName
   445	
   446	          try {
   447	            const inserted = await knex(tableName)
   448	              .insert(recordsToInsert)
   449	              .returning('*')
   450	
   451	            // Transform back to API format
   452	            const results = []
   453	            for (const record of inserted) {
   454	              const apiRecord = await scope.transformFromDatabase({ record })
   455	              results.push(apiRecord)
   456	            }
   457	
   458	            // Return optimized result
   459	            return {
   460	              data: results,
   461	              meta: {
   462	                total: inputRecords.length,
   463	                succeeded: results.length,
   464	                failed: 0,
   465	                atomic: true,
   466	                optimized: true
   467	              },
   468	              skipDefault: true // Tell the default handler to skip
   469	            }
   470	          } catch (error) {
   471	            // Fall back to default handling
   472	            log.warn('Optimized bulk insert failed, falling back to default', { error: error.message })
   473	          }
   474	        }
   475	      })
   476	    }
   477	
   478	    log.info('Bulk Operations plugin installed successfully')
   479	  }
   480	}
