improvement(search): bound indexing and simplify source recovery

Author: waleedlatif1

.github/workflows/test-build.yml

@@ -153,6 +153,16 @@ jobs:
           lib/table/rows/secret-provenance.postgres.test.ts
           lib/memory/message-provenance.postgres.test.ts
 
+      - name: Verify Search dispatch, progress, and pagination in PostgreSQL
+        working-directory: apps/sim
+        env:
+          KNOWLEDGE_ACL_TEST_DATABASE_URL: postgresql://postgres:postgres@127.0.0.1:5432/sim_auth_scim
+        run: >-
+          bunx vitest run --mode integration
+          lib/knowledge/__integration__/document-dispatch.integration.ts
+          lib/knowledge/__integration__/search-source-progress.integration.ts
+          lib/knowledge/__integration__/search-source-pagination.integration.ts
+
   test-build:
     name: Lint and Test
     runs-on: ${{ (vars.CI_PROVIDER == '' || vars.CI_PROVIDER == 'blacksmith') && 'blacksmith-8vcpu-ubuntu-2404' || 'ubuntu-latest' }}

apps/docs/content/docs/platform/self-hosting/background-jobs.mdx

@@ -5,7 +5,7 @@ description: Scheduled workflows, polling triggers, and the cron endpoints that
 
 import { Callout } from 'fumadocs-ui/components/callout'
 
-A large part of Sim runs on a schedule rather than in response to a user request: scheduled workflows, every polling trigger, connector syncs, the outbox, data drains, and retention. All of it is driven by **HTTP endpoints that something external must call on a timer**.
+A large part of Sim runs on a schedule rather than in response to a user request: scheduled workflows, every polling trigger, connector syncs, the outbox, data drains, and retention. The default self-hosted scheduler calls **HTTP endpoints on a timer**. Deployments using Trigger.dev also run the document-indexing tasks described below.
 
 Both deployments ship a scheduler and enable it by default: Kubernetes as CronJobs, Docker Compose as a `cron` service. Both authenticate with `CRON_SECRET`, and both use the same schedules.
 
@@ -146,3 +146,16 @@ Scheduled execution volume is bounded per app instance by:
 </Callout>
 
 Raise the schedule limit only alongside memory headroom: concurrent executions run in the app process, so throughput is bounded by the pod's memory before it is bounded by this number.
+
+## Trigger.dev document indexing
+
+If Trigger.dev is enabled, document indexing uses a durable dispatch queue. Large imports wait in the database, and organizations share the existing worker budget. Temporary provider-capacity waits retry later without keeping a worker occupied. Default self-hosted installs continue to process documents in the app.
+
+When upgrading a deployment that uses Trigger.dev:
+
+1. Apply the database migrations before starting the new workers or application.
+2. Deploy the Trigger tasks from the same revision before deploying the application. New workers accept existing queued jobs as well as the new compact document references.
+3. Confirm the application and workers use the same database and Trigger environment. Verify that the `knowledge-document-dispatch-schedule` task is enabled; it runs every minute to resume delayed work and recover missed dispatch notifications.
+4. Sync a small source and confirm its documents become searchable. A successful sync only confirms the source was fetched; indexing can still be running.
+
+Enqueue and completion notifications normally start the next work promptly. The minute schedule provides recovery when a notification is missed; it does not repeatedly download documents. Keep the updated workers running until their durable queue has drained before rolling them back.

apps/docs/content/docs/search/index.mdx

@@ -95,6 +95,14 @@ Workspace Search remains separate. Workspace admins add sources through **Search
 3. Ask a teammate with different source access to repeat the search. Documents restricted to you should not appear for them.
 4. Change or remove a test document's access in the source and check again after the next completed content and permission refresh.
 
