fix aggregate mongo function

Author: tysonrm

src/adapters/datasources/datasource-mongodb.js

@@ -259,6 +259,8 @@ export class DataSourceMongoDb extends DataSource {
    */
   async mongoFind ({ filter, sort, limit, aggregate, skip } = {}) {
     console.log({ fn: this.mongoFind.name, filter })
+    console.log({ aggregate })
+
     let cursor = aggregate
       ? (await this.collection()).aggregate(aggregate)
       : (await this.collection()).find(filter)
@@ -294,36 +296,10 @@ export class DataSourceMongoDb extends DataSource {
    * }} param0
    * @returns
    */
-  streamList ({ serialize, options }) {
-    try {
-      const serializer = new Transform({
-        writableObjectMode: true,
-
-        // start of array
-        construct (callback) {
-          this.first = true
-          this.push('[')
-          callback()
-        },
-
-        // each chunk is a record
-        transform (chunk, _encoding, next) {
-          // comma-separate
-          if (this.first) this.first = false
-          else this.push(',')
-
-          // serialize record
-          this.push(JSON.stringify(chunk))
-          next()
-        },
-
-        // end of array
-        flush (callback) {
-          this.push(']')
-          callback()
-        }
-      })
+  async streamList (options) {
+    const pipeArgs = []
 
+    try {
       const paginate = new Transform({
         writableObjectMode: true,
 
@@ -346,29 +322,25 @@ export class DataSourceMongoDb extends DataSource {
         }
       })
 
-      return new Promise(async (resolve, reject) => {
-        const pipeArgs = []
-        const readable = (await this.mongoFind(options)).stream()
-
-        // optionally transform db stream then pipe to output
-        pipeArgs.push(readable)
+      const readable = (await this.mongoFind(options)).stream()
+      // optionally transform db stream then pipe to output
+      pipeArgs.push(readable)
 
-        if (options.page) {
-          const count = ~~(await this.mongoCount(options.filter))
+      if (options.page) {
+        const count = ~~(await this.mongoCount(options.filter))
 
-          paginate._transform(
-            `"count":${count}, "total":${Math.ceil(
-              count / options.limit
-            )}, ${JSON.stringify(options).slice(1, -1)}, "data":`,
-            'utf8',
-            () => console.log('paginated query', options)
-          )
+        paginate._transform(
+          `"count":${count}, "total":${Math.ceil(
+            count / options.limit
+          )}, ${JSON.stringify(options).slice(1, -1)}, "data":`,
+          'utf8',
+          () => console.log('paginated query', options)
+        )
 
-          pipeArgs.push(paginate)
-        }
+        pipeArgs.push(paginate)
+      }
 
-        return pipeArgs
-      })
+      return pipeArgs
     } catch (error) {
       console.error({
         fn: this.streamList.name,
@@ -378,6 +350,7 @@ export class DataSourceMongoDb extends DataSource {
   }
 
   processOptions ({ options = {}, query = {} }) {
+    console.log(processQuery(query))
     return { ...processQuery(query), ...options } // options must overwite the query not otherwise
   }
 
@@ -388,34 +361,35 @@ export class DataSourceMongoDb extends DataSource {
    */
   async list (param) {
     try {
+      console.log({ param })
       let result
       const options = this.processOptions(param)
+
       if (0 < ~~options.__page) {
         // qpm > processOptions weeds out __page - add it back properly as an integer
         options.page = ~~options.__page
         options.skip = (options.page - 1) * options.limit || 0
       }
-      if (options.__aggregate) {
+
+      if (param.query.__aggregate) {
         // qpm > processOptions weeds out __aggregate - add it back properly as parsed json
         try {
-          const aggregateQuery = JSON.parse(query.__aggregate)
-          options.aggregate = aggregateQuery
+          options.aggregate = JSON.parse(param.query.__aggregate)
         } catch (e) {
           console.error(e, 'invalid Aggregate')
         }
       }
       console.log({ options })
 
-      if (options.streamRequested) this.streamList({ serialize, options })
-      else {
-        const data = (await this.mongoFind(options)).toArray()
-        const count = data?.length
-        result = {
-          ...options,
-          data,
-          count,
-          total: Math.ceil(count / options.limit)
-        }
+      if (options.streamRequested) return this.streamList(options)
+
+      const data = (await this.mongoFind(options)).toArray()
+      const count = data?.length
+      result = {
+        ...options,
+        data,
+        count,
+        total: Math.ceil(count / options.limit)
       }
 
       return options?.page ? result : result?.data

src/domain/datasource-factory.js

@@ -38,15 +38,15 @@ const DefaultDataSource =
  */
 const DsCoreExtensions = superclass =>
   class extends superclass {
-    constructor(map, name, namespace, options) {
+    constructor (map, name, namespace, options) {
       super(map, name, namespace, options)
     }
 
-    set factory(value) {
+    set factory (value) {
       this[FACTORY] = value
     }
 
-    get factory() {
+    get factory () {
       return this[FACTORY]
     }
 
@@ -61,7 +61,7 @@ const DsCoreExtensions = superclass =>
      * @param {string} id
      * @param {Model} data
      */
-    async save(id, data) {
+    async save (id, data) {
       try {
         this.saveSync(id, data)
         await super.save(id, JSON.parse(JSON.stringify(data)))
@@ -78,7 +78,7 @@ const DsCoreExtensions = superclass =>
      * @param {string} id
      * @returns {Promise<Model>|undefined}
      */
-    async find(id) {
+    async find (id) {
       try {
         const cached = this.findSync(id)
         if (cached) return cached
@@ -98,33 +98,33 @@ const DsCoreExtensions = superclass =>
       }
     }
 
-    hydrate() {
+    hydrate () {
       const ctx = this
 
       return new Transform({
         objectMode: true,
 
-        transform(chunk, encoding, next) {
+        transform (chunk, encoding, next) {
           this.push(ModelFactory.loadModel(broker, ctx, chunk, ctx.name))
           next()
-        },
+        }
       })
     }
 
-    serialize() {
+    serialize () {
       let first = true
 
       return new Transform({
         objectMode: true,
 
         // start of array
-        construct(callback) {
+        construct (callback) {
           this.push('[')
           callback()
         },
 
         // each chunk is a record
-        transform(chunk, encoding, next) {
+        transform (chunk, encoding, next) {
           // comma-separate
           if (first) first = false
           else this.push(',')
@@ -135,10 +135,10 @@ const DsCoreExtensions = superclass =>
         },
 
         // end of array
-        flush(callback) {
+        flush (callback) {
           this.push(']')
           callback()
-        },
+        }
       })
     }
 
@@ -148,7 +148,7 @@ const DsCoreExtensions = superclass =>
      * @param {import('./datasource').listOptions} options
      * @returns {Array<Readable|Transform>}
      */
-    stream(list, options) {
+    stream (list, options) {
       return new Promise((resolve, reject) => {
         options.writable.on('error', reject)
         options.writable.on('end', resolve)
@@ -173,12 +173,15 @@ const DsCoreExtensions = superclass =>
      * @override
      * @param {import('../../domain/datasource').listOptions} param
      */
-    async list(options) {
+    async list (options) {
       try {
         if (options?.query?.__count) return this.count()
         if (options?.query?.__cached) return this.listSync(options.query)
 
-        const opts = { ...options, streamRequested: options?.writable }
+        const opts = {
+          ...options,
+          streamRequested: options?.writable ? true : false
+        }
         const list = [await super.list(opts)].flat()
 
         if (list.length < 1) return []
@@ -202,7 +205,7 @@ const DsCoreExtensions = superclass =>
      * @param {*} id
      * @returns
      */
-    async delete(id) {
+    async delete (id) {
       try {
         await super.delete(id)
         // only if super succeeds
@@ -238,11 +241,11 @@ const DataSourceFactory = (() => {
    * @param {*} name
    * @returns
    */
-  function hasDataSource(name) {
+  function hasDataSource (name) {
     return dataSources.has(name)
   }
 
-  function listDataSources() {
+  function listDataSources () {
     return [...dataSources.keys()]
   }
 
@@ -267,7 +270,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} options
    * @returns {typeof DataSource}
    */
-  function createDataSourceClass(spec, options) {
+  function createDataSourceClass (spec, options) {
     const { memoryOnly, ephemeral, adapterName } = options
 
     if (memoryOnly || ephemeral) return dsClasses['DataSourceMemory']
@@ -295,7 +298,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} options
    * @returns {typeof DataSource}
    */
-  function extendDataSourceClass(DsClass, options = {}) {
+  function extendDataSourceClass (DsClass, options = {}) {
     const mixins = [extendClass].concat(options.mixins || [])
     return compose(...mixins)(DsClass)
   }
@@ -306,7 +309,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} [options]
    * @returns {DataSource}
    */
-  function createDataSource(name, namespace, options) {
+  function createDataSource (name, namespace, options) {
     const spec = ModelFactory.getModelSpec(name)
     const dsMap = options.dsMap || new Map()
 
@@ -332,7 +335,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} options
    * @returns {import('./datasource').default}
    */
-  function getDataSource(name, namespace = null, options = {}) {
+  function getDataSource (name, namespace = null, options = {}) {
     if (!dataSources) dataSources = new Map()
     if (!namespace) return dataSources.get(name)
     if (dataSources.has(name)) return dataSources.get(name)
@@ -347,7 +350,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} [options]
    * @returns
    */
-  function getSharedDataSource(name, namespace = null, options = {}) {
+  function getSharedDataSource (name, namespace = null, options = {}) {
     if (!dataSources) dataSources = new Map()
     if (!namespace) return dataSources.get(name)
     if (dataSources.has(name)) return dataSources.get(name)
@@ -362,20 +365,20 @@ const DataSourceFactory = (() => {
    * @param {string} name
    * @returns {ProxyHandler<DataSource>}
    */
-  function getRestrictedDataSource(name, namespace, options) {
+  function getRestrictedDataSource (name, namespace, options) {
     return new Proxy(getDataSource(name, namespace, options), {
-      get(target, key) {
+      get (target, key) {
         if (key === 'factory') {
           throw new Error('unauthorized')
         }
       },
-      ownKeys(target) {
+      ownKeys (target) {
         return []
-      },
+      }
     })
   }
 
-  function close() {
+  function close () {
     dataSources.forEach(ds => ds.close())
   }
 
@@ -390,7 +393,7 @@ const DataSourceFactory = (() => {
     getRestrictedDataSource,
     hasDataSource,
     listDataSources,
-    close,
+    close
   })
 })()