simstudioai--sim/apps/sim/app/api/knowledge/[id]/documents/upsert/route.ts

import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
import { db } from '@sim/db'
import { document } from '@sim/db/schema'
import { createLogger } from '@sim/logger'
import { authorizeWorkflowByWorkspacePermission } from '@sim/platform-authz/workflow'
import { getErrorMessage } from '@sim/utils/errors'
import { generateId } from '@sim/utils/id'
import { and, eq, isNull } from 'drizzle-orm'
import { type NextRequest, NextResponse } from 'next/server'
import { upsertKnowledgeDocumentContract } from '@/lib/api/contracts/knowledge'
import { parseRequest } from '@/lib/api/server'
import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
import { checkActorUsageLimits } from '@/lib/billing/calculations/usage-monitor'
import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
import {
  createDocumentRecords,
  deleteDocument,
  getProcessingConfig,
  KnowledgeBaseFileOwnershipError,
  processDocumentsWithQueue,
} from '@/lib/knowledge/documents/service'
import { checkKnowledgeBaseWriteAccess } from '@/app/api/knowledge/utils'

const logger = createLogger('DocumentUpsertAPI')

export const POST = withRouteHandler(
  async (req: NextRequest, context: { params: Promise<{ id: string }> }) => {
    const requestId = generateId().slice(0, 8)
    const { id: knowledgeBaseId } = await context.params

    try {
      const parsed = await parseRequest(upsertKnowledgeDocumentContract, req, context)
      if (!parsed.success) return parsed.response
      const validatedData = parsed.data.body

      logger.info(`[${requestId}] Knowledge base document upsert request`, {
        knowledgeBaseId,
        hasDocumentId: !!validatedData.documentId,
        filename: validatedData.filename,
      })

      const auth = await checkSessionOrInternalAuth(req, { requireWorkflowId: false })
      if (!auth.success || !auth.userId) {
        logger.warn(`[${requestId}] Authentication failed: ${auth.error || 'Unauthorized'}`)
        return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
      }
      const userId = auth.userId

      if (validatedData.workflowId) {
        const authorization = await authorizeWorkflowByWorkspacePermission({
          workflowId: validatedData.workflowId,
          userId,
          action: 'write',
        })
        if (!authorization.allowed) {
          return NextResponse.json(
            { error: authorization.message || 'Access denied' },
            { status: authorization.status }
          )
        }
      }

      const accessCheck = await checkKnowledgeBaseWriteAccess(knowledgeBaseId, userId)

      if (!accessCheck.hasAccess) {
        if ('notFound' in accessCheck && accessCheck.notFound) {
          logger.warn(`[${requestId}] Knowledge base not found: ${knowledgeBaseId}`)
          return NextResponse.json({ error: 'Knowledge base not found' }, { status: 404 })
        }
        logger.warn(
          `[${requestId}] User ${userId} attempted to upsert document in unauthorized knowledge base ${knowledgeBaseId}`
        )
        return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
      }

      // Gate usage before any create/delete so an over-limit upsert is rejected up
      // front and never deletes the existing (already-indexed) document. Runs even
      // for legacy KBs with no workspace — the uploader's pooled/frozen status is
      // still enforced (per-member is skipped when there's no org workspace).
      const kbWorkspaceId = accessCheck.knowledgeBase?.workspaceId
      const usage = await checkActorUsageLimits(userId, kbWorkspaceId)
      if (usage.isExceeded) {
        return NextResponse.json(
          {
            error: usage.message || 'Usage limit exceeded. Please upgrade your plan to continue.',
          },
          { status: 402 }
        )
      }

      let existingDocumentId: string | null = null
      let isUpdate = false

      if (validatedData.documentId) {
        const existingDoc = await db
          .select({ id: document.id })
          .from(document)
          .where(
            and(
              eq(document.id, validatedData.documentId),
              eq(document.knowledgeBaseId, knowledgeBaseId),
              isNull(document.deletedAt)
            )
          )
          .limit(1)

        if (existingDoc.length > 0) {
          existingDocumentId = existingDoc[0].id
        }
      } else {
        const docsByFilename = await db
          .select({ id: document.id })
          .from(document)
          .where(
            and(
              eq(document.filename, validatedData.filename),
              eq(document.knowledgeBaseId, knowledgeBaseId),
              isNull(document.deletedAt)
            )
          )
          .limit(1)

        if (docsByFilename.length > 0) {
          existingDocumentId = docsByFilename[0].id
        }
      }

      if (existingDocumentId) {
        isUpdate = true
        logger.info(
          `[${requestId}] Found existing document ${existingDocumentId}, creating replacement before deleting old`
        )
      }

      const createdDocuments = await createDocumentRecords(
        [
          {
            filename: validatedData.filename,
            fileUrl: validatedData.fileUrl,
            fileSize: validatedData.fileSize,
            mimeType: validatedData.mimeType,
            ...(validatedData.documentTagsData && {
              documentTagsData: validatedData.documentTagsData,
            }),
          },
        ],
        knowledgeBaseId,
        requestId,
        userId
      )

      const firstDocument = createdDocuments[0]
      if (!firstDocument) {
        logger.error(`[${requestId}] createDocumentRecords returned empty array unexpectedly`)
        return NextResponse.json({ error: 'Failed to create document record' }, { status: 500 })
      }

      if (existingDocumentId) {
        try {
          await deleteDocument(existingDocumentId, requestId)
        } catch (deleteError) {
          logger.error(
            `[${requestId}] Failed to delete old document ${existingDocumentId}, rolling back new record`,
            deleteError
          )
          await deleteDocument(firstDocument.documentId, requestId).catch(() => {})
          return NextResponse.json(
            { error: 'Failed to replace existing document' },
            { status: 500 }
          )
        }
      }

      processDocumentsWithQueue(
        createdDocuments,
        knowledgeBaseId,
        validatedData.processingOptions ?? {},
        requestId
      ).catch((error: unknown) => {
        logger.error(`[${requestId}] Critical error in document processing pipeline:`, error)
      })

      try {
        const { PlatformEvents } = await import('@/lib/core/telemetry')
        PlatformEvents.knowledgeBaseDocumentsUploaded({
          knowledgeBaseId,
          documentsCount: 1,
          uploadType: 'single',
          recipe: validatedData.processingOptions?.recipe,
        })
      } catch (_e) {
        // Silently fail
      }

      recordAudit({
        workspaceId: accessCheck.knowledgeBase?.workspaceId ?? null,
        actorId: userId,
        actorName: auth.userName,
        actorEmail: auth.userEmail,
        action: isUpdate ? AuditAction.DOCUMENT_UPDATED : AuditAction.DOCUMENT_UPLOADED,
        resourceType: AuditResourceType.DOCUMENT,
        resourceId: knowledgeBaseId,
        resourceName: validatedData.filename,
        description: isUpdate
          ? `Upserted (replaced) document "${validatedData.filename}" in knowledge base "${knowledgeBaseId}"`
          : `Upserted (created) document "${validatedData.filename}" in knowledge base "${knowledgeBaseId}"`,
        metadata: {
          knowledgeBaseName: accessCheck.knowledgeBase?.name,
          fileName: validatedData.filename,
          fileType: validatedData.mimeType,
          fileSize: validatedData.fileSize,
          previousDocumentId: existingDocumentId,
          isUpdate,
        },
        request: req,
      })

      return NextResponse.json({
        success: true,
        data: {
          documentsCreated: [
            {
              documentId: firstDocument.documentId,
              filename: firstDocument.filename,
              status: 'pending',
            },
          ],
          isUpdate,
          previousDocumentId: existingDocumentId,
          processingMethod: 'background',
          processingConfig: {
            maxConcurrentDocuments: getProcessingConfig().maxConcurrentDocuments,
            batchSize: getProcessingConfig().batchSize,
          },
        },
      })
    } catch (error) {
      logger.error(`[${requestId}] Error upserting document`, error)

      if (error instanceof KnowledgeBaseFileOwnershipError) {
        return NextResponse.json(
          { error: 'File URL does not reference a file owned by this knowledge base' },
          { status: 403 }
        )
      }

      const errorMessage = getErrorMessage(error, 'Failed to upsert document')
      const isStorageLimitError =
        errorMessage.includes('Storage limit exceeded') || errorMessage.includes('storage limit')
      const isMissingKnowledgeBase = errorMessage === 'Knowledge base not found'

      return NextResponse.json(
        { error: errorMessage },
        { status: isMissingKnowledgeBase ? 404 : isStorageLimitError ? 413 : 500 }
      )
    }
  }
)