DRY

Author: waleedlatif1

apps/sim/app/api/chat/[identifier]/route.test.ts

@@ -337,7 +337,6 @@ describe('Chat Identifier API Route', () => {
     })
 
     it('should return 503 when workflow is not available', async () => {
-      // Mock preprocessing to return workflow not deployed error
       const { preprocessExecution } = await import('@/lib/execution/preprocessing')
       const originalImplementation = vi.mocked(preprocessExecution).getMockImplementation()
 

apps/sim/app/api/webhooks/test/[id]/route.ts

@@ -67,8 +67,6 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
     return authError
   }
 
-  // Wrap preprocessing in try-catch to handle unexpected exceptions that might prevent
-  // proper execution log completion (e.g., programming errors, out-of-memory, etc.)
   let preprocessError: NextResponse | null = null
   try {
     preprocessError = await checkWebhookPreprocessing(
@@ -81,15 +79,13 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
       return preprocessError
     }
   } catch (error) {
-    // Handle unexpected preprocessing failures to ensure proper error response
     logger.error(`[${requestId}] Unexpected error during webhook preprocessing`, {
       error: error instanceof Error ? error.message : String(error),
       stack: error instanceof Error ? error.stack : undefined,
       webhookId: foundWebhook.id,
       workflowId: foundWorkflow.id,
     })
 
-    // Return provider-specific error response
     if (foundWebhook.provider === 'microsoft-teams') {
       return NextResponse.json(
         {

apps/sim/app/api/webhooks/trigger/[path]/route.ts

@@ -123,8 +123,6 @@ export async function POST(
     return authError
   }
 
-  // Wrap preprocessing in try-catch to handle unexpected exceptions that might prevent
-  // proper execution log completion (e.g., programming errors, out-of-memory, etc.)
   let preprocessError: NextResponse | null = null
   try {
     preprocessError = await checkWebhookPreprocessing(
@@ -137,15 +135,13 @@ export async function POST(
       return preprocessError
     }
   } catch (error) {
-    // Handle unexpected preprocessing failures to ensure proper error response
     logger.error(`[${requestId}] Unexpected error during webhook preprocessing`, {
       error: error instanceof Error ? error.message : String(error),
       stack: error instanceof Error ? error.stack : undefined,
       webhookId: foundWebhook.id,
       workflowId: foundWorkflow.id,
     })
 
-    // Return provider-specific error response
     if (foundWebhook.provider === 'microsoft-teams') {
       return NextResponse.json(
         {

apps/sim/background/schedule-execution.ts

@@ -381,11 +381,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
     if (!preprocessResult.success) {
       const statusCode = preprocessResult.error?.statusCode || 500
 
-      // Handle specific error types with appropriate retry/disable logic
       switch (statusCode) {
         case 401: {
-          // Authentication error - credentials invalid or expired
-          // Don't retry as authentication issues require manual intervention
           logger.warn(
             `[${requestId}] Authentication error during preprocessing, disabling schedule`
           )
@@ -405,8 +402,6 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
         }
 
         case 403: {
-          // Authorization error - typically workflow not deployed
-          // Don't retry as authorization issues (like undeployed workflows) require manual intervention
           logger.warn(
             `[${requestId}] Authorization error during preprocessing, disabling schedule: ${preprocessResult.error?.message}`
           )
@@ -426,7 +421,6 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
         }
 
         case 404: {
-          // Workflow not found - permanent error, disable schedule
           logger.warn(`[${requestId}] Workflow not found, disabling schedule`)
           await applyScheduleUpdate(
             payload.scheduleId,
@@ -443,7 +437,6 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
         }
 
         case 429: {
-          // Rate limit exceeded - temporary error, schedule retry after delay
           logger.warn(`[${requestId}] Rate limit exceeded, scheduling retry`)
           const retryDelay = 5 * 60 * 1000
           const nextRetryAt = new Date(now.getTime() + retryDelay)
@@ -462,7 +455,6 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
         }
 
         case 402: {
-          // Usage limit exceeded - schedule next run based on schedule configuration
           logger.warn(`[${requestId}] Usage limit exceeded, scheduling next run`)
           const nextRunAt = await calculateNextRunFromDeployment(payload, requestId)
           if (nextRunAt) {
@@ -481,7 +473,6 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
         }
 
         default: {
-          // Other errors (500, etc.) - track failures and disable after max consecutive failures
           logger.error(`[${requestId}] Preprocessing failed: ${preprocessResult.error?.message}`)
           const nextRunAt = await determineNextRunAfterError(payload, now, requestId)
           const newFailedCount = (payload.failedCount || 0) + 1

apps/sim/lib/execution/preprocessing.ts

@@ -10,6 +10,100 @@ import { RateLimiter } from '@/services/queue/RateLimiter'
 
 const logger = createLogger('ExecutionPreprocessing')
 
+const BILLING_ERROR_MESSAGES = {
+  BILLING_REQUIRED:
+    'Unable to resolve billing account. This workflow cannot execute without a valid billing account.',
+  BILLING_ERROR_GENERIC: 'Error resolving billing account',
+} as const
+
+/**
+ * Attempts to resolve billing actor with fallback for resume contexts.
+ * Returns the resolved actor user ID or null if resolution fails and should block execution.
+ *
+ * For resume contexts, this function allows fallback to the workflow owner if workspace
+ * billing cannot be resolved, ensuring users can complete their paused workflows even
+ * if billing configuration changes mid-execution.
+ *
+ * @returns Object containing actorUserId (null if should block) and shouldBlock flag
+ */
+async function resolveBillingActorWithFallback(params: {
+  requestId: string
+  workflowId: string
+  workspaceId: string
+  executionId: string
+  triggerType: string
+  workflowRecord: WorkflowRecord
+  userId: string
+  isResumeContext: boolean
+  baseActorUserId: string | null
+  failureReason: 'null' | 'error'
+  error?: unknown
+  loggingSession?: LoggingSession
+}): Promise<
+  { actorUserId: string; shouldBlock: false } | { actorUserId: null; shouldBlock: true }
+> {
+  const {
+    requestId,
+    workflowId,
+    workspaceId,
+    executionId,
+    triggerType,
+    workflowRecord,
+    userId,
+    isResumeContext,
+    baseActorUserId,
+    failureReason,
+    error,
+    loggingSession,
+  } = params
+
+  if (baseActorUserId) {
+    return { actorUserId: baseActorUserId, shouldBlock: false }
+  }
+
+  const workflowOwner = workflowRecord.userId?.trim()
+  if (isResumeContext && workflowOwner) {
+    const logMessage =
+      failureReason === 'null'
+        ? '[BILLING_FALLBACK] Workspace billing account is null. Using workflow owner for billing.'
+        : '[BILLING_FALLBACK] Exception during workspace billing resolution. Using workflow owner for billing.'
+
+    logger.warn(`[${requestId}] ${logMessage}`, {
+      workflowId,
+      workspaceId,
+      fallbackUserId: workflowOwner,
+      ...(error ? { error } : {}),
+    })
+
+    return { actorUserId: workflowOwner, shouldBlock: false }
+  }
+
+  const fallbackUserId = workflowRecord.userId || userId || 'unknown'
+  const errorMessage =
+    failureReason === 'null'
+      ? BILLING_ERROR_MESSAGES.BILLING_REQUIRED
+      : BILLING_ERROR_MESSAGES.BILLING_ERROR_GENERIC
+
+  logger.warn(`[${requestId}] ${errorMessage}`, {
+    workflowId,
+    workspaceId,
+    ...(error ? { error } : {}),
+  })
+
+  await logPreprocessingError({
+    workflowId,
+    executionId,
+    triggerType,
+    requestId,
+    userId: fallbackUserId,
+    workspaceId,
+    errorMessage,
+    loggingSession,
+  })
+
+  return { actorUserId: null, shouldBlock: true }
+}
+
 export interface PreprocessExecutionOptions {
   // Required fields
   workflowId: string
@@ -84,7 +178,6 @@ export async function preprocessExecution(
     if (records.length === 0) {
       logger.warn(`[${requestId}] Workflow not found: ${workflowId}`)
 
-      // Log error to database
       await logPreprocessingError({
         workflowId,
         executionId,
@@ -176,47 +269,21 @@ export async function preprocessExecution(
     }
 
     if (!actorUserId) {
-      // Special handling for resume context: allow fallback to workflow owner
-      if (isResumeContext && workflowRecord.userId) {
-        actorUserId = workflowRecord.userId
-        logger.warn(
-          `[${requestId}] Billing account resolution failed in resume context. Falling back to workflow owner.`,
-          {
-            workflowId,
-            workspaceId,
-            fallbackUserId: actorUserId,
-          }
-        )
-        // Log warning but don't block execution
-        await logPreprocessingError({
-          workflowId,
-          executionId,
-          triggerType,
-          requestId,
-          userId: actorUserId,
-          workspaceId,
-          errorMessage:
-            'Warning: Workspace billing account could not be resolved. Using workflow owner for billing. Please verify workspace billing settings.',
-          loggingSession: providedLoggingSession,
-        })
-      } else {
-        logger.warn(`[${requestId}] Unable to resolve billing account`, {
-          workflowId,
-          workspaceId,
-        })
-
-        await logPreprocessingError({
-          workflowId,
-          executionId,
-          triggerType,
-          requestId,
-          userId: workflowRecord.userId || userId || 'unknown',
-          workspaceId,
-          errorMessage:
-            'Unable to resolve billing account. This workflow cannot execute without a valid billing account.',
-          loggingSession: providedLoggingSession,
-        })
+      const result = await resolveBillingActorWithFallback({
+        requestId,
+        workflowId,
+        workspaceId,
+        executionId,
+        triggerType,
+        workflowRecord,
+        userId,
+        isResumeContext,
+        baseActorUserId: actorUserId,
+        failureReason: 'null',
+        loggingSession: providedLoggingSession,
+      })
 
+      if (result.shouldBlock) {
         return {
           success: false,
           error: {
@@ -226,47 +293,28 @@ export async function preprocessExecution(
           },
         }
       }
+
+      actorUserId = result.actorUserId
     }
   } catch (error) {
     logger.error(`[${requestId}] Error resolving billing actor`, { error, workflowId })
 
-    // Special handling for resume context: allow fallback on error
-    if (isResumeContext && workflowRecord.userId) {
-      actorUserId = workflowRecord.userId
-      logger.warn(
-        `[${requestId}] Billing account resolution error in resume context. Falling back to workflow owner.`,
-        {
-          workflowId,
-          workspaceId,
-          fallbackUserId: actorUserId,
-          error,
-        }
-      )
-      // Log warning but continue execution
-      await logPreprocessingError({
-        workflowId,
-        executionId,
-        triggerType,
-        requestId,
-        userId: actorUserId,
-        workspaceId,
-        errorMessage:
-          'Warning: Error resolving workspace billing account. Using workflow owner for billing. Please verify workspace billing settings.',
-        loggingSession: providedLoggingSession,
-      })
-      // Continue to next step
-    } else {
-      await logPreprocessingError({
-        workflowId,
-        executionId,
-        triggerType,
-        requestId,
-        userId: workflowRecord.userId || userId || 'unknown',
-        workspaceId,
-        errorMessage: 'Error resolving billing account',
-        loggingSession: providedLoggingSession,
-      })
+    const result = await resolveBillingActorWithFallback({
+      requestId,
+      workflowId,
+      workspaceId,
+      executionId,
+      triggerType,
+      workflowRecord,
+      userId,
+      isResumeContext,
+      baseActorUserId: null,
+      failureReason: 'error',
+      error,
+      loggingSession: providedLoggingSession,
+    })
 
+    if (result.shouldBlock) {
       return {
         success: false,
         error: {
@@ -276,6 +324,8 @@ export async function preprocessExecution(
         },
       }
     }
+
+    actorUserId = result.actorUserId
   }
 
   // ========== STEP 4: Get User Subscription ==========