salesforce Add support for asyncMode in bulk operation

.changeset/moody-crabs-fly.md

@@ -0,0 +1,5 @@
+---
+'@openfn/language-salesforce': minor
+---
+
+Add `asyncMode` option in `bulk()` function

packages/salesforce/ast.json

@@ -84,6 +84,10 @@
             "title": "state",
             "description": "{SalesforceResultState}"
           },
+          {
+            "title": "state",
+            "description": "data.batches - Array of batch Infomation"
+          },
           {
             "title": "returns",
             "description": null,

packages/salesforce/src/Adaptor.js

@@ -66,6 +66,7 @@ import flatten from 'lodash/flatten';
  * @property {boolean} [failOnError=false] - Fail the operation on error.
  * @property {integer} [pollTimeout=240000] - Polling timeout in milliseconds.
  * @property {integer} [pollInterval=6000] - Polling interval in milliseconds.
+ * @property {boolean} [asynMode=false] - Use asyn mode.
  */
 
 /**
@@ -159,6 +160,7 @@ export function execute(...operations) {
  * @param {array} records - an array of records, or a function which returns an array.
  * @param {BulkOptions} [options] - Options to configure the request. In addition to these, you can pass any of the options supported by the {@link https://bit.ly/41tyvVU jsforce API}.
  * @state {SalesforceResultState}
+ * @state data.batches - Array of batch Infomation
  * @returns {Operation}
  */
 export function bulk(sObjectName, operation, records, options = {}) {
@@ -177,6 +179,7 @@ export function bulk(sObjectName, operation, records, options = {}) {
       allowNoOp = false,
       pollTimeout = 240000,
       pollInterval = 6000,
+      asyncMode = false,
     } = resolvedOptions;
 
     const flatRecords = util.removeNestings(resolvedRecords);
@@ -220,36 +223,54 @@ export function bulk(sObjectName, operation, records, options = {}) {
               reject(err);
             });
 
-            return batch
-              .on('queue', function (batchInfo) {
-                const batchId = batchInfo.id;
-                var batch = job.batch(batchId);
-                batch.poll(pollInterval, pollTimeout);
-              })
-              .then(async res => {
+            if (asyncMode) {
+              return batch.on('queue', async function (batchInfo) {
+                console.info('Batch queued. Batch ID:', batchInfo.id);
+                resolve({ batchInfo });
                 await job.close();
-                const errors = res
-                  .map((r, i) => ({ ...r, position: i + 1 }))
-                  .filter(item => {
-                    return !item.success;
+              });
+            } else {
+              return batch
+                .on('queue', function (batchInfo) {
+                  const batchId = batchInfo.id;
+                  var batch = job.batch(batchId);
+                  batch.poll(pollInterval, pollTimeout);
+                })
+                .then(async res => {
+                  await job.close();
+                  const errors = res
+                    .map((r, i) => ({ ...r, position: i + 1 }))
+                    .filter(item => {
+                      return !item.success;
+                    });
+
+                  errors.forEach(err => {
+                    err[`${resolvedOptions.extIdField}`] =
+                      chunkedBatch[err.position - 1][
+                        resolvedOptions.extIdField
+                      ];
                   });
 
-                errors.forEach(err => {
-                  err[`${resolvedOptions.extIdField}`] =
-                    chunkedBatch[err.position - 1][resolvedOptions.extIdField];
+                  if (failOnError && errors.length > 0) {
+                    console.error('Errors detected:');
+                    reject(JSON.stringify(errors, null, 2));
+                  } else {
+                    console.log('Result : ' + JSON.stringify(res, null, 2));
+                    resolve(res);
+                  }
                 });
-
-                if (failOnError && errors.length > 0) {
-                  console.error('Errors detected:');
-                  reject(JSON.stringify(errors, null, 2));
-                } else {
-                  console.log('Result : ' + JSON.stringify(res, null, 2));
-                  resolve(res);
-                }
-              });
+            }
           })
       )
     ).then(results => {
+      // Handle batch results
+      if (asyncMode && results.some(obj => 'batchInfo' in obj)) {
+        return composeNextState(state, {
+          success: true,
+          batches: results.map(r => r.batchInfo).filter(Boolean),
+          errors: [],
+        });
+      }
       const allResults = util.formatResults(results.flat());
       console.log('Merging results arrays.');
       return composeNextState(state, allResults);

packages/salesforce/test/integration.js

@@ -17,6 +17,14 @@ describe('Integration tests', () => {
     before(async () => {
       state.data = [{ name: 'Coco', vera__Active__c: 'No' }];
     });
+    it('should return batchInfo if asyncMode: true', async () => {
+      const { data } = await execute(
+        bulk('Account', 'insert', state => state.data, { asyncMode: true })
+      )(state);
+
+      expect(data.batches.length).to.eql(1);
+      expect(data.success).to.eql(true);
+    }).timeout(10000);
     it('should create multiple sobjects', async () => {
       const { data } = await execute(
         bulk('Account', 'insert', state => state.data)