fix(appconfig): share the polled snapshot with Trigger.dev runs (#9053)

* fix(appconfig): share the polled snapshot with Trigger.dev runs

Every Trigger.dev run starts a fresh process, and each new AppConfig session is
billed for the full configuration it receives. Profiles that opt in
(feature-flags, block-visibility, access-control) are now published to Redis by
app processes (latest validatedAt wins), and a cold cache inside a Trigger.dev
run adopts a snapshot under 120s old (tolerating 5s of clock skew) instead of
opening a session. Missing, stale, unreadable, or unparseable snapshots fall back
to polling as before. Trigger.dev runs never publish, and strict snapshots
(outbound routing) never use the shared copy.

* test(appconfig): guard the Redis URL, reuse flushMacrotask, await the older publish

Author: waleedlatif1

apps/sim/lib/auth/access-control.test.ts

@@ -93,7 +93,8 @@ describe('getAccessControlConfig', () => {
       expect(result.blockedEmailMxHosts).toEqual([])
       expect(mockFetch).toHaveBeenCalledWith(
         { application: 'sim-staging', environment: 'staging', profile: 'access-control' },
-        expect.any(Function)
+        expect.any(Function),
+        { shareWithTriggerRuns: true }
       )
     })
 

apps/sim/lib/auth/access-control.ts

@@ -100,7 +100,8 @@ export async function getAccessControlConfig(): Promise<AccessControlConfig> {
       environment: env.APPCONFIG_ENVIRONMENT as string,
       profile: ACCESS_CONTROL_PROFILE,
     },
-    parseConfig
+    parseConfig,
+    { shareWithTriggerRuns: true }
   )
 
   return value ?? fromEnv()

apps/sim/lib/core/config/appconfig.integration.ts

@@ -0,0 +1,119 @@
+/**
+ * The shared AppConfig snapshot against real Redis: a long-lived process publishes what it polled,
+ * a fresh Trigger.dev run process adopts it without an AppConfig session, and an older snapshot
+ * never replaces a newer one. Skipped without `TEST_REDIS_URL`.
+ */
+import { readTestRedisUrl } from '@sim/db/testing/test-infrastructure'
+import { redisConfigMock, redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
+import { sleep } from '@sim/utils/helpers'
+import { generateId } from '@sim/utils/id'
+import Redis from 'ioredis'
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  markInsideTriggerRun,
+  resetInsideTriggerRunForTests,
+} from '@/lib/core/config/trigger-runtime'
+
+const { mockSend } = vi.hoisted(() => ({ mockSend: vi.fn() }))
+
+vi.mock('@/lib/core/config/redis', () => redisConfigMock)
+vi.mock('@aws-sdk/client-appconfigdata', () => ({
+  AppConfigDataClient: class {
+    send = mockSend
+  },
+  StartConfigurationSessionCommand: class {
+    __type = 'start'
+  },
+  GetLatestConfigurationCommand: class {
+    __type = 'get'
+  },
+}))
+
+const redisUrl = readTestRedisUrl()
+
+/** Each import is a separate process's module-level cache. */
+async function freshProcess() {
+  vi.resetModules()
+  return import('@/lib/core/config/appconfig')
+}
+
+function respondWith(revision: string) {
+  mockSend.mockImplementation((command: { __type: string }) =>
+    Promise.resolve(
+      command.__type === 'start'
+        ? { InitialConfigurationToken: 'token' }
+        : {
+            Configuration: new TextEncoder().encode(JSON.stringify({ revision })),
+            NextPollConfigurationToken: 'next',
+          }
+    )
+  )
+}
+
+describe.runIf(Boolean(redisUrl))('shared AppConfig snapshot with Redis', () => {
+  let redis: Redis
+  let ids: { application: string; environment: string; profile: string }
+  const key = () => `appconfig:snapshot:v1:${ids.application}/${ids.environment}/${ids.profile}`
+
+  async function waitForPublished(revision: string) {
+    for (let attempt = 0; attempt < 50; attempt++) {
+      const stored = await redis.get(key())
+      if (stored?.includes(revision)) return
+      await sleep(20)
+    }
+    throw new Error(`snapshot ${revision} was never published`)
+  }
+
+  beforeAll(async () => {
+    if (!redisUrl) throw new Error('TEST_REDIS_URL is required for this suite')
+    redis = new Redis(redisUrl, { lazyConnect: true, maxRetriesPerRequest: 0 })
+    await redis.connect()
+  })
+
+  beforeEach(() => {
+    ids = { application: generateId(), environment: 'integration', profile: 'feature-flags' }
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(redis)
+  })
+
+  afterEach(async () => {
+    resetInsideTriggerRunForTests()
+    await redis.del(key())
+  })
+
+  afterAll(async () => {
+    await redis?.quit()
+  })
+
+  it('lets a Trigger.dev run adopt what an app process polled, with no AppConfig session', async () => {
+    respondWith('polled')
+    const app = await freshProcess()
+    expect(
+      await app.fetchAppConfigProfile(ids, (json) => json, { shareWithTriggerRuns: true })
+    ).toEqual({ revision: 'polled' })
+    await waitForPublished('polled')
+
+    mockSend.mockClear()
+    markInsideTriggerRun()
+    const run = await freshProcess()
+    expect(
+      await run.fetchAppConfigProfile(ids, (json) => json, { shareWithTriggerRuns: true })
+    ).toEqual({ revision: 'polled' })
+    expect(mockSend).not.toHaveBeenCalled()
+  })
+
+  it('keeps a newer snapshot when a process that polled earlier publishes', async () => {
+    const newer = { raw: JSON.stringify({ revision: 'newer' }), validatedAt: Date.now() + 60_000 }
+    await redis.set(key(), JSON.stringify(newer))
+
+    respondWith('older')
+    const publish = vi.spyOn(redis, 'eval')
+    const app = await freshProcess()
+    expect(
+      await app.fetchAppConfigProfile(ids, (json) => json, { shareWithTriggerRuns: true })
+    ).toEqual({ revision: 'older' })
+    await vi.waitFor(() => expect(publish).toHaveBeenCalledTimes(1))
+    expect(await publish.mock.results[0]?.value).toBe(0)
+
+    expect(JSON.parse((await redis.get(key())) ?? 'null')).toEqual(newer)
+  })
+})

