fix(knowledge): keep a failed dispatch from stopping a live import, audit each import once, and close review gaps

- a dispatch never settles the import itself; a retry or the stale-import janitor does
- projected audit entries take an idempotency key, recorded through recordAuditOnce, so a retried completion audits the import exactly once
- a cancel that races completion stops the record the completion wrote
- refuse empty rebuilt chunk-only documents; knowledge_base_import.user_id references the user

Author: waleedlatif1

apps/sim/lib/core/application/authorized-workspace-use-case.ts

@@ -1,6 +1,12 @@
-import { type AuditActionType, type AuditResourceTypeValue, recordAudit } from '@sim/audit'
+import {
+  type AuditActionType,
+  type AuditResourceTypeValue,
+  recordAudit,
+  recordAuditOnce,
+} from '@sim/audit'
 import type { Principal, PrincipalAuditAttribution } from '@sim/auth/principal'
 import { resolvePrincipalAuditAttribution } from '@sim/auth/principal'
+import { createLogger } from '@sim/logger'
 import { runDeclaredContentOperation } from '@/lib/core/application/content-delivery'
 import type { ApplicationOperation, OperationUseCase } from '@/lib/core/application/operation'
 import {
@@ -18,6 +24,8 @@ import { runWithOutboundOrganization } from '@/lib/core/network/context.server'
 import type { OrchestrationRequestContext } from '@/lib/core/orchestration/types'
 import type { ResourcePolicyBinding } from '@/lib/resource-policies/registry'
 
+const logger = createLogger('WorkspaceUseCaseAudit')
+
 export interface WorkspaceUseCaseAuditEntry {
   /** Canonical affected workspace; null keeps an organization event outside the authorization workspace. */
   workspaceId?: string | null
@@ -27,6 +35,11 @@ export interface WorkspaceUseCaseAuditEntry {
   resourceName?: string
   description?: string
   metadata?: Record<string, unknown>
+  /**
+   * Records the entry at most once for this key, for an operation a caller may
+   * retry after losing its response: each retry projects the same entry again.
+   */
+  idempotencyKey?: string
 }
 
 export interface AuthorizedWorkspaceUseCaseContext<
@@ -106,8 +119,8 @@ export function recordProjectedUseCaseAuditEntries(
   organizationId?: string
 ): void {
   const attribution: PrincipalAuditAttribution = resolvePrincipalAuditAttribution(principal)
-  for (const entry of entries) {
-    recordAudit({
+  for (const { idempotencyKey, ...entry } of entries) {
+    const params = {
       workspaceId: entry.workspaceId === undefined ? workspaceId : entry.workspaceId,
       actorId: attribution.actorId,
       actorName: attribution.actorName,
@@ -123,6 +136,13 @@ export function recordProjectedUseCaseAuditEntries(
         actor: attribution.actor,
       },
       request,
+    }
+    if (!idempotencyKey) {
+      recordAudit(params)
+      continue
+    }
+    recordAuditOnce(idempotencyKey, params).catch((error) => {
+      logger.error('Failed to record audit log', { error, action: params.action })
     })
   }
 }

apps/sim/lib/knowledge/__integration__/bundle-import.integration.ts

@@ -661,6 +661,19 @@ describe('knowledge base import requests', () => {
         .from(knowledgeBase)
         .where(eq(knowledgeBase.name, 'Redispatched import'))
     ).toEqual([{ id: created.import.id }])
+    await expect
+      .poll(async () =>
+        db
+          .select({ resourceId: auditLog.resourceId })
+          .from(auditLog)
+          .where(
+            and(
+              eq(auditLog.workspaceId, ids.workspaceId),
+              eq(auditLog.resourceId, created.import.id)
+            )
+          )
+      )
+      .toEqual([{ resourceId: created.import.id }])
   })
 
   it('refuses a bundle that is not a zip archive before creating an upload', async () => {

apps/sim/lib/knowledge/application/imports.ts

@@ -24,7 +24,6 @@ import {
 import { knowledgeOperations } from '@/lib/knowledge/application/operations'
 import {
   findKnowledgeImport,
-  finishKnowledgeImport,
   getKnowledgeImport,
   insertKnowledgeImport,
   KNOWLEDGE_IMPORT_STORAGE_CONTEXT,
@@ -37,7 +36,6 @@ import {
   runKnowledgeImport,
 } from '@/lib/knowledge/transfer/import-runner'
 import { assertWorkspaceCapability } from '@/lib/permission-groups/capability-assertions'
-import { deleteFile } from '@/lib/uploads/core/storage-service'
 import { requestOrigin } from '@/lib/uploads/upload-session/application'
 import {
   abortUploadSession,
@@ -101,11 +99,6 @@ interface KnowledgeImportResult {
   import: KnowledgeImportResource
 }
 
-interface CompleteKnowledgeImportResult extends KnowledgeImportResult {
-  /** The base this completion started importing, or `null` when the import was already started. */
-  startedKnowledgeBaseId: string | null
-}
-
 interface KnowledgeImportRecordContext extends KnowledgeWorkspaceContext {
   record: KnowledgeImportRecord
 }
@@ -270,33 +263,13 @@ function startKnowledgeImport(payload: KnowledgeImportPayload): Promise<void> {
   })
 }
 
-/** Starts the runner; a dispatch that fails settles the import and drops its bundle. */
-async function dispatchKnowledgeImport(payload: KnowledgeImportPayload): Promise<void> {
-  try {
-    await startKnowledgeImport(payload)
-  } catch (error) {
-    await finishKnowledgeImport(payload.importId, {
-      status: 'failed',
-      error: 'The import could not be started',
-    })
-    await deleteFile({ key: payload.sourceKey, context: KNOWLEDGE_IMPORT_STORAGE_CONTEXT }).catch(
-      (cleanupError: unknown) => {
-        logger.error('Failed to delete a knowledge bundle after dispatch failure', {
-          importId: payload.importId,
-          error: getErrorMessage(cleanupError, 'Unknown error'),
-        })
-      }
-    )
-    throw error
-  }
-}
-
 /**
  * Dispatches again an import whose record no worker has claimed: its first
- * dispatch was lost with the request that wrote the record, or is still
- * queued. Either way a second run is safe, since only one worker can claim it,
- * and a failure here leaves the import to the stale-import janitor rather than
- * failing one that may already be on its way.
+ * dispatch failed or was lost with the request that wrote the record, or is
+ * still queued. Either way a second run is safe, since only one worker can
+ * claim it. A dispatch never settles the import itself, since another one may
+ * already be on its way; one that never runs is failed by the stale-import
+ * janitor, which also deletes its bundle.
  */
 async function redispatchUnclaimedImport(
   principal: Principal,
@@ -399,11 +372,11 @@ export const completeKnowledgeImport = defineAuthorizedKnowledgeUseCase({
     principal: Principal
     input: KnowledgeImportUploadInput
   }) => resolveUploadContext(principal, input),
-  async execute({ principal, context }): Promise<CompleteKnowledgeImportResult> {
+  async execute({ principal, context }): Promise<KnowledgeImportResult> {
     const existing = await findStartedImport(context)
     if (existing) {
       await redispatchUnclaimedImport(principal, context, existing)
-      return { import: fromRecord(existing), startedKnowledgeBaseId: null }
+      return { import: fromRecord(existing) }
     }
     let started: KnowledgeImportPayload | null
     try {
@@ -427,11 +400,11 @@ export const completeKnowledgeImport = defineAuthorizedKnowledgeUseCase({
       // A concurrent completion holds the session; report the import it started, if any.
       const raced =
         asOrchestrationError(error)?.code === 'conflict' ? await findStartedImport(context) : null
-      if (raced) return { import: fromRecord(raced), startedKnowledgeBaseId: null }
+      if (raced) return { import: fromRecord(raced) }
       throw error
     }
     if (started) {
-      await dispatchKnowledgeImport(started)
+      await startKnowledgeImport(started)
       logger.info('Started knowledge import', {
         importId: started.importId,
         workspaceId: context.workspaceId,
@@ -442,20 +415,21 @@ export const completeKnowledgeImport = defineAuthorizedKnowledgeUseCase({
       importId: context.upload.id,
       assertedWorkspaceId: context.workspaceId,
     })
-    return { import: fromRecord(record), startedKnowledgeBaseId: started?.importId ?? null }
+    return { import: fromRecord(record) }
   },
-  /** Recorded once per import: a repeated completion only reads back the one already started. */
-  projectAudit: ({ context, result }) =>
-    result.startedKnowledgeBaseId
-      ? {
-          action: AuditAction.KNOWLEDGE_BASE_IMPORTED,
-          resourceType: AuditResourceType.KNOWLEDGE_BASE,
-          resourceId: result.startedKnowledgeBaseId,
-          resourceName: importedName(context.upload) ?? context.upload.fileName,
-          description: `Imported a knowledge base from bundle "${context.upload.fileName}"`,
-          metadata: { workspaceId: context.workspaceId, importId: result.import.id },
-        }
-      : [],
+  /**
+   * Keyed by the import, so the completion that started it and any retry of it
+   * record one event between them. The base takes the import's id.
+   */
+  projectAudit: ({ context, result }) => ({
+    action: AuditAction.KNOWLEDGE_BASE_IMPORTED,
+    resourceType: AuditResourceType.KNOWLEDGE_BASE,
+    resourceId: result.import.id,
+    resourceName: importedName(context.upload) ?? context.upload.fileName,
+    description: `Imported a knowledge base from bundle "${context.upload.fileName}"`,
+    metadata: { workspaceId: context.workspaceId, importId: result.import.id },
+    idempotencyKey: `knowledge-import:${result.import.id}`,
+  }),
 })
 
 /**
@@ -473,6 +447,10 @@ export const cancelKnowledgeImport = defineAuthorizedKnowledgeUseCase({
     const started = await findStartedImport(context)
     if (started) return { import: fromRecord(await stopKnowledgeImport(started)) }
     assertUploadSessionAuthBinding(context.upload, principal)
-    return { import: fromUpload(await abortUploadSession(context.upload), context.workspaceId) }
+    const aborted = await abortUploadSession(context.upload)
+    // A completion finalizing alongside this abort may have written the record first.
+    const late = await findStartedImport(context)
+    if (late) return { import: fromRecord(await stopKnowledgeImport(late)) }
+    return { import: fromUpload(aborted, context.workspaceId) }
   },
 })

apps/sim/lib/knowledge/transfer/import-runner.ts

@@ -40,7 +40,10 @@ import {
 } from '@/lib/knowledge/transfer/import-jobs'
 import { generateKnowledgeBaseFileKey } from '@/lib/uploads/contexts/knowledge-base/file-manager'
 import { deleteFile, downloadFileStream } from '@/lib/uploads/core/storage-service'
-import { MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE } from '@/lib/uploads/shared/types'
+import {
+  EMPTY_KNOWLEDGE_DOCUMENT_MESSAGE,
+  MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE,
+} from '@/lib/uploads/shared/types'
 import { getFileExtension } from '@/lib/uploads/utils/file'
 import { isSupportedExtension, validateFileType } from '@/lib/uploads/utils/validation'
 
@@ -166,6 +169,9 @@ async function importDocument(
     }
   } else if (prepared) {
     const bytes = Buffer.from(prepared.chunks.map((chunk) => chunk.text).join('\n\n'))
+    if (bytes.byteLength === 0) {
+      throw new OrchestrationError('validation', EMPTY_KNOWLEDGE_DOCUMENT_MESSAGE)
+    }
     if (bytes.byteLength > MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE) {
       throw new OrchestrationError(
         'payload_too_large',

packages/db/migrations/0436_knowledge_base_import.sql

@@ -19,6 +19,7 @@ CREATE TABLE "knowledge_base_import" (
 );
 --> statement-breakpoint
 ALTER TABLE "knowledge_base_import" ADD CONSTRAINT "knowledge_base_import_workspace_id_workspace_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspace"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "knowledge_base_import" ADD CONSTRAINT "knowledge_base_import_user_id_user_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."user"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
 ALTER TABLE "knowledge_base_import" ADD CONSTRAINT "knowledge_base_import_knowledge_base_id_knowledge_base_id_fk" FOREIGN KEY ("knowledge_base_id") REFERENCES "public"."knowledge_base"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint
 CREATE INDEX "knowledge_base_import_watchdog_idx" ON "knowledge_base_import" USING btree ("status","updated_at");--> statement-breakpoint
 CREATE INDEX "knowledge_base_import_knowledge_base_id_idx" ON "knowledge_base_import" USING btree ("knowledge_base_id");

packages/db/migrations/meta/0436_snapshot.json

@@ -1,5 +1,5 @@
 {
-  "id": "eabb7aa2-0cfb-4872-aa0c-38733757019b",
+  "id": "a7a19a4e-8207-4fbf-918c-f5ae171bd9cc",
   "prevId": "2a4095de-e341-4264-becd-1714d26404f5",
   "version": "7",
   "dialect": "postgresql",
@@ -10223,6 +10223,15 @@
           "onDelete": "cascade",
           "onUpdate": "no action"
         },
+        "knowledge_base_import_user_id_user_id_fk": {
+          "name": "knowledge_base_import_user_id_user_id_fk",
+          "tableFrom": "knowledge_base_import",
+          "tableTo": "user",
+          "columnsFrom": ["user_id"],
+          "columnsTo": ["id"],
+          "onDelete": "cascade",
+          "onUpdate": "no action"
+        },
         "knowledge_base_import_knowledge_base_id_knowledge_base_id_fk": {
           "name": "knowledge_base_import_knowledge_base_id_knowledge_base_id_fk",
           "tableFrom": "knowledge_base_import",

packages/db/migrations/meta/_journal.json

@@ -3050,7 +3050,7 @@
     {
       "idx": 436,
       "version": "7",
-      "when": 1792719631197,
+      "when": 1792720602032,
       "tag": "0436_knowledge_base_import",
       "breakpoints": true
     }

packages/db/schema.ts

@@ -3865,7 +3865,9 @@ export const knowledgeBaseImport = pgTable(
     workspaceId: text('workspace_id')
       .notNull()
       .references(() => workspace.id, { onDelete: 'cascade' }),
-    userId: text('user_id').notNull(),
+    userId: text('user_id')
+      .notNull()
+      .references(() => user.id, { onDelete: 'cascade' }),
     /** The uploaded archive, deleted once the import settles. */
     sourceKey: text('source_key').notNull(),
     knowledgeBaseId: text('knowledge_base_id').references(() => knowledgeBase.id, {