Meteor aggregate with support for Decimal fields, custom types...

aggregate.js

// var MongoDB = NpmModuleMongodb;
var MongoDB =  MongoInternals.NpmModules.mongodb.module;

// just beautiful copy paste from mongo_driver.js

// This is used to add or remove EJSON from the beginning of everything nested
// inside an EJSON custom type. It should only be called on pure JSON!
var replaceNames = function (filter, thing) {
    if (typeof thing === "object" && thing !== null) {
        if (_.isArray(thing)) {
            return _.map(thing, _.bind(replaceNames, null, filter));
        }
        var ret = {};
        _.each(thing, function (value, key) {
            ret[filter(key)] = replaceNames(filter, value);
        });
        return ret;
    }
    return thing;
};

var unmakeMongoLegal = function (name) { return name.substr(5); };

var replaceTypes = function (document, atomTransformer) {
    if (typeof document !== 'object' || document === null)
        return document;

    var replacedTopLevelAtom = atomTransformer(document);
    if (replacedTopLevelAtom !== undefined)
        return replacedTopLevelAtom;

    var ret = document;
    _.each(document, function (val, key) {
        var valReplaced = replaceTypes(val, atomTransformer);
        if (val !== valReplaced) {
            // Lazy clone. Shallow copy.
            if (ret === document)
                ret = _.clone(document);
            ret[key] = valReplaced;
        }
    });
    return ret;
};

var replaceMongoAtomWithMeteor = function (document) {
    if (document instanceof MongoDB.Binary) {
        var buffer = document.value(true);
        return new Uint8Array(buffer);
    }
    if (document instanceof MongoDB.ObjectID) {
        return new Mongo.ObjectID(document.toHexString());
    }
    if (document instanceof MongoDB.Decimal128) {
        return Decimal(document.toString());
    }
    if (document["EJSON$type"] && document["EJSON$value"] && _.size(document) === 2) {
        return EJSON.fromJSONValue(replaceNames(unmakeMongoLegal, document));
    }
    if (document instanceof MongoDB.Timestamp) {
        // For now, the Meteor representation of a Mongo timestamp type (not a date!
        // this is a weird internal thing used in the oplog!) is the same as the
        // Mongo representation. We need to do this explicitly or else we would do a
        // structural clone and lose the prototype.
        return document;
    }
    return undefined;
};

var replaceTypes = function (document, atomTransformer) {
    if (typeof document !== 'object' || document === null)
        return document;

    var replacedTopLevelAtom = atomTransformer(document);
    if (replacedTopLevelAtom !== undefined)
        return replacedTopLevelAtom;

    var ret = document;
    _.each(document, function (val, key) {
        var valReplaced = replaceTypes(val, atomTransformer);
        if (val !== valReplaced) {
            // Lazy clone. Shallow copy.
            if (ret === document)
                ret = _.clone(document);
            ret[key] = valReplaced;
        }
    });
    return ret;
};

wrapAsync = Meteor.wrapAsync || Meteor._wrapAsync;

Mongo.Collection.prototype.aggregate = function(pipelines, options) {
    var coll;
    if (this.rawCollection) {
        // >= Meteor 1.0.4
        coll = this.rawCollection();
    } else {
        // < Meteor 1.0.4
        coll = this._getCollection();
    }
    if (MongoInternals.NpmModules.mongodb.version[0] === '3') {
        var cursor = wrapAsync(coll.aggregate, coll)(pipelines, options);
        return wrapAsync(cursor.toArray, cursor)()
            .map(r => replaceTypes(r, replaceMongoAtomWithMeteor));
    }
    return wrapAsync(coll.aggregate.bind(coll))(pipelines, options)
        .map(r => replaceTypes(r, replaceMongoAtomWithMeteor));
};

Solution based on:
sakulstra:aggregate
and
Meteor Mongo driver

Usage

Just import aggregate.js on server initialization.

leaving a big THANK YOU here for this... awesome work.

i simplified it a little bit for our use case (we don't need old meteor support) - also removed a (seemingly) redundant replaceTypes function:

let MongoDB = MongoInternals.NpmModules.mongodb.module;

// just beautiful copy paste from mongo_driver.js

// This is used to add or remove EJSON from the beginning of everything nested
// inside an EJSON custom type. It should only be called on pure JSON!
const replaceNames = (filter, thing) => {
  if (typeof thing === "object" && thing !== null) {
    if (_.isArray(thing)) {
      return _.map(thing, _.bind(replaceNames, null, filter));
    }
    let ret = {};
    _.each(thing, function (value, key) {
      ret[filter(key)] = replaceNames(filter, value);
    });
    return ret;
  }
  return thing;
};

const unmakeMongoLegal = name => {
  return name.substr(5);
};

const replaceTypes = (document, atomTransformer) => {
  if (typeof document !== "object" || document === null) return document;

  const replacedTopLevelAtom = atomTransformer(document);
  if (replacedTopLevelAtom !== undefined) return replacedTopLevelAtom;

  let ret = document;
  _.each(document, function (val, key) {
    const valReplaced = replaceTypes(val, atomTransformer);
    if (val !== valReplaced) {
      // Lazy clone. Shallow copy.
      if (ret === document) ret = _.clone(document);
      ret[key] = valReplaced;
    }
  });
  return ret;
};

const replaceMongoAtomWithMeteor = document => {
  if (document instanceof MongoDB.Binary) {
    const buffer = document.value(true);
    return new Uint8Array(buffer);
  }
  if (document instanceof MongoDB.ObjectID) {
    return new Mongo.ObjectID(document.toHexString());
  }
  if (document instanceof MongoDB.Decimal128) {
    return Decimal(document.toString());
  }
  if (document["EJSON$type"] && document["EJSON$value"] && _.size(document) === 2) {
    return EJSON.fromJSONValue(replaceNames(unmakeMongoLegal, document));
  }
  if (document instanceof MongoDB.Timestamp) {
    // For now, the Meteor representation of a Mongo timestamp type (not a date!
    // this is a weird internal thing used in the oplog!) is the same as the
    // Mongo representation. We need to do this explicitly or else we would do a
    // structural clone and lose the prototype.
    return document;
  }
  return undefined;
};

Mongo.Collection.prototype.aggregate = function (pipelines, options) {
  const coll = this.rawCollection();
  return coll.aggregate
    .bind(coll)(pipelines, options)
    .map(r => replaceTypes(r, replaceMongoAtomWithMeteor));
};