refactor: crud insert many

README.md

@@ -362,9 +362,11 @@ Where the script arguments are the following:
 To simplify these operations, you can execute the command `npm run bench:init` from your shell. This command starts a container with a MongoDB 6.0 instance, a container with the CRUD Service (built from your current branch), and populates the _customers_ collection with 100,000 documents.
 
 To execute any test, start the k6 service with the following command:
-```
+
+```bash
 docker compose -f bench/dc-k6.yml up <service name>
 ```
+
 Remember to replace `<service name>` with one of the following:
 | Service Name                   | Description                                                                                     | File name containing the test            | 
 |--------------------------------|-------------------------------------------------------------------------------------------------|------------------------------------------|
@@ -373,6 +375,7 @@ Remember to replace `<service name>` with one of the following:
 | k6-stress-test-on-collections  | Executes a Stress Test (GET requests for 90 seconds by 250 users) on the _customers_ collection | [stress-test-on-collections.js](bench/scripts/stress-test-on-collections.js)         |
 | k6-stress-test-on-view         | Executes a Stress Test (GET requests for 90 seconds by 250 users) on the _registered-customers_ view | [stress-test-on-view.js](bench/scripts/stress-test-on-view.js)         |
 | k6-spike-test                  | Executes a Spike Test (simulate a spike of 500 concurrent users for GET requests) on the _customers_ collection | [spike-test.js](bench/scripts/spike-test.js)         |
+| k6-bulk-test                  | Executes a Bulk Test (simulate a spike of 40 `POST /bulk` requests and 40 `PATCH /bulk` requests of 10k records each) on the _items_ collection | [bulk-test.js](bench/scripts/bulk-test.js)         |
 | runner                         | An empty test that can be populated for tests on local environment | [runner.js](bench/scripts/runner.js)         |
 
 We suggest you use the runner to execute customized tests for your research.

bench/dc-k6.yml

@@ -16,6 +16,23 @@ services:
       "json=runner-results.json",
       "/app/runner.js",
     ]
+  k6-bulk-test:
+    image: grafana/k6:0.48.0
+    deploy:
+      resources:
+        limits:
+          memory: 512Mb
+          cpus: "1"
+    volumes:
+      - ./scripts:/app
+    networks:
+      - k6-net
+    command: [ 
+      "run", 
+      "--out",
+      "json=bulk-test-results.json",
+      "/app/bulk-test.js",
+    ]
   k6-load-test:
     image: grafana/k6:0.48.0
     deploy:

bench/scripts/bulk-test.js

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2023 Mia s.r.l.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import http from 'k6/http';
+import { sleep, check } from 'k6';
+import {
+    randomIntBetween,
+    randomString,
+} from 'https://jslib.k6.io/k6-utils/1.4.0/index.js';
+
+// 
+// This file represent a stub of a k6 test. Feel free to modify based on your needs.
+// 
+
+const generateItems = (length = 10000) => {
+
+    return Array.from(Array(length).keys()).map((counter) =>({
+        string: randomString(5),
+        number: randomIntBetween(1, 10),
+        boolean: randomIntBetween(0, 1) === 1,
+        date: new Date(randomIntBetween(0, 10)),
+        object: {
+            string: `object-${randomString(5)}`,
+            number: randomIntBetween(1, 10),
+            boolean: randomIntBetween(0, 1) === 1,
+            counter
+        },
+        array: Array.from(Array(randomIntBetween(1, 10)).keys()).map((i) => ({
+            string: `array-${i}-${randomString(5)}`,
+            number: randomIntBetween(1, 10),
+            boolean: randomIntBetween(0, 1) === 1,
+        }))
+    }))
+}
+
+export const options = {
+    stages: [
+        { duration: '10s', target: 5 },
+        { duration: '20s', target: 20 },
+        { duration: '80s', target: 40 },
+        { duration: '20s', target: 20 },
+        { duration: '10s', target: 5 },
+    ],
+    thresholds: {
+        checks: ['rate==1'],
+        http_req_failed: ['rate<0.01'],
+        'http_req_duration{type:bulk}': ['p(90)<4500']
+    }
+}
+
+export function setup() {
+    // Here it goes any code we want to execute before running our tests
+    const getProbe = http.get(`http://crud-service:3000/items/`, { 
+        headers: { 'Content-Type': 'application/json', 'accept': 'application/json' },
+        tags: { type: 'get' }
+    })
+    check(getProbe, { 'GET /items returns status 200': res => res.status === 200 })
+}
+
+export default function () {
+    const items = generateItems()
+
+    const postList = http.post(`http://crud-service:3000/items/bulk`, JSON.stringify(items), { 
+        headers: { 'Content-Type': 'application/json', 'accept': 'application/json' },
+        tags: { type: 'post-bulk' }
+    })
+    check(postList, { 'POST /items/bulk returns status 200': res => res.status === 200 }) 
+    // get ids from post
+    const ids = JSON.parse(postList.body) 
+    // perform patch bulk
+    const patchResult = http.patch(
+        `http://crud-service:3000/items/bulk`, 
+        JSON.stringify(ids.map(({_id}) => ({
+            filter: { _id },
+            update: {$set: {'object.boolean': true}}
+        }))), { 
+        headers: { 'Content-Type': 'application/json' },
+        tags: { type: 'patch-bulk' }
+    })
+    check(patchResult, { 
+        'PATCH /items/bulk returns status 200 after updating all items': (res) => {
+            return res.status === 200 && Number(res.body) === items.length 
+        }
+    }) 
+}
+
+export function teardown(data) {
+    // Here it goes any code we want to execute after running our tests
+}

