59 add selected fields (#60)

Fixed issue with 431 and 414 errors in Get Updated Objects Polling trigger: new configuration field Selected Fields added

Author: shulkaolka

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 2.5.1 (January 31, 2023)
+* Fixed [issue](https://github.com/elasticio/salesforce-component-v2/issues/59) with 431 and 414 errors in `Get Updated Objects Polling` trigger: new configuration field `Selected Fields` added
+
 ## 2.5.0 (January 13, 2023)
 * Fixed issue with attachments in `Bulk Create/Update/Delete/Upsert` action
 * Added ability to directly provide url to csv file in `Bulk Create/Update/Delete/Upsert` action

README.md

@@ -38,13 +38,13 @@
 The component uses Salesforce - API Version 46.0 by defaults but can be overwritten by the environment variable `SALESFORCE_API_VERSION`
 
 ### Environment variables
-Name|Mandatory|Description|Values|
-|----|---------|-----------|------|
-|SALESFORCE_API_VERSION| false | Determines API version of Salesforce to use | Default: `46.0` |
-|REFRESH_TOKEN_RETRIES| false | Determines how many retries to refresh token should be done before throwing an error | Default: `10` |
-|HASH_LIMIT_TIME| false | Hash expiration time in ms  | Default: `600000` |
-|HASH_LIMIT_ELEMENTS| false | Hash size number limit  | Default: `10` |
-|UPSERT_TIME_OUT| false | Time out for `Upsert Object` action in ms | Default: `120000` (2min) |
+| Name                   | Mandatory | Description                                                                          | Values                   |
+|------------------------|-----------|--------------------------------------------------------------------------------------|--------------------------|
+| SALESFORCE_API_VERSION | false     | Determines API version of Salesforce to use                                          | Default: `46.0`          |
+| REFRESH_TOKEN_RETRIES  | false     | Determines how many retries to refresh token should be done before throwing an error | Default: `10`            |
+| HASH_LIMIT_TIME        | false     | Hash expiration time in ms                                                           | Default: `600000`        |
+| HASH_LIMIT_ELEMENTS    | false     | Hash size number limit                                                               | Default: `10`            |
+| UPSERT_TIME_OUT        | false     | Time out for `Upsert Object` action in ms                                            | Default: `120000` (2min) |
 
 ## Credentials
 Authentication occurs via OAuth 2.0.
@@ -56,13 +56,13 @@ During credentials creation you would need to:
 - select existing Auth Client from drop-down list ``Choose Auth Client`` or create the new one. 
 For creating Auth Client you should specify following fields:
 
-Field name|Mandatory|Description|
-|----|---------|-----------|
-|Name| true | your Auth Client's name |
-|Client ID| true | your OAuth client key |
-|Client Secret| true | your OAuth client secret |
-|Authorization Endpoint| true | your OAuth authorization endpoint. For production use `https://login.salesforce.com/services/oauth2/authorize`, for sandbox - `https://test.salesforce.com/services/oauth2/authorize`|
-|Token Endpoint| true | your OAuth Token endpoint for refreshing access token. For production use `https://login.salesforce.com/services/oauth2/token`, for sandbox - `https://test.salesforce.com/services/oauth2/token`|
+| Field name             | Mandatory | Description                                                                                                                                                                                       |
+|------------------------|-----------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| Name                   | true      | your Auth Client's name                                                                                                                                                                           |
+| Client ID              | true      | your OAuth client key                                                                                                                                                                             |
+| Client Secret          | true      | your OAuth client secret                                                                                                                                                                          |
+| Authorization Endpoint | true      | your OAuth authorization endpoint. For production use `https://login.salesforce.com/services/oauth2/authorize`, for sandbox - `https://test.salesforce.com/services/oauth2/authorize`             |
+| Token Endpoint         | true      | your OAuth Token endpoint for refreshing access token. For production use `https://login.salesforce.com/services/oauth2/token`, for sandbox - `https://test.salesforce.com/services/oauth2/token` |
 
 - fill field ``Name Your Credential``
 - click on ``Authenticate`` button - if you have not logged in Salesforce before then log in by entering data in the login window that appears
@@ -74,6 +74,7 @@ Field name|Mandatory|Description|
 ### Get Updated Objects Polling
 ### Config Fields
  * **Object Type** Dropdown: Indicates Object Type to be fetched
+ * **Selected Fields** Multiselect dropdown: list with all Object Fields. Select fields, which will be returned in response. That can prevent [431 and 414 Errors](https://developer.salesforce.com/docs/atlas.en-us.salesforce_app_limits_cheatsheet.meta/salesforce_app_limits_cheatsheet/salesforce_app_limits_platform_api.htm).
  * **Include linked objects** Multiselect dropdown: list with all the related child and parent objects of the selected object type. List entries are given as `Object Name/Reference To (Relationship Name)`. Select one or more related objects, which will be join queried and included in the response from your Salesforce Organization. Please see the **Limitations** section below for use case advisories.
  * **Emit behavior** Dropdown: Indicates emit objects individually or emit by page
  * **Start Time** - TextField (string, optional): Indicates the beginning time to start retrieving events from in ISO 8601 Date time utc format - YYYY-MM-DDThh:mm:ssZ

component.json

@@ -3,7 +3,7 @@
   "description": "Customer relationship management (CRM) software & cloud computing from the leader in CRM solutions for businesses large & small.",
   "docsUrl": "https://github.com/elasticio/salesforce-component-v2",
   "url": "http://www.salesforce.com/",
-  "version": "2.5.0",
+  "version": "2.5.1",
   "authClientTypes": [
     "oauth2"
   ],
@@ -48,6 +48,19 @@
           "model": "objectTypes",
           "prompt": "Please select a Salesforce Object"
         },
+        "selectedFields": {
+          "label": "Selected Fields",
+          "viewClass": "MultiSelectView",
+          "order": 17,
+          "required": false,
+          "require": [
+            "sobject"
+          ],
+          "help": {
+            "description": "Select fields that will be returned"
+          },
+          "model": "getObjectFields"
+        },
         "linkedObjects": {
           "label": "Include linked objects",
           "viewClass": "MultiSelectView",

lib/triggers/getUpdatedObjectsPolling.js

@@ -9,15 +9,28 @@ const isDateValid = (date) => new Date(date).toString() !== 'Invalid Date';
 const isNumberNaN = (num) => Number(num).toString() === 'NaN';
 const MAX_FETCH = 10000;
 
+function getSelectedFields(cfg) {
+  const { selectedFields = [] } = cfg;
+  if (selectedFields.length === 0) {
+    return '*';
+  }
+  if (!selectedFields.includes('LastModifiedDate')) {
+    selectedFields.push('LastModifiedDate');
+  }
+  return selectedFields.toString();
+}
+
 exports.process = async function processTrigger(_msg, cfg, snapshot) {
   this.logger.info('Start processing "Get Updated Objects Polling" trigger');
   const currentTime = new Date();
 
   const {
     sobject,
-    pageSize = MAX_FETCH,
     linkedObjects = [],
+    pageSize = MAX_FETCH,
     emitBehavior = 'emitIndividually',
+  } = cfg;
+  let {
     singlePagePerInterval,
   } = cfg;
 
@@ -39,12 +52,20 @@ exports.process = async function processTrigger(_msg, cfg, snapshot) {
 
   this.logger.info(`Filter "${sobject}" updated from ${timeToString(from)} to ${timeToString(to)}, page limit ${pageSize}`);
 
+  const selectedFields = getSelectedFields(cfg);
   const options = {
     sobject,
-    selectedObjects: linkedObjects.reduce((query, obj) => (obj.startsWith('!') ? query : `${query}, ${obj}.*`), '*'),
+    selectedObjects: linkedObjects.reduce((query, obj) => (obj.startsWith('!') ? query : `${query}, ${obj}.*`), selectedFields),
     linkedObjects,
   };
 
+  const isDebugFlow = process.env.ELASTICIO_FLOW_TYPE === 'debug';
+  if (isDebugFlow) {
+    options.maxFetch = 10;
+    singlePagePerInterval = true;
+    this.logger.info('Debug flow detected, set maxFetch to 10');
+  }
+
   let proceed = true;
   let hasNextPage = false;
   let iteration = 1;
@@ -53,53 +74,69 @@ exports.process = async function processTrigger(_msg, cfg, snapshot) {
     do {
       if (!hasNextPage) {
         options.whereCondition = `LastModifiedDate >= ${timeToString(from)} AND LastModifiedDate < ${timeToString(to)}`;
+        this.logger.debug('Start poll object with options: %j', options);
         results = await callJSForceMethod.call(this, cfg, 'pollingSelectQuery', options);
         this.logger.info(`Polling iteration ${iteration} - ${results.length} results found`);
         iteration++;
       }
       if (results.length !== 0) {
+        this.logger.debug('New records found, check other options...');
         nextStartTime = currentTime;
         if (singlePagePerInterval && snapshot.lastElementId) {
+          this.logger.debug('Snapshot contain lastElementId, going to delete records that have already been emitted');
           const lastElement = results.filter((item) => item.Id === snapshot.lastElementId)[0];
           const lastElementIndex = results.indexOf(lastElement);
           results = results.slice(lastElementIndex + 1);
+          this.logger.debug('Emitted records deleted. Current results length is %s', results.length);
         }
         if (results.length === MAX_FETCH) {
+          this.logger.debug('The size of the resulting array is equal to MAX_FETCH, so all entries that have the same LastModifiedDate as the last entry will be deleted from the resulting array to prevent emitting duplicates');
           nextStartTime = results[results.length - 1].LastModifiedDate;
           const filteredResults = results.filter((item) => item.LastModifiedDate === nextStartTime);
           results = results.slice(0, MAX_FETCH - filteredResults.length);
+          this.logger.debug('Entries that have the same LastModifiedDate as the last entry deleted. Current size of the resulting array is %s', results.length);
         }
         if (results.length >= Number(pageSize)) {
+          this.logger.debug('The size of the resulting array >= pageSize. Going to process one page...');
           hasNextPage = true;
           const pageResults = results.slice(0, pageSize);
           results = results.slice(pageSize);
           const lastElementLastModifiedDate = pageResults[pageSize - 1].LastModifiedDate;
           const lastElementId = pageResults[pageSize - 1].Id;
           if (emitBehavior === 'fetchPage') {
+            this.logger.debug('Emit Behavior set as Fetch Page, going to emit one page...');
             await this.emit('data', messages.newMessageWithBody({ results: pageResults }));
           } else if (emitBehavior === 'emitIndividually') {
+            this.logger.debug('Emit Behavior set as Emit Individually, going to emit records one by one');
             for (const record of pageResults) {
               await this.emit('data', messages.newMessageWithBody(record));
             }
           }
+          this.logger.debug('Page processing is finished.');
 
           if (singlePagePerInterval) {
+            this.logger.debug('Single Page Per Interval option is set. Going to emit snapshot %j', { nextStartTime: lastElementLastModifiedDate, lastElementId });
             await this.emit('snapshot', { nextStartTime: lastElementLastModifiedDate, lastElementId });
             proceed = false;
           }
         } else {
+          this.logger.debug('The size of the resulting array < pageSize. Going to process all found results');
           hasNextPage = false;
           if (emitBehavior === 'fetchPage') {
+            this.logger.debug('Emit Behavior set as Fetch Page, going to emit one page...');
             await this.emit('data', messages.newMessageWithBody({ results }));
           } else if (emitBehavior === 'emitIndividually') {
+            this.logger.debug('Emit Behavior set as Emit Individually, going to emit records one by one');
             for (const record of results) {
               await this.emit('data', messages.newMessageWithBody(record));
             }
           }
+          this.logger.debug('All results processed. going to emit snapshot: %j', { nextStartTime });
           await this.emit('snapshot', { nextStartTime });
           proceed = false;
         }
       } else {
+        this.logger.debug('The size of the resulting array is 0, going to emit snapshot: %j', { nextStartTime });
         await this.emit('snapshot', { nextStartTime });
         await this.emit('end');
         proceed = false;
@@ -118,6 +155,13 @@ module.exports.objectTypes = async function getObjectTypes(configuration) {
   return callJSForceMethod.call(this, configuration, 'getObjectTypes');
 };
 
+module.exports.getObjectFields = async function getObjectFields(configuration) {
+  const fields = await callJSForceMethod.call(this, configuration, 'getObjectFieldsMetaData');
+  const result = {};
+  fields.forEach((field) => { result[field.name] = field.label; });
+  return result;
+};
+
 module.exports.linkedObjectTypes = async function linkedObjectTypes(configuration) {
   const meta = await callJSForceMethod.call(this, configuration, 'describe');
   return getLinkedObjectTypes(meta);
@@ -126,7 +170,16 @@ module.exports.linkedObjectTypes = async function linkedObjectTypes(configuratio
 module.exports.getMetaModel = async function getMetaModel(configuration) {
   const describeObject = await callJSForceMethod.call(this, configuration, 'describe');
   const metadata = await processMeta(describeObject, 'polling');
-  const { outputMethod } = configuration;
+  const { outputMethod, selectedFields = [] } = configuration;
+  if (selectedFields.length > 0) {
+    const filteredOutMetadata = {};
+    for (const outProperty of Object.keys(metadata.out.properties)) {
+      if (selectedFields.includes(outProperty)) {
+        filteredOutMetadata[outProperty] = metadata.out.properties[outProperty];
+      }
+    }
+    metadata.out.properties = filteredOutMetadata;
+  }
   if (outputMethod !== 'emitIndividually') {
     const outputArray = {
       type: 'object',

spec-integration/triggers/polling.spec.js

@@ -1,4 +1,5 @@
 /* eslint-disable no-return-assign */
+process.env.ELASTICIO_FLOW_TYPE = 'debug';
 const { expect } = require('chai');
 const polling = require('../../lib/triggers/getUpdatedObjectsPolling');
 const { getContext, testsCommon } = require('../../spec/common');
@@ -40,10 +41,46 @@ describe('polling', () => {
     await polling.process.call(getContext(), msg, testCfg, snapshot);
   });
 
+  it('Custom Object 300+ custom fields polling', async () => {
+    const testCfg = {
+      oauth: getOauth(),
+      sobject: 'TestBook__c',
+      emitBehavior: 'fetchPage',
+      pageSize: 10,
+      linkedObjects: ['CreatedBy'],
+      selectedFields: ['Book_Description__c', 'Test_Book_Custom_Field_1__c'],
+    };
+    const snapshot = {};
+    const msg = { body: {} };
+    const context = getContext();
+    await polling.process.call(context, msg, testCfg, snapshot);
+    expect(context.emit.firstCall.args[0]).to.deep.equal('data');
+    expect(context.emit.firstCall.args[1].body.results[0].Book_Description__c).to.deep.equal('First book');
+  });
+
   it('Get metadata', async () => {
-    const testCfg = { oauth: getOauth(), sobject: 'Contact', outputMethod: 'emitAll' };
+    const testCfg = {
+      oauth: getOauth(),
+      sobject: 'Contact',
+      outputMethod: 'emitAll',
+      selectedFields: ['Id', 'LastName'],
+    };
 
     const result = await polling.getMetaModel.call(getContext(), testCfg);
     expect(result).to.be.equal(true);
   });
+
+  it('Get object Fields', async () => {
+    const testCfg = { oauth: getOauth(), sobject: 'TestBook__c' };
+
+    const result = await polling.getObjectFields.call(getContext(), testCfg);
+    expect(result.Id).to.be.equal('Record ID');
+  });
+
+  it('Get linkedObjectTypes', async () => {
+    const testCfg = { oauth: getOauth(), sobject: 'TestBook__c' };
+
+    const result = await polling.linkedObjectTypes.call(getContext(), testCfg);
+    expect(result.CreatedBy).to.be.equal('User (CreatedBy)');
+  });
 });

spec/triggers/getUpdatedObjectsPolling.spec.js

@@ -31,6 +31,27 @@ describe('getUpdatedObjectsPolling trigger', () => {
       expect(context.emit.getCall(0).lastArg.body).to.deep.equal(duplicateRecords[0]);
     });
 
+    it('emitIndividually with selected fields', async () => {
+      execRequest = sinon.stub(callJSForceMethod, 'call').returns(duplicateRecords);
+      const cfg = {
+        sobject: 'Document',
+        pageSize: 10,
+        linkedObjects: [],
+        selectedFields: ['Id', 'Name'],
+        emitBehavior: 'emitIndividually',
+        singlePagePerInterval: false,
+      };
+      const snapshot = {};
+      const msg = {};
+      const context = getContext();
+      await process.call(context, msg, cfg, snapshot);
+      expect(execRequest.firstCall.args[3].selectedObjects).to.be.equal('Id,Name,LastModifiedDate');
+      expect(context.emit.callCount).to.be.equal(28);
+      expect(context.emit.getCall(27).firstArg).to.be.equal('snapshot');
+      expect(execRequest.callCount).to.be.equal(1);
+      expect(context.emit.getCall(0).lastArg.body).to.deep.equal(duplicateRecords[0]);
+    });
+
     it('fetchPage sizePage = 10, singlePagePerInterval = false', async () => {
       execRequest = sinon.stub(callJSForceMethod, 'call').returns(duplicateRecords);
       const cfg = {