fix(data-drains): abortable webhook backoff + S3/webhook optional form fields

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/ee/data-drains/destinations/registry.tsx

@@ -1,7 +1,7 @@
 'use client'
 
 import type { ComponentType } from 'react'
-import { FormField, Input, SecretInput } from '@/components/emcn'
+import { FormField, Input, SecretInput, Switch } from '@/components/emcn'
 import type { CreateDataDrainBody } from '@/lib/api/contracts/data-drains'
 import type { DestinationType } from '@/lib/data-drains/types'
 
@@ -25,6 +25,8 @@ interface S3State {
   bucket: string
   region: string
   prefix: string
+  endpoint: string
+  forcePathStyle: boolean
   accessKeyId: string
   secretAccessKey: string
 }
@@ -35,6 +37,8 @@ const s3FormSpec: DestinationFormSpec<S3State> = {
     bucket: '',
     region: 'us-east-1',
     prefix: '',
+    endpoint: '',
+    forcePathStyle: false,
     accessKeyId: '',
     secretAccessKey: '',
   },
@@ -59,6 +63,19 @@ const s3FormSpec: DestinationFormSpec<S3State> = {
           placeholder='exports/sim'
         />
       </FormField>
+      <FormField label='Endpoint (optional, S3-compatible stores)'>
+        <Input
+          value={state.endpoint}
+          onChange={(e) => setState({ ...state, endpoint: e.target.value })}
+          placeholder='https://s3.example.com'
+        />
+      </FormField>
+      <FormField label='Force path style (MinIO, Ceph)'>
+        <Switch
+          checked={state.forcePathStyle}
+          onCheckedChange={(v) => setState({ ...state, forcePathStyle: v })}
+        />
+      </FormField>
       <FormField label='Access key ID'>
         <SecretInput
           value={state.accessKeyId}
@@ -84,6 +101,8 @@ const s3FormSpec: DestinationFormSpec<S3State> = {
       bucket: s.bucket,
       region: s.region,
       prefix: s.prefix || undefined,
+      endpoint: s.endpoint || undefined,
+      forcePathStyle: s.forcePathStyle || undefined,
     },
     destinationCredentials: {
       accessKeyId: s.accessKeyId,
@@ -94,12 +113,14 @@ const s3FormSpec: DestinationFormSpec<S3State> = {
 
 interface WebhookState {
   url: string
+  signatureHeader: string
   signingSecret: string
+  bearerToken: string
 }
 
 const webhookFormSpec: DestinationFormSpec<WebhookState> = {
   displayName: 'HTTPS webhook',
-  initialState: { url: '', signingSecret: '' },
+  initialState: { url: '', signatureHeader: '', signingSecret: '', bearerToken: '' },
   FormFields: ({ state, setState }) => (
     <>
       <FormField label='URL'>
@@ -109,19 +130,38 @@ const webhookFormSpec: DestinationFormSpec<WebhookState> = {
           placeholder='https://example.com/sim-drain'
         />
       </FormField>
+      <FormField label='Signature header (optional)'>
+        <Input
+          value={state.signatureHeader}
+          onChange={(e) => setState({ ...state, signatureHeader: e.target.value })}
+          placeholder='X-Sim-Signature'
+        />
+      </FormField>
       <FormField label='Signing secret'>
         <SecretInput
           value={state.signingSecret}
           onChange={(v) => setState({ ...state, signingSecret: v })}
         />
       </FormField>
+      <FormField label='Bearer token (optional)'>
+        <SecretInput
+          value={state.bearerToken}
+          onChange={(v) => setState({ ...state, bearerToken: v })}
+        />
+      </FormField>
     </>
   ),
   isComplete: (s) => s.url.length > 0 && s.signingSecret.length >= 8,
   toDestinationBranch: (s) => ({
     destinationType: 'webhook',
-    destinationConfig: { url: s.url },
-    destinationCredentials: { signingSecret: s.signingSecret },
+    destinationConfig: {
+      url: s.url,
+      signatureHeader: s.signatureHeader || undefined,
+    },
+    destinationCredentials: {
+      signingSecret: s.signingSecret,
+      bearerToken: s.bearerToken || undefined,
+    },
   }),
 }
 

apps/sim/lib/data-drains/destinations/webhook.ts

@@ -1,7 +1,6 @@
 import { createHmac } from 'node:crypto'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { sleep } from '@sim/utils/helpers'
 import { z } from 'zod'
 import { validateExternalUrl } from '@/lib/core/security/input-validation'
 import {
@@ -91,6 +90,27 @@ function sign(body: Buffer, secret: string, timestamp: number): string {
   return `t=${timestamp},${SIGNATURE_VERSION}=${hmac}`
 }
 
+/**
+ * Resolves after `ms` or as soon as `signal` aborts, whichever happens first.
+ * The caller checks `signal.aborted` at the top of the next iteration to
+ * surface the abort — keeping resolution side-effect-free here.
+ */
+function sleepUntilAborted(ms: number, signal: AbortSignal): Promise<void> {
+  if (signal.aborted) return Promise.resolve()
+  return new Promise((resolve) => {
+    let timeoutId: ReturnType<typeof setTimeout>
+    const onAbort = () => {
+      clearTimeout(timeoutId)
+      resolve()
+    }
+    timeoutId = setTimeout(() => {
+      signal.removeEventListener('abort', onAbort)
+      resolve()
+    }, ms)
+    signal.addEventListener('abort', onAbort, { once: true })
+  })
+}
+
 function backoffWithJitter(attempt: number, retryAfterMs?: number): number {
   if (retryAfterMs !== undefined) {
     // Floor at 500ms so a misbehaving server returning Retry-After: 0 cannot
@@ -249,7 +269,7 @@ export const webhookDestination: DrainDestination<
             retryAfterMs = parseRetryAfter(response.headers.get('retry-after'))
           }
           if (attempt < MAX_ATTEMPTS) {
-            await sleep(backoffWithJitter(attempt, retryAfterMs))
+            await sleepUntilAborted(backoffWithJitter(attempt, retryAfterMs), signal)
           }
         }
         throw lastError instanceof Error