move upload strategy to settings

Author: varadarajan-tw

packages/destination-actions/src/destinations/sftp/__tests__/client.test.ts

@@ -1,6 +1,6 @@
 import { DEFAULT_REQUEST_TIMEOUT } from '@segment/actions-core'
 import { normalizeSSHKey, testSFTPConnection, uploadSFTP } from '../client'
-import { SFTP_DEFAULT_PORT } from '../constants'
+import { SFTP_DEFAULT_PORT, UploadStrategy } from '../constants'
 import { Settings } from '../generated-types'
 import { SFTPWrapper } from '../sftp-wrapper'
 
@@ -77,8 +77,13 @@ describe('SFTP Client', () => {
       expect(mockSftpInstance.end).toHaveBeenCalled()
     })
 
-    it('uploads using fastPutFromBuffer when useConcurrency is true', async () => {
-      await uploadSFTP(sshKeySettings, 'sftp_folder_path', 'filename', Buffer.from('test content'), true)
+    it('uploads using fastPutFromBuffer when upload strategy is concurrent', async () => {
+      await uploadSFTP(
+        { ...sshKeySettings, uploadStrategy: UploadStrategy.CONCURRENT },
+        'sftp_folder_path',
+        'filename',
+        Buffer.from('test content')
+      )
 
       expect(SFTPWrapper).toHaveBeenCalled()
       expect(mockSftpInstance.connect).toHaveBeenCalledWith({

packages/destination-actions/src/destinations/sftp/__tests__/functions.test.ts

@@ -1,6 +1,6 @@
 import { MultiStatusResponse } from '@segment/actions-core'
 import { uploadSFTP } from '../client'
-import { SFTP_DEFAULT_PORT } from '../constants'
+import { SFTP_DEFAULT_PORT, UploadStrategy } from '../constants'
 import {
   clean,
   createFilename,
@@ -416,7 +416,6 @@ describe('send', () => {
         enable_batching: true,
         file_extension: 'csv',
         batch_size: 100000,
-        useConcurrentWrites: true,
         columns: {
           email: '[email protected]',
           name: 'John Doe'
@@ -425,15 +424,15 @@ describe('send', () => {
     ]
 
     const signal = AbortSignal.timeout(0)
+    const settings = { ...mockSettings, uploadStrategy: UploadStrategy.CONCURRENT }
 
-    await send(payloads, mockSettings, mockRawMapping, undefined, signal)
+    await send(payloads, settings, mockRawMapping, undefined, signal)
 
     expect(mockUploadSFTP).toHaveBeenCalledWith(
-      mockSettings,
+      settings,
       '/uploads',
       expect.any(String), // filename
       expect.any(Buffer), // file content,
-      true, // useConcurrentWrites
       undefined, // logger
       signal
     )

packages/destination-actions/src/destinations/sftp/client.ts

@@ -7,7 +7,7 @@ import {
 } from '@segment/actions-core'
 import path from 'path'
 import Client from 'ssh2-sftp-client'
-import { SFTP_DEFAULT_PORT } from './constants'
+import { SFTP_DEFAULT_PORT, UploadStrategy } from './constants'
 import { Settings } from './generated-types'
 import { sftpConnectionConfig } from './types'
 import { SFTPWrapper } from './sftp-wrapper'
@@ -27,18 +27,21 @@ interface SFTPError extends Error {
  * @param sftpFolderPath - The target folder path on the SFTP server.
  * @param filename - The name of the file to upload.
  * @param fileContent - The content of the file to upload as a Buffer.
+ * @param uploadStrategy - The upload strategy to use (standard or concurrent).
+ * @param logger - Optional logger for logging messages.
+ * @param signal - Optional AbortSignal to handle request cancellation.
  * @returns A promise that resolves when the file is successfully uploaded.
  */
 async function uploadSFTP(
   settings: Settings,
   sftpFolderPath: string,
   filename: string,
   fileContent: Buffer,
-  useConcurrentWrites?: boolean,
   logger?: Logger,
   signal?: AbortSignal
 ) {
   const sftp = new SFTPWrapper('uploadSFTP', logger)
+
   signal?.throwIfAborted() // exit early if already aborted
   // Set up abort listener to clean up SFTP connection on abort
   const abortListener = () => {
@@ -48,10 +51,11 @@ async function uploadSFTP(
     throw new RequestTimeoutError()
   }
   signal?.addEventListener('abort', abortListener, { once: true })
+
   try {
     await sftp.connect(createConnectionConfig(settings))
     const remoteFilePath = path.posix.join(sftpFolderPath, filename)
-    if (useConcurrentWrites) {
+    if (settings.uploadStrategy === UploadStrategy.CONCURRENT) {
       return await sftp.fastPutFromBuffer(fileContent, remoteFilePath)
     } else {
       return await sftp.put(fileContent, remoteFilePath)

packages/destination-actions/src/destinations/sftp/constants.ts

@@ -1 +1,6 @@
 export const SFTP_DEFAULT_PORT = 22
+
+export const UploadStrategy = {
+  STANDARD: 'standard',
+  CONCURRENT: 'concurrent'
+}

packages/destination-actions/src/destinations/sftp/fields.ts

@@ -245,14 +245,7 @@ const audienceFields: Record<string, InputField> = {
   traits_or_props,
   computation_key
 }
-const useConcurrentWrites: InputField = {
-  label: 'Use Concurrent Writes',
-  description:
-    'Enable concurrent writes when uploading files to SFTP. This can improve upload performance for large files. However, it may not be supported by all SFTP servers. Test with your server for compatibility.',
-  type: 'boolean',
-  required: false,
-  default: false
-}
+
 export const baseFields: Record<string, InputField> = {
   columns: columnsNoDefaultMappings,
   filename_prefix,
@@ -261,8 +254,7 @@ export const baseFields: Record<string, InputField> = {
   sftp_folder_path,
   enable_batching,
   batch_size,
-  batch_size_column_name,
-  useConcurrentWrites
+  batch_size_column_name
 }
 export const commonFields: Record<string, InputField> = {
   ...baseFields,

packages/destination-actions/src/destinations/sftp/functions.ts

@@ -18,8 +18,7 @@ async function send(
     batch_size_column_name,
     filename_prefix,
     file_extension,
-    sftp_folder_path,
-    useConcurrentWrites
+    sftp_folder_path
   } = payloads[0]
 
   const headers: ColumnHeader[] = createHeaders(
@@ -41,7 +40,7 @@ async function send(
 
   const msResponse = new MultiStatusResponse()
   try {
-    await uploadSFTP(settings, sftp_folder_path, filename, fileContent, useConcurrentWrites, logger, signal)
+    await uploadSFTP(settings, sftp_folder_path, filename, fileContent, logger, signal)
     payloads.forEach((payload, index) => {
       const row = rowsObservabilityArray[index] ?? ''
       msResponse.setSuccessResponseAtIndex(index, {

packages/destination-actions/src/destinations/sftp/generated-types.ts

@@ -2,7 +2,7 @@ import type { DestinationDefinition } from '@segment/actions-core'
 import type { Settings } from './generated-types'
 
 import { testSFTPConnection } from './client'
-import { SFTP_DEFAULT_PORT } from './constants'
+import { SFTP_DEFAULT_PORT, UploadStrategy } from './constants'
 import syncEvents from './syncEvents'
 import syncModelToSFTP from './syncModelToSFTP'
 
@@ -89,6 +89,21 @@ const destination: DestinationDefinition<Settings> = {
             }
           ]
         }
+      },
+      uploadStrategy: {
+        label: 'Upload Strategy',
+        description: 'Select the upload strategy to use when uploading files to the SFTP server.',
+        type: 'string',
+        choices: [
+          { label: 'Standard [Supported by all SFTP servers]', value: UploadStrategy.STANDARD },
+          {
+            label:
+              'Concurrent [Improves upload time but may not be supported by all SFTP servers. Test and enable if supported.]',
+            value: UploadStrategy.CONCURRENT
+          }
+        ],
+        default: UploadStrategy.STANDARD,
+        required: false
       }
     },
     testAuthentication: async (_, { settings }) => await testSFTPConnection(settings)

packages/destination-actions/src/destinations/sftp/index.ts

@@ -68,6 +68,7 @@ export class SFTPWrapper {
   ): Promise<void> {
     const fsize = input.length
     return new Promise<void>((resolve, reject) => {
+      // Open the connection to the remote file
       this.client?.open(remoteFilePath, 'w', (err, handle) => {
         if (err) {
           return reject(new Error(`Error opening remote file: ${err.message}`))
@@ -78,6 +79,7 @@ export class SFTPWrapper {
         let position = 0
         let writeRequests: Promise<void>[] = []
 
+        // function that writes a chunk to the remote file
         const writeChunk = (chunkPos: number): Promise<void> => {
           return new Promise((chunkResolve, chunkReject) => {
             const bytesToWrite = Math.min(chunkSize, fsize - chunkPos)
@@ -93,6 +95,7 @@ export class SFTPWrapper {
           })
         }
 
+        // function that splits the writes into concurrent requests and processes them
         const processWrites = async () => {
           while (position < fsize) {
             writeRequests.push(writeChunk(position))
@@ -121,6 +124,7 @@ export class SFTPWrapper {
   }
 
   async end() {
+    // Close the SFTP connection
     return this.sftp.end()
   }
 }