records equal to pageSize in trigger (#52)

if0s · web-flow · commit 64e46b632f08 · 2022-10-07T17:35:36.000+03:00
2.4.1
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 2.4.1 (October 07, 2022)
+* Fixed loop issue when records equal to `Size of Polling Page` and have same `LastModifiedDate` in `Get Updated Objects Polling` trigger
+* Update Sailor version to 2.7.0
+
 ## 2.4.0 (August 30, 2022)
 * New `Get New and Updated Objects Polling` trigger, old one set to deprecated
 * Update Sailor version to 2.6.29
diff --git a/component.json b/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.4.0",
+  "version": "2.4.1",
   "authClientTypes": [
     "oauth2"
   ],
@@ -614,4 +614,4 @@
       }
     }
   }
-}
+}
diff --git a/lib/triggers/getUpdatedObjectsPolling.js b/lib/triggers/getUpdatedObjectsPolling.js
@@ -7,14 +7,15 @@ const timestamp = (date) => new Date(date).getTime();
 const timeToString = (date) => JSON.stringify(new Date(date)).replace(/"/g, '');
 const isDateValid = (date) => new Date(date).toString() !== 'Invalid Date';
 const isNumberNaN = (num) => Number(num).toString() === 'NaN';
+const MAX_FETCH = 10000;
 
 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 = 10000,
+    pageSize = MAX_FETCH,
     linkedObjects = [],
     emitBehavior = 'emitIndividually',
     singlePagePerInterval,
@@ -25,8 +26,8 @@ exports.process = async function processTrigger(_msg, cfg, snapshot) {
   if (!endTime) endTime = currentTime;
   if (!isDateValid(startTime)) throw new Error('invalid "Start Time" date format, use ISO 8601 Date time utc format - YYYY-MM-DDThh:mm:ssZ');
   if (!isDateValid(endTime)) throw new Error('invalid "End Time" date format, use ISO 8601 Date time utc format - YYYY-MM-DDThh:mm:ssZ');
-  if (pageSize > 10000 || isNumberNaN(pageSize) || Number(pageSize) < 0) throw new Error('"Size of Polling Page" must be valid number between 0 and 10000');
-  let from = snapshot?.nextStartTime || startTime;
+  if (pageSize > MAX_FETCH || isNumberNaN(pageSize) || Number(pageSize) < 0) throw new Error(`"Size of Polling Page" must be valid number between 0 and ${MAX_FETCH}`);
+  const from = snapshot?.nextStartTime || startTime;
   const to = endTime || currentTime;
   let nextStartTime = currentTime;
 
@@ -42,45 +43,59 @@ exports.process = async function processTrigger(_msg, cfg, snapshot) {
     sobject,
     selectedObjects: linkedObjects.reduce((query, obj) => (obj.startsWith('!') ? query : `${query}, ${obj}.*`), '*'),
     linkedObjects,
-    maxFetch: pageSize,
   };
 
   let proceed = true;
+  let hasNextPage = false;
   let iteration = 1;
+  let results = [];
   try {
     do {
-      options.whereCondition = `LastModifiedDate >= ${timeToString(from)} AND LastModifiedDate < ${timeToString(to)}`;
-
-      let results = await callJSForceMethod.call(this, cfg, 'pollingSelectQuery', options);
-      this.logger.info(`Polling iteration ${iteration} - ${results.length} results found`);
-      iteration++;
-
+      if (!hasNextPage) {
+        options.whereCondition = `LastModifiedDate >= ${timeToString(from)} AND LastModifiedDate < ${timeToString(to)}`;
+        results = await callJSForceMethod.call(this, cfg, 'pollingSelectQuery', options);
+        this.logger.info(`Polling iteration ${iteration} - ${results.length} results found`);
+        iteration++;
+      }
       if (results.length !== 0) {
-        if (results.length === Number(pageSize)) {
-          const currentResultsLength = results.length;
-          nextStartTime = results[results.length - 1].LastModifiedDate;
-          const filteredResults = results.filter((item) => item.LastModifiedDate !== nextStartTime);
-          if (currentResultsLength - filteredResults.length === 0) {
-            this.logger.warn(`All results from this iteration have same "LastModifiedDate" - ${timeToString(nextStartTime)}, they will be proceed, but all objects in the next iteration will start from date strictly greater than this`);
-            nextStartTime = new Date(new Date(nextStartTime).getTime() + 1);
-          } else {
-            results = filteredResults;
-            this.logger.warn(`Founded ${currentResultsLength} results, which is equal to page limit.. New start time - ${timeToString(nextStartTime)}, ${currentResultsLength - results.length} result(s) excluded from this iteration`);
-          }
-          from = nextStartTime;
-        } else {
-          nextStartTime = currentTime;
-          await this.emit('snapshot', { nextStartTime });
-          proceed = false;
+        nextStartTime = currentTime;
+        if (singlePagePerInterval && snapshot.lastElementId) {
+          const lastElement = results.filter((item) => item.Id === snapshot.lastElementId)[0];
+          const lastElementIndex = results.indexOf(lastElement);
+          results = results.slice(lastElementIndex + 1);
         }
-
-        if (emitBehavior === 'fetchPage') {
-          await this.emit('data', messages.newMessageWithBody({ results }));
-        } else if (emitBehavior === 'emitIndividually') {
-          for (const record of results) { await this.emit('data', messages.newMessageWithBody(record)); }
+        if (results.length === MAX_FETCH) {
+          nextStartTime = results[results.length - 1].LastModifiedDate;
+          const filteredResults = results.filter((item) => item.LastModifiedDate === nextStartTime);
+          results = results.slice(0, MAX_FETCH - filteredResults.length);
         }
+        if (results.length >= Number(pageSize)) {
+          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') {
+            await this.emit('data', messages.newMessageWithBody({ results: pageResults }));
+          } else if (emitBehavior === 'emitIndividually') {
+            for (const record of pageResults) {
+              await this.emit('data', messages.newMessageWithBody(record));
+            }
+          }
 
-        if (singlePagePerInterval) {
+          if (singlePagePerInterval) {
+            await this.emit('snapshot', { nextStartTime: lastElementLastModifiedDate, lastElementId });
+            proceed = false;
+          }
+        } else {
+          hasNextPage = false;
+          if (emitBehavior === 'fetchPage') {
+            await this.emit('data', messages.newMessageWithBody({ results }));
+          } else if (emitBehavior === 'emitIndividually') {
+            for (const record of results) {
+              await this.emit('data', messages.newMessageWithBody(record));
+            }
+          }
           await this.emit('snapshot', { nextStartTime });
           proceed = false;
         }
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -31,7 +31,7 @@
     "dotenv": "8.2.0",
     "elasticio-node": "0.0.9",
     "elasticio-rest-node": "1.2.7",
-    "elasticio-sailor-nodejs": "2.6.29",
+    "elasticio-sailor-nodejs": "2.7.0",
     "jsforce": "1.10.0",
     "mime-types": "2.1.27",
     "nyc": "15.1.0",
diff --git a/spec-integration/triggers/polling.spec.js b/spec-integration/triggers/polling.spec.js
@@ -1,16 +1,27 @@
 /* eslint-disable no-return-assign */
 const { expect } = require('chai');
-const polling = require('../../lib/entry');
+const polling = require('../../lib/triggers/getUpdatedObjectsPolling');
 const { getContext, testsCommon } = require('../../spec/common');
+const enrtyMetaModel = require('../../spec/triggers/outMetadataSchemas/entryObjectTypes.json');
 
 const { getOauth } = testsCommon;
 
 describe('polling', () => {
   it('Account polling with empty snapshot', async () => {
-    const testCfg = { oauth: getOauth(), sobject: 'Contact' };
+    const testCfg = {
+      oauth: getOauth(),
+      sobject: 'Contact',
+      emitBehavior: 'fetchPage',
+      pageSize: 500,
+      singlePagePerInterval: true,
+    };
+    const snapshot = {
+      lastElementId: '0032R00002La0YtQAJ',
+    };
     const msg = { body: {} };
-
-    await polling.process.call(getContext(), msg, testCfg, {});
+    const context = getContext();
+    await polling.process.call(context, msg, testCfg, snapshot);
+    expect(context).to.deep.equal(enrtyMetaModel);
   });
 
   it('Account polling with not empty snapshot', async () => {
diff --git a/spec/testData/trigger.results.1612.json b/spec/testData/trigger.results.1612.json
@@ -0,0 +1,137 @@
+[
+  {
+    "Id": "0034400001uxwXZAAY",
+    "LastModifiedDate": "2021-04-19T04:35:12.000+0000",
+    "name": 1
+  },
+  {
+    "Id": "0034400001uxXuyAAE",
+    "LastModifiedDate": "2021-11-29T16:51:59.000+0000",
+    "name": 2
+  },
+  {
+    "Id": "00344000020qT3KAAU",
+    "LastModifiedDate": "2022-06-10T10:23:15.000+0000",
+    "name": 3
+  },
+  {
+    "Id": "0032R00002La0YtQAJ",
+    "LastModifiedDate": "2022-07-05T09:14:53.000+0000",
+    "name": 4
+  },
+  {
+    "Id": "0032R00002La0ZXQAZ",
+    "LastModifiedDate": "2022-07-05T09:25:14.000+0000",
+    "name": 5
+  },
+  {
+    "Id": "0032R00002La0ZcQAJ",
+    "LastModifiedDate": "2022-07-05T09:25:14.000+0000",
+    "name": 6
+  },
+  {
+    "Id": "0032R00002La0ZhQAJ",
+    "LastModifiedDate": "2022-07-05T09:25:15.000+0000",
+    "name": 7
+  },
+  {
+    "Id": "0032R00002La0ZrQAJ",
+    "LastModifiedDate": "2022-07-05T09:28:36.000+0000",
+    "name": 8
+  },
+  {
+    "Id": "0032R00002La0a1QAB",
+    "LastModifiedDate": "2022-07-05T09:28:37.000+0000",
+    "name": 9
+  },
+  {
+    "Id": "0032R00002La0aaQAB",
+    "LastModifiedDate": "2022-07-05T09:28:37.000+0000",
+    "name": 10
+  },
+  {
+    "Id": "0032R00002La0apQAB",
+    "LastModifiedDate": "2022-07-05T09:28:37.000+0000",
+    "name": 11
+  },
+  {
+    "Id": "0032R00002La0bxQAB",
+    "LastModifiedDate": "2022-07-05T09:28:37.000+0000",
+    "name": 12
+  },
+  {
+    "Id": "0032R00002La0dFQAR",
+    "LastModifiedDate": "2022-07-05T10:05:56.000+0000",
+    "name": 13
+  },
+  {
+    "Id": "0032R00002La0doQAB",
+    "LastModifiedDate": "2022-07-05T10:09:05.000+0000",
+    "name": 14
+  },
+  {
+    "Id": "0032R00002La0dyQAB",
+    "LastModifiedDate": "2022-07-05T10:09:06.000+0000",
+    "name": 15
+  },
+  {
+    "Id": "0032R00002La0ffQAB",
+    "LastModifiedDate": "2022-07-05T10:25:03.000+0000",
+    "name": 16
+  },
+  {
+    "Id": "0032R00002La0fkQAB",
+    "LastModifiedDate": "2022-07-05T10:25:04.000+0000",
+    "name": 17
+  },
+  {
+    "Id": "0032R00002La0fpQAB",
+    "LastModifiedDate": "2022-07-05T10:25:05.000+0000",
+    "name": 18
+  },
+  {
+    "Id": "0032R00002La0gxQAB",
+    "LastModifiedDate": "2022-07-05T10:38:35.000+0000",
+    "name": 19
+  },
+  {
+    "Id": "0032R00002La0h2QAB",
+    "LastModifiedDate": "2022-07-05T10:44:45.000+0000",
+    "name": 20
+  },
+  {
+    "Id": "0032R00002La0hCQAR",
+    "LastModifiedDate": "2022-07-05T10:44:45.000+0000",
+    "name": 21
+  },
+  {
+    "Id": "0032R00002La0hHQAR",
+    "LastModifiedDate": "2022-07-05T10:44:46.000+0000",
+    "name": 22
+  },
+  {
+    "Id": "0032R00002La0hgQAB",
+    "LastModifiedDate": "2022-07-05T10:54:18.000+0000",
+    "name": 23
+  },
+  {
+    "Id": "0032R00002La0hqQAB",
+    "LastModifiedDate": "2022-07-05T10:54:19.000+0000",
+    "name": 24
+  },
+  {
+    "Id": "0032R00002La0itQAB",
+    "LastModifiedDate": "2022-07-05T11:05:29.000+0000",
+    "name": 25
+  },
+  {
+    "Id": "0032R00002La0j3QAB",
+    "LastModifiedDate": "2022-07-05T11:05:30.000+0000",
+    "name": 26
+  },
+  {
+    "Id": "0032R00002La0iyQAB",
+    "LastModifiedDate": "2022-07-05T11:05:30.000+0000",
+    "name": 27
+  }
+]
diff --git a/spec/triggers/getUpdatedObjectsPolling.spec.js b/spec/triggers/getUpdatedObjectsPolling.spec.js

Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,7 @@`
`3`	`3`	`"description": "Customer relationship management (CRM) software & cloud computing from the leader in CRM solutions for businesses large & small.",`
`4`	`4`	`"docsUrl": "https://github.com/elasticio/salesforce-component-v2",`
`5`	`5`	`"url": "http://www.salesforce.com/",`
`6`		`- "version": "2.4.0",`
	`6`	`+ "version": "2.4.1",`
`7`	`7`	`"authClientTypes": [`
`8`	`8`	`"oauth2"`
`9`	`9`	`],`
`@@ -614,4 +614,4 @@`
`614`	`614`	`}`
`615`	`615`	`}`
`616`	`616`	`}`
`617`		`-}`
	`617`	`+}`