refactor(knowledge): tighten administrator access after the mechanics review

- directory refresh idempotency keys on the sync interval, so a cron wrapper
  retry cannot start a second walk of the same directory; the scheduler
  offers every admin connector and lets the tenant-level freshness check
  dedupe, since a connector type does not imply one tenant
- config validation seeds the run context from the token, so a Confluence
  service account validates without a discovery call it cannot make
- the domain-share vocabulary (group id, wildcard member, domain fold) lives
  in one module; the Google directory drains pages through the shared
  Google pagination helper; Confluence's getJson owns the not-found branch
- access-mode predicates take the union, and the engine narrows the locked
  row once; runnable statuses are one constant everywhere
- a knowledge base with a mirrored connector reports hasPermissionScopedConnector
- dead export, unused import, and stale docblocks removed

Author: waleedlatif1

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

@@ -18,7 +18,9 @@ vi.mock('@sim/db', () => ({
   db: {
     select: () => ({
       from: () => ({
-        innerJoin: () => ({ where: () => ({ orderBy: () => mockConnectorRows() }) }),
+        innerJoin: () => ({
+          where: () => ({ orderBy: () => ({ limit: () => mockConnectorRows() }) }),
+        }),
       }),
     }),
   },
@@ -27,12 +29,7 @@ vi.mock('@sim/db', () => ({
 import { GET } from '@/app/api/knowledge/connectors/directory-sync/route'
 
 function connector(overrides: Record<string, unknown> = {}) {
-  return {
-    id: 'connector-1',
-    connectorType: 'google_drive',
-    workspaceId: 'ws-1',
-    ...overrides,
-  }
+  return { id: 'connector-1', ...overrides }
 }
 
 async function run() {
@@ -47,50 +44,27 @@ describe('connector directory sync scheduler', () => {
     mockDispatch.mockResolvedValue(undefined)
   })
 
-  it('dispatches one refresh per directory an admin-mode connector mirrors', async () => {
-    mockConnectorRows.mockResolvedValue([
-      connector(),
-      connector({ id: 'connector-2', workspaceId: 'ws-2' }),
-    ])
-
-    await expect(run()).resolves.toMatchObject({
-      considered: 2,
-      directories: 2,
-      dispatched: 2,
-      failed: 0,
-    })
-    expect(mockDispatch).toHaveBeenCalledTimes(2)
-  })
-
   /**
-   * Two connectors of one type in one workspace mirror one directory; walking
-   * it twice per tick would double the Admin SDK cost for nothing.
+   * Every eligible connector is offered under one tick time; the tenant-level
+   * freshness check in the refresh, not the scheduler, decides which walk.
    */
-  it('dispatches once for connectors that share a directory', async () => {
+  it('dispatches a refresh for every admin-mode connector under the same tick', async () => {
     mockConnectorRows.mockResolvedValue([connector(), connector({ id: 'connector-2' })])
 
-    await expect(run()).resolves.toMatchObject({ considered: 2, directories: 1, dispatched: 1 })
-    expect(mockDispatch).toHaveBeenCalledTimes(1)
-    expect(mockDispatch).toHaveBeenCalledWith('connector-1', expect.anything())
+    await expect(run()).resolves.toMatchObject({ considered: 2, dispatched: 2, failed: 0 })
+    expect(mockDispatch).toHaveBeenCalledTimes(2)
+    const [, first] = mockDispatch.mock.calls[0]
+    const [, second] = mockDispatch.mock.calls[1]
+    expect(first.tickAt).toBe(second.tickAt)
   })
 
-  it('contains a dispatch failure to the directory that caused it', async () => {
-    mockConnectorRows.mockResolvedValue([
-      connector(),
-      connector({ id: 'connector-2', workspaceId: 'ws-2' }),
-    ])
+  it('contains a dispatch failure to the connector that caused it', async () => {
+    mockConnectorRows.mockResolvedValue([connector(), connector({ id: 'connector-2' })])
     mockDispatch.mockRejectedValueOnce(new Error('queue unreachable'))
 
     await expect(run()).resolves.toMatchObject({ dispatched: 1, failed: 1 })
   })
 
-  it('skips a connector whose knowledge base has no workspace', async () => {
-    mockConnectorRows.mockResolvedValue([connector({ workspaceId: null })])
-
-    await expect(run()).resolves.toMatchObject({ directories: 0, dispatched: 0 })
-    expect(mockDispatch).not.toHaveBeenCalled()
-  })
-
   it('refuses an unauthenticated tick', async () => {
     mockVerifyCronAuth.mockReturnValue(new Response('nope', { status: 401 }))
 

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

@@ -2,7 +2,7 @@ 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, asc, eq, inArray, isNull } from 'drizzle-orm'
+import { and, asc, eq, inArray, isNotNull, isNull } from 'drizzle-orm'
 import type { NextRequest } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { generateRequestId } from '@/lib/core/utils/request'
@@ -14,7 +14,7 @@ export const dynamic = 'force-dynamic'
 
 const logger = createLogger('ConnectorDirectorySyncSchedulerAPI')
 
-/** Directories dispatched per tick. */
+/** Connectors offered per tick. */
 const MAX_DIRECTORIES_PER_TICK = 200
 
 /**
@@ -26,11 +26,13 @@ const MAX_DIRECTORIES_PER_TICK = 200
  * 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.
  *
- * Connectors sharing a directory cost one refresh between them: the tick
- * dispatches one connector per workspace and provider, and
- * `syncExternalDirectoryGroups` decides whether that directory is actually
- * due. The walk itself runs in the background, like every other connector
- * job, because a large domain takes longer than a scheduler request lives.
+ * Every eligible connector is offered each tick, and
+ * `syncExternalDirectoryGroups` decides whether its directory is actually due:
+ * a tenant is the credential's own site or domain, which the row does not
+ * carry, so connectors sharing one cost a refresh and a skip rather than a
+ * refresh each. The walk itself runs in the background, like every other
+ * connector job, because a large domain takes longer than a scheduler request
+ * lives.
  */
 export const GET = withRouteHandler(async (request: NextRequest) => {
   const requestId = generateRequestId()
@@ -41,11 +43,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
   if (authError) return authError
 
   const connectors = await db
-    .select({
-      id: knowledgeConnector.id,
-      connectorType: knowledgeConnector.connectorType,
-      workspaceId: knowledgeBase.workspaceId,
-    })
+    .select({ id: knowledgeConnector.id })
     .from(knowledgeConnector)
     .innerJoin(knowledgeBase, eq(knowledgeConnector.knowledgeBaseId, knowledgeBase.id))
     .where(
@@ -54,27 +52,16 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         inArray(knowledgeConnector.status, RUNNABLE_CONNECTOR_STATUSES),
         isNull(knowledgeConnector.archivedAt),
         isNull(knowledgeConnector.deletedAt),
-        isNull(knowledgeBase.deletedAt)
+        isNull(knowledgeBase.deletedAt),
+        isNotNull(knowledgeBase.workspaceId)
       )
     )
     .orderBy(asc(knowledgeConnector.createdAt))
-
-  /**
-   * One connector per directory. A connector type implies its provider, and
-   * two connectors of one type in one workspace share a directory by
-   * construction — the first to be created stands for it.
-   */
-  const representatives = new Map<string, string>()
-  for (const connector of connectors) {
-    if (!connector.workspaceId) continue
-    const key = `${connector.workspaceId}:${connector.connectorType}`
-    if (!representatives.has(key)) representatives.set(key, connector.id)
-  }
-  const due = [...representatives.values()].slice(0, MAX_DIRECTORIES_PER_TICK)
+    .limit(MAX_DIRECTORIES_PER_TICK)
 
   let dispatched = 0
   let failed = 0
-  for (const connectorId of due) {
+  for (const { id: connectorId } of connectors) {
     try {
       await dispatchDirectorySync(connectorId, { requestId, tickAt })
       dispatched += 1
@@ -87,7 +74,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     }
   }
 
-  const summary = { considered: connectors.length, directories: due.length, dispatched, failed }
+  const summary = { considered: connectors.length, dispatched, failed }
   logger.info(`[${requestId}] Connector directory sync scheduler finished`, summary)
   return Response.json({ success: true, ...summary })
 })

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

@@ -20,6 +20,7 @@ import {
   MAX_CONSECUTIVE_FAILURES,
   MEMBER_SYNC_STALE_LOCK_TTL_MS,
 } from '@/lib/knowledge/connectors/sync-limits'
+import { RUNNABLE_CONNECTOR_STATUSES } from '@/lib/knowledge/connectors/sync-lock'
 
 export const dynamic = 'force-dynamic'
 
@@ -168,7 +169,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       .where(
         and(
           eq(knowledgeConnector.accessMode, 'members'),
-          inArray(knowledgeConnector.status, ['active', 'error']),
+          inArray(knowledgeConnector.status, RUNNABLE_CONNECTOR_STATUSES),
           inArray(knowledgeConnector.memberSyncStatus, QUEUEABLE_MEMBER_SYNC_STATUSES),
           lte(knowledgeConnector.nextMemberSyncAt, now),
           isNull(knowledgeConnector.archivedAt),

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

@@ -17,6 +17,7 @@ import {
   CONNECTOR_SYNC_STALE_LOCK_TTL_MS,
   MAX_CONSECUTIVE_FAILURES,
 } from '@/lib/knowledge/connectors/sync-limits'
+import { RUNNABLE_CONNECTOR_STATUSES } from '@/lib/knowledge/connectors/sync-lock'
 
 export const dynamic = 'force-dynamic'
 
@@ -304,7 +305,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       .innerJoin(knowledgeBase, eq(knowledgeConnector.knowledgeBaseId, knowledgeBase.id))
       .where(
         and(
-          inArray(knowledgeConnector.status, ['active', 'error']),
+          inArray(knowledgeConnector.status, RUNNABLE_CONNECTOR_STATUSES),
           inArray(knowledgeConnector.accessMode, [...CONTENT_ENGINE_ACCESS_MODES]),
           lte(knowledgeConnector.nextSyncAt, now),
           isNull(knowledgeConnector.archivedAt),

apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connector-access-field/connector-access-field.tsx

@@ -24,7 +24,7 @@ interface ConnectorAccessFieldProps {
   onChange: (value: ConnectorAccessSelection) => void
   /** From `useConnectorMemberGroupOptions`; shared with the modal so both agree on what is required. */
   groupOptions: ConnectorMemberGroupOptions
-  /** Only an admin may put a connector into members mode. */
+  /** Only an admin may move a connector out of workspace mode. */
   canAdmin: boolean
   disabled?: boolean
   /** Whether per-member access may be chosen; false leaves only the way back to workspace access. */
@@ -60,12 +60,15 @@ function accessHint(input: {
 }
 
 /**
- * The Access section of a connector's settings: sync as the workspace, or
- * crawl once per member so each person sees only what the source lets them
- * read. Per-member access needs nothing from the admin: a Credential Group is
- * found or created for the connector's provider, everyone in the workspace is
+ * The Access section of a connector's settings: sync as the workspace; crawl
+ * once per member so each person sees only what the source lets them read; or
+ * crawl once as an administrator and mirror each document's own permissions.
+ * Per-member access needs nothing from the admin: a Credential Group is found
+ * or created for the connector's provider, everyone in the workspace is
  * invited, and each person connects their own account. Only a workspace with
- * several matching groups is asked which one to use.
+ * several matching groups is asked which one to use. Administrator access
+ * needs a connector that can read the source's permissions, and the
+ * administrator it crawls as is part of the connector's own config.
  */
 export function ConnectorAccessField({
   connectorConfig,

apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/edit-connector-modal/edit-connector-modal.tsx

@@ -20,10 +20,8 @@ import {
 import { RefreshCw, SquareArrowUpRight } from '@sim/emcn/icons'
 import { createLogger } from '@sim/logger'
 import { useParams } from 'next/navigation'
-import {
-  type ConnectorAccessMode,
-  isCredentialBackedAccessMode,
-} from '@/lib/knowledge/connectors/access-modes'
+import type { ConnectorAccessMode } from '@/lib/api/contracts/knowledge/connectors'
+import { isCredentialBackedAccessMode } from '@/lib/knowledge/connectors/access-modes'
 import { getProviderIdFromServiceId, type OAuthProvider } from '@/lib/oauth'
 import {
   ConnectorAccessField,

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/knowledge-base-selector/knowledge-base-selector.tsx

@@ -184,7 +184,9 @@ export function KnowledgeBaseSelector({
   const label =
     subBlock.placeholder || (isMultiSelect ? 'Select knowledge bases' : 'Select knowledge base')
 
-  const hasMemberScopedSelection = selectedKnowledgeBases.some((kb) => kb.hasMemberScopedConnector)
+  const hasMemberScopedSelection = selectedKnowledgeBases.some(
+    (kb) => kb.hasPermissionScopedConnector
+  )
 
   return (
     <div className='w-full'>

apps/sim/connectors/confluence/confluence.ts

@@ -679,7 +679,8 @@ export const confluenceConnector: ConnectorConfig = {
 
   validateConfig: async (
     accessToken: string,
-    sourceConfig: Record<string, unknown>
+    sourceConfig: Record<string, unknown>,
+    syncContext?: Record<string, unknown>
   ): Promise<{ valid: boolean; error?: string }> => {
     const domain = sourceConfig.domain as string
     const spaceKeys = parseMultiValue(sourceConfig.spaceKey)
@@ -694,7 +695,11 @@ export const confluenceConnector: ConnectorConfig = {
     }
 
     try {
-      const cloudId = await getConfluenceCloudId(domain, accessToken, VALIDATE_RETRY_OPTIONS)
+      const seededCloudId = syncContext?.cloudId
+      const cloudId =
+        typeof seededCloudId === 'string' && seededCloudId
+          ? seededCloudId
+          : await getConfluenceCloudId(domain, accessToken, VALIDATE_RETRY_OPTIONS)
       const params = new URLSearchParams()
       for (const key of spaceKeys) params.append('keys', key)
       params.append('limit', String(Math.max(spaceKeys.length, 1)))

apps/sim/connectors/confluence/permissions.ts

@@ -22,12 +22,26 @@ function apiBase(cloudId: string): string {
   return `https://api.atlassian.com/ex/confluence/${cloudId}/wiki`
 }
 
-/** A GET with the same transient-error retry every other Confluence call gets. */
-async function getJson<T>(url: string, accessToken: string): Promise<T> {
+/**
+ * A GET with the same transient-error retry every other Confluence call gets.
+ * With `allowNotFound`, a 404 resolves to null instead of throwing.
+ */
+async function getJson<T>(url: string, accessToken: string): Promise<T>
+async function getJson<T>(
+  url: string,
+  accessToken: string,
+  options: { allowNotFound: true }
+): Promise<T | null>
+async function getJson<T>(
+  url: string,
+  accessToken: string,
+  options?: { allowNotFound: true }
+): Promise<T | null> {
   const response = await fetchWithRetry(url, {
     method: 'GET',
     headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json' },
   })
+  if (response.status === 404 && options?.allowNotFound) return null
   if (!response.ok) {
     throw new Error(`Confluence request failed: ${response.status} ${response.statusText}`)
   }
@@ -79,7 +93,7 @@ async function drainV1<T>(url: string, accessToken: string, what: string): Promi
     const results = body.results ?? []
     items.push(...results)
     if (!body._links?.next || results.length === 0) return items
-    start += body.size ?? results.length
+    start += body.size || results.length
   }
   throw new Error(`Confluence ${what} exceeded ${MAX_PAGES} pages`)
 }
@@ -249,19 +263,12 @@ export async function describeContent(
 ): Promise<{ spaceId: string; contentType: 'page' | 'blogpost' } | null> {
   for (const contentType of ['page', 'blogpost'] as const) {
     const collection = contentType === 'page' ? 'pages' : 'blogposts'
-    const response = await fetchWithRetry(
+    const body = await getJson<{ spaceId?: string | number }>(
       `${apiBase(cloudId)}/api/v2/${collection}/${encodeURIComponent(contentId)}`,
-      {
-        method: 'GET',
-        headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json' },
-      }
+      accessToken,
+      { allowNotFound: true }
     )
-    if (response.status === 404) continue
-    if (!response.ok) {
-      throw new Error(`Confluence request failed: ${response.status} ${response.statusText}`)
-    }
-    const body = (await response.json()) as { spaceId?: string | number }
-    if (body.spaceId !== undefined) return { spaceId: String(body.spaceId), contentType }
+    if (body?.spaceId !== undefined) return { spaceId: String(body.spaceId), contentType }
   }
   return null
 }
@@ -322,7 +329,7 @@ export async function resolveUserEmails(
 }
 
 /** Every group on the site, by the id its permissions and restrictions name. */
-export async function listSiteGroups(
+async function listSiteGroups(
   cloudId: string,
   accessToken: string
 ): Promise<ConnectorDirectoryGroup[]> {
@@ -360,7 +367,7 @@ export async function listGroupMemberEmails(
     accessToken,
     'group membership'
   )
-  const accountIds = [...new Set(members.map((m) => m.accountId).filter(Boolean) as string[])]
+  const accountIds = [...new Set(members.flatMap((m) => (m.accountId ? [m.accountId] : [])))]
   const emails = await resolveUserEmails(cloudId, accessToken, accountIds)
   return {
     group,