Support for mongo transactions

README.md

@@ -99,6 +99,33 @@ const Post = Class.create({
 });
 ```
 
+## Mongo Transactions Support
+
+```js
+import Post from './post'
+import User from './user'
+import { MongoInternals } from 'meteor/mongo';
+
+let post = Post.find(postQuery);
+let user = User.find(userQuery);
+
+post.userId = user._id;
+user.posts.push(post);
+
+const { client } = MongoInternals.defaultRemoteCollectionDriver().mongo;
+const session = await client.startSession();
+await session.startTransaction();
+try {
+  post.save({session});
+  user.save({session});
+  await session.commitTransaction();
+} catch (e) {
+  await session.abortTransaction();
+} finally {
+  session.endSession();
+}
+```
+
 ## Supporters
 
 [<img src="http://jagi.github.io/meteor-astronomy/images/usefulio.png" />](http://useful.io/)

lib/modules/fields/module.js

@@ -6,6 +6,7 @@ import './types/mongo_object_id.js';
 import './types/number.js';
 import './types/object.js';
 import './types/string.js';
+import './types/decimal.js';
 // Utils.
 import castNested from './utils/castNested';
 import getAll from './utils/getAll';
@@ -49,4 +50,4 @@ Module.create({
     setOne,
     traverse
   }
-});
+});

lib/modules/fields/types/decimal.js

@@ -0,0 +1,14 @@
+import Type from '../type.js';
+import Validators from '../../validators/validators.js';
+import { Decimal } from 'meteor/mongo-decimal';
+
+Type.create({
+  name: 'Decimal',
+  class: Decimal,
+  cast(value) {
+    Decimal(value.toString())
+  },
+  validate(args) {
+    Decimal.isDecimal(args)
+  }
+});

lib/modules/storage/class_prototype_methods/remove.js

@@ -57,7 +57,8 @@ function remove(args = {}, callback) {
     let methodArgs = {
       doc,
       simulation,
-      trusted: true
+      trusted: true,
+      session: args.session
     };
     let result = documentRemove(methodArgs);
     if (callback) {
@@ -74,4 +75,4 @@ function remove(args = {}, callback) {
   }
 }
 
-export default remove;
+export default remove;

lib/modules/storage/class_prototype_methods/save.js

@@ -139,6 +139,7 @@ function save(options = {}, callback) {
       stopOnFirstError: options.stopOnFirstError,
       simulation: options.simulation,
       trusted: true,
+      session: options.session
     };
     if (inserting) {
       let result = documentInsert(methodArgs);

lib/modules/storage/class_static_methods/insert.js

@@ -2,16 +2,26 @@ import isRemote from '../utils/is_remote.js';
 import callMeteorMethod from '../utils/call_meteor_method.js';
 import classInsert from '../utils/class_insert.js';
 
-function insert(rawDoc, callback) {
+function insert(rawDoc, options, callback) {
   const Class = this;
   const Collection = Class.getCollection();
 
+  // If we omit options argument then it may be a callback function.
+  if (options instanceof Function) {
+    callback = options;
+    options = {};
+  }
+
   // Prepare arguments.
   const args = {
     className: Class.getName(),
     rawDoc
   };
 
+  if (options && options.session) {
+    args.session = options.session
+  }
+
   // Generate ID if not provided.
   if (!rawDoc._id) {
     let generateId = true;
@@ -71,4 +81,4 @@ function insert(rawDoc, callback) {
   }
 }
 
-export default insert;
+export default insert;

lib/modules/storage/utils/class_insert.js

@@ -9,6 +9,7 @@ function classInsert(args = {}) {
     fields,
     simulation = true,
     trusted = false,
+    session
   } = args;
 
   // Stop execution, if we are not on the server, when the "simulation" flag is
@@ -27,7 +28,8 @@ function classInsert(args = {}) {
     stopOnFirstError,
     simulation,
     trusted,
+    session
   });
 };
 
-export default classInsert;
+export default classInsert;

lib/modules/storage/utils/class_remove.js

@@ -9,7 +9,7 @@ function classRemove(args = {}) {
     selector,
     options,
     simulation = true,
-    trusted = false
+    trusted = false,
   } = args;
 
   // Stop execution, if we are not on the server, when the "simulation" flag is
@@ -35,11 +35,12 @@ function classRemove(args = {}) {
     result += documentRemove({
       doc,
       simulation,
-      trusted
+      trusted,
+      session: options && options.session
     });
   });
 
   return result;
 };
 
-export default classRemove;
+export default classRemove;

lib/modules/storage/utils/class_update.js

@@ -86,11 +86,12 @@ function classUpdate(args = {}) {
       simulation,
       fields,
       trusted,
-      oldDoc
+      oldDoc,
+      session: options.session
     });
   });
 
   return result;
 };
 
-export default classUpdate;
+export default classUpdate;

lib/modules/storage/utils/class_upsert.js

@@ -74,7 +74,8 @@ function classUpsert(args = {}) {
         stopOnFirstError,
         simulation,
         fields,
-        trusted
+        trusted,
+        session: options.session
       });
     });
   }
@@ -102,10 +103,11 @@ function classUpsert(args = {}) {
       stopOnFirstError,
       simulation,
       trusted,
+      session: options.session
     });
   }
 
   return result;
 };
 
-export default classUpsert;
+export default classUpsert;

lib/modules/storage/utils/document_insert.js

@@ -6,14 +6,16 @@ import triggerBeforeInsert from './trigger_before_insert';
 import triggerAfterSave from './trigger_after_save';
 import triggerAfterInsert from './trigger_after_insert';
 import documentValidate from '../../validators/utils/document_validate';
+import { replaceTypes, replaceMeteorAtomWithMongo } from './replace_types';
 
 function documentInsert(args = {}) {
   const {
     doc,
     stopOnFirstError,
     fields,
     simulation = true,
-    trusted = false
+    trusted = false,
+    session
   } = args;
 
   // Stop execution, if we are not on the server, when the "simulation" flag is
@@ -22,6 +24,9 @@ function documentInsert(args = {}) {
     return;
   }
 
+  const options = {}
+  if (session) options.session = session;
+
   const Class = doc.constructor;
   const Collection = Class.getCollection();
 
@@ -76,7 +81,13 @@ function documentInsert(args = {}) {
     // server it returns array of inserted documents. So we always return the
     // generated id. We can't send an entire document because it could be a
     // security issue if we are not subscribed to all fields of a document.
-    Collection._collection.insert(values);
+    // if (session) {
+    //   values = replaceTypes(values, replaceMeteorAtomWithMongo);
+    //   Promise.await(Collection.rawCollection().insert(values, {session}));
+    // } else {
+    //   Collection._collection.insert(values);
+    // }
+    Collection._collection.insert(values, options);
 
     // Change the "_isNew" flag to "false". Mark a document as not new.
     doc._isNew = false;

lib/modules/storage/utils/document_remove.js

@@ -5,7 +5,8 @@ function documentRemove(args = {}) {
   const {
     doc,
     simulation = true,
-    trusted = false
+    trusted = false,
+    session
   } = args;
 
   // Stop execution, if we are not on the server, when the "simulation" flag is
@@ -14,6 +15,9 @@ function documentRemove(args = {}) {
     return;
   }
 
+  const options = {}
+  if (session) options.session = session;
+
   const Class = doc.constructor;
   const Collection = Class.getCollection();
 
@@ -32,9 +36,19 @@ function documentRemove(args = {}) {
 
   // Remove a document.
   try {
-    const result = Collection._collection.remove({
+    let result
+    // if (session) {
+    //   result = Promise.await(Collection.rawCollection().remove({
+    //     _id: doc._id
+    //   }, {session}));
+    // } else {
+    //   result = Collection._collection.remove({
+    //     _id: doc._id
+    //   });
+    // }
+    result = Collection._collection.remove({
       _id: doc._id
-    });
+    }, options);
 
     // Mark a document as new, so it will be possible to save it again.
     doc._isNew = true;
@@ -54,4 +68,4 @@ function documentRemove(args = {}) {
   }
 };
 
-export default documentRemove;
+export default documentRemove;

lib/modules/storage/utils/document_update.js

@@ -8,6 +8,7 @@ import triggerAfterUpdate from "./trigger_after_update";
 import isModified from "./isModified";
 import getModifier from "./getModifier";
 import documentValidate from "../../validators/utils/document_validate";
+import { replaceTypes, replaceMeteorAtomWithMongo } from './replace_types';
 
 function documentUpdate(args = {}) {
   let {
@@ -17,7 +18,8 @@ function documentUpdate(args = {}) {
     simulation = true,
     forceUpdate = false,
     trusted = false,
-    oldDoc
+    oldDoc,
+    session
   } = args;
 
   // Stop execution, if we are not on the server, when the "simulation" flag is
@@ -26,6 +28,9 @@ function documentUpdate(args = {}) {
     return;
   }
 
+  const options = {}
+  if (session) options.session = session;
+
   let Class = doc.constructor;
   let Collection = Class.getCollection();
 
@@ -97,13 +102,32 @@ function documentUpdate(args = {}) {
   }
   // Update a document.
   try {
-    const result = Collection._collection.update(
+    let result;
+    // if (session && Meteor.isServer) {
+    //   let _modifier = replaceTypes(modifier, replaceMeteorAtomWithMongo);
+    //   result = Promise.await(Collection.rawCollection().updateOne(
+    //     {
+    //       _id: doc._id
+    //     },
+    //     _modifier,
+    //     {session}
+    //   ));
+    // } else {
+    //   result = Collection._collection.update(
+    //     {
+    //       _id: doc._id
+    //     },
+    //     modifier
+    //   );
+    // }
+    // console.log('document_update', modifier)
+    result = Collection._collection.update(
       {
         _id: doc._id
       },
-      modifier
+      modifier,
+      options
     );
-
     // Trigger after events.
     triggerAfterUpdate(args);
     triggerAfterSave(args);

lib/modules/storage/utils/getModifier.js

@@ -5,6 +5,7 @@ import _isPlainObject from "lodash/isPlainObject";
 import _omitBy from "lodash/omitBy";
 import _size from "lodash/size";
 import { EJSON } from "meteor/ejson";
+import { Decimal } from 'meteor/mongo-decimal'
 import throwParseError from "../../core/utils/throw_parse_error.js";
 import rawMany from "../../fields/utils/rawMany";
 import diff from "./diff";
@@ -115,6 +116,9 @@ handlers.onScalarDiff = function({ oldValue, newValue, prefix, result }) {
       !_isNaN(newValue)
     ) {
       result.$inc[prefix] = newValue - oldValue;
+    } else if (oldValue instanceof Decimal || newValue instanceof Decimal) {
+      // result.$inc[prefix] = Decimal(newValue || '0').minus(oldValue || '0')
+      result.$set[prefix] = newValue;
     } else {
       result.$set[prefix] = newValue;
     }