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

TheodoreSpeaks · TheodoreSpeaks · commit b48c4c382cd7 · 2026-06-22T15:03:40.000-07:00
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.
diff --git a/apps/sim/app/api/table/[tableId]/delete-async/route.ts b/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
diff --git a/apps/sim/app/api/table/[tableId]/export-async/route.ts b/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
diff --git a/apps/sim/app/api/table/[tableId]/import-async/route.ts b/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
diff --git a/apps/sim/app/api/table/import-async/route.ts b/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
diff --git a/apps/sim/app/api/webhooks/agentmail/route.ts b/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
diff --git a/apps/sim/lib/a2a/push-notifications.ts b/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 })
diff --git a/apps/sim/lib/billing/cleanup-dispatcher.ts b/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,13 +315,15 @@ 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) => ({
           payload,
           options: {
             tags: [`plan:${payload.plan}`, `jobType:${jobType}`],
             concurrencyKey: getCleanupConcurrencyKey(jobType),
+            region,
           },
         }))
       )
diff --git a/apps/sim/lib/copilot/tools/server/table/user-table.ts b/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(() => {})
diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts b/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
     })
 
diff --git a/apps/sim/lib/core/async-jobs/region.test.ts b/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'])
+  })
+})
diff --git a/apps/sim/lib/core/async-jobs/region.ts b/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
+}
diff --git a/apps/sim/lib/core/config/env.ts b/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)
diff --git a/apps/sim/lib/core/config/feature-flags.test.ts b/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 () => {
diff --git a/apps/sim/lib/core/config/feature-flags.ts b/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>
 
 /**
diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/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 {
diff --git a/apps/sim/lib/knowledge/documents/service.ts b/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,
           },
         }))
       )
diff --git a/apps/sim/lib/messaging/lifecycle.ts b/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(),
     }
   )
 
diff --git a/apps/sim/lib/table/backfill-runner.ts b/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.
diff --git a/apps/sim/lib/table/workflow-columns.ts b/apps/sim/lib/table/workflow-columns.ts

Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,7 @@ import {`
`19`	`19`	`agentMailMessageSchema,`
`20`	`20`	`webhookSvixHeadersSchema,`
`21`	`21`	`} from '@/lib/api/contracts/webhooks'`
	`22`	`+import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'`
`22`	`23`	`import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'`
`23`	`24`	`import {`
`24`	`25`	`assertContentLengthWithinLimit,`
`@@ -234,6 +235,7 @@ export const POST = withRouteHandler(async (req: Request) => {`
`234`	`235`	`{ taskId },`
`235`	`236`	`{`
`236`	`237`	tags: [`workspaceId:${result.id}`, `taskId:${taskId}`],
	`238`	`+ region: await resolveTriggerRegion(),`
`237`	`239`	`}`
`238`	`240`	`)`
`239`	`241`	`await db`
Original file line number	Diff line number	Diff line change
`@@ -111,13 +111,15 @@ export async function notifyTaskStateChange(taskId: string, state: TaskState): P`
`111`	`111`
`112`	`112`	`if (isTriggerDevEnabled) {`
`113`	`113`	`try {`
`114`		`- const { a2aPushNotificationTask } = await import(`
`115`		`- '@/background/a2a-push-notification-delivery'`
`116`		`- )`
	`114`	`+ const [{ a2aPushNotificationTask }, { resolveTriggerRegion }] = await Promise.all([`
	`115`	`+ import('@/background/a2a-push-notification-delivery'),`
	`116`	`+ import('@/lib/core/async-jobs/region'),`
	`117`	`+ ])`
`117`	`118`	`await a2aPushNotificationTask.trigger(`
`118`	`119`	`{ taskId, state },`
`119`	`120`	`{`
`120`	`121`	tags: [`taskId:${taskId}`],
	`122`	`+ region: await resolveTriggerRegion(),`
`121`	`123`	`}`
`122`	`124`	`)`
`123`	`125`	`logger.info('Push notification queued to trigger.dev', { taskId, state })`