improvement(knowledge): batch trigger dispatch, prune redundant DB roundtrips

Connector sync was dispatching Trigger.dev document-processing jobs one
HTTP roundtrip at a time. processDocumentsWithQueue now uses
tasks.batchTrigger when Trigger.dev is available, collapsing N roundtrips
to ceil(N/1000). Idempotency keys protect against duplicate runs on retry.

Also trims DB roundtrips inside the sync loop:
- Per-batch isConnectorDeleted + isKnowledgeBaseDeleted collapsed into a
  single checkSyncLiveness JOIN (one SELECT instead of two per batch).
- Dropped redundant pre-upload isKnowledgeBaseDeleted checks from
  addDocument/updateDocument: the batch-boundary liveness check already
  catches pre-batch deletions and the in-tx FOR UPDATE is authoritative
  for races during the batch.
- Removed dead processDocumentsWithTrigger helper (never called).

Author: waleedlatif1

apps/sim/lib/knowledge/connectors/sync-engine.ts

@@ -58,22 +58,33 @@ type DocOp =
   | { type: 'add'; extDoc: ExternalDocument }
   | { type: 'update'; existingId: string; extDoc: ExternalDocument }
 
-async function isConnectorDeleted(connectorId: string): Promise<boolean> {
+/**
+ * Combined liveness check used between batches. One JOIN query checks both
+ * connector and knowledge base state in a single roundtrip.
+ */
+async function checkSyncLiveness(
+  connectorId: string,
+  knowledgeBaseId: string
+): Promise<{ connectorDeleted: boolean; knowledgeBaseDeleted: boolean }> {
   const rows = await db
-    .select({ archivedAt: knowledgeConnector.archivedAt, deletedAt: knowledgeConnector.deletedAt })
+    .select({
+      connectorArchivedAt: knowledgeConnector.archivedAt,
+      connectorDeletedAt: knowledgeConnector.deletedAt,
+      kbDeletedAt: knowledgeBase.deletedAt,
+    })
     .from(knowledgeConnector)
-    .where(eq(knowledgeConnector.id, connectorId))
+    .innerJoin(knowledgeBase, eq(knowledgeBase.id, knowledgeConnector.knowledgeBaseId))
+    .where(and(eq(knowledgeConnector.id, connectorId), eq(knowledgeBase.id, knowledgeBaseId)))
     .limit(1)
-  return rows.length === 0 || rows[0].archivedAt !== null || rows[0].deletedAt !== null
-}
 
-async function isKnowledgeBaseDeleted(knowledgeBaseId: string): Promise<boolean> {
-  const rows = await db
-    .select({ deletedAt: knowledgeBase.deletedAt })
-    .from(knowledgeBase)
-    .where(eq(knowledgeBase.id, knowledgeBaseId))
-    .limit(1)
-  return rows.length === 0 || rows[0].deletedAt !== null
+  if (rows.length === 0) {
+    return { connectorDeleted: true, knowledgeBaseDeleted: true }
+  }
+  const row = rows[0]
+  return {
+    connectorDeleted: row.connectorArchivedAt !== null || row.connectorDeletedAt !== null,
+    knowledgeBaseDeleted: row.kbDeletedAt !== null,
+  }
 }
 
 async function isKnowledgeBaseActiveInTx(
@@ -502,10 +513,11 @@ export async function executeSync(
     }
 
     for (let i = 0; i < pendingOps.length; i += SYNC_BATCH_SIZE) {
-      if (await isConnectorDeleted(connectorId)) {
+      const liveness = await checkSyncLiveness(connectorId, connector.knowledgeBaseId)
+      if (liveness.connectorDeleted) {
         throw new ConnectorDeletedException(connectorId)
       }
-      if (await isKnowledgeBaseDeleted(connector.knowledgeBaseId)) {
+      if (liveness.knowledgeBaseDeleted) {
         throw new Error(`Knowledge base ${connector.knowledgeBaseId} was deleted during sync`)
       }
 
@@ -642,11 +654,12 @@ export async function executeSync(
       }
     }
 
-    // Check if connector was deleted before retrying stuck documents
-    if (await isConnectorDeleted(connectorId)) {
+    // Check if connector/KB were deleted before retrying stuck documents
+    const postBatchLiveness = await checkSyncLiveness(connectorId, connector.knowledgeBaseId)
+    if (postBatchLiveness.connectorDeleted) {
       throw new ConnectorDeletedException(connectorId)
     }
-    if (await isKnowledgeBaseDeleted(connector.knowledgeBaseId)) {
+    if (postBatchLiveness.knowledgeBaseDeleted) {
       throw new Error(`Knowledge base ${connector.knowledgeBaseId} was deleted during sync`)
     }
 
@@ -881,9 +894,6 @@ async function addDocument(
   extDoc: ExternalDocument,
   sourceConfig?: Record<string, unknown>
 ): Promise<DocumentData> {
-  if (await isKnowledgeBaseDeleted(knowledgeBaseId)) {
-    throw new Error(`Knowledge base ${knowledgeBaseId} is deleted`)
-  }
   const documentId = generateId()
   const contentBuffer = Buffer.from(extDoc.content, 'utf-8')
   const safeTitle = sanitizeStorageTitle(extDoc.title)
@@ -963,9 +973,6 @@ async function updateDocument(
   extDoc: ExternalDocument,
   sourceConfig?: Record<string, unknown>
 ): Promise<DocumentData> {
-  if (await isKnowledgeBaseDeleted(knowledgeBaseId)) {
-    throw new Error(`Knowledge base ${knowledgeBaseId} is deleted`)
-  }
   // Fetch old file URL before uploading replacement
   const existingRows = await db
     .select({ fileUrl: document.fileUrl })

apps/sim/lib/knowledge/documents/service.ts

@@ -48,7 +48,6 @@ import { estimateTokenCount } from '@/lib/tokenization/estimators'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
 import { extractStorageKey } from '@/lib/uploads/utils/file-utils'
 import { getWorkspaceBilledAccountUserId } from '@/lib/workspaces/utils'
-import type { DocumentProcessingPayload } from '@/background/knowledge-processing'
 import { calculateCost } from '@/providers/utils'
 
 const logger = createLogger('DocumentService')
@@ -314,12 +313,16 @@ async function processDocumentTags(
   return result
 }
 
+const TRIGGER_BATCH_SIZE = 1000
+
 export async function processDocumentsWithQueue(
   createdDocuments: DocumentData[],
   knowledgeBaseId: string,
   processingOptions: ProcessingOptions,
   requestId: string
 ): Promise<void> {
+  if (createdDocuments.length === 0) return
+
   const jobPayloads = createdDocuments.map<DocumentJobData>((doc) => ({
     knowledgeBaseId,
     documentId: doc.documentId,
@@ -333,13 +336,52 @@ export async function processDocumentsWithQueue(
     requestId,
   }))
 
+  const useTrigger = isTriggerAvailable()
   logger.info(
     `[${requestId}] Dispatching background processing for ${jobPayloads.length} documents`,
-    {
-      backend: isTriggerAvailable() ? 'trigger-dev' : 'direct',
-    }
+    { backend: useTrigger ? 'trigger-dev' : 'direct' }
   )
 
+  if (useTrigger) {
+    /**
+     * Single batched dispatch per chunk of up to TRIGGER_BATCH_SIZE — collapses
+     * N HTTP roundtrips to ceil(N / TRIGGER_BATCH_SIZE). Idempotency keys allow
+     * safe re-dispatch on retry without duplicating runs.
+     */
+    let dispatched = 0
+    for (let i = 0; i < jobPayloads.length; i += TRIGGER_BATCH_SIZE) {
+      const chunk = jobPayloads.slice(i, i + TRIGGER_BATCH_SIZE)
+      try {
+        await tasks.batchTrigger(
+          'knowledge-process-document',
+          chunk.map((payload) => ({
+            payload,
+            options: {
+              idempotencyKey: `doc-process-${payload.documentId}-${requestId}`,
+              tags: [
+                `knowledgeBaseId:${payload.knowledgeBaseId}`,
+                `documentId:${payload.documentId}`,
+              ],
+            },
+          }))
+        )
+        dispatched += chunk.length
+      } catch (error) {
+        logger.error(`[${requestId}] Failed to batchTrigger ${chunk.length} document jobs`, {
+          error: getErrorMessage(error),
+        })
+        if (dispatched === 0 && i + TRIGGER_BATCH_SIZE >= jobPayloads.length) {
+          throw new Error(`All ${jobPayloads.length} document processing dispatches failed`)
+        }
+      }
+    }
+
+    logger.info(
+      `[${requestId}] Document dispatch complete: ${dispatched}/${jobPayloads.length} succeeded`
+    )
+    return
+  }
+
   const results = await Promise.allSettled(
     jobPayloads.map((payload) => dispatchDocumentProcessingJob(payload))
   )
@@ -358,8 +400,6 @@ export async function processDocumentsWithQueue(
   if (failures.length === results.length) {
     throw new Error(`All ${failures.length} document processing dispatches failed`)
   }
-
-  return
 }
 
 export async function processDocumentAsync(
@@ -698,54 +738,6 @@ export function isTriggerAvailable(): boolean {
   return Boolean(env.TRIGGER_SECRET_KEY) && isTriggerDevEnabled
 }
 
-async function processDocumentsWithTrigger(
-  documents: DocumentProcessingPayload[],
-  requestId: string
-): Promise<{ success: boolean; message: string; batchIds?: string[] }> {
-  if (!isTriggerAvailable()) {
-    throw new Error('Trigger.dev is not configured - TRIGGER_SECRET_KEY missing')
-  }
-
-  try {
-    logger.info(`[${requestId}] Triggering background processing for ${documents.length} documents`)
-
-    const MAX_BATCH_SIZE = 1000
-    const batchIds: string[] = []
-
-    for (let i = 0; i < documents.length; i += MAX_BATCH_SIZE) {
-      const chunk = documents.slice(i, i + MAX_BATCH_SIZE)
-      const batchResult = await tasks.batchTrigger(
-        'knowledge-process-document',
-        chunk.map((doc) => ({
-          payload: doc,
-          options: {
-            idempotencyKey: `doc-process-${doc.documentId}-${requestId}`,
-            tags: [`knowledgeBaseId:${doc.knowledgeBaseId}`, `documentId:${doc.documentId}`],
-          },
-        }))
-      )
-      batchIds.push(batchResult.batchId)
-    }
-
-    logger.info(
-      `[${requestId}] Triggered ${documents.length} document processing jobs in ${batchIds.length} batch(es)`
-    )
-
-    return {
-      success: true,
-      message: `${documents.length} document processing jobs triggered`,
-      batchIds,
-    }
-  } catch (error) {
-    logger.error(`[${requestId}] Failed to trigger document processing jobs:`, error)
-
-    return {
-      success: false,
-      message: getErrorMessage(error, 'Failed to trigger background jobs'),
-    }
-  }
-}
-
 export async function createDocumentRecords(
   documents: Array<{
     filename: string