diff --git a/apps/sim/app/api/auth/oauth/token/route.ts b/apps/sim/app/api/auth/oauth/token/route.ts index b767b59325e..6b5c2adad6d 100644 --- a/apps/sim/app/api/auth/oauth/token/route.ts +++ b/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) { diff --git a/apps/sim/app/api/auth/oauth/utils.test.ts b/apps/sim/app/api/auth/oauth/utils.test.ts index 0e6550d44ea..43ea672f9d1 100644 --- a/apps/sim/app/api/auth/oauth/utils.test.ts +++ b/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 = {}) { + 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( diff --git a/apps/sim/app/api/auth/oauth/utils.ts b/apps/sim/app/api/auth/oauth/utils.ts index 4fd56f0e0a7..7f6fe1eb461 100644 --- a/apps/sim/app/api/auth/oauth/utils.ts +++ b/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 { + /** + * 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,14 +718,49 @@ async function performCoalescedRefresh({ return null } - const lockKey = `oauth:refresh:${accountId}` + const lockKey = `oauth:refresh:${scopeKey}` const refreshPromise = coalesceLocally(lockKey, () => withLeaderLock({ 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', { @@ -700,24 +768,49 @@ async function performCoalescedRefresh({ 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 = { - 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 = { + 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, }) diff --git a/apps/sim/lib/auth/auth.ts b/apps/sim/lib/auth/auth.ts index a4aeafc2343..a8d11f31cd5 100644 --- a/apps/sim/lib/auth/auth.ts +++ b/apps/sim/lib/auth/auth.ts @@ -95,6 +95,8 @@ import { getMicrosoftRefreshTokenExpiry, isMicrosoftProvider, } from '@/lib/oauth/microsoft' +import { extractSlackTeamId, fanOutSlackTokenChain } from '@/lib/oauth/slack' +import { clearDeadFlag } from '@/lib/oauth/terminal-errors' import { getCanonicalScopesForProvider } from '@/lib/oauth/utils' import { captureServerEvent, getPostHogClient } from '@/lib/posthog/server' import { disableUserResources } from '@/lib/workflows/lifecycle' @@ -444,6 +446,41 @@ export const auth = betterAuth({ }) } + /** + * A fresh Slack connect re-issues the installation's rotating token + * chain, invalidating the copies held by sibling account rows for the + * same team (Slack bot tokens are per-installation, not per-grant). + * Propagate the new chain so every sibling is valid again, and clear + * the installation's dead flag. + */ + if (account.providerId === 'slack' && account.accessToken) { + try { + const teamId = extractSlackTeamId(account.accountId) + if (teamId) { + // Clear the dead flag before fanning out: the connect itself + // proves the installation has live tokens, and a fan-out + // failure must not leave the hour-long flag blocking refreshes. + await clearDeadFlag(`slack:${teamId}`) + await fanOutSlackTokenChain(teamId, { + accessToken: account.accessToken, + refreshToken: account.refreshToken ?? null, + accessTokenExpiresAt: account.accessTokenExpiresAt ?? null, + }) + logger.info('[account.create.after] Propagated Slack installation token chain', { + userId: account.userId, + teamId, + newAccountId: account.id, + }) + } + } catch (error) { + logger.error('[account.create.after] Failed to propagate Slack token chain', { + userId: account.userId, + accountId: account.id, + error, + }) + } + } + try { await processCredentialDraft({ userId: account.userId, diff --git a/apps/sim/lib/oauth/slack.test.ts b/apps/sim/lib/oauth/slack.test.ts new file mode 100644 index 00000000000..e38c919f8d3 --- /dev/null +++ b/apps/sim/lib/oauth/slack.test.ts @@ -0,0 +1,35 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest' +import { extractSlackTeamId } from '@/lib/oauth/slack' + +describe('extractSlackTeamId', () => { + it('extracts the team id from a scoped account id', () => { + expect( + extractSlackTeamId('T08CM6ZNYBE-usr_U08USBQ9B1T-cbf46a7e-ca75-4a2e-bef5-fd467299eaae') + ).toBe('T08CM6ZNYBE') + }) + + it('extracts the team id from a legacy bot-segment account id', () => { + expect(extractSlackTeamId('T08CM6ZNYBE-U08USBQ9B1T-599a2a79-2543-42fd-9a74-9e2c466a8b19')).toBe( + 'T08CM6ZNYBE' + ) + }) + + it('accepts enterprise grid ids', () => { + expect(extractSlackTeamId('E0123ABCD-usr_U1-aaaa')).toBe('E0123ABCD') + }) + + it('returns null for pasted custom-bot account ids', () => { + expect(extractSlackTeamId('slack-bot-1764756583292')).toBeNull() + }) + + it('returns null for lowercase or malformed ids', () => { + expect(extractSlackTeamId('t123-usr_U1-x')).toBeNull() + expect(extractSlackTeamId('T08CM6ZNYBE')).toBeNull() + expect(extractSlackTeamId('')).toBeNull() + expect(extractSlackTeamId(null)).toBeNull() + expect(extractSlackTeamId(undefined)).toBeNull() + }) +}) diff --git a/apps/sim/lib/oauth/slack.ts b/apps/sim/lib/oauth/slack.ts new file mode 100644 index 00000000000..25f5811859c --- /dev/null +++ b/apps/sim/lib/oauth/slack.ts @@ -0,0 +1,140 @@ +import { db } from '@sim/db' +import { account } from '@sim/db/schema' +import { and, eq, gt, isNotNull, like, max, sql } from 'drizzle-orm' + +/** + * Slack bot tokens belong to the installation (team × app), not to the OAuth + * grant: every connect of the same Slack workspace hands back the same rotating + * token chain, so all `account` rows for one team are copies of one credential. + * These helpers let the refresh path treat those rows as a single unit. + * + * External account ids are `${teamId}-${userSegment}-${uuid}` (see the Slack + * getUserInfo in lib/auth/auth.ts). The team segment charset `[TE][A-Z0-9]+` + * contains no LIKE metacharacters, and the trailing `-` in the prefix match + * prevents `T123` from matching `T1234-...`. + */ +const SLACK_TEAM_PREFIX_RE = /^([TE][A-Z0-9]+)-/ + +export function isSlackProvider(providerId: string): boolean { + return providerId === 'slack' +} + +/** + * Extracts the Slack installation (team/enterprise) id from an external account + * id. Returns null for ids that don't carry a team prefix (e.g. legacy pasted + * `slack-bot-...` rows), which keep per-row refresh behavior. + */ +export function extractSlackTeamId(externalAccountId: string | null | undefined): string | null { + if (!externalAccountId) return null + const match = SLACK_TEAM_PREFIX_RE.exec(externalAccountId) + return match ? match[1] : null +} + +function installationFilter(teamId: string) { + return and(eq(account.providerId, 'slack'), like(account.accountId, `${teamId}-%`)) +} + +interface SlackTokenChain { + accessToken: string + refreshToken: string | null + accessTokenExpiresAt: Date | null +} + +interface FanOutOptions { + /** + * Version guard: skip the write entirely when any sibling row was updated + * after this timestamp. A refresh leader holds its chain snapshot across a + * multi-second provider call while a concurrent OAuth connect (whose + * `account.create.after` fan-out takes no lock) may land the newly issued + * chain first — an unconditional write would then overwrite the live chain + * with a stale one. Omit for connect-time fan-out, whose chain is by + * definition the newest. + */ + ifChainUnchangedSince?: Date +} + +/** + * Writes a token chain to every account row of a Slack installation. Rotation + * revokes whatever the sibling rows were holding, so a successful refresh or a + * fresh connect must overwrite all copies or the stale ones fail with + * `token_revoked` at call time. + */ +export async function fanOutSlackTokenChain( + teamId: string, + chain: SlackTokenChain, + options?: FanOutOptions +): Promise { + const since = options?.ifChainUnchangedSince + await db + .update(account) + .set({ + accessToken: chain.accessToken, + accessTokenExpiresAt: chain.accessTokenExpiresAt, + ...(chain.refreshToken ? { refreshToken: chain.refreshToken } : {}), + updatedAt: new Date(), + }) + .where( + and( + installationFilter(teamId), + since + ? sql`NOT EXISTS (SELECT 1 FROM ${account} sibling WHERE sibling.provider_id = 'slack' AND sibling.account_id LIKE ${`${teamId}-%`} AND sibling.updated_at > ${since})` + : undefined + ) + ) +} + +/** + * True when any account row of the installation was updated after `since` — + * i.e. another writer (a fresh connect or a competing refresh) landed a newer + * chain while the caller was working from an older snapshot. + */ +export async function hasSlackChainMoved(teamId: string, since: Date): Promise { + const [row] = await db + .select({ moved: max(account.updatedAt) }) + .from(account) + .where(and(installationFilter(teamId), gt(account.updatedAt, since))) + .limit(1) + return row?.moved != null +} + +interface FreshestSlackChain { + accessToken: string | null + refreshToken: string + accessTokenExpiresAt: Date | null + /** + * Max `updated_at` across the installation's rows at read time — the version + * guard passed back into {@link fanOutSlackTokenChain} / consulted via + * {@link hasSlackChainMoved} after the provider round-trip. + */ + chainVersion: Date +} + +/** + * Reads the freshest token chain for an installation: the sibling row with the + * latest access-token expiry that still holds a refresh token. The caller's own + * copy may already be rotated away; the installation's live refresh token is + * the most recently issued one. + */ +export async function getFreshestSlackChain(teamId: string): Promise { + const [row] = await db + .select({ + accessToken: account.accessToken, + refreshToken: account.refreshToken, + accessTokenExpiresAt: account.accessTokenExpiresAt, + chainVersion: sql< + Date | string | null + >`(SELECT max(sibling.updated_at) FROM ${account} sibling WHERE sibling.provider_id = 'slack' AND sibling.account_id LIKE ${`${teamId}-%`})`, + }) + .from(account) + .where(and(installationFilter(teamId), isNotNull(account.refreshToken))) + .orderBy(sql`${account.accessTokenExpiresAt} DESC NULLS LAST`) + .limit(1) + + if (!row?.refreshToken) return null + return { + accessToken: row.accessToken, + refreshToken: row.refreshToken, + accessTokenExpiresAt: row.accessTokenExpiresAt, + chainVersion: row.chainVersion ? new Date(row.chainVersion) : new Date(0), + } +} diff --git a/apps/sim/lib/oauth/terminal-errors.ts b/apps/sim/lib/oauth/terminal-errors.ts index 25fba73205c..81d915531f6 100644 --- a/apps/sim/lib/oauth/terminal-errors.ts +++ b/apps/sim/lib/oauth/terminal-errors.ts @@ -12,6 +12,7 @@ const TERMINAL_ERRORS = new Set([ 'invalid_client_id', 'invalid_client', 'bad_redirect_uri', + 'token_revoked', ]) const DEAD_CACHE_TTL_SEC = 60 * 60