feat(kb): harden sync engine and add connector audit logging

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

@@ -53,6 +53,11 @@ vi.mock('@/lib/auth/hybrid', () => ({
 vi.mock('@/lib/core/utils/request', () => ({
   generateRequestId: vi.fn().mockReturnValue('test-req-id'),
 }))
+vi.mock('@/lib/audit/log', () => ({
+  recordAudit: vi.fn(),
+  AuditAction: {},
+  AuditResourceType: {},
+}))
 
 import { GET, PATCH } from '@/app/api/knowledge/[id]/connectors/[connectorId]/documents/route'
 
@@ -168,8 +173,16 @@ describe('Connector Documents API Route', () => {
     })
 
     it('returns success for restore operation', async () => {
-      mockCheckSession.mockResolvedValue({ success: true, userId: 'user-1' })
-      mockCheckWriteAccess.mockResolvedValue({ hasAccess: true })
+      mockCheckSession.mockResolvedValue({
+        success: true,
+        userId: 'user-1',
+        userName: 'Test',
+        userEmail: 'test@test.com',
+      })
+      mockCheckWriteAccess.mockResolvedValue({
+        hasAccess: true,
+        knowledgeBase: { workspaceId: 'ws-1', name: 'Test KB' },
+      })
       mockDbChain.limit.mockResolvedValueOnce([{ id: 'conn-456' }])
       mockDbChain.returning.mockResolvedValueOnce([{ id: 'doc-1' }])
 
@@ -182,8 +195,16 @@ describe('Connector Documents API Route', () => {
     })
 
     it('returns success for exclude operation', async () => {
-      mockCheckSession.mockResolvedValue({ success: true, userId: 'user-1' })
-      mockCheckWriteAccess.mockResolvedValue({ hasAccess: true })
+      mockCheckSession.mockResolvedValue({
+        success: true,
+        userId: 'user-1',
+        userName: 'Test',
+        userEmail: 'test@test.com',
+      })
+      mockCheckWriteAccess.mockResolvedValue({
+        hasAccess: true,
+        knowledgeBase: { workspaceId: 'ws-1', name: 'Test KB' },
+      })
       mockDbChain.limit.mockResolvedValueOnce([{ id: 'conn-456' }])
       mockDbChain.returning.mockResolvedValueOnce([{ id: 'doc-2' }, { id: 'doc-3' }])
 

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

@@ -4,6 +4,7 @@ import { createLogger } from '@sim/logger'
 import { and, eq, inArray, isNull } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { z } from 'zod'
+import { AuditAction, AuditResourceType, recordAudit } from '@/lib/audit/log'
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { checkKnowledgeBaseAccess, checkKnowledgeBaseWriteAccess } from '@/app/api/knowledge/utils'
@@ -184,6 +185,19 @@ export async function PATCH(request: NextRequest, { params }: RouteParams) {
 
       logger.info(`[${requestId}] Restored ${updated.length} excluded documents`, { connectorId })
 
+      recordAudit({
+        workspaceId: writeCheck.knowledgeBase.workspaceId,
+        actorId: auth.userId,
+        actorName: auth.userName,
+        actorEmail: auth.userEmail,
+        action: AuditAction.CONNECTOR_DOCUMENT_RESTORED,
+        resourceType: AuditResourceType.CONNECTOR,
+        resourceId: connectorId,
+        description: `Restored ${updated.length} excluded document(s) for knowledge base "${writeCheck.knowledgeBase.name}"`,
+        metadata: { knowledgeBaseId, documentCount: updated.length },
+        request,
+      })
+
       return NextResponse.json({
         success: true,
         data: { restoredCount: updated.length, documentIds: updated.map((d) => d.id) },
@@ -206,6 +220,19 @@ export async function PATCH(request: NextRequest, { params }: RouteParams) {
 
     logger.info(`[${requestId}] Excluded ${updated.length} documents`, { connectorId })
 
+    recordAudit({
+      workspaceId: writeCheck.knowledgeBase.workspaceId,
+      actorId: auth.userId,
+      actorName: auth.userName,
+      actorEmail: auth.userEmail,
+      action: AuditAction.CONNECTOR_DOCUMENT_EXCLUDED,
+      resourceType: AuditResourceType.CONNECTOR,
+      resourceId: connectorId,
+      description: `Excluded ${updated.length} document(s) from knowledge base "${writeCheck.knowledgeBase.name}"`,
+      metadata: { knowledgeBaseId, documentCount: updated.length },
+      request,
+    })
+
     return NextResponse.json({
       success: true,
       data: { excludedCount: updated.length, documentIds: updated.map((d) => d.id) },

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

@@ -75,6 +75,11 @@ vi.mock('@/lib/knowledge/tags/service', () => ({
 vi.mock('@/lib/knowledge/documents/service', () => ({
   deleteDocumentStorageFiles: vi.fn().mockResolvedValue(undefined),
 }))
+vi.mock('@/lib/audit/log', () => ({
+  recordAudit: vi.fn(),
+  AuditAction: {},
+  AuditResourceType: {},
+}))
 
 import { DELETE, GET, PATCH } from '@/app/api/knowledge/[id]/connectors/[connectorId]/route'
 
@@ -183,8 +188,16 @@ describe('Knowledge Connector By ID API Route', () => {
     })
 
     it('returns 200 and updates status', async () => {
-      mockCheckSession.mockResolvedValue({ success: true, userId: 'user-1' })
-      mockCheckWriteAccess.mockResolvedValue({ hasAccess: true })
+      mockCheckSession.mockResolvedValue({
+        success: true,
+        userId: 'user-1',
+        userName: 'Test',
+        userEmail: 'test@test.com',
+      })
+      mockCheckWriteAccess.mockResolvedValue({
+        hasAccess: true,
+        knowledgeBase: { workspaceId: 'ws-1', name: 'Test KB' },
+      })
 
       const updatedConnector = { id: 'conn-456', status: 'paused', syncIntervalMinutes: 120 }
       mockDbChain.limit.mockResolvedValueOnce([updatedConnector])
@@ -210,8 +223,16 @@ describe('Knowledge Connector By ID API Route', () => {
     })
 
     it('returns 200 on successful hard-delete', async () => {
-      mockCheckSession.mockResolvedValue({ success: true, userId: 'user-1' })
-      mockCheckWriteAccess.mockResolvedValue({ hasAccess: true })
+      mockCheckSession.mockResolvedValue({
+        success: true,
+        userId: 'user-1',
+        userName: 'Test',
+        userEmail: 'test@test.com',
+      })
+      mockCheckWriteAccess.mockResolvedValue({
+        hasAccess: true,
+        knowledgeBase: { workspaceId: 'ws-1', name: 'Test KB' },
+      })
       mockDbChain.where
         .mockReturnValueOnce(mockDbChain)
         .mockResolvedValueOnce([{ id: 'doc-1', fileUrl: '/api/uploads/test.txt' }])

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

@@ -11,6 +11,7 @@ import { and, desc, eq, inArray, isNull, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { z } from 'zod'
 import { decryptApiKey } from '@/lib/api-key/crypto'
+import { AuditAction, AuditResourceType, recordAudit } from '@/lib/audit/log'
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { deleteDocumentStorageFiles } from '@/lib/knowledge/documents/service'
@@ -233,6 +234,21 @@ export async function PATCH(request: NextRequest, { params }: RouteParams) {
       .limit(1)
 
     const { encryptedApiKey: __, ...updatedData } = updated[0]
+
+    recordAudit({
+      workspaceId: writeCheck.knowledgeBase.workspaceId,
+      actorId: auth.userId,
+      actorName: auth.userName,
+      actorEmail: auth.userEmail,
+      action: AuditAction.CONNECTOR_UPDATED,
+      resourceType: AuditResourceType.CONNECTOR,
+      resourceId: connectorId,
+      resourceName: updatedData.connectorType,
+      description: `Updated connector for knowledge base "${writeCheck.knowledgeBase.name}"`,
+      metadata: { knowledgeBaseId, updatedFields: Object.keys(parsed.data) },
+      request,
+    })
+
     return NextResponse.json({ success: true, data: updatedData })
   } catch (error) {
     logger.error(`[${requestId}] Error updating connector`, error)
@@ -260,7 +276,7 @@ export async function DELETE(request: NextRequest, { params }: RouteParams) {
     }
 
     const existingConnector = await db
-      .select({ id: knowledgeConnector.id })
+      .select({ id: knowledgeConnector.id, connectorType: knowledgeConnector.connectorType })
       .from(knowledgeConnector)
       .where(
         and(
@@ -323,6 +339,20 @@ export async function DELETE(request: NextRequest, { params }: RouteParams) {
 
     logger.info(`[${requestId}] Hard-deleted connector ${connectorId} and its documents`)
 
+    recordAudit({
+      workspaceId: writeCheck.knowledgeBase.workspaceId,
+      actorId: auth.userId,
+      actorName: auth.userName,
+      actorEmail: auth.userEmail,
+      action: AuditAction.CONNECTOR_DELETED,
+      resourceType: AuditResourceType.CONNECTOR,
+      resourceId: connectorId,
+      resourceName: existingConnector[0].connectorType,
+      description: `Deleted connector from knowledge base "${writeCheck.knowledgeBase.name}"`,
+      metadata: { knowledgeBaseId, documentsDeleted: connectorDocuments.length },
+      request,
+    })
+
     return NextResponse.json({ success: true })
   } catch (error) {
     logger.error(`[${requestId}] Error deleting connector`, error)

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

@@ -43,6 +43,11 @@ vi.mock('@/lib/core/utils/request', () => ({
 vi.mock('@/lib/knowledge/connectors/sync-engine', () => ({
   dispatchSync: mockDispatchSync,
 }))
+vi.mock('@/lib/audit/log', () => ({
+  recordAudit: vi.fn(),
+  AuditAction: {},
+  AuditResourceType: {},
+}))
 
 import { POST } from '@/app/api/knowledge/[id]/connectors/[connectorId]/sync/route'
 
@@ -92,8 +97,16 @@ describe('Connector Manual Sync API Route', () => {
   })
 
   it('dispatches sync on valid request', async () => {
-    mockCheckSession.mockResolvedValue({ success: true, userId: 'user-1' })
-    mockCheckWriteAccess.mockResolvedValue({ hasAccess: true })
+    mockCheckSession.mockResolvedValue({
+      success: true,
+      userId: 'user-1',
+      userName: 'Test',
+      userEmail: 'test@test.com',
+    })
+    mockCheckWriteAccess.mockResolvedValue({
+      hasAccess: true,
+      knowledgeBase: { workspaceId: 'ws-1', name: 'Test KB' },
+    })
     mockDbChain.limit.mockResolvedValueOnce([{ id: 'conn-456', status: 'active' }])
 
     const req = createMockRequest('POST')

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

@@ -3,6 +3,7 @@ import { knowledgeConnector } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, eq, isNull } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
+import { AuditAction, AuditResourceType, recordAudit } from '@/lib/audit/log'
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { dispatchSync } from '@/lib/knowledge/connectors/sync-engine'
@@ -54,6 +55,20 @@ export async function POST(request: NextRequest, { params }: RouteParams) {
 
     logger.info(`[${requestId}] Manual sync triggered for connector ${connectorId}`)
 
+    recordAudit({
+      workspaceId: writeCheck.knowledgeBase.workspaceId,
+      actorId: auth.userId,
+      actorName: auth.userName,
+      actorEmail: auth.userEmail,
+      action: AuditAction.CONNECTOR_SYNCED,
+      resourceType: AuditResourceType.CONNECTOR,
+      resourceId: connectorId,
+      resourceName: connectorRows[0].connectorType,
+      description: `Triggered manual sync for connector on knowledge base "${writeCheck.knowledgeBase.name}"`,
+      metadata: { knowledgeBaseId },
+      request,
+    })
+
     dispatchSync(connectorId, { requestId }).catch((error) => {
       logger.error(
         `[${requestId}] Failed to dispatch manual sync for connector ${connectorId}`,

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

@@ -5,6 +5,7 @@ import { and, desc, eq, isNull, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { z } from 'zod'
 import { encryptApiKey } from '@/lib/api-key/crypto'
+import { AuditAction, AuditResourceType, recordAudit } from '@/lib/audit/log'
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { dispatchSync } from '@/lib/knowledge/connectors/sync-engine'
@@ -226,6 +227,20 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
 
     logger.info(`[${requestId}] Created connector ${connectorId} for KB ${knowledgeBaseId}`)
 
+    recordAudit({
+      workspaceId: writeCheck.knowledgeBase.workspaceId,
+      actorId: auth.userId,
+      actorName: auth.userName,
+      actorEmail: auth.userEmail,
+      action: AuditAction.CONNECTOR_CREATED,
+      resourceType: AuditResourceType.CONNECTOR,
+      resourceId: connectorId,
+      resourceName: connectorType,
+      description: `Created ${connectorType} connector for knowledge base "${writeCheck.knowledgeBase.name}"`,
+      metadata: { knowledgeBaseId, connectorType, syncIntervalMinutes },
+      request,
+    })
+
     dispatchSync(connectorId, { requestId }).catch((error) => {
       logger.error(
         `[${requestId}] Failed to dispatch initial sync for connector ${connectorId}`,

apps/sim/app/api/knowledge/[id]/restore/route.ts

@@ -3,6 +3,7 @@ import { knowledgeBase } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { eq } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
+import { AuditAction, AuditResourceType, recordAudit } from '@/lib/audit/log'
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { restoreKnowledgeBase } from '@/lib/knowledge/service'
@@ -23,6 +24,7 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
     const [kb] = await db
       .select({
         id: knowledgeBase.id,
+        name: knowledgeBase.name,
         workspaceId: knowledgeBase.workspaceId,
         userId: knowledgeBase.userId,
       })
@@ -47,6 +49,19 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
 
     logger.info(`[${requestId}] Restored knowledge base ${id}`)
 
+    recordAudit({
+      workspaceId: kb.workspaceId,
+      actorId: auth.userId,
+      actorName: auth.userName,
+      actorEmail: auth.userEmail,
+      action: AuditAction.KNOWLEDGE_BASE_RESTORED,
+      resourceType: AuditResourceType.KNOWLEDGE_BASE,
+      resourceId: id,
+      resourceName: kb.name,
+      description: `Restored knowledge base "${kb.name}"`,
+      request,
+    })
+
     return NextResponse.json({ success: true })
   } catch (error) {
     logger.error(`[${requestId}] Error restoring knowledge base ${id}`, error)

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

@@ -27,6 +27,34 @@ export async function GET(request: NextRequest) {
   try {
     const now = new Date()
 
+    const STALE_SYNC_TTL_MS = 120 * 60 * 1000
+    const staleCutoff = new Date(now.getTime() - STALE_SYNC_TTL_MS)
+
+    const recoveredConnectors = await db
+      .update(knowledgeConnector)
+      .set({
+        status: 'error',
+        lastSyncError: 'Sync timed out (stale lock recovered)',
+        nextSyncAt: new Date(now.getTime() + 10 * 60 * 1000),
+        updatedAt: now,
+      })
+      .where(
+        and(
+          eq(knowledgeConnector.status, 'syncing'),
+          lte(knowledgeConnector.updatedAt, staleCutoff),
+          isNull(knowledgeConnector.archivedAt),
+          isNull(knowledgeConnector.deletedAt)
+        )
+      )
+      .returning({ id: knowledgeConnector.id })
+
+    if (recoveredConnectors.length > 0) {
+      logger.warn(
+        `[${requestId}] Recovered ${recoveredConnectors.length} stale syncing connectors`,
+        { ids: recoveredConnectors.map((c) => c.id) }
+      )
+    }
+
     const dueConnectors = await db
       .select({
         id: knowledgeConnector.id,

apps/sim/app/api/table/[tableId]/restore/route.ts

@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
+import { AuditAction, AuditResourceType, recordAudit } from '@/lib/audit/log'
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { getTableById, restoreTable } from '@/lib/table'
@@ -34,6 +35,19 @@ export async function POST(
 
     logger.info(`[${requestId}] Restored table ${tableId}`)
 
+    recordAudit({
+      workspaceId: table.workspaceId,
+      actorId: auth.userId,
+      actorName: auth.userName,
+      actorEmail: auth.userEmail,
+      action: AuditAction.TABLE_RESTORED,
+      resourceType: AuditResourceType.TABLE,
+      resourceId: tableId,
+      resourceName: table.name,
+      description: `Restored table "${table.name}"`,
+      request,
+    })
+
     return NextResponse.json({ success: true })
   } catch (error) {
     logger.error(`[${requestId}] Error restoring table ${tableId}`, error)