fix(oauth): coalesce slack token refresh per installation and preserve provider refresh errors (#5723)

* fix(oauth): coalesce slack token refresh per installation and preserve provider refresh errors

* fix(oauth): keep transient refresh failures errorless and size slack lock budgets past the provider timeout

* fix(oauth): let slack refresh followers poll for the lock's full lifetime

* chore(oauth): drop provider refresh-error surfacing to keep this PR slack-only

* fix(oauth): version-guard slack chain writes against concurrent connects

* fix(oauth): clear slack dead flag before connect fan-out

Author: TheodoreSpeaks

apps/sim/app/api/auth/oauth/token/route.ts

@@ -410,7 +410,8 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         },
         { status: 200 }
       )
-    } catch (_error) {
+    } catch (error) {
+      logger.error(`[${requestId}] Failed to refresh access token:`, error)
       return NextResponse.json({ error: 'Failed to refresh access token' }, { status: 401 })
     }
   } catch (error) {

apps/sim/app/api/auth/oauth/utils.test.ts

@@ -282,6 +282,146 @@ describe('OAuth Utils', () => {
     })
   })
 
+  describe('Slack installation-scoped refresh', () => {
+    const SLACK_ACCOUNT_ID = 'T08CM6ZNYBE-usr_U08USBQ9B1T-cbf46a7e-ca75-4a2e-bef5-fd467299eaae'
+    const past = new Date(Date.now() - 3600 * 1000)
+    const future = new Date(Date.now() + 3600 * 1000)
+
+    /** Select chain for getFreshestSlackChain: where() -> orderBy() -> limit(). */
+    function mockSelectOrderedChain(limitResult: unknown[]) {
+      const mockLimit = vi.fn().mockReturnValue(limitResult)
+      const mockOrderBy = vi.fn().mockReturnValue({ limit: mockLimit })
+      const mockWhere = vi.fn().mockReturnValue({ orderBy: mockOrderBy, limit: mockLimit })
+      const mockFrom = vi.fn().mockReturnValue({ where: mockWhere })
+      mockDb.select.mockReturnValueOnce({ from: mockFrom })
+      return { mockWhere, mockOrderBy, mockLimit }
+    }
+
+    function slackCredential(overrides: Record<string, unknown> = {}) {
+      return {
+        id: 'row-1',
+        resolvedCredentialId: 'row-1',
+        accountId: SLACK_ACCOUNT_ID,
+        accessToken: 'stale-at',
+        refreshToken: 'stale-rt',
+        accessTokenExpiresAt: past,
+        providerId: 'slack',
+        ...overrides,
+      }
+    }
+
+    it('locks per installation and refreshes with the freshest sibling refresh token', async () => {
+      mockSelectOrderedChain([
+        { accessToken: 'stale-at', refreshToken: 'live-rt', accessTokenExpiresAt: past },
+      ])
+      mockRefreshOAuthToken.mockResolvedValueOnce({
+        ok: true,
+        accessToken: 'new-at',
+        expiresIn: 43200,
+        refreshToken: 'new-rt',
+      })
+      const { mockSet } = mockUpdateChain()
+
+      const result = await refreshTokenIfNeeded('request-id', slackCredential(), 'row-1')
+
+      expect(result).toEqual({ accessToken: 'new-at', refreshed: true })
+      expect(redisConfigMockFns.mockAcquireLock.mock.calls[0][0]).toBe(
+        'oauth:refresh:slack:T08CM6ZNYBE'
+      )
+      expect(redisConfigMockFns.mockAcquireLock.mock.calls[0][2]).toBe(30)
+      expect(mockRefreshOAuthToken).toHaveBeenCalledWith('slack', 'live-rt')
+      expect(mockSet).toHaveBeenCalledWith(
+        expect.objectContaining({ accessToken: 'new-at', refreshToken: 'new-rt' })
+      )
+    })
+
+    it('returns the freshest sibling token without refreshing when it is still valid', async () => {
+      mockSelectOrderedChain([
+        { accessToken: 'sibling-at', refreshToken: 'live-rt', accessTokenExpiresAt: future },
+      ])
+      const { mockSet } = mockUpdateChain()
+
+      const result = await refreshTokenIfNeeded('request-id', slackCredential(), 'row-1')
+
+      expect(result).toEqual({ accessToken: 'sibling-at', refreshed: true })
+      expect(mockRefreshOAuthToken).not.toHaveBeenCalled()
+      expect(mockSet).toHaveBeenCalledWith(
+        expect.objectContaining({ accessToken: 'sibling-at', refreshToken: 'live-rt' })
+      )
+    })
+
+    it('keeps per-row behavior for pasted custom-bot account ids', async () => {
+      mockRefreshOAuthToken.mockResolvedValueOnce({
+        ok: true,
+        accessToken: 'new-at',
+        expiresIn: 43200,
+        refreshToken: 'new-rt',
+      })
+      mockUpdateChain()
+
+      const result = await refreshTokenIfNeeded(
+        'request-id',
+        slackCredential({ accountId: 'slack-bot-1764756583292' }),
+        'row-1'
+      )
+
+      expect(result).toEqual({ accessToken: 'new-at', refreshed: true })
+      expect(redisConfigMockFns.mockAcquireLock.mock.calls[0][0]).toBe('oauth:refresh:row-1')
+      expect(mockRefreshOAuthToken).toHaveBeenCalledWith('slack', 'stale-rt')
+    })
+
+    it('dead-flags the installation, not the row, on terminal refresh errors', async () => {
+      const fakeRedis = {
+        set: vi.fn().mockResolvedValue('OK'),
+        get: vi.fn().mockResolvedValue(null),
+        del: vi.fn().mockResolvedValue(1),
+      }
+      redisConfigMockFns.mockGetRedisClient.mockReturnValue(fakeRedis)
+      mockSelectOrderedChain([
+        { accessToken: 'stale-at', refreshToken: 'live-rt', accessTokenExpiresAt: past },
+      ])
+      mockRefreshOAuthToken.mockResolvedValueOnce({
+        ok: false,
+        errorCode: 'token_revoked',
+      })
+      mockSelectChain([])
+
+      await expect(refreshTokenIfNeeded('request-id', slackCredential(), 'row-1')).rejects.toThrow(
+        'Failed to refresh token'
+      )
+
+      expect(fakeRedis.set).toHaveBeenCalledWith(
+        'oauth:dead:slack:T08CM6ZNYBE',
+        'token_revoked',
+        'EX',
+        3600
+      )
+    })
+
+    it('skips the dead flag when the chain moved during the failed refresh', async () => {
+      const fakeRedis = {
+        set: vi.fn().mockResolvedValue('OK'),
+        get: vi.fn().mockResolvedValue(null),
+        del: vi.fn().mockResolvedValue(1),
+      }
+      redisConfigMockFns.mockGetRedisClient.mockReturnValue(fakeRedis)
+      mockSelectOrderedChain([
+        { accessToken: 'stale-at', refreshToken: 'live-rt', accessTokenExpiresAt: past },
+      ])
+      mockRefreshOAuthToken.mockResolvedValueOnce({
+        ok: false,
+        errorCode: 'token_revoked',
+      })
+      mockSelectChain([{ moved: new Date() }])
+
+      await expect(refreshTokenIfNeeded('request-id', slackCredential(), 'row-1')).rejects.toThrow(
+        'Failed to refresh token'
+      )
+
+      expect(fakeRedis.set).not.toHaveBeenCalled()
+    })
+  })
+
   describe('resolveServiceAccountToken', () => {
     it('throws loudly for an unknown provider (never silently attempts Google)', async () => {
       await expect(resolveServiceAccountToken('cred-1', 'mystery-provider')).rejects.toThrow(

apps/sim/app/api/auth/oauth/utils.ts

@@ -27,6 +27,13 @@ import {
   isMicrosoftProvider,
   PROACTIVE_REFRESH_THRESHOLD_DAYS,
 } from '@/lib/oauth/microsoft'
+import {
+  extractSlackTeamId,
+  fanOutSlackTokenChain,
+  getFreshestSlackChain,
+  hasSlackChainMoved,
+  isSlackProvider,
+} from '@/lib/oauth/slack'
 import {
   getRecentTerminalError,
   isTerminalRefreshError,
@@ -658,25 +665,51 @@ interface CoalescedRefreshOptions {
   accountId: string
   providerId: string
   refreshToken: string
+  /** External provider account id (`account.accountId`), used to scope Slack refreshes per installation. */
+  providerAccountId?: string | null
   requestId?: string
   userId?: string
 }
 
+/**
+ * Slack lock budgets sized past `TOKEN_REFRESH_TIMEOUT_MS` (15s) in
+ * lib/oauth/oauth.ts: installation-keyed locks make every sibling row's request
+ * a follower of one refresh, so the TTL covers the provider call plus generous
+ * headroom for the surrounding DB reads and the fan-out write, and followers
+ * poll for the lock's full lifetime so a slow-but-successful refresh is still
+ * observed rather than reported as a failure. These budgets are latency knobs,
+ * not correctness guarantees — chain integrity under lock expiry or unlocked
+ * concurrent writers is enforced by the version-guarded fan-out
+ * (`ifChainUnchangedSince` in lib/oauth/slack.ts).
+ */
+const SLACK_LOCK_TTL_SEC = 30
+const SLACK_FOLLOWER_MAX_WAIT_MS = SLACK_LOCK_TTL_SEC * 1000
+
 async function performCoalescedRefresh({
   accountId,
   providerId,
   refreshToken,
+  providerAccountId,
   requestId,
   userId,
 }: CoalescedRefreshOptions): Promise<string | null> {
+  /**
+   * Slack bot tokens are per-installation (team × app): every account row for
+   * one team holds a copy of the same rotating chain, so refreshes are locked,
+   * dead-flagged, and written per installation rather than per row.
+   */
+  const slackTeamId = isSlackProvider(providerId) ? extractSlackTeamId(providerAccountId) : null
+  const scopeKey = slackTeamId ? `slack:${slackTeamId}` : accountId
+
   const logContext = {
     ...(requestId ? { requestId } : {}),
     ...(userId ? { userId } : {}),
+    ...(slackTeamId ? { slackTeamId } : {}),
     providerId,
     accountId,
   }
 
-  const deadCode = await getRecentTerminalError(accountId)
+  const deadCode = await getRecentTerminalError(scopeKey)
   if (deadCode) {
     logger.warn('Skipping refresh: credential recently failed', {
       ...logContext,
@@ -685,39 +718,99 @@ async function performCoalescedRefresh({
     return null
   }
 
-  const lockKey = `oauth:refresh:${accountId}`
+  const lockKey = `oauth:refresh:${scopeKey}`
 
   const refreshPromise = coalesceLocally(lockKey, () =>
     withLeaderLock<string>({
       key: lockKey,
+      // Installation-keyed Slack locks gather followers from every sibling row,
+      // so their wait and the lock TTL must outlast the 15s provider timeout —
+      // the 3s/10s defaults would fail followers early and let a second leader
+      // start a concurrent rotation mid-refresh.
+      ...(slackTeamId ? { maxWaitMs: SLACK_FOLLOWER_MAX_WAIT_MS, ttlSec: SLACK_LOCK_TTL_SEC } : {}),
       onLeader: async () => {
         try {
-          const result = await refreshOAuthToken(providerId, refreshToken)
+          let refreshTokenToUse = refreshToken
+          let slackChainVersion: Date | null = null
+          if (slackTeamId) {
+            const freshest = await getFreshestSlackChain(slackTeamId)
+            if (!freshest) {
+              throw new Error(
+                `No refresh-capable account row found for Slack installation ${slackTeamId}`
+              )
+            }
+            slackChainVersion = freshest.chainVersion
+            if (
+              freshest.accessToken &&
+              freshest.accessTokenExpiresAt &&
+              freshest.accessTokenExpiresAt > new Date()
+            ) {
+              await fanOutSlackTokenChain(
+                slackTeamId,
+                {
+                  accessToken: freshest.accessToken,
+                  refreshToken: freshest.refreshToken,
+                  accessTokenExpiresAt: freshest.accessTokenExpiresAt,
+                },
+                { ifChainUnchangedSince: freshest.chainVersion }
+              )
+              logger.info('Reused freshest Slack installation token', logContext)
+              return freshest.accessToken
+            }
+            refreshTokenToUse = freshest.refreshToken
+          }
+
+          const result = await refreshOAuthToken(providerId, refreshTokenToUse)
 
           if (!result.ok) {
             logger.error('Failed to refresh token', {
               ...logContext,
               errorCode: result.errorCode,
             })
             if (result.errorCode && isTerminalRefreshError(result.errorCode)) {
-              await markCredentialDead(accountId, result.errorCode)
+              // A refresh that lost a race with a concurrent connect fails with
+              // a revoked/rotated-out token even though the installation just
+              // got a live chain — dead-flagging then would take down a healthy
+              // credential for an hour.
+              if (
+                slackChainVersion &&
+                (await hasSlackChainMoved(slackTeamId!, slackChainVersion))
+              ) {
+                logger.info('Skipping dead flag: Slack chain moved during refresh', logContext)
+              } else {
+                await markCredentialDead(scopeKey, result.errorCode)
+              }
             }
             return null
           }
 
-          const updateData: Record<string, unknown> = {
-            accessToken: result.accessToken,
-            accessTokenExpiresAt: new Date(Date.now() + result.expiresIn * 1000),
-            updatedAt: new Date(),
-          }
-          if (result.refreshToken && result.refreshToken !== refreshToken) {
-            updateData.refreshToken = result.refreshToken
-          }
-          if (isMicrosoftProvider(providerId)) {
-            updateData.refreshTokenExpiresAt = getMicrosoftRefreshTokenExpiry()
-          }
+          const accessTokenExpiresAt = new Date(Date.now() + result.expiresIn * 1000)
+
+          if (slackTeamId) {
+            await fanOutSlackTokenChain(
+              slackTeamId,
+              {
+                accessToken: result.accessToken,
+                refreshToken: result.refreshToken || refreshTokenToUse,
+                accessTokenExpiresAt,
+              },
+              { ifChainUnchangedSince: slackChainVersion ?? undefined }
+            )
+          } else {
+            const updateData: Record<string, unknown> = {
+              accessToken: result.accessToken,
+              accessTokenExpiresAt,
+              updatedAt: new Date(),
+            }
+            if (result.refreshToken && result.refreshToken !== refreshToken) {
+              updateData.refreshToken = result.refreshToken
+            }
+            if (isMicrosoftProvider(providerId)) {
+              updateData.refreshTokenExpiresAt = getMicrosoftRefreshTokenExpiry()
+            }
 
-          await db.update(account).set(updateData).where(eq(account.id, accountId))
+            await db.update(account).set(updateData).where(eq(account.id, accountId))
+          }
 
           logger.info('Successfully refreshed access token', logContext)
           return result.accessToken
@@ -774,6 +867,7 @@ export async function getOAuthToken(userId: string, providerId: string): Promise
   const connections = await db
     .select({
       id: account.id,
+      providerAccountId: account.accountId,
       accessToken: account.accessToken,
       refreshToken: account.refreshToken,
       accessTokenExpiresAt: account.accessTokenExpiresAt,
@@ -813,6 +907,7 @@ export async function getOAuthToken(userId: string, providerId: string): Promise
       accountId: credential.id,
       providerId,
       refreshToken: credential.refreshToken!,
+      providerAccountId: credential.providerAccountId,
       userId,
     })
     if (fresh) return fresh
@@ -915,6 +1010,7 @@ export async function resolveCredentialAccessToken(
       accountId: resolvedCredentialId,
       providerId: credential.providerId,
       refreshToken: credential.refreshToken!,
+      providerAccountId: credential.accountId,
       requestId,
       userId: credential.userId,
     })
@@ -1019,6 +1115,7 @@ export async function refreshTokenIfNeeded(
     accountId: resolvedCredentialId,
     providerId: credential.providerId,
     refreshToken: credential.refreshToken!,
+    providerAccountId: credential.accountId,
     requestId,
     userId: credential.userId,
   })