feat(knowledge): refresh mirrored directories on their own clock

Group membership decides who can read an already-indexed document, so it has to
move independently of the corpus: someone leaving a group should lose access in
minutes, not on whatever schedule their documents happen to be re-crawled on.
Until now the only thing that refreshed a directory was the admin crawl itself,
which made the five-minute interval a ceiling rather than a cadence — on a
connector syncing daily, a revoked membership stood for a day, bounded only by
the staleness ratchet.

A scheduler now offers every admin-mode connector each tick and lets
`syncExternalDirectoryGroups` decide whether its directory is actually due. The
crawl keeps its own refresh, which is a floor rather than a duplicate: it is
what guarantees a crawl never publishes grants against membership nobody has
read.

Connectors sharing a directory cost one refresh between them — the first brings
it up to date and the rest skip on the interval gate. Two ticks overlapping on
one directory would both enumerate and write the same rows, which is wasteful
and never wrong, so it takes no lease to prevent; every write on this path is
idempotent, and a lease would be new state to keep correct for no behavioural
gain.

Failure is contained per connector. One workspace whose credential lapsed must
not stop the tick refreshing every other workspace's directory, and a test
covers exactly that.

The refresh moves out of the sync engine so both callers share it, and the
Confluence connector drops a concurrency helper it should never have had —
`mapWithConcurrency` already existed in `lib/core/utils`.

Author: waleedlatif1

apps/sim/app/api/knowledge/connectors/directory-sync/route.test.ts

