feat(trigger): add trigger-eu-region flag to switch runs to eu-central-1 (#5173)

* feat(trigger): add trigger-eu-region flag to switch runs to eu-central-1

Global on/off feature flag routing every Trigger.dev run from the default
us-east-1 to eu-central-1 via the per-trigger region option, resolved at
each dispatch site through resolveTriggerRegion.

* test(trigger): mock resolveTriggerRegion in delete-async route test

The route now pulls in feature-flags (which imports isAppConfigEnabled from
env-flags); the test's partial env-flags mock made that access throw. Stub the
region module and assert the region option on the dispatch.

apps/sim/app/api/table/[tableId]/delete-async/route.test.ts

@@ -39,6 +39,9 @@ vi.mock('@/lib/core/config/env-flags', () => ({
   },
 }))
 vi.mock('@/background/table-delete', () => ({ tableDeleteTask: { id: 'table-delete' } }))
+vi.mock('@/lib/core/async-jobs/region', () => ({
+  resolveTriggerRegion: vi.fn().mockResolvedValue('us-east-1'),
+}))
 vi.mock('@trigger.dev/sdk', () => ({
   tasks: { trigger: mockTasksTrigger },
   task: (config: unknown) => config,
@@ -196,7 +199,7 @@ describe('POST /api/table/[tableId]/delete-async', () => {
         excludeRowIds: ['row_keep'],
         cutoff: expect.any(String),
       }),
-      { tags: ['tableId:tbl_1', 'jobId:job-id-xyz'] }
+      { tags: ['tableId:tbl_1', 'jobId:job-id-xyz'], region: 'us-east-1' }
     )
   })
 

apps/sim/app/api/table/[tableId]/delete-async/route.ts