-Search runs background syncs on an hourly schedule. Large sources, provider limits, and indexing queues can delay completion. Results are indexed copies, so edits and access changes are not fetched live for every query. Admins can inspect errors and progress under **Manage**.
+Search runs background syncs on an hourly schedule. Large sources, provider limits, and indexing queues can delay completion. Results are indexed copies, so edits and access changes are not fetched live for every query.
+
+## If indexing needs attention
+
+A completed sync means the source was checked; some documents may still be indexing. The source row shows how many documents you can search and whether indexing failed for any documents you can access.
+
+As an admin, open **Manage → Settings → Documents → Failed** to inspect those files. Select **Retry indexing** beside a file to try again. **Exclude** removes a file from search; use the **Excluded** tab and **Restore** to include it again. Fix a disconnected account or source configuration before retrying a sync that needs attention.
+
+Temporary indexing-capacity waits retry automatically in the background. If automatic retries are exhausted, the document appears as failed so an admin can retry it. Other searchable documents remain available while this work finishes.
 
 These guides cover permission-aware Search sources. For a general knowledge base used by workflows, see [Knowledge-base connectors](/knowledgebase/connectors); its workspace access settings are a separate choice.

apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/documents/route.ts

@@ -25,11 +25,12 @@ export const GET = defineInternalJsonRoute({
     knowledgeBaseId: params.id,
     connectorId: params.connectorId,
     includeExcluded: query.includeExcluded,
+    failedOnly: query.failedOnly,
     limit: query.limit,
     offset: query.offset,
   }),
   useCase: listKnowledgeConnectorDocuments,
-  present: ({ documents, counts }) => ({
+  present: ({ documents, counts, hasMore }) => ({
     success: true as const,
     data: {
       documents: documents.map((document) => ({
@@ -38,6 +39,7 @@ export const GET = defineInternalJsonRoute({
         uploadedAt: document.uploadedAt.toISOString(),
       })),
       counts,
+      hasMore,
     },
   }),
 })

apps/sim/app/api/knowledge/migrated-routes.test.ts

@@ -299,7 +299,8 @@ describe('migrated internal Knowledge routes', () => {
           processingStatus: 'completed',
         },
       ],
-      counts: { active: 1, excluded: 0 },
+      counts: { active: 1, excluded: 0, failed: 0 },
+      hasMore: false,
     })
     const params = Promise.resolve({ id: 'knowledge-1', connectorId: 'connector-1' })
     const listResponse = await listConnectorDocuments(
@@ -314,12 +315,18 @@ describe('migrated internal Knowledge routes', () => {
         documents: [
           expect.objectContaining({ id: 'document-1', uploadedAt: '2026-01-01T00:00:00.000Z' }),
         ],
-        counts: { active: 1, excluded: 0 },
+        counts: { active: 1, excluded: 0, failed: 0 },
+        hasMore: false,
       },
     })
     expect(mocks.listConnectorDocuments).toHaveBeenLastCalledWith(
       expect.objectContaining({
-        input: expect.objectContaining({ includeExcluded: true, limit: 25, offset: 50 }),
+        input: expect.objectContaining({
+          includeExcluded: true,
+          failedOnly: false,
+          limit: 25,
+          offset: 50,
+        }),
       })
     )
 

apps/sim/app/api/knowledge/sim-search/sources/overview/route.ts

@@ -0,0 +1,23 @@
+import { readSearchSourceOverviewContract } from '@/lib/api/contracts/knowledge/connectors'
+import {
+  defineInternalJsonRoute,
+  internalRateLimits,
+  internalSessionAuth,
+} from '@/lib/api/server/routes'
+import { internalKnowledgeErrorPolicies } from '@/lib/knowledge/api/route-policies'
+import { knowledgeOperations } from '@/lib/knowledge/application/operations'
+import { readSearchSourceOverview } from '@/lib/knowledge/application/search-source-overview'
+
+export const GET = defineInternalJsonRoute({
+  contract: readSearchSourceOverviewContract,
+  auth: internalSessionAuth,
+  operation: knowledgeOperations.readSearchSourceOverview,
+  rateLimit: internalRateLimits.none({
+    reason: 'Bounded provider existence probes for source setup and indexing progress',
+  }),
+  errorPolicy: internalKnowledgeErrorPolicies.connectors,
+  mapInput: ({ query }) => query,
+  useCase: readSearchSourceOverview,
+  present: (overview) => ({ success: true as const, data: overview }),
+  staticResponseHeaders: { 'Cache-Control': 'private, no-store' },
+})

