diff --git a/packages/ezsLodex/src/index.js b/packages/ezsLodex/src/index.js index 09a4ff552..36fe639ed 100644 --- a/packages/ezsLodex/src/index.js +++ b/packages/ezsLodex/src/index.js @@ -25,6 +25,7 @@ import buildContext from './buildContext'; import aggregateQuery from './aggregateQuery'; import LodexJoinQuery from './joinQuery'; import saveDocuments from './saveDocuments.js'; +import precomputedSelect from './precomputedSelect.js'; const funcs = { flattenPatch, @@ -53,6 +54,7 @@ const funcs = { aggregateQuery, LodexJoinQuery, saveDocuments, + precomputedSelect, // aliases fixFlatten: flattenPatch.flattenPatch, LodexContext: disabled.disabled, @@ -71,6 +73,7 @@ const funcs = { LodexInjectCountFrom: injectCountFrom.injectCountFrom, LodexAggregateQuery: aggregateQuery.aggregateQuery, LodexSaveDocuments: saveDocuments.saveDocuments, + LodexPrecomputedSelect: precomputedSelect.precomputedSelect, }; export default funcs; diff --git a/packages/ezsLodex/src/precomputedSelect.js b/packages/ezsLodex/src/precomputedSelect.js new file mode 100644 index 000000000..7da68db5e --- /dev/null +++ b/packages/ezsLodex/src/precomputedSelect.js @@ -0,0 +1,71 @@ +import set from 'lodash/set'; +import mongoDatabase from './mongoDatabase'; + +/** + * Take `Object` Request to precomputed collection + * + * + * @name LodexPRecomputedSelect + * @param {String} [path=value] Default path to set result + * @param {Object} [filter] MongoDB filter + * @param {Number} [limit] limit the result + * @param {Number} [skip] skip the result + * @returns {Object} + */ +export const createFunction = () => + async function LodexPrecomputedSelect(data, feed) { + if (this.isLast()) { + return feed.close(); + } + const { ezs } = this; + const path = this.getParam('path', 'value'); + const precomputedName = this.getParam('name'); + const filter = this.getParam('filter', data.filter || {}); + const limit = Number(this.getParam('limit', data.limit || 1000000)); + const skip = Number(this.getParam('skip', data.skip || 0)); + const connectionStringURI = this.getParam( + 'connectionStringURI', + this.getEnv('connectionStringURI'), + ); + const db = await mongoDatabase(connectionStringURI); + + const [precomputedObject] = await db + .collection('precomputed') + .find({ name: precomputedName }) + .toArray(); + if (!precomputedObject || !precomputedObject._id) { + return feed.stop(new Error(`${precomputedName} is unknown`)); + } + const collectionName = `pc_${precomputedObject._id.toString()}`; + + const collection = db.collection(collectionName); + let cursor = collection.find(filter); + + const total = await cursor.count(); + if (total === 0) { + return feed.send({ total: 0 }); + } + const func = (d, f, cxt) => { + if (cxt.isLast()) { + return f.close(); + } + if (!d._id) { + return f.end(); + } + delete d._id; + set(data, path, d); + f.send(data); + }; + + const stream = cursor + .skip(skip) + .limit(limit) + .stream() + .on('error', (e) => feed.stop(e)) + .pipe(ezs(func)); + await feed.flow(stream); + }; + +export default { + precomputedSelect: createFunction(), +}; diff --git a/src/api/services/enrichment/enrichment.js b/src/api/services/enrichment/enrichment.js index 88323b4b3..d04bc4c22 100644 --- a/src/api/services/enrichment/enrichment.js +++ b/src/api/services/enrichment/enrichment.js @@ -3,6 +3,7 @@ import path from 'path'; import ezs from '@ezs/core'; import progress from '../../services/progress'; import localConfig from '../../../../config.json'; +import { mongoConnectionString } from '../mongoClient'; import { ObjectId } from 'mongodb'; import from from 'from'; @@ -197,10 +198,13 @@ async function postcheck(data, feed) { const processEzsEnrichment = (entries, commands, ctx, preview = false) => { return new Promise((resolve, reject) => { + const environment = { + connectionStringURI: mongoConnectionString(ctx.tenant), + }; const values = []; from(entries) .pipe(ezs(preformat)) - .pipe(ezs('delegate', { commands }, {})) + .pipe(ezs('delegate', { commands }, environment)) .pipe(ezs(postcheck, { preview }, ctx)) .on('data', (data) => { if (data instanceof Error) {