Function processDocumentsWithQueue

apps/sim/lib/knowledge/documents/service.ts:403–432

(
  createdDocuments: DocumentData[],
  knowledgeBaseId: string,
  processingOptions: ProcessingOptions,
  requestId: string
)

 * partial failures are logged and recovered by the next sync's stuck-doc pass.
 */
export async function processDocumentsWithQueue(
  createdDocuments: DocumentData[],
  knowledgeBaseId: string,
  processingOptions: ProcessingOptions,
  requestId: string
): Promise<void> {
  if (createdDocuments.length === 0) return

  const jobPayloads = createdDocuments.map((doc) =>
    buildJobPayload(doc, knowledgeBaseId, processingOptions, requestId)
  )

  const useTrigger = isTriggerAvailable()
  logger.info(
    `[${requestId}] Dispatching background processing for ${jobPayloads.length} documents`,
    { backend: useTrigger ? 'trigger-dev' : 'direct' }
  )

  const dispatched = useTrigger
    ? await dispatchViaBatchTrigger(jobPayloads, requestId)
    : await dispatchInProcess(jobPayloads, requestId)

  logger.info(
    `[${requestId}] Document dispatch complete: ${dispatched}/${jobPayloads.length} succeeded`
  )

  if (dispatched === 0) {
    throw new Error(`All ${jobPayloads.length} document processing dispatches failed`)
  }
}

async function dispatchViaBatchTrigger(
  jobPayloads: DocumentProcessingPayload[],