apps/sim/app/api/knowledge/sim-search/sources/progress/route.ts

@@ -0,0 +1,23 @@
+import { readSearchSourceProgressContract } from '@/lib/api/contracts/knowledge/connectors'
+import {
+  defineInternalJsonRoute,
+  internalRateLimits,
+  internalSessionAuth,
+} from '@/lib/api/server/routes'
+import { internalKnowledgeErrorPolicies } from '@/lib/knowledge/api/route-policies'
+import { knowledgeOperations } from '@/lib/knowledge/application/operations'
+import { readSearchSourceProgress } from '@/lib/knowledge/application/search-source-progress'
+
+export const POST = defineInternalJsonRoute({
+  contract: readSearchSourceProgressContract,
+  auth: internalSessionAuth,
+  operation: knowledgeOperations.readSearchSourceProgress,
+  rateLimit: internalRateLimits.none({
+    reason: 'Bounded viewer-authorized indexing progress polling',
+  }),
+  errorPolicy: internalKnowledgeErrorPolicies.connectors,
+  mapInput: ({ body }) => body,
+  useCase: readSearchSourceProgress,
+  present: ({ sources }) => ({ success: true as const, data: sources }),
+  staticResponseHeaders: { 'Cache-Control': 'private, no-store' },
+})

apps/sim/app/api/knowledge/sim-search/sources/route.test.ts

@@ -2,7 +2,7 @@
 import { authMockFns, createMockRequest } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const mocks = vi.hoisted(() => ({ execute: vi.fn(), connect: vi.fn() }))
