From 2b0ec1f4ae490efc46df897151a95ddba35f3e08 Mon Sep 17 00:00:00 2001 From: Youssef Dakkak Date: Sun, 18 Oct 2026 04:04:09 +0000 Subject: [PATCH] feat(bulk): add BulkOperationsService and Express bulk handlers (#522) Co-authored-by: Cursor --- src/__tests__/bulkOperations.test.js | 128 +++++++++++++++ src/services/bulkOperations.js | 235 +++++++++++++++++++++++++++ 2 files changed, 363 insertions(+) create mode 100644 src/__tests__/bulkOperations.test.js create mode 100644 src/services/bulkOperations.js diff --git a/src/__tests__/bulkOperations.test.js b/src/__tests__/bulkOperations.test.js new file mode 100644 index 0000000..e349c65 --- /dev/null +++ b/src/__tests__/bulkOperations.test.js @@ -0,0 +1,128 @@ +'use strict'; + +const { BulkOperationsService } = require('../services/bulkOperations'); + +describe('BulkOperationsService (#522)', () => { + let service; + let mockStore; + + beforeEach(() => { + service = new BulkOperationsService({ maxBatchSize: 5, concurrency: 2 }); + const records = new Map(); + let seq = 1; + + mockStore = { + records, + async create(data) { + if (!data.name) throw new Error('name is required'); + const id = `item_${seq++}`; + const rec = { id, ...data }; + records.set(id, rec); + return rec; + }, + async update(id, patch) { + const existing = records.get(id); + if (!existing) return null; + const updated = { ...existing, ...patch }; + records.set(id, updated); + return updated; + }, + async remove(id) { + const existing = records.get(id); + if (!existing) return null; + records.delete(id); + return existing; + }, + }; + }); + + it('executes bulkCreate and returns summary with all fulfilled items', async () => { + const summary = await service.bulkCreate( + [{ name: 'Alpha' }, { name: 'Beta' }, { name: 'Gamma' }], + mockStore + ); + + expect(summary.total).toBe(3); + expect(summary.succeeded).toBe(3); + expect(summary.failed).toBe(0); + expect(summary.results.map((r) => r.data.name)).toEqual([ + 'Alpha', + 'Beta', + 'Gamma', + ]); + }); + + it('supports partial success in bulkUpdate and bulkDelete', async () => { + const created = await service.bulkCreate( + [{ name: 'One' }, { name: 'Two' }], + mockStore + ); + const id1 = created.results[0].data.id; + + const updateSummary = await service.bulkUpdate( + [ + { id: id1, data: { name: 'One Updated' } }, + { id: 'missing_id', data: { name: 'Ghost' } }, + ], + mockStore + ); + + expect(updateSummary.succeeded).toBe(1); + expect(updateSummary.failed).toBe(1); + expect(updateSummary.results[0].status).toBe('fulfilled'); + expect(updateSummary.results[1].status).toBe('rejected'); + + const deleteSummary = await service.bulkDelete( + [id1, 'missing_id'], + mockStore + ); + expect(deleteSummary.succeeded).toBe(1); + expect(deleteSummary.failed).toBe(1); + }); + + it('enforces maxBatchSize and non-empty array validation', async () => { + await expect(service.bulkCreate([], mockStore)).rejects.toThrow( + /cannot be empty/ + ); + await expect( + service.bulkCreate( + [{ name: '1' }, { name: '2' }, { name: '3' }, { name: '4' }, { name: '5' }, { name: '6' }], + mockStore + ) + ).rejects.toThrow(/exceeds maximum batch size/); + }); + + it('returns HTTP 201 on full create and HTTP 207 on partial success via handlers', async () => { + const handlers = service.createHandlers(mockStore); + + const makeRes = () => ({ + statusCode: null, + payload: null, + status(code) { + this.statusCode = code; + return this; + }, + json(body) { + this.payload = body; + return this; + }, + }); + + const resFull = makeRes(); + await handlers.create( + { body: { items: [{ name: 'A' }, { name: 'B' }] } }, + resFull + ); + expect(resFull.statusCode).toBe(201); + expect(resFull.payload.succeeded).toBe(2); + + const resPartial = makeRes(); + await handlers.create( + { body: { items: [{ name: 'C' }, {}] } }, + resPartial + ); + expect(resPartial.statusCode).toBe(207); + expect(resPartial.payload.succeeded).toBe(1); + expect(resPartial.payload.failed).toBe(1); + }); +}); diff --git a/src/services/bulkOperations.js b/src/services/bulkOperations.js new file mode 100644 index 0000000..fc51ad2 --- /dev/null +++ b/src/services/bulkOperations.js @@ -0,0 +1,235 @@ +'use strict'; + +/** + * Bulk operations service & Express handler factory (#522). + * + * Eliminates N-request round-trip overhead by executing batch create, update, + * and delete operations in a single request with bounded batch sizes, + * configurable concurrency, and detailed per-item status reporting (including + * HTTP 207 Multi-Status support for partial successes). + */ + +const DEFAULT_MAX_BATCH_SIZE = 500; +const DEFAULT_CONCURRENCY = 10; + +class BulkOperationsService { + constructor(options = {}) { + this.maxBatchSize = options.maxBatchSize ?? DEFAULT_MAX_BATCH_SIZE; + this.concurrency = Math.max(1, options.concurrency ?? DEFAULT_CONCURRENCY); + } + + /** + * Validates that `items` is a non-empty array within `maxBatchSize`. + */ + validateBatch(items) { + if (!Array.isArray(items)) { + const err = new Error('Bulk payload "items" must be an array'); + err.statusCode = 400; + err.code = 'INVALID_BULK_PAYLOAD'; + throw err; + } + if (items.length === 0) { + const err = new Error('Bulk payload "items" cannot be empty'); + err.statusCode = 400; + err.code = 'EMPTY_BULK_PAYLOAD'; + throw err; + } + if (items.length > this.maxBatchSize) { + const err = new Error( + `Bulk payload exceeds maximum batch size of ${this.maxBatchSize} items` + ); + err.statusCode = 413; + err.code = 'BULK_BATCH_LIMIT_EXCEEDED'; + throw err; + } + } + + /** + * Executes an async `operationFn(item, index)` across `items` with bounded + * concurrency. + * + * @param {Array} items - Batch items to process + * @param {Function} operationFn - Async function `(item, index) => result` + * @param {Object} [options={}] - `{ atomic: boolean }` + * @returns {Promise<{ total: number, succeeded: number, failed: number, results: Array, durationMs: number }>} + */ + async executeBatch(items, operationFn, options = {}) { + this.validateBatch(items); + if (typeof operationFn !== 'function') { + throw new Error('operationFn must be a function'); + } + + const startTime = Date.now(); + const atomic = Boolean(options.atomic); + const results = new Array(items.length); + let succeeded = 0; + let failed = 0; + let cursor = 0; + let abortedError = null; + + const workerCount = Math.min(this.concurrency, items.length); + + const runWorker = async () => { + while (cursor < items.length) { + if (atomic && abortedError) return; + const index = cursor; + cursor += 1; + const item = items[index]; + + try { + const data = await operationFn(item, index); + results[index] = { + index, + status: 'fulfilled', + data, + }; + succeeded += 1; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + results[index] = { + index, + status: 'rejected', + error: message, + code: err?.code || 'OPERATION_FAILED', + }; + failed += 1; + if (atomic && !abortedError) { + abortedError = err instanceof Error ? err : new Error(message); + } + } + } + }; + + const workers = []; + for (let i = 0; i < workerCount; i += 1) { + workers.push(runWorker()); + } + await Promise.all(workers); + + if (atomic && abortedError) { + const atomicErr = new Error( + `Atomic bulk operation aborted: ${abortedError.message}` + ); + atomicErr.statusCode = abortedError.statusCode || 400; + atomicErr.code = 'ATOMIC_BULK_ABORTED'; + atomicErr.results = results.filter(Boolean); + throw atomicErr; + } + + return { + total: items.length, + succeeded, + failed, + results, + durationMs: Math.max(0, Date.now() - startTime), + }; + } + + async bulkCreate(items, store, options = {}) { + return this.executeBatch(items, (item) => store.create(item), options); + } + + async bulkUpdate(items, store, options = {}) { + return this.executeBatch( + items, + async (item) => { + if (!item || !item.id) { + throw new Error('Each bulk update item requires an "id"'); + } + const updated = await store.update(item.id, item.data ?? item); + if (!updated) { + const notFound = new Error(`Resource not found: ${item.id}`); + notFound.code = 'NOT_FOUND'; + throw notFound; + } + return updated; + }, + options + ); + } + + async bulkDelete(ids, store, options = {}) { + return this.executeBatch( + ids, + async (id) => { + const rawId = typeof id === 'object' && id !== null ? id.id : id; + if (!rawId || typeof rawId !== 'string') { + throw new Error('Each bulk delete entry must be a non-empty id'); + } + const removed = await store.remove(rawId); + if (!removed) { + const notFound = new Error(`Resource not found: ${rawId}`); + notFound.code = 'NOT_FOUND'; + throw notFound; + } + return { id: rawId, deleted: true }; + }, + options + ); + } + + /** + * Creates Express route handlers (`create`, `update`, `delete`) backed by a + * target store (`{ create, update, remove }`). + */ + createHandlers(store) { + const respond = (res, summary, successCode = 200) => { + if (summary.failed === 0) { + return res.status(successCode).json(summary); + } + if (summary.succeeded > 0) { + return res.status(207).json(summary); + } + return res.status(400).json(summary); + }; + + const handleError = (res, err) => { + const status = err.statusCode || 400; + return res.status(status).json({ + error: err.message, + code: err.code || 'BULK_ERROR', + }); + }; + + return { + create: async (req, res) => { + try { + const summary = await this.bulkCreate(req.body?.items, store, { + atomic: req.body?.atomic, + }); + return respond(res, summary, 201); + } catch (err) { + return handleError(res, err); + } + }, + update: async (req, res) => { + try { + const summary = await this.bulkUpdate(req.body?.items, store, { + atomic: req.body?.atomic, + }); + return respond(res, summary, 200); + } catch (err) { + return handleError(res, err); + } + }, + delete: async (req, res) => { + try { + const items = req.body?.ids ?? req.body?.items; + const summary = await this.bulkDelete(items, store, { + atomic: req.body?.atomic, + }); + return respond(res, summary, 200); + } catch (err) { + return handleError(res, err); + } + }, + }; + } +} + +const defaultBulkOperations = new BulkOperationsService(); + +module.exports = { + BulkOperationsService, + bulkOperations: defaultBulkOperations, +};