bench/scripts/runner.js

@@ -15,27 +15,69 @@
  */
 
 import http from 'k6/http';
-import { sleep } from 'k6';
+import { sleep, check } from 'k6';
+import {
+    randomIntBetween,
+    randomString,
+} from 'https://jslib.k6.io/k6-utils/1.4.0/index.js';
 
 // 
 // This file represent a stub of a k6 test. Feel free to modify based on your needs.
 // 
 
+const generateItems = (length = 10000) => {
+
+    return Array.from(Array(length).keys()).map((counter) =>({
+        string: randomString(5),
+        number: randomIntBetween(1, 10),
+        boolean: randomIntBetween(0, 1) === 1,
+        date: new Date(randomIntBetween(0, 10)),
+        object: {
+            string: `object-${randomString(5)}`,
+            number: randomIntBetween(1, 10),
+            boolean: randomIntBetween(0, 1) === 1,
+            counter
+        },
+        array: Array.from(Array(randomIntBetween(1, 10)).keys()).map((i) => ({
+            string: `array-${i}-${randomString(5)}`,
+            number: randomIntBetween(1, 10),
+            boolean: randomIntBetween(0, 1) === 1,
+        }))
+    }))
+}
+
 export const options = {
     stages: [
-        { duration: '30s', target: 5 },
-        { duration: '2m', target: 10 },
-        { duration: '30s', target: 0 },
+        { duration: '10s', target: 5 },
+        { duration: '20s', target: 20 },
+        { duration: '150s', target: 50 },
+        { duration: '20s', target: 20 },
+        { duration: '10s', target: 5 },
     ],
+    thresholds: {
+        checks: ['rate==1'],
+        http_req_failed: ['rate<0.01'],
+        'http_req_duration{type:bulk}': ['p(90)<4500']
+    }
 }
 
 export function setup() {
     // Here it goes any code we want to execute before running our tests
+    const getProbe = http.get(`http://crud-service:3000/items/`, { 
+        headers: { 'Content-Type': 'application/json', 'accept': 'application/json' },
+        tags: { type: 'get' }
+    })
+    check(getProbe, { 'GET /items returns status 200': res => res.status === 200 })
 }
 
 export default function () {
-    http.get('http://crud-service:3000/users/export?shopID=2')
-    sleep(1)
+    const items = generateItems()
+
+    const postList = http.post(`http://crud-service:3000/items/bulk`, JSON.stringify(items), { 
+        headers: { 'Content-Type': 'application/json', 'accept': 'application/json' },
+        tags: { type: 'bulk' }
+    })
+    check(postList, { 'POST / returns status 200': res => res.status === 200 }) 
 }
 
 export function teardown(data) {

lib/CrudService.js

@@ -36,10 +36,13 @@ const {
   PUSHCMD,
   ADDTOSETCMD,
   PULLCMD,
+  QUERY,
+  STATE,
 } = require('./consts')
 const { getStateQuery } = require('./CrudService.utils')
 const { JSONPath } = require('jsonpath-plus')
 const { getPathFromPointer } = require('./JSONPath.utils')
+const resolveMongoQuery = require('./resolveMongoQuery')
 
 
 const ALLOWED_COMMANDS = [
@@ -96,6 +99,10 @@ function getQueryOptions(crudServiceOptions) {
 class CrudService {
   constructor(mongoCollection, stateOnInsert, defaultSorting, options = {}) {
     assert(STATES_FINITE_STATE_MACHINE[stateOnInsert], 'Invalid `stateOnInsert`')
+
+    /**
+     * @type {import('mongodb').Collection} collection used by this CRUD instance
+     */
     this._mongoCollection = mongoCollection
     this._stateOnInsert = stateOnInsert
     this._defaultSorting = defaultSorting
@@ -218,11 +225,22 @@ class CrudService {
     return doc
   }
 
-  async insertMany(context, docs) {
+  /**
+   * performs a insert many operation over a CRUD resource
+   * @param {unknown} context context of http handler
+   * @param {unknown[]} docs array of documents
+   * @param {import('./QueryParser')} queryParser QueryParser instance
+   * @param {{ idOnly: boolean}?} opts
+   * @returns {Promise<unknown[] | import('mongodb').ObjectId[]>} if flag idOnly is set to true,
+   * returns just the mongodb object ids
+   */
+  async insertMany(context, docs, queryParser, opts = {}) {
+    const { idOnly } = opts
     context.log.debug({ docs }, 'insertMany operation requested')
     assert(docs.length > 0, 'At least one element is required')
 
     for (const doc of docs) {
+      queryParser.parseAndCastBody(doc)
       assertDocHasNotStandardField(doc)
       doc[CREATORID] = context.userId
       doc[UPDATERID] = context.userId
@@ -232,18 +250,19 @@ class CrudService {
     }
 
     const writeOpResult = await this._mongoCollection.insertMany(docs)
+    context.log.debug({ docIds: writeOpResult.insertedIds }, 'insertMany operation executed')
 
-    const indexKeys = Object.keys(writeOpResult.insertedIds)
-    const resultDocs = new Array(indexKeys.length)
-    for (const indexKey of indexKeys) {
-      resultDocs[indexKey] = {
-        _id: writeOpResult.insertedIds[indexKey],
-        ...docs[indexKey],
+    if (!idOnly) {
+      for (const indexKey of Object.keys(writeOpResult.insertedIds)) {
+        docs[indexKey] = {
+          _id: writeOpResult.insertedIds[indexKey],
+          ...docs[indexKey],
+        }
       }
+      return docs
     }
 
-    context.log.debug({ docIds: writeOpResult.insertedIds }, 'insertMany operation executed')
-    return resultDocs
+    return Object.values(writeOpResult.insertedIds).map((_id) => ({ _id }))
   }
 
   async deleteById(context, id, query, _states) {
@@ -347,12 +366,13 @@ class CrudService {
     return writeOpResult
   }
 
-  async upsertMany(context, documents) {
+  async upsertMany(context, documents, queryParser) {
     context.log.debug(documents, 'upsertMany operation requested')
     assert(documents.length > 0, 'At least one element is required')
 
     const operations = []
     for (const document of documents) {
+      queryParser.parseAndCastBody(document)
       assertDocHasNotStandardField(document)
 
       const operation = {
@@ -385,24 +405,57 @@ class CrudService {
     return upsertedCount
   }
 
-  async patchBulk(context, filterUpdateCommands) {
+  /**
+   * Performs a patch bulk operation over the CRUD resource
+   * @param {unknown} context
+   * @param {unknown[]} filterUpdateCommands list of commands that needs to be executed
+   * @param {import('./QueryParser')} queryParser
+   * @param {Function} castCollectionId function that casts _id field (if defined in commands)
+   * @param {*} editableFields
+   * @param {*} aclRows
+   * @returns {Promise<number>} the number of modified documents
+   */
+  // eslint-disable-next-line max-statements
+  async patchBulk(
+    context, filterUpdateCommands, queryParser, castCollectionId, editableFields, aclRows
+  ) {
     context.log.debug(filterUpdateCommands, 'patchBulk operation requested')
     assert(filterUpdateCommands.length > 0, 'At least one element is required')
 
     const unorderedBulkOp = this._mongoCollection.initializeUnorderedBulkOp()
+    const parsedAndCastedCommands = new Array(filterUpdateCommands.length)
     for (let i = 0; i < filterUpdateCommands.length; i++) {
-      const { _id, commands, query, state: allowedStates } = filterUpdateCommands[i]
+      const { filter, update } = filterUpdateCommands[i]
+      const {
+        _id,
+        [QUERY]: clientQueryString,
+        [STATE]: state,
+        ...otherParams
+      } = filter
+
+      const commands = update
+
+      const mongoQuery = resolveMongoQuery(queryParser, clientQueryString, aclRows, otherParams, false)
+      queryParser.parseAndCastCommands(commands, editableFields)
+
+      const allowedStates = state.split(',')
+
+      parsedAndCastedCommands[i] = {
+        commands,
+        state: allowedStates,
+        query: mongoQuery,
+      }
+
       const stateQuery = getStateQuery(allowedStates)
-      const isQueryValid = query && Object.keys(query).length > 0
+      const isQueryValid = mongoQuery && Object.keys(mongoQuery).length > 0
 
-      const searchQuery = isQueryValid ? { $and: [query, stateQuery] } : { $and: [stateQuery] }
+      const searchQuery = isQueryValid ? { $and: [mongoQuery, stateQuery] } : { $and: [stateQuery] }
 
       if (_id) {
-        searchQuery.$and.push({ _id })
+        searchQuery.$and.push({ _id: castCollectionId(_id) })
       }
 
       assertCommands(commands)
-
       commands.$set = commands.$set || {}
       commands.$set.updaterId = context.userId
       commands.$set.updatedAt = context.now

lib/consts.js

@@ -132,4 +132,8 @@ module.exports = Object.freeze({
   UNIQUE_INDEX_ERROR_STATUS_CODE: 409,
   UNSUPPORTED_MIME_TYPE_STATUS_CODE: 415,
   INTERNAL_SERVER_ERROR_STATUS_CODE: 500,
+
+  // headers
+  ACL_ROWS: 'acl_rows',
+  ACL_WRITE_COLUMNS: 'acl_write_columns',
 })