Skip to content
3 changes: 2 additions & 1 deletion apps/sim/app/api/auth/oauth/token/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
116 changes: 116 additions & 0 deletions apps/sim/app/api/auth/oauth/utils.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,122 @@ 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(20)
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',
})

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
)
})
})

describe('resolveServiceAccountToken', () => {
it('throws loudly for an unknown provider (never silently attempts Google)', async () => {
await expect(resolveServiceAccountToken('cred-1', 'mystery-provider')).rejects.toThrow(
Expand Down
104 changes: 88 additions & 16 deletions apps/sim/app/api/auth/oauth/utils.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,12 @@ import {
isMicrosoftProvider,
PROACTIVE_REFRESH_THRESHOLD_DAYS,
} from '@/lib/oauth/microsoft'
import {
extractSlackTeamId,
fanOutSlackTokenChain,
getFreshestSlackChain,
isSlackProvider,
} from '@/lib/oauth/slack'
import {
getRecentTerminalError,
isTerminalRefreshError,
Expand Down Expand Up @@ -658,25 +664,48 @@ 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. The lock TTL must not expire under a live refresh
* (15s provider call plus 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.
*/
const SLACK_LOCK_TTL_SEC = 20
const SLACK_FOLLOWER_MAX_WAIT_MS = SLACK_LOCK_TTL_SEC * 1000
Comment thread
TheodoreSpeaks marked this conversation as resolved.
Outdated

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,
Expand All @@ -685,39 +714,78 @@ 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
if (slackTeamId) {
const freshest = await getFreshestSlackChain(slackTeamId)
if (!freshest) {
throw new Error(
`No refresh-capable account row found for Slack installation ${slackTeamId}`
)
}
if (
freshest.accessToken &&
freshest.accessTokenExpiresAt &&
freshest.accessTokenExpiresAt > new Date()
) {
await fanOutSlackTokenChain(slackTeamId, {
accessToken: freshest.accessToken,
refreshToken: freshest.refreshToken,
accessTokenExpiresAt: freshest.accessTokenExpiresAt,
})
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)
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,
})
Comment thread
TheodoreSpeaks marked this conversation as resolved.
Outdated
} 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
Expand Down Expand Up @@ -774,6 +842,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,
Expand Down Expand Up @@ -813,6 +882,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
Expand Down Expand Up @@ -915,6 +985,7 @@ export async function resolveCredentialAccessToken(
accountId: resolvedCredentialId,
providerId: credential.providerId,
refreshToken: credential.refreshToken!,
providerAccountId: credential.accountId,
requestId,
userId: credential.userId,
})
Expand Down Expand Up @@ -1019,6 +1090,7 @@ export async function refreshTokenIfNeeded(
accountId: resolvedCredentialId,
providerId: credential.providerId,
refreshToken: credential.refreshToken!,
providerAccountId: credential.accountId,
requestId,
userId: credential.userId,
})
Expand Down
34 changes: 34 additions & 0 deletions apps/sim/lib/auth/auth.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -444,6 +446,38 @@ 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) {
await fanOutSlackTokenChain(teamId, {
accessToken: account.accessToken,
refreshToken: account.refreshToken ?? null,
accessTokenExpiresAt: account.accessTokenExpiresAt ?? null,
})
Comment thread
cursor[bot] marked this conversation as resolved.
await clearDeadFlag(`slack:${teamId}`)
Comment thread
TheodoreSpeaks marked this conversation as resolved.
Outdated
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,
Expand Down
Loading
Loading