Add bulk1 functions in salesforce

packages/salesforce/src/bulk1.js

@@ -0,0 +1,248 @@
+import { connection } from './Adaptor';
+import { composeNextState } from '@openfn/language-common';
+import { expandReferences } from '@openfn/language-common/util';
+
+/**
+ * State object for Salesforce Bulk API 1.0 operations
+ * @typedef {Object} Bulk1QueryState
+ * @property {Array} data - The processed records or query results from the Bulk API operation
+ * @property {Array} references - Array of previous state data objects used in the job
+ * @private
+ */
+
+/**
+ * @typedef {Object} Bulk1LoadState - State object for bulk load operations
+ * @property {Object} data - The processed records or results from the bulk load operation
+ * @property {Object} data.successfulResults - Array of successful results
+ * @property {Object} data.failedResults - Array of failed results
+ * @property {Object} data.unprocessedRecords - Array of unprocessed records
+ * @property {Array} references - Array of previous state data objects used in the job
+ * @private
+ */
+
+/**
+ * Options for configuring Salesforce Bulk API 1.0 operations
+ * @typedef {Object} Bulk1Options
+ * @public
+ * @property {string} [extIdField] - External id field. Required for upsert operations
+ * @property {boolean} [allowNoOp=false] - Skipping bulk operation if no records
+ * @property {boolean} [failOnError=false] - Fail the operation on error
+ * @property {number} [pollTimeout=240000] - Polling timeout in milliseconds. Default: 240000 (4 minutes)
+ * @property {number} [pollInterval=6000] - Polling interval in milliseconds. Default: 6000 (6 seconds)
+ * @property {boolean} [concurrencyMode='Parallel'] - Concurrency mode: 'Parallel' or 'Serial'
+ */
+
+/**
+ * Executes a bulk query using Salesforce Bulk API 1.0
+ * @public
+ * @example <caption>Query records</caption>
+ * bulk1.query('SELECT Id, Name FROM Account');
+ * @example <caption>Query with custom options</caption>
+ * bulk1.query('SELECT Id, Name FROM Account', {
+ *   pollInterval: 1000,
+ *   pollTimeout: 30000
+ * });
+ * @function
+ * @param {string} query - SOQL query string to execute
+ * @state {Bulk1QueryState}
+ * @returns {Operation}
+ */
+export function query(query) {
+  return async state => {
+    const [resolvedQuery] = expandReferences(state, query);
+
+    console.log(`Executing bulk query: ${resolvedQuery}`);
+
+    try {
+      const stream = await connection.bulk.query(resolvedQuery);
+      const records = await new Promise((resolve, reject) => {
+        const recs = [];
+        stream
+          .on('record', rec => recs.push(rec))
+          .on('error', reject)
+          .on('finish', () => resolve(recs));
+      });
+
+      return composeNextState(state, records);
+    } catch (error) {
+      console.error('Bulk query error:', error);
+      throw error;
+    }
+  };
+}
+
+/**
+ * Helper function for bulk load operations (insert, update, upsert, delete)
+ * @private
+ * @param {string} operation - The bulk operation to perform
+ * @param {string} sObject - The Salesforce object type
+ * @param {Array} records - Array of records to process
+ * @param {Bulk1Options} options - Bulk operation options
+ * @returns {Operation}
+ */
+const DEFAULT_POLL_INTERVAL = 6e3; // 6 seconds
+const DEFAULT_POLL_TIMEOUT = 3e4; // 30 seconds
+function bulk1Load(operation, sObject, records, options = {}) {
+  return async state => {
+    const [resolvedSObject, resolvedRecords, resolvedOptions] =
+      expandReferences(state, sObject, records, options);
+    const {
+      extIdField,
+      failOnError = false,
+      concurrencyMode = 'Parallel',
+      pollInterval = DEFAULT_POLL_INTERVAL,
+      pollTimeout = DEFAULT_POLL_TIMEOUT,
+    } = resolvedOptions;
+
+    connection.bulk.pollTimeout = pollTimeout;
+    connection.bulk.pollInterval = pollInterval;
+    if (!resolvedRecords || resolvedRecords.length === 0) {
+      if (resolvedOptions.allowNoOp) {
+        console.log(
+          `No items in ${resolvedSObject} array. Skipping bulk ${operation} operation.`
+        );
+        return composeNextState(state, { message: 'No records to process' });
+      }
+      throw new Error(`No records provided for bulk ${operation} operation`);
+    }
+
+    console.log(
+      `Creating bulk ${operation} job for ${resolvedSObject} with ${resolvedRecords.length} records`
+    );
+
+    try {
+      const opts = {
+        concurrencyMode,
+        ...(operation === 'upsert' && { extIdField }),
+      };
+
+      const results = await connection.bulk.load(
+        resolvedSObject,
+        operation,
+        opts,
+        resolvedRecords
+      );
+
+      const successfulResults = results.filter(r => r.success);
+      const failedResults = results.filter(r => !r.success);
+      const unprocessedRecords = failedResults.map((r, i) => ({
+        ...resolvedRecords[i],
+        error: r.errors ? r.errors.join(', ') : 'Unknown error',
+      }));
+
+      const resultData = {
+        successfulResults,
+        failedResults,
+        unprocessedRecords,
+        totalProcessed: results.length,
+        successCount: successfulResults.length,
+        failureCount: failedResults.length,
+      };
+
+      if (failOnError && failedResults.length > 0) {
+        throw new Error(
+          `Bulk ${operation} failed with ${
+            failedResults.length
+          } errors: ${JSON.stringify(failedResults)}`
+        );
+      }
+
+      return composeNextState(state, resultData);
+    } catch (error) {
+      console.error(`Bulk ${operation} error:`, error);
+      throw error;
+    }
+  };
+}
+
+/**
+ * Bulk inserts records using Salesforce Bulk API 1.0
+ * @public
+ * @example <caption>Insert multiple records</caption>
+ * bulk1.insert('Account', [{ Name: 'Coco' }, { Name: 'Melon' }]);
+ * @example <caption>Insert with custom options</caption>
+ * bulk1.insert('Account', [{ Name: 'Coco' }, { Name: 'Melon' }], {
+ *   pollInterval: 1000,
+ *   pollTimeout: 30000,
+ *   failOnError: true
+ * });
+ * @function
+ * @param {string} sObject - The Salesforce object type
+ * @param {Array} records - Array of records to insert
+ * @param {Bulk1Options} [options] - Bulk insert options
+ * @state {Bulk1LoadState}
+ * @returns {Operation}
+ */
+export function insert(sObject, records, options = {}) {
+  return bulk1Load('insert', sObject, records, options);
+}
+
+/**
+ * Bulk updates records using Salesforce Bulk API 1.0
+ * @public
+ * @example <caption>Update multiple records</caption>
+ * bulk1.update('Account', [{ Id: '001xx', Name: 'Updated Name' }]);
+ * @example <caption>Update with custom options</caption>
+ * bulk1.update('Account', [{ Id: '001xx', Name: 'Updated Name' }], {
+ *   pollInterval: 1000,
+ *   pollTimeout: 30000,
+ *   failOnError: true
+ * });
+ * @function
+ * @param {string} sObject - The Salesforce object type
+ * @param {Array} records - Array of records to update
+ * @param {Bulk1Options} [options] - Bulk update options
+ * @state {Bulk1LoadState}
+ * @returns {Operation}
+ */
+export function update(sObject, records, options = {}) {
+  return bulk1Load('update', sObject, records, options);
+}
+
+/**
+ * Bulk upserts records using Salesforce Bulk API 1.0
+ * @public
+ * @example <caption>Upsert multiple records</caption>
+ * bulk1.upsert('Account', [{ External_Id__c: 'EXT001', Name: 'Upserted Name' }], { extIdField: 'External_Id__c' });
+ * @example <caption>Upsert with custom options</caption>
+ * bulk1.upsert('Account', [{ External_Id__c: 'EXT001', Name: 'Upserted Name' }], {
+ *   extIdField: 'External_Id__c',
+ *   pollInterval: 1000,
+ *   pollTimeout: 30000,
+ *   failOnError: true
+ * });
+ * @function
+ * @param {string} sObject - The Salesforce object type
+ * @param {Array} records - Array of records to upsert
+ * @param {Bulk1Options} options - Bulk upsert options (extIdField is required)
+ * @state {Bulk1LoadState}
+ * @returns {Operation}
+ */
+export function upsert(sObject, records, options = {}) {
+  if (!options.extIdField) {
+    throw new Error('extIdField is required for upsert operations');
+  }
+  return bulk1Load('upsert', sObject, records, options);
+}
+
+/**
+ * Bulk deletes records using Salesforce Bulk API 1.0
+ * @public
+ * @example <caption>Delete multiple records</caption>
+ * bulk1.delete('Account', [{ Id: '001xx' }]);
+ * @example <caption>Delete with custom options</caption>
+ * bulk1.delete('Account', [{ Id: '001xx' }], {
+ *   pollInterval: 1000,
+ *   pollTimeout: 30000,
+ *   failOnError: true
+ * });
+ * @function
+ * @param {string} sObject - The Salesforce object type
+ * @param {Array} records - Array of records to delete (must include Id field)
+ * @param {Bulk1Options} [options] - Bulk delete options
+ * @state {Bulk1LoadState}
+ * @returns {Operation}
+ */
+export function destroy(sObject, records, options = {}) {
+  return bulk1Load('delete', sObject, records, options);
+}

packages/salesforce/src/index.js

@@ -7,4 +7,5 @@ export * from './Adaptor';
 
 export * as util from './util';
 export * as bulk2 from './bulk2';
+export * as bulk1 from './bulk1';
 export * as http from './http';