@@ -0,0 +1,132 @@
+/**
+ * @vitest-environment node
+ */
+import { createMockRequest } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const {
+  mockVerifyCronAuth,
+  mockConnectorRows,
+  mockResolveIdentity,
+  mockResolveToken,
+  mockRefreshDirectory,
+} = vi.hoisted(() => ({
+  mockVerifyCronAuth: vi.fn(() => null),
+  mockConnectorRows: vi.fn(),
+  mockResolveIdentity: vi.fn(),
+  mockResolveToken: vi.fn(),
+  mockRefreshDirectory: vi.fn(),
+}))
+
+vi.mock('@/lib/auth/internal', () => ({ verifyCronAuth: mockVerifyCronAuth }))
+vi.mock('@/lib/credentials/access', () => ({
+  resolveCredentialTokenIdentity: mockResolveIdentity,
+}))
+vi.mock('@/lib/knowledge/connectors/access-token', () => ({
+  resolveConnectorAccessToken: mockResolveToken,
+}))
+vi.mock('@/lib/knowledge/connectors/external-group-sync', () => ({
+  refreshMirroredDirectory: mockRefreshDirectory,
+}))
+vi.mock('@/connectors/registry.server', () => ({
+  CONNECTOR_REGISTRY: {
+    google_drive: { auth: { mode: 'oauth', provider: 'google-drive' }, openDirectory: vi.fn() },
+    notion: { auth: { mode: 'oauth', provider: 'notion' } },
+  },
+}))
+vi.mock('@sim/db', () => ({
+  db: {
+    select: () => ({
+      from: () => ({
+        innerJoin: () => ({ where: () => ({ limit: () => mockConnectorRows() }) }),
+      }),
+    }),
+  },
+}))
+
+import { GET } from '@/app/api/knowledge/connectors/directory-sync/route'
+
+function connector(overrides: Record<string, unknown> = {}) {
+  return {
+    id: 'connector-1',
+    connectorType: 'google_drive',
+    credentialId: 'credential-1',
+    encryptedApiKey: null,
+    sourceConfig: { adminEmail: 'admin@corp.com' },
+    workspaceId: 'ws-1',
+    knowledgeBaseOwnerId: 'owner-1',
+    ...overrides,
+  }
+}
+
+async function run() {
+  const response = await GET(createMockRequest('GET'))
+  return response.json()
+}
+
+describe('connector directory sync scheduler', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockVerifyCronAuth.mockReturnValue(null)
+    mockResolveIdentity.mockResolvedValue({ kind: 'service_account' })
+    mockResolveToken.mockResolvedValue({ accessToken: 'token' })
+    mockRefreshDirectory.mockResolvedValue(undefined)
+  })
+
+  it('refreshes the directory of every admin-mode connector it finds', async () => {
+    mockConnectorRows.mockResolvedValue([connector(), connector({ id: 'connector-2' })])
+
+    await expect(run()).resolves.toMatchObject({ considered: 2, refreshed: 2, failed: 0 })
+    expect(mockRefreshDirectory).toHaveBeenCalledTimes(2)
+  })
+
+  /**
+   * The tick refreshes every workspace's directory, so one workspace's lapsed
+   * credential or unreachable source must not stop the others.
+   */
+  it('contains a failure to the connector that caused it', async () => {
+    mockConnectorRows.mockResolvedValue([connector(), connector({ id: 'connector-2' })])
+    mockRefreshDirectory.mockRejectedValueOnce(new Error('directory unreachable'))
+
+    await expect(run()).resolves.toMatchObject({ considered: 2, refreshed: 1, failed: 1 })
+  })
+
+  it('reports a connector whose credential no longer resolves rather than failing', async () => {
+    mockConnectorRows.mockResolvedValue([connector()])
+    mockResolveToken.mockResolvedValue(null)
+
+    await expect(run()).resolves.toMatchObject({ unusable: 1, refreshed: 0 })
+    expect(mockRefreshDirectory).not.toHaveBeenCalled()
+  })
+
+  it('skips a connector whose source has no directory to read', async () => {
+    mockConnectorRows.mockResolvedValue([connector({ connectorType: 'notion' })])
+
+    await expect(run()).resolves.toMatchObject({ skipped: 1, refreshed: 0 })
+    expect(mockRefreshDirectory).not.toHaveBeenCalled()
+  })
+
+  /**
+   * Token reads are scoped to the credential's own account owner, not the
+   * knowledge base owner, who is routinely a different member.
+   */
+  it('resolves the token as the credential owner for an OAuth credential', async () => {
+    mockConnectorRows.mockResolvedValue([connector()])
+    mockResolveIdentity.mockResolvedValue({ kind: 'oauth', userId: 'credential-owner' })
+
+    await run()
+
+    expect(mockResolveToken).toHaveBeenCalledWith(
+      expect.objectContaining({ userId: 'credential-owner' })
+    )
+  })
+
+  it('refuses an unauthenticated tick', async () => {
+    mockVerifyCronAuth.mockReturnValue(new Response('nope', { status: 401 }))
+
+    const response = await GET(createMockRequest('GET'))
+
+    expect(response.status).toBe(401)
+    expect(mockConnectorRows).not.toHaveBeenCalled()
+  })
+})

apps/sim/app/api/knowledge/connectors/directory-sync/route.ts