apps/sim/lib/core/config/appconfig.test.ts

@@ -1,3 +1,6 @@
+import { flushMacrotask } from '@sim/testing/helpers/async'
+import { redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
+import { toRecord } from '@sim/utils/object'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const { mockSend } = vi.hoisted(() => ({
@@ -19,6 +22,10 @@ vi.mock('@aws-sdk/client-appconfigdata', () => ({
 }))
 
 import { fetchAppConfigProfile, fetchAppConfigSnapshot } from '@/lib/core/config/appconfig'
+import {
+  markInsideTriggerRun,
+  resetInsideTriggerRunForTests,
+} from '@/lib/core/config/trigger-runtime'
 
 const encode = (value: unknown) => new TextEncoder().encode(JSON.stringify(value))
 
@@ -241,3 +248,114 @@ describe('fetchAppConfigSnapshot freshness', () => {
     expect(mockSend.mock.calls).toHaveLength(3)
   })
 })
+
+describe('shared snapshot in Trigger.dev runs', () => {
+  const shared = { shareWithTriggerRuns: true }
+  const sessionResponses = (command: { __type: string }) =>
+    Promise.resolve(
+      command.__type === 'start'
+        ? { InitialConfigurationToken: 'token' }
+        : { Configuration: encode({ revision: 'appconfig' }), NextPollConfigurationToken: 'next' }
+    )
+  const snapshotAt = (validatedAt: number) =>
+    JSON.stringify({ raw: JSON.stringify({ revision: 'shared' }), validatedAt })
+  const redisReturning = (stored: string | null) => {
+    const redis = { get: vi.fn().mockResolvedValue(stored), eval: vi.fn().mockResolvedValue(1) }
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(redis)
+    return redis
+  }
+
+  beforeEach(() => {
+    mockSend.mockImplementation(sessionResponses)
+    markInsideTriggerRun()
+  })
+  afterEach(() => resetInsideTriggerRunForTests())
+
+  it('adopts a fresh snapshot without opening an AppConfig session', async () => {
+    redisReturning(snapshotAt(Date.now()))
+    expect(await fetchAppConfigProfile(uniqueIds(), (json) => json, shared)).toEqual({
+      revision: 'shared',
+    })
+    expect(mockSend).not.toHaveBeenCalled()
+  })
+
+  it('tolerates a publisher clock a few seconds ahead, but not more', async () => {
+    redisReturning(snapshotAt(Date.now() + 4_000))
+    expect(await fetchAppConfigProfile(uniqueIds(), (json) => json, shared)).toEqual({
+      revision: 'shared',
+    })
+    redisReturning(snapshotAt(Date.now() + 60_000))
+    expect(await fetchAppConfigProfile(uniqueIds(), (json) => json, shared)).toEqual({
+      revision: 'appconfig',
+    })
+  })
+
+  it('never reads the snapshot outside a Trigger.dev run', async () => {
+    resetInsideTriggerRunForTests()
+    redisReturning(snapshotAt(Date.now()))
+    expect(await fetchAppConfigProfile(uniqueIds(), (json) => json, shared)).toEqual({
+      revision: 'appconfig',
+    })
+  })
+
+  it('never reads or publishes a profile that did not opt in', async () => {
+    const redis = redisReturning(snapshotAt(Date.now()))
+    expect(await fetchAppConfigProfile(uniqueIds(), (json) => json)).toEqual({
+      revision: 'appconfig',
+    })
+    resetInsideTriggerRunForTests()
+    await fetchAppConfigProfile(uniqueIds(), (json) => json)
+    await flushMacrotask()
+    expect(redis.eval).not.toHaveBeenCalled()
+  })
+
+  it('polls AppConfig when the snapshot is older than the adoption window', async () => {
+    redisReturning(snapshotAt(Date.now() - 120_000))
+    expect(await fetchAppConfigProfile(uniqueIds(), (json) => json, shared)).toEqual({
+      revision: 'appconfig',
+    })
+  })
+
+  it('polls AppConfig when Redis fails or the snapshot does not parse', async () => {
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue({
+      get: vi.fn().mockRejectedValue(new Error('redis down')),
+      eval: vi.fn().mockRejectedValue(new Error('redis down')),
+    })
+    expect(await fetchAppConfigProfile(uniqueIds(), (json) => json, shared)).toEqual({
+      revision: 'appconfig',
+    })
+
+    redisReturning(snapshotAt(Date.now()))
+    const rejectShared = (json: unknown) => {
+      if (toRecord(json).revision === 'shared') throw new Error('invalid')
+      return json
+    }
+    expect(await fetchAppConfigProfile(uniqueIds(), rejectShared, shared)).toEqual({
+      revision: 'appconfig',
+    })
+  })
+
+  it('publishes only from app processes', async () => {
+    const redis = redisReturning(null)
+    expect(await fetchAppConfigProfile(uniqueIds(), (json) => json, shared)).toEqual({
+      revision: 'appconfig',
+    })
+    await flushMacrotask()
+    expect(redis.eval).not.toHaveBeenCalled()
+
+    resetInsideTriggerRunForTests()
+    await fetchAppConfigProfile(uniqueIds(), (json) => json, shared)
+    await vi.waitFor(() => expect(redis.eval).toHaveBeenCalledTimes(1))
+  })
+
+  it('never serves or publishes a strict snapshot through the shared copy', async () => {
+    const redis = redisReturning(snapshotAt(Date.now()))
+    expect((await fetchAppConfigSnapshot(uniqueIds(), (json) => json)).value).toEqual({
+      revision: 'appconfig',
+    })
+    resetInsideTriggerRunForTests()
+    await fetchAppConfigSnapshot(uniqueIds(), (json) => json)
+    await flushMacrotask()
+    expect(redis.eval).not.toHaveBeenCalled()
+  })
+})