@@ -86,14 +86,15 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
     // Trigger.dev runs the delete outside the web container (survives deploys) and retries —
     // safe: the keyset + cutoff walk just deletes whatever remains.
     try {
-      const [{ tableDeleteTask }, { tasks }] = await Promise.all([
+      const [{ tableDeleteTask }, { tasks }, { resolveTriggerRegion }] = await Promise.all([
         import('@/background/table-delete'),
         import('@trigger.dev/sdk'),
+        import('@/lib/core/async-jobs/region'),
       ])
       await tasks.trigger<typeof tableDeleteTask>(
         'table-delete',
         { jobId, tableId, workspaceId, filter, excludeRowIds, cutoff: cutoff.toISOString() },
-        { tags: [`tableId:${tableId}`, `jobId:${jobId}`] }
+        { tags: [`tableId:${tableId}`, `jobId:${jobId}`], region: await resolveTriggerRegion() }
       )
     } catch (error) {
       // A failed dispatch must not leave a ghost `running` job holding the

apps/sim/app/api/table/[tableId]/export-async/route.ts

@@ -61,12 +61,14 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
   const payload: TableExportPayload = { jobId, tableId, workspaceId, format }
   if (isTriggerDevEnabled) {
     try {
-      const [{ tableExportTask }, { tasks }] = await Promise.all([
+      const [{ tableExportTask }, { tasks }, { resolveTriggerRegion }] = await Promise.all([
         import('@/background/table-export'),
         import('@trigger.dev/sdk'),
+        import('@/lib/core/async-jobs/region'),
       ])
       await tasks.trigger<typeof tableExportTask>('table-export', payload, {
         tags: [`tableId:${tableId}`, `jobId:${jobId}`],
+        region: await resolveTriggerRegion(),
       })
     } catch (error) {
       // A failed dispatch must not leave a ghost `running` job holding the

apps/sim/app/api/table/[tableId]/import-async/route.ts

@@ -83,12 +83,14 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
   if (isTriggerDevEnabled) {
     // Trigger.dev runs the import outside the web container, so it survives app deploys.
     try {
-      const [{ tableImportTask }, { tasks }] = await Promise.all([
+      const [{ tableImportTask }, { tasks }, { resolveTriggerRegion }] = await Promise.all([
         import('@/background/table-import'),
         import('@trigger.dev/sdk'),
+        import('@/lib/core/async-jobs/region'),
       ])
       await tasks.trigger<typeof tableImportTask>('table-import', importPayload, {
         tags: [`tableId:${tableId}`, `jobId:${importId}`],
+        region: await resolveTriggerRegion(),
       })
     } catch (error) {
       // A failed dispatch must not leave a ghost `running` job holding the

apps/sim/app/api/table/import-async/route.ts

@@ -115,12 +115,14 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
   if (isTriggerDevEnabled) {
     // Trigger.dev runs the import outside the web container, so it survives app deploys.
     try {
-      const [{ tableImportTask }, { tasks }] = await Promise.all([
+      const [{ tableImportTask }, { tasks }, { resolveTriggerRegion }] = await Promise.all([
         import('@/background/table-import'),
         import('@trigger.dev/sdk'),
+        import('@/lib/core/async-jobs/region'),
       ])
       await tasks.trigger<typeof tableImportTask>('table-import', importPayload, {
         tags: [`tableId:${table.id}`, `jobId:${importId}`],
+        region: await resolveTriggerRegion(),
       })
     } catch (error) {
       // A failed dispatch must not leave a ghost `running` job holding the

apps/sim/app/api/webhooks/agentmail/route.ts

@@ -19,6 +19,7 @@ import {
   agentMailMessageSchema,
   webhookSvixHeadersSchema,
 } from '@/lib/api/contracts/webhooks'
+import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
 import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
 import {
   assertContentLengthWithinLimit,
@@ -234,6 +235,7 @@ export const POST = withRouteHandler(async (req: Request) => {
           { taskId },
           {
             tags: [`workspaceId:${result.id}`, `taskId:${taskId}`],
+            region: await resolveTriggerRegion(),
           }
         )
         await db

apps/sim/lib/a2a/push-notifications.ts

@@ -111,13 +111,15 @@ export async function notifyTaskStateChange(taskId: string, state: TaskState): P
 
   if (isTriggerDevEnabled) {
     try {
-      const { a2aPushNotificationTask } = await import(
-        '@/background/a2a-push-notification-delivery'
-      )
+      const [{ a2aPushNotificationTask }, { resolveTriggerRegion }] = await Promise.all([
+        import('@/background/a2a-push-notification-delivery'),
+        import('@/lib/core/async-jobs/region'),
+      ])
       await a2aPushNotificationTask.trigger(
         { taskId, state },
         {
           tags: [`taskId:${taskId}`],
+          region: await resolveTriggerRegion(),
         }
       )
       logger.info('Push notification queued to trigger.dev', { taskId, state })

apps/sim/lib/billing/cleanup-dispatcher.ts

@@ -10,6 +10,7 @@ import { getPlanType, type PlanCategory } from '@/lib/billing/plan-helpers'
 import { chunkArray } from '@/lib/cleanup/batch-delete'
 import { getJobQueue } from '@/lib/core/async-jobs'
 import { shouldExecuteInline } from '@/lib/core/async-jobs/config'
+import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
 import type { EnqueueOptions } from '@/lib/core/async-jobs/types'
 import { isTriggerAvailable } from '@/lib/knowledge/documents/service'
 import { isOrganizationWorkspace, WORKSPACE_MODE } from '@/lib/workspaces/policy'
@@ -314,6 +315,7 @@ export async function dispatchCleanupJobs(jobType: CleanupJobType): Promise<{
       if (batch.length === 0) return
       const currentBatch = batch
       batch = []
+      const region = await resolveTriggerRegion()
       const batchResult = await tasks.batchTrigger(
         jobType,
         currentBatch.map((payload) => ({
@@ -321,6 +323,7 @@ export async function dispatchCleanupJobs(jobType: CleanupJobType): Promise<{
           options: {
             tags: [`plan:${payload.plan}`, `jobType:${jobType}`],
             concurrencyKey: getCleanupConcurrencyKey(jobType),
+            region,
           },
         }))
       )

apps/sim/lib/copilot/tools/server/table/user-table.ts

@@ -134,12 +134,14 @@ function shouldImportInBackground(record: { name: string; size: number }): boole
 async function dispatchImportJob(payload: TableImportPayload): Promise<void> {
   if (isTriggerDevEnabled) {
     try {
-      const [{ tableImportTask }, { tasks }] = await Promise.all([
+      const [{ tableImportTask }, { tasks }, { resolveTriggerRegion }] = await Promise.all([
         import('@/background/table-import'),
         import('@trigger.dev/sdk'),
+        import('@/lib/core/async-jobs/region'),
       ])
       await tasks.trigger<typeof tableImportTask>('table-import', payload, {
         tags: [`tableId:${payload.tableId}`, `jobId:${payload.importId}`],
+        region: await resolveTriggerRegion(),
       })
     } catch (error) {
       await releaseJobClaim(payload.tableId, payload.importId).catch(() => {})
@@ -166,14 +168,15 @@ async function dispatchDeleteJob(params: {
   const { jobId, tableId, workspaceId, filter, cutoff, maxRows } = params
   if (isTriggerDevEnabled) {
     try {
-      const [{ tableDeleteTask }, { tasks }] = await Promise.all([
+      const [{ tableDeleteTask }, { tasks }, { resolveTriggerRegion }] = await Promise.all([
         import('@/background/table-delete'),
         import('@trigger.dev/sdk'),
+        import('@/lib/core/async-jobs/region'),
       ])
       await tasks.trigger<typeof tableDeleteTask>(
         'table-delete',
         { jobId, tableId, workspaceId, filter, cutoff: cutoff.toISOString(), maxRows },
-        { tags: [`tableId:${tableId}`, `jobId:${jobId}`] }
+        { tags: [`tableId:${tableId}`, `jobId:${jobId}`], region: await resolveTriggerRegion() }
       )
     } catch (error) {
       await releaseJobClaim(tableId, jobId).catch(() => {})
@@ -208,14 +211,15 @@ async function dispatchUpdateJob(params: {
   const { jobId, tableId, workspaceId, filter, data, cutoff, maxRows } = params
   if (isTriggerDevEnabled) {
     try {
-      const [{ tableUpdateTask }, { tasks }] = await Promise.all([
+      const [{ tableUpdateTask }, { tasks }, { resolveTriggerRegion }] = await Promise.all([
         import('@/background/table-update'),
         import('@trigger.dev/sdk'),
+        import('@/lib/core/async-jobs/region'),
       ])
       await tasks.trigger<typeof tableUpdateTask>(
         'table-update',
         { jobId, tableId, workspaceId, filter, data, cutoff: cutoff.toISOString(), maxRows },
-        { tags: [`tableId:${tableId}`, `jobId:${jobId}`] }
+        { tags: [`tableId:${tableId}`, `jobId:${jobId}`], region: await resolveTriggerRegion() }
       )
     } catch (error) {
       await releaseJobClaim(tableId, jobId).catch(() => {})

apps/sim/lib/core/async-jobs/backends/trigger-dev.ts

@@ -1,6 +1,7 @@
 import { createLogger } from '@sim/logger'
 import { taskContext } from '@trigger.dev/core/v3'
 import { runs, type TriggerOptions, tasks } from '@trigger.dev/sdk'
+import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
 import {
   type EnqueueOptions,
   JOB_STATUS,
@@ -84,6 +85,7 @@ export class TriggerDevJobQueue implements JobQueueBackend {
     if (options?.delayMs && options.delayMs > 0) {
       triggerOptions.delay = new Date(Date.now() + options.delayMs)
     }
+    triggerOptions.region = await resolveTriggerRegion()
     const handle = await tasks.trigger(taskId, enrichedPayload, triggerOptions)
 
     logger.debug('Enqueued job via trigger.dev', { jobId: handle.id, type, taskId, tags })
@@ -125,6 +127,7 @@ export class TriggerDevJobQueue implements JobQueueBackend {
     const taskId = JOB_TYPE_TO_TASK_ID[type]
     if (!taskId) throw new Error(`Unknown job type: ${type}`)
 
+    const region = await resolveTriggerRegion()
     const batchItems = items.map(({ payload, options }) => {
       const enrichedPayload =
         options?.metadata && typeof payload === 'object' && payload !== null
@@ -133,12 +136,12 @@ export class TriggerDevJobQueue implements JobQueueBackend {
       const tags = buildTags(options)
       const batchItem: {
         payload: unknown
-        options?: { concurrencyKey?: string; tags?: string[] }
+        options?: { concurrencyKey?: string; tags?: string[]; region?: string }
       } = { payload: enrichedPayload }
-      const batchOpts: { concurrencyKey?: string; tags?: string[] } = {}
+      const batchOpts: { concurrencyKey?: string; tags?: string[]; region?: string } = { region }
       if (options?.concurrencyKey) batchOpts.concurrencyKey = options.concurrencyKey
       if (tags.length > 0) batchOpts.tags = tags
-      if (Object.keys(batchOpts).length > 0) batchItem.options = batchOpts
+      batchItem.options = batchOpts
       return batchItem
     })
 

apps/sim/lib/core/async-jobs/region.test.ts

@@ -0,0 +1,42 @@
+/**
+ * @vitest-environment node
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockIsFeatureEnabled } = vi.hoisted(() => ({
+  mockIsFeatureEnabled: vi.fn(),
+}))
+
+vi.mock('@/lib/core/config/feature-flags', () => ({
+  isFeatureEnabled: mockIsFeatureEnabled,
+}))
+
+import {
+  resolveTriggerRegion,
+  TRIGGER_REGION_EU_CENTRAL,
+  TRIGGER_REGION_US_EAST,
+} from '@/lib/core/async-jobs/region'
+
+describe('resolveTriggerRegion', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('returns eu-central-1 when the flag is enabled', async () => {
+    mockIsFeatureEnabled.mockResolvedValue(true)
+    expect(await resolveTriggerRegion()).toBe(TRIGGER_REGION_EU_CENTRAL)
+    expect(mockIsFeatureEnabled).toHaveBeenCalledWith('trigger-eu-region')
+  })
+
+  it('returns us-east-1 when the flag is disabled', async () => {
+    mockIsFeatureEnabled.mockResolvedValue(false)
+    expect(await resolveTriggerRegion()).toBe(TRIGGER_REGION_US_EAST)
+  })
+
+  it('evaluates globally, passing no gating context', async () => {
+    mockIsFeatureEnabled.mockResolvedValue(false)
+    await resolveTriggerRegion()
+    expect(mockIsFeatureEnabled).toHaveBeenCalledTimes(1)
+    expect(mockIsFeatureEnabled.mock.calls[0]).toEqual(['trigger-eu-region'])
+  })
+})

apps/sim/lib/core/async-jobs/region.ts

@@ -0,0 +1,21 @@
+import { isFeatureEnabled } from '@/lib/core/config/feature-flags'
+
+/** Default Trigger.dev region — the project default when the eu-central flag is off. */
+export const TRIGGER_REGION_US_EAST = 'us-east-1'
+
+/** Target region when the `trigger-eu-region` flag is enabled. */
+export const TRIGGER_REGION_EU_CENTRAL = 'eu-central-1'
+
+/**
+ * Resolve which Trigger.dev region a run should execute in. Gated globally by the
+ * `trigger-eu-region` feature flag (all-or-nothing — no per-user/org targeting):
+ * `eu-central-1` when enabled, otherwise `us-east-1`.
+ *
+ * The result is passed as the `region` option to `tasks.trigger` / `batchTrigger`,
+ * overriding the project's dashboard default per run.
+ */
+export async function resolveTriggerRegion(): Promise<string> {
+  return (await isFeatureEnabled('trigger-eu-region'))
+    ? TRIGGER_REGION_EU_CENTRAL
+    : TRIGGER_REGION_US_EAST
+}

apps/sim/lib/core/config/env.ts

@@ -74,6 +74,7 @@ export const env = createEnv({
     TABLES_FRACTIONAL_ORDERING:            z.boolean().optional(),                 // Order table rows by fractional order_key (O(1) insert/delete) instead of integer position
     TABLE_SNAPSHOT_CACHE:                  z.boolean().optional(),                 // Mount tables into sandboxes by reference via a version-keyed CSV snapshot in object storage instead of draining the whole table into web-process heap
     PII_REDACTION:                         z.boolean().optional(),                 // Redact PII from workflow logs via configurable Data Retention rules (Presidio at the logger persist choke point) and expose the Data Retention config UI
+    TRIGGER_EU_REGION:                     z.boolean().optional(),                 // Route Trigger.dev runs to eu-central-1 instead of the default us-east-1 (fallback for the trigger-eu-region flag when AppConfig is not the source of truth)
 
     // Table feature limits (per plan). Apply when billing is disabled (free tier defaults) or for billed plans.
     FREE_TABLES_LIMIT:                     z.number().optional(),                  // Max user tables per workspace on free tier (default: 5)

apps/sim/lib/core/config/feature-flags.test.ts

@@ -63,6 +63,7 @@ describe('getFeatureFlags', () => {
     expect(flags['tables-fractional-ordering']).toEqual({ enabled: false })
     expect(flags['mothership-beta']).toEqual({ enabled: false })
     expect(flags['pii-redaction']).toEqual({ enabled: false })
+    expect(flags['trigger-eu-region']).toEqual({ enabled: false })
     expect(mockFetch).not.toHaveBeenCalled()
   })
 
@@ -90,6 +91,7 @@ describe('getFeatureFlags', () => {
     expect(flags['tables-fractional-ordering']).toEqual({ enabled: false })
     expect(flags['mothership-beta']).toEqual({ enabled: false })
     expect(flags['pii-redaction']).toEqual({ enabled: false })
+    expect(flags['trigger-eu-region']).toEqual({ enabled: false })
   })
 
   it('degrades gracefully on a malformed document', async () => {

apps/sim/lib/core/config/feature-flags.ts

@@ -90,6 +90,13 @@ const FEATURE_FLAGS = {
       'agree.',
     fallback: 'PII_REDACTION',
   },
+  'trigger-eu-region': {
+    description:
+      'Route Trigger.dev runs to eu-central-1 instead of the default us-east-1. Global on/off ' +
+      'only — resolved without user/org context at every task-trigger call site via ' +
+      'resolveTriggerRegion, so the whole deployment switches regions together.',
+    fallback: 'TRIGGER_EU_REGION',
+  },
 } satisfies Record<string, FeatureFlagDefinition>
 
 /**

apps/sim/lib/knowledge/connectors/sync-engine.ts

@@ -12,6 +12,7 @@ import { generateId } from '@sim/utils/id'
 import { randomInt } from '@sim/utils/random'
 import { and, eq, gt, inArray, isNotNull, isNull, lt, ne, or, sql } from 'drizzle-orm'
 import { decryptApiKey } from '@/lib/api-key/crypto'
+import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
 import { getInternalApiBaseUrl } from '@/lib/core/utils/urls'
 import type { DocumentData } from '@/lib/knowledge/documents/service'
 import {
@@ -325,7 +326,7 @@ export async function dispatchSync(
         fullSync: options?.fullSync,
         requestId,
       },
-      { tags }
+      { tags, region: await resolveTriggerRegion() }
     )
     logger.info(`Dispatched connector sync to Trigger.dev`, { connectorId, requestId })
   } else {

apps/sim/lib/knowledge/documents/service.ts

@@ -32,6 +32,7 @@ import { checkActorUsageLimits } from '@/lib/billing/calculations/usage-monitor'
 import { recordUsage } from '@/lib/billing/core/usage-log'
 import { checkAndBillOverageThreshold } from '@/lib/billing/threshold-billing'
 import type { ChunkingStrategy, StrategyOptions } from '@/lib/chunkers/types'
+import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
 import { env, envNumber } from '@/lib/core/config/env'
 import { getCostMultiplier, isTriggerDevEnabled } from '@/lib/core/config/env-flags'
 import { processDocument } from '@/lib/knowledge/documents/document-processor'
@@ -447,6 +448,7 @@ async function dispatchViaBatchTrigger(
 ): Promise<number> {
   let dispatched = 0
   const batchIds: string[] = []
+  const region = await resolveTriggerRegion()
   for (let i = 0; i < jobPayloads.length; i += TRIGGER_BATCH_SIZE) {
     const chunk = jobPayloads.slice(i, i + TRIGGER_BATCH_SIZE)
     try {
@@ -462,6 +464,7 @@ async function dispatchViaBatchTrigger(
               `knowledgeBaseId:${payload.knowledgeBaseId}`,
               `documentId:${payload.documentId}`,
             ],
+            region,
           },
         }))
       )

apps/sim/lib/messaging/lifecycle.ts

@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { tasks } from '@trigger.dev/sdk'
+import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
 import { env } from '@/lib/core/config/env'
 import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
 
@@ -41,6 +42,7 @@ export async function scheduleLifecycleEmail({
     {
       delay: delayUntil,
       idempotencyKey: `lifecycle-${type}-${userId}`,
+      region: await resolveTriggerRegion(),
     }
   )
 

apps/sim/lib/table/backfill-runner.ts

@@ -319,12 +319,14 @@ export async function maybeBackfillGroupOutputs(opts: {
   }
   if (isTriggerDevEnabled) {
     try {
-      const [{ tableBackfillTask }, { tasks }] = await Promise.all([
+      const [{ tableBackfillTask }, { tasks }, { resolveTriggerRegion }] = await Promise.all([
         import('@/background/table-backfill'),
         import('@trigger.dev/sdk'),
+        import('@/lib/core/async-jobs/region'),
       ])
       await tasks.trigger<typeof tableBackfillTask>('table-backfill', payload, {
         tags: [`tableId:${table.id}`, `jobId:${jobId}`],
+        region: await resolveTriggerRegion(),
       })
     } catch (error) {
       // Release the claim so a ghost `running` job doesn't block imports/deletes.

apps/sim/lib/table/workflow-columns.ts

@@ -764,14 +764,15 @@ export async function runWorkflowColumn(opts: {
   if (isTriggerDevEnabled) {
     // Trigger.dev runs `tableRunDispatcherTask`, which loops `dispatcherStep`
     // until done with CRIU-checkpointed waits between windows.
-    const [{ tableRunDispatcherTask }, { tasks }] = await Promise.all([
+    const [{ tableRunDispatcherTask }, { tasks }, { resolveTriggerRegion }] = await Promise.all([
       import('@/background/table-run-dispatcher'),
       import('@trigger.dev/sdk'),
+      import('@/lib/core/async-jobs/region'),
     ])
     await tasks.trigger<typeof tableRunDispatcherTask>(
       'table-run-dispatcher',
       { dispatchId },
-      { concurrencyKey: dispatchId }
+      { concurrencyKey: dispatchId, region: await resolveTriggerRegion() }
     )
   } else {
     // Local / no-trigger.dev: drive the same loop in-process, fire-and-forget