+const mocks = vi.hoisted(() => ({ execute: vi.fn(), connect: vi.fn(), overview: vi.fn() }))
 vi.mock('@/lib/knowledge/application/sim-search', () => ({
   connectSimSearchConnector: {
     operation: { id: 'knowledge.simSearch.connect' },
@@ -12,6 +12,12 @@ vi.mock('@/lib/knowledge/application/sim-search', () => ({
 vi.mock('@/lib/knowledge/application/search-sources', () => ({
   listSearchSources: { operation: { id: 'knowledge.search.sources.list' }, execute: mocks.execute },
 }))
+vi.mock('@/lib/knowledge/application/search-source-overview', () => ({
+  readSearchSourceOverview: {
+    operation: { id: 'knowledge.search.sources.overview' },
+    execute: mocks.overview,
+  },
+}))
 vi.mock('@/lib/knowledge/application/search', () => ({
   KnowledgeSearchProvenanceUnavailableError: class extends Error {},
 }))
@@ -21,6 +27,7 @@ vi.mock('@/lib/knowledge/application/upload-sessions', () => ({
 
 import { NoWorkspaceAccessError } from '@/lib/core/application/workspace-authorization'
 import { POST as connectSource } from '@/app/api/knowledge/sim-search/connect/route'
+import { GET as getOverview } from '@/app/api/knowledge/sim-search/sources/overview/route'
 import { GET } from '@/app/api/knowledge/sim-search/sources/route'
 
 const WORKSPACE_ID = '7d28e5e2-fb03-4118-9c52-4ab77ccff369'
@@ -36,6 +43,7 @@ const source = {
   lastSyncAt: null,
   hasSyncError: false,
   viewerDocumentCount: 0,
+  viewerFailedDocumentCount: 0,
   viewerEmailVerified: true,
   connectionRequired: false,
   viewerMembership: null,
@@ -47,7 +55,7 @@ beforeEach(() => {
     user: { id: 'reader' },
     session: { id: 'session' },
   })
-  mocks.execute.mockResolvedValue({ sources: [source] })
+  mocks.execute.mockResolvedValue({ sources: [source], nextCursor: null })
 })
 
 describe('GET Search sources', () => {
@@ -93,6 +101,7 @@ describe('GET Search sources', () => {
 
   it('passes the authenticated subject into the registered operation and projects only the contract fields', async () => {
     mocks.execute.mockResolvedValue({
+      nextCursor: null,
       sources: [
         {
           ...source,
@@ -113,8 +122,8 @@ describe('GET Search sources', () => {
     expect(response.status).toBe(200)
     expect(response.headers.get('Cache-Control')).toBe('private, no-store')
     const body = await response.json()
-    expect(body).toMatchObject({ success: true, data: [source] })
-    expect(body.data[0]).toEqual(source)
+    expect(body).toMatchObject({ success: true, data: { sources: [source], nextCursor: null } })
+    expect(body.data.sources[0]).toEqual(source)
     expect(mocks.execute).toHaveBeenCalledWith(
       expect.objectContaining({
         principal: { kind: 'session', userId: 'reader', sessionId: 'session' },
@@ -153,3 +162,73 @@ describe('GET Search sources', () => {
     expect(body).not.toHaveProperty('data')
   })
 })
+
+describe('Search pagination boundary', () => {
+  it('forwards bounded source filters and opaque cursors to the authorized operation', async () => {
+    const response = await GET(
+      createMockRequest(
+        'GET',
+        undefined,
+        {},
+        `http://localhost/api/knowledge/sim-search/sources?workspaceId=${WORKSPACE_ID}&search=Handbook&mine=true&cursor=opaque`
+      )
+    )
+    expect(response.status).toBe(200)
+    expect(mocks.execute).toHaveBeenCalledWith(
+      expect.objectContaining({
+        input: { workspaceId: WORKSPACE_ID, search: 'Handbook', mine: true, cursor: 'opaque' },
+      })
+    )
+  })
+  it.each([`search=${'x'.repeat(201)}`, `cursor=${'x'.repeat(1025)}`])(
+    'rejects an oversized filter or cursor before source reads',
+    async (filter) => {
+      const response = await GET(
+        createMockRequest(
+          'GET',
+          undefined,
+          {},
+          `http://localhost/api/knowledge/sim-search/sources?workspaceId=${WORKSPACE_ID}&${filter}`
+        )
+      )
+      expect(response.status).toBe(400)
+      expect(mocks.execute).not.toHaveBeenCalled()
+    }
+  )
+  it('serves only the bounded provider overview through the registered operation', async () => {
+    mocks.overview.mockResolvedValue({
+      providers: [{ connectorType: 'google_drive', isSyncing: true }],
+      hasSearchableDocuments: false,
+      credentials: 'private',
+    })
+    const response = await getOverview(
+      createMockRequest(
+        'GET',
+        undefined,
+        {},
+        `http://localhost/api/knowledge/sim-search/sources/overview?workspaceId=${WORKSPACE_ID}`
+      )
+    )
+    expect(response.status).toBe(200)
+    expect(response.headers.get('Cache-Control')).toBe('private, no-store')
+    expect(await response.json()).toEqual({
+      success: true,
+      data: {
+        providers: [{ connectorType: 'google_drive', isSyncing: true }],
+        hasSearchableDocuments: false,
+      },
+    })
+    expect(mocks.overview).toHaveBeenCalledWith(
+      expect.objectContaining({
+        principal: { kind: 'session', userId: 'reader', sessionId: 'session' },
+        input: { workspaceId: WORKSPACE_ID },
+      })
+    )
+  })
+  it('authenticates before parsing the overview scope', async () => {
+    authMockFns.mockGetSession.mockResolvedValue(null)
+    const response = await getOverview(createMockRequest('GET'))
+    expect(response.status).toBe(401)
+    expect(mocks.overview).not.toHaveBeenCalled()
+  })
+})

apps/sim/app/api/knowledge/sim-search/sources/route.ts

@@ -18,6 +18,6 @@ export const GET = defineInternalJsonRoute({
   errorPolicy: internalKnowledgeErrorPolicies.connectors,
   mapInput: ({ query }) => query,
   useCase: listSearchSources,
-  present: ({ sources }) => ({ success: true as const, data: sources }),
+  present: (page) => ({ success: true as const, data: page }),
   staticResponseHeaders: { 'Cache-Control': 'private, no-store' },
 })