Function withMcpOauthRefreshLock

apps/sim/lib/mcp/oauth/storage.ts:272–310

(rowId: string, fn: () => Promise<T>)

const inflightChains = new Map<string, Promise<unknown>>()

export async function withMcpOauthRefreshLock<T>(rowId: string, fn: () => Promise<T>): Promise<T> {
  const lockKey = `mcp:oauth:refresh:${rowId}`
  const prev = inflightChains.get(lockKey) ?? Promise.resolve()
  const prevSettled = prev.catch(() => undefined)

  let queueTimedOut = false
  const next = prevSettled.then(() => {
    if (queueTimedOut) {
      throw new Error(`MCP OAuth refresh queue for ${rowId} abandoned after timeout`)
    }
    return runWithRedisMutex(lockKey, rowId, fn)
  })
  inflightChains.set(lockKey, next)
  const cleanup = () => {
    if (inflightChains.get(lockKey) === next) inflightChains.delete(lockKey)
  }
  next.then(cleanup, cleanup)

  let queueTimer: ReturnType<typeof setTimeout> | undefined
  const queueDeadline = new Promise<never>((_, reject) => {
    queueTimer = setTimeout(() => {
      queueTimedOut = true
      reject(
        new Error(
          `MCP OAuth refresh queue for ${rowId} stalled for ${REFRESH_QUEUE_WAIT_TIMEOUT_MS}ms`
        )
      )
    }, REFRESH_QUEUE_WAIT_TIMEOUT_MS)
    queueTimer.unref?.()
  })

  try {
    await Promise.race([prevSettled, queueDeadline])
  } finally {
    clearTimeout(queueTimer)
  }

  return next
}

async function runWithRedisMutex<T>(
  lockKey: string,