feat(idempotency): added generalized idempotency service for all triggers/webhooks (#1330)

* update infra and remove railway

* feat(webhooks): add idempotency service for all triggers/webhooks

* Revert "update infra and remove railway"

This reverts commit abfa2f8.

* cleanup

* ack PR comments

Author: waleedlatif1

apps/sim/app/api/webhooks/cleanup/idempotency/route.ts

@@ -0,0 +1,64 @@
+import { type NextRequest, NextResponse } from 'next/server'
+import { verifyCronAuth } from '@/lib/auth/internal'
+import { cleanupExpiredIdempotencyKeys, getIdempotencyKeyStats } from '@/lib/idempotency/cleanup'
+import { createLogger } from '@/lib/logs/console/logger'
+import { generateRequestId } from '@/lib/utils'
+
+const logger = createLogger('IdempotencyCleanupAPI')
+
+export const dynamic = 'force-dynamic'
+export const maxDuration = 300 // Allow up to 5 minutes for cleanup
+
+export async function GET(request: NextRequest) {
+  const requestId = generateRequestId()
+  logger.info(`Idempotency cleanup triggered (${requestId})`)
+
+  try {
+    const authError = verifyCronAuth(request, 'Idempotency key cleanup')
+    if (authError) {
+      return authError
+    }
+
+    const statsBefore = await getIdempotencyKeyStats()
+    logger.info(
+      `Pre-cleanup stats: ${statsBefore.totalKeys} keys across ${Object.keys(statsBefore.keysByNamespace).length} namespaces`
+    )
+
+    const result = await cleanupExpiredIdempotencyKeys({
+      maxAgeSeconds: 7 * 24 * 60 * 60, // 7 days
+      batchSize: 1000,
+    })
+
+    const statsAfter = await getIdempotencyKeyStats()
+    logger.info(`Post-cleanup stats: ${statsAfter.totalKeys} keys remaining`)
+
+    return NextResponse.json({
+      success: true,
+      message: 'Idempotency key cleanup completed',
+      requestId,
+      result: {
+        deleted: result.deleted,
+        errors: result.errors,
+        statsBefore: {
+          totalKeys: statsBefore.totalKeys,
+          keysByNamespace: statsBefore.keysByNamespace,
+        },
+        statsAfter: {
+          totalKeys: statsAfter.totalKeys,
+          keysByNamespace: statsAfter.keysByNamespace,
+        },
+      },
+    })
+  } catch (error) {
+    logger.error(`Error during idempotency cleanup (${requestId}):`, error)
+    return NextResponse.json(
+      {
+        success: false,
+        message: 'Idempotency cleanup failed',
+        error: error instanceof Error ? error.message : 'Unknown error',
+        requestId,
+      },
+      { status: 500 }
+    )
+  }
+}

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

@@ -4,6 +4,7 @@ import { type NextRequest, NextResponse } from 'next/server'
 import { checkServerSideUsageLimits } from '@/lib/billing'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
 import { env, isTruthy } from '@/lib/env'
+import { IdempotencyService, webhookIdempotency } from '@/lib/idempotency/service'
 import { createLogger } from '@/lib/logs/console/logger'
 import { generateRequestId } from '@/lib/utils'
 import {
@@ -328,7 +329,7 @@ export async function POST(
     // Continue processing - better to risk usage limit bypass than fail webhook
   }
 
-  // --- PHASE 5: Queue webhook execution (trigger.dev or direct based on env) ---
+  // --- PHASE 5: Idempotent webhook execution ---
   try {
     const payload = {
       webhookId: foundWebhook.id,
@@ -341,22 +342,44 @@ export async function POST(
       blockId: foundWebhook.blockId,
     }
 
-    const useTrigger = isTruthy(env.TRIGGER_DEV_ENABLED)
+    const idempotencyKey = IdempotencyService.createWebhookIdempotencyKey(
+      foundWebhook.id,
+      body,
+      Object.fromEntries(request.headers.entries())
+    )
 
-    if (useTrigger) {
-      const handle = await tasks.trigger('webhook-execution', payload)
-      logger.info(
-        `[${requestId}] Queued webhook execution task ${handle.id} for ${foundWebhook.provider} webhook`
-      )
-    } else {
-      // Fire-and-forget direct execution to avoid blocking webhook response
-      void executeWebhookJob(payload).catch((error) => {
-        logger.error(`[${requestId}] Direct webhook execution failed`, error)
-      })
-      logger.info(
-        `[${requestId}] Queued direct webhook execution for ${foundWebhook.provider} webhook (Trigger.dev disabled)`
-      )
-    }
+    const result = await webhookIdempotency.executeWithIdempotency(
+      foundWebhook.provider,
+      idempotencyKey,
+      async () => {
+        const useTrigger = isTruthy(env.TRIGGER_DEV_ENABLED)
+
+        if (useTrigger) {
+          const handle = await tasks.trigger('webhook-execution', payload)
+          logger.info(
+            `[${requestId}] Queued webhook execution task ${handle.id} for ${foundWebhook.provider} webhook`
+          )
+          return {
+            method: 'trigger.dev',
+            taskId: handle.id,
+            status: 'queued',
+          }
+        }
+        // Fire-and-forget direct execution to avoid blocking webhook response
+        void executeWebhookJob(payload).catch((error) => {
+          logger.error(`[${requestId}] Direct webhook execution failed`, error)
+        })
+        logger.info(
+          `[${requestId}] Queued direct webhook execution for ${foundWebhook.provider} webhook (Trigger.dev disabled)`
+        )
+        return {
+          method: 'direct',
+          status: 'queued',
+        }
+      }
+    )
+
+    logger.debug(`[${requestId}] Webhook execution result:`, result)
 
     // Return immediate acknowledgment with provider-specific format
     if (foundWebhook.provider === 'microsoftteams') {

apps/sim/background/webhook-execution.ts

@@ -3,6 +3,7 @@ import { eq, sql } from 'drizzle-orm'
 import { v4 as uuidv4 } from 'uuid'
 import { checkServerSideUsageLimits } from '@/lib/billing'
 import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
+import { IdempotencyService, webhookIdempotency } from '@/lib/idempotency'
 import { createLogger } from '@/lib/logs/console/logger'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
@@ -41,11 +42,29 @@ export async function executeWebhookJob(payload: WebhookExecutionPayload) {
     executionId,
   })
 
-  // Initialize logging session outside try block so it's available in catch
+  const idempotencyKey = IdempotencyService.createWebhookIdempotencyKey(
+    payload.webhookId,
+    payload.body,
+    payload.headers
+  )
+
+  return await webhookIdempotency.executeWithIdempotency(
+    payload.provider,
+    idempotencyKey,
+    async () => {
+      return await executeWebhookJobInternal(payload, executionId, requestId)
+    }
+  )
+}
+
+async function executeWebhookJobInternal(
+  payload: WebhookExecutionPayload,
+  executionId: string,
+  requestId: string
+) {
   const loggingSession = new LoggingSession(payload.workflowId, executionId, 'webhook', requestId)
 
   try {
-    // Check usage limits first
     const usageCheck = await checkServerSideUsageLimits(payload.userId)
     if (usageCheck.isExceeded) {
       logger.warn(
@@ -62,15 +81,13 @@ export async function executeWebhookJob(payload: WebhookExecutionPayload) {
       )
     }
 
-    // Load workflow from normalized tables
     const workflowData = await loadWorkflowFromNormalizedTables(payload.workflowId)
     if (!workflowData) {
       throw new Error(`Workflow not found: ${payload.workflowId}`)
     }
 
     const { blocks, edges, loops, parallels } = workflowData
 
-    // Get environment variables with workspace precedence
     const wfRows = await db
       .select({ workspaceId: workflowTable.workspaceId })
       .from(workflowTable)

apps/sim/db/migrations/0090_fearless_zaladane.sql

@@ -0,0 +1,10 @@
+CREATE TABLE "idempotency_key" (
+	"key" text NOT NULL,
+	"namespace" text DEFAULT 'default' NOT NULL,
+	"result" json NOT NULL,
+	"created_at" timestamp DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+CREATE UNIQUE INDEX "idempotency_key_namespace_unique" ON "idempotency_key" USING btree ("key","namespace");--> statement-breakpoint
+CREATE INDEX "idempotency_key_created_at_idx" ON "idempotency_key" USING btree ("created_at");--> statement-breakpoint
+CREATE INDEX "idempotency_key_namespace_idx" ON "idempotency_key" USING btree ("namespace");