Merge pull request #5065 from Automattic/3998

`Model.bulkWrite()`
Automattic · Mar 12, 2017 · 27e461e · 27e461e
2 parents c3d7429 + 06d8509
commit 27e461e
Show file tree

Hide file tree

Showing 4 changed files with 576 additions and 272 deletions.
diff --git a/lib/model.js b/lib/model.js
@@ -15,6 +15,8 @@ var VersionError = require('./error').VersionError;
 var applyHooks = require('./services/model/applyHooks');
 var applyMethods = require('./services/model/applyMethods');
 var applyStatics = require('./services/model/applyStatics');
+var cast = require('./cast');
+var castUpdate = require('./services/query/castUpdate');
 var discriminator = require('./services/model/discriminator');
 var isPathSelectedInclusive = require('./services/projection/isPathSelectedInclusive');
 var parallel = require('async/parallel');
@@ -2010,6 +2012,170 @@ Model.insertMany = function(arr, options, callback) {
   });
 };
 
+/**
+ * Sends multiple `insertOne`, `updateOne`, `updateMany`, `replaceOne`,
+ * `deleteOne`, and/or `deleteMany` operations to the MongoDB server in one
+ * command. This is faster than sending multiple independent operations because
+ * with `bulkWrite()` there is only one round trip to the server.
+ *
+ * Mongoose will perform casting on all operations you provide.
+ *
+ * This function does **not** trigger any middleware.
+ *
+ * ####Example:
+ *
+ *     Character.bulkWrite([
+ *       {
+ *         insertOne: {
+ *           document: {
+ *             name: 'Eddard Stark',
+ *             title: 'Warden of the North'
+ *           }
+ *         }
+ *       },
+ *       {
+ *         updateOne: {
+ *           filter: { name: 'Eddard Stark' },
+ *           // Mongoose inserts `$set` for you in the update below
+ *           update: { title: 'Hand of the King' }
+ *         }
+ *       },
+ *       {
+ *         deleteOne: {
+ *           { name: 'Eddard Stark' }
+ *         }
+ *       }
+ *     ]).then(handleResult);
+ *
+ * @param {Array} ops
+ * @param {Object} [options]
+ * @param {Function} [callback] callback `function(error, bulkWriteOpResult) {}`
+ * @return {Promise} resolves to a `BulkWriteOpResult` if the operation succeeds
+ * @see writeOpResult http://mongodb.github.io/node-mongodb-native/2.2/api/Collection.html#~BulkWriteOpResult
+ * @api public
+ */
+
+Model.bulkWrite = function(ops, options, callback) {
+  var Promise = PromiseProvider.get();
+  var _this = this;
+  if (typeof options === 'function') {
+    callback = options;
+    options = null;
+  }
+  if (callback) {
+    callback = this.$wrapCallback(callback);
+  }
+  options = options || {};
+
+  var validations = ops.map(function(op) {
+    if (op['insertOne']) {
+      return function(callback) {
+        op['insertOne']['document'] = new _this(op['insertOne']['document']);
+        op['insertOne']['document'].validate({ __noPromise: true }, function(error) {
+          if (error) {
+            return callback(error);
+          }
+          callback(null);
+        });
+      };
+    } else if (op['updateOne']) {
+      return function(callback) {
+        try {
+          op['updateOne']['filter'] = cast(_this.schema,
+            op['updateOne']['filter']);
+          op['updateOne']['update'] = castUpdate(_this.schema,
+            op['updateOne']['update'], _this.schema.options.strict);
+        } catch (error) {
+          return callback(error);
+        }
+
+        callback(null);
+      };
+    } else if (op['updateMany']) {
+      return function(callback) {
+        try {
+          op['updateMany']['filter'] = cast(_this.schema,
+            op['updateMany']['filter']);
+          op['updateMany']['update'] = castUpdate(_this.schema, op['updateMany']['filter'], {
+            strict: _this.schema.options.strict,
+            overwrite: false
+          });
+        } catch (error) {
+          return callback(error);
+        }
+
+        callback(null);
+      };
+    } else if (op['replaceOne']) {
+      return function(callback) {
+        try {
+          op['replaceOne']['filter'] = cast(_this.schema,
+            op['replaceOne']['filter']);
+        } catch (error) {
+          return callback(error);
+        }
+
+        // set `skipId`, otherwise we get "_id field cannot be changed"
+        op['replaceOne']['replacement'] =
+          new _this(op['replaceOne']['replacement'], null, true);
+        op['replaceOne']['replacement'].validate({ __noPromise: true }, function(error) {
+          if (error) {
+            return callback(error);
+          }
+          callback(null);
+        });
+      };
+    } else if (op['deleteOne']) {
+      return function(callback) {
+        try {
+          op['deleteOne']['filter'] = cast(_this.schema,
+            op['deleteOne']['filter']);
+        } catch (error) {
+          return callback(error);
+        }
+
+        callback(null);
+      };
+    } else if (op['deleteMany']) {
+      return function(callback) {
+        try {
+          op['deleteMany']['filter'] = cast(_this.schema,
+            op['deleteMany']['filter']);
+        } catch (error) {
+          return callback(error);
+        }
+
+        callback(null);
+      };
+    } else {
+      return function(callback) {
+        callback(new Error('Invalid op passed to `bulkWrite()`'));
+      };
+    }
+  });
+
+  var promise = new Promise.ES6(function(resolve, reject) {
+    parallel(validations, function(error) {
+      if (error) {
+        callback && callback(error);
+        return reject(error);
+      }
+
+      _this.collection.bulkWrite(ops, options, function(error, res) {
+        if (error) {
+          callback && callback(error);
+          return reject(error);
+        }
+
+        callback && callback(null, res);
+        resolve(res);
+      });
+    });
+  });
+
+  return promise;
+};
+
 /**
  * Shortcut for creating a new Document from existing raw data, pre-saved in the DB.
  * The document returned has no paths marked as modified initially.