@@ -0,0 +1,130 @@
+import { db } from '@sim/db'
+import { knowledgeBase, knowledgeConnector } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import { and, eq, isNull } from 'drizzle-orm'
+import type { NextRequest } from 'next/server'
+import { verifyCronAuth } from '@/lib/auth/internal'
+import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
+import { generateRequestId } from '@/lib/core/utils/request'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { resolveCredentialTokenIdentity } from '@/lib/credentials/access'
+import { resolveConnectorAccessToken } from '@/lib/knowledge/connectors/access-token'
+import { refreshMirroredDirectory } from '@/lib/knowledge/connectors/external-group-sync'
+import { CONNECTOR_REGISTRY } from '@/connectors/registry.server'
+
+export const dynamic = 'force-dynamic'
+
+const logger = createLogger('ConnectorDirectorySyncSchedulerAPI')
+
+/** Directories refreshed per tick, and how many at once. */
+const MAX_DIRECTORIES_PER_TICK = 200
+const REFRESH_CONCURRENCY = 4
+
+/**
+ * Refreshes the external directories that admin-mode connectors mirror.
+ *
+ * Group membership decides who can read an already-indexed document, so it has
+ * to move on its own clock: someone leaving a group should lose access in
+ * minutes, not on whatever schedule the corpus happens to be re-crawled on. The
+ * admin crawl refreshes the directory too — so a crawl can never publish grants
+ * against membership nobody has read — but that is a floor, not the cadence.
+ *
+ * Every eligible connector is offered each tick; `syncExternalDirectoryGroups`
+ * decides whether its directory is actually due. Connectors sharing a directory
+ * therefore cost one refresh between them: the first brings it up to date and
+ * the rest skip. Two ticks overlapping on one directory would both enumerate
+ * and write the same rows — wasteful, never wrong, and not worth a lease to
+ * prevent, since every write here is idempotent.
+ */
+export const GET = withRouteHandler(async (request: NextRequest) => {
+  const requestId = generateRequestId()
+  logger.info(`[${requestId}] Connector directory sync scheduler triggered`)
+
+  const authError = verifyCronAuth(request, 'Connector directory sync scheduler')
+  if (authError) return authError
+
+  const connectors = await db
+    .select({
+      id: knowledgeConnector.id,
+      connectorType: knowledgeConnector.connectorType,
+      credentialId: knowledgeConnector.credentialId,
+      encryptedApiKey: knowledgeConnector.encryptedApiKey,
+      sourceConfig: knowledgeConnector.sourceConfig,
+      workspaceId: knowledgeBase.workspaceId,
+      knowledgeBaseOwnerId: knowledgeBase.userId,
+    })
+    .from(knowledgeConnector)
+    .innerJoin(knowledgeBase, eq(knowledgeConnector.knowledgeBaseId, knowledgeBase.id))
+    .where(
+      and(
+        eq(knowledgeConnector.accessMode, 'admin'),
+        isNull(knowledgeConnector.archivedAt),
+        isNull(knowledgeConnector.deletedAt),
+        isNull(knowledgeBase.deletedAt)
+      )
+    )
+    .limit(MAX_DIRECTORIES_PER_TICK)
+
+  const outcomes = await mapWithConcurrency(connectors, REFRESH_CONCURRENCY, async (connector) => {
+    /**
+     * Every failure is contained here. One workspace whose credential lapsed
+     * must not stop the tick refreshing every other workspace's directory.
+     */
+    try {
+      if (!connector.workspaceId) return 'skipped'
+      const connectorConfig = CONNECTOR_REGISTRY[connector.connectorType]
+      if (!connectorConfig?.openDirectory) return 'skipped'
+
+      /**
+       * The credential's own account owner, not the knowledge base owner —
+       * token reads are scoped to `account.userId`, and a service account
+       * ignores the argument entirely.
+       */
+      let credentialUserId = connector.knowledgeBaseOwnerId
+      if (connector.credentialId) {
+        const identity = await resolveCredentialTokenIdentity(
+          connector.credentialId,
+          connector.workspaceId
+        )
+        if (!identity) return 'unusable'
+        if (identity.kind === 'oauth') credentialUserId = identity.userId
+      }
+
+      const sourceConfig = connector.sourceConfig as Record<string, unknown>
+      const token = await resolveConnectorAccessToken({
+        auth: connectorConfig.auth,
+        connector,
+        userId: credentialUserId,
+        requestId,
+        sourceConfig,
+      })
+      if (!token) return 'unusable'
+
+      await refreshMirroredDirectory({
+        workspaceId: connector.workspaceId,
+        connectorConfig,
+        sourceConfig,
+        syncContext: {},
+        accessToken: token.accessToken,
+      })
+      return 'refreshed'
+    } catch (error) {
+      logger.error(`[${requestId}] Directory refresh failed for a connector`, {
+        connectorId: connector.id,
+        error: getErrorMessage(error),
+      })
+      return 'failed'
+    }
+  })
+
+  const summary = {
+    considered: connectors.length,
+    refreshed: outcomes.filter((outcome) => outcome === 'refreshed').length,
+    skipped: outcomes.filter((outcome) => outcome === 'skipped').length,
+    unusable: outcomes.filter((outcome) => outcome === 'unusable').length,
+    failed: outcomes.filter((outcome) => outcome === 'failed').length,
+  }
+  logger.info(`[${requestId}] Connector directory sync scheduler finished`, summary)
+  return Response.json({ success: true, ...summary })
+})

apps/sim/connectors/confluence/confluence.ts

@@ -5,6 +5,7 @@ import {
   AtlassianSiteNotAccessibleError,
   AtlassianSiteNotMatchedError,
 } from '@/lib/atlassian/discovery'
+import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import {
   type ConfluencePrincipal,
   type ConfluenceRestriction,
@@ -424,23 +425,6 @@ async function resolveConfluenceAcls(
   return acls
 }
 
-/** Runs `worker` over `items`, at most `limit` at a time, preserving no order. */
-async function mapWithConcurrency<T>(
-  items: readonly T[],
-  limit: number,
-  worker: (item: T) => Promise<void>
-): Promise<void> {
-  let cursor = 0
-  const runners = Array.from({ length: Math.min(limit, items.length) }, async () => {
-    for (;;) {
-      const index = cursor++
-      if (index >= items.length) return
-      await worker(items[index])
-    }
-  })
-  await Promise.all(runners)
-}
-
 export const confluenceConnector: ConnectorConfig = {
   ...confluenceConnectorMeta,
 

apps/sim/lib/knowledge/connectors/external-group-sync.ts

@@ -5,7 +5,11 @@ import { getErrorMessage } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { and, eq, isNull, lt, notInArray, or } from 'drizzle-orm'
 import { EXTERNAL_GROUP_SYNC_INTERVAL_MS } from '@/lib/knowledge/access/external-groups'
-import type { ConnectorDirectory, ConnectorDirectoryGroup } from '@/connectors/types'
+import type {
+  ConnectorConfig,
+  ConnectorDirectory,
+  ConnectorDirectoryGroup,
+} from '@/connectors/types'
 
 const logger = createLogger('ExternalGroupSync')
 
@@ -229,3 +233,58 @@ async function pruneRemovedGroups(input: {
     .returning({ id: knowledgeExternalGroup.id })
   return removed.length
 }
+
+/**
+ * Refreshes the directory groups the mirrored ACLs refer to.
+ *
+ * A `g:` token grants nobody until the directory says who is in that group, so
+ * the refresh runs in the same pass that writes the tokens — a crawl can never
+ * publish grants against membership this workspace has never read.
+ *
+ * It is rate-limited on its own clock rather than the connector's, so a
+ * frequently-syncing connector does not re-read the whole directory every run.
+ * A failure is logged rather than thrown: last-known-good membership is still
+ * serving reads, and failing the content sync over it would strand the
+ * documents as well as the groups.
+ */
+export async function refreshMirroredDirectory(input: {
+  workspaceId: string
+  connectorConfig: ConnectorConfig
+  sourceConfig: Record<string, unknown>
+  syncContext: Record<string, unknown>
+  accessToken: string
+}): Promise<void> {
+  const { workspaceId, connectorConfig } = input
+  if (connectorConfig.auth.mode !== 'oauth' || !connectorConfig.openDirectory) return
+
+  try {
+    const directory = await connectorConfig.openDirectory(
+      input.accessToken,
+      input.sourceConfig,
+      input.syncContext
+    )
+    if (!directory) {
+      logger.warn('Skipping directory refresh: the connector names no directory', {
+        workspaceId,
+        connector: connectorConfig.id,
+      })
+      return
+    }
+    const result = await syncExternalDirectoryGroups({
+      workspaceId,
+      providerId: connectorConfig.auth.provider,
+      directory,
+    })
+    logger.info('Refreshed mirrored directory groups', {
+      workspaceId,
+      tenantId: directory.tenantId,
+      ...result,
+    })
+  } catch (error) {
+    logger.error('Directory refresh failed; serving last-known-good group membership', {
+      workspaceId,
+      connector: connectorConfig.id,
+      error: getErrorMessage(error),
+    })
+  }
+}