import { createLogger } from '@sim/logger'
import { type NextRequest, NextResponse } from 'next/server'
import { z } from 'zod'
import { AuthType, checkHybridAuth, hasExternalApiCredentials } from '@/lib/auth/hybrid'
import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate'
import { getJobQueue, shouldExecuteInline, shouldUseBullMQ } from '@/lib/core/async-jobs'
import { createBullMQJobData } from '@/lib/core/bullmq'
import {
  createTimeoutAbortController,
  getTimeoutErrorMessage,
  isTimeoutError,
} from '@/lib/core/execution-limits'
import { generateRequestId } from '@/lib/core/utils/request'
import { SSE_HEADERS } from '@/lib/core/utils/sse'
import { getBaseUrl } from '@/lib/core/utils/urls'
import { generateId, isValidUuid } from '@/lib/core/utils/uuid'
import {
  DispatchQueueFullError,
  enqueueWorkspaceDispatch,
  type WorkspaceDispatchLane,
  waitForDispatchJob,
} from '@/lib/core/workspace-dispatch'
import { createBufferedExecutionStream } from '@/lib/execution/buffered-stream'
import {
  buildNextCallChain,
  parseCallChain,
  SIM_VIA_HEADER,
  validateCallChain,
} from '@/lib/execution/call-chain'
import { createExecutionEventWriter, setExecutionMeta } from '@/lib/execution/event-buffer'
import { processInputFileFields } from '@/lib/execution/files'
import {
  registerManualExecutionAborter,
  unregisterManualExecutionAborter,
} from '@/lib/execution/manual-cancellation'
import { preprocessExecution } from '@/lib/execution/preprocessing'
import { LoggingSession } from '@/lib/logs/execution/logging-session'
import {
  cleanupExecutionBase64Cache,
  hydrateUserFilesWithBase64,
} from '@/lib/uploads/utils/user-file-base64.server'
import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
import { type ExecutionEvent, encodeSSEEvent } from '@/lib/workflows/executor/execution-events'
import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
import {
  DIRECT_WORKFLOW_JOB_NAME,
  type QueuedWorkflowExecutionPayload,
  type QueuedWorkflowExecutionResult,
} from '@/lib/workflows/executor/queued-workflow-execution'
import {
  loadDeployedWorkflowState,
  loadWorkflowFromNormalizedTables,
} from '@/lib/workflows/persistence/utils'
import { createStreamingResponse } from '@/lib/workflows/streaming/streaming'
import {
  authorizeWorkflowByWorkspacePermission,
  createHttpResponseFromBlock,
  workflowHasResponseBlock,
} from '@/lib/workflows/utils'
import { executeWorkflowJob, type WorkflowExecutionPayload } from '@/background/workflow-execution'
import { normalizeName } from '@/executor/constants'
import { ExecutionSnapshot } from '@/executor/execution/snapshot'
import type {
  ChildWorkflowContext,
  ExecutionMetadata,
  IterationContext,
  SerializableExecutionState,
} from '@/executor/execution/types'
import type { NormalizedBlockOutput, StreamingExecution } from '@/executor/types'
import { hasExecutionResult } from '@/executor/utils/errors'
import { Serializer } from '@/serializer'
import { CORE_TRIGGER_TYPES, type CoreTriggerType } from '@/stores/logs/filters/types'

const logger = createLogger('WorkflowExecuteAPI')

const ExecuteWorkflowSchema = z.object({
  selectedOutputs: z.array(z.string()).optional().default([]),
  triggerType: z.enum(CORE_TRIGGER_TYPES).optional(),
  stream: z.boolean().optional(),
  useDraftState: z.boolean().optional(),
  input: z.any().optional(),
  isClientSession: z.boolean().optional(),
  includeFileBase64: z.boolean().optional().default(true),
  base64MaxBytes: z.number().int().positive().optional(),
  workflowStateOverride: z
    .object({
      blocks: z.record(z.any()),
      edges: z.array(z.any()),
      loops: z.record(z.any()).optional(),
      parallels: z.record(z.any()).optional(),
    })
    .optional(),
  stopAfterBlockId: z.string().optional(),
  runFromBlock: z
    .object({
      startBlockId: z.string().min(1, 'Start block ID is required'),
      sourceSnapshot: z
        .object({
          blockStates: z.record(z.any()),
          executedBlocks: z.array(z.string()),
          blockLogs: z.array(z.any()),
          decisions: z.object({
            router: z.record(z.string()),
            condition: z.record(z.string()),
          }),
          completedLoops: z.array(z.string()),
          loopExecutions: z.record(z.any()).optional(),
          parallelExecutions: z.record(z.any()).optional(),
          parallelBlockMapping: z.record(z.any()).optional(),
          activeExecutionPath: z.array(z.string()),
        })
        .optional(),
      executionId: z.string().optional(),
    })
    .optional(),
})

export const runtime = 'nodejs'
export const dynamic = 'force-dynamic'

const INLINE_TRIGGER_TYPES = new Set<CoreTriggerType>(['manual', 'workflow'])

function resolveOutputIds(
  selectedOutputs: string[] | undefined,
  blocks: Record<string, any>
): string[] | undefined {
  if (!selectedOutputs || selectedOutputs.length === 0) {
    return selectedOutputs
  }

  return selectedOutputs.map((outputId) => {
    const underscoreIndex = outputId.indexOf('_')
    const dotIndex = outputId.indexOf('.')
    if (underscoreIndex > 0) {
      const maybeUuid = outputId.substring(0, underscoreIndex)
      if (isValidUuid(maybeUuid)) {
        return outputId
      }
    }

    if (dotIndex > 0) {
      const maybeUuid = outputId.substring(0, dotIndex)
      if (isValidUuid(maybeUuid)) {
        return `${outputId.substring(0, dotIndex)}_${outputId.substring(dotIndex + 1)}`
      }
    }

    if (isValidUuid(outputId)) {
      return outputId
    }

    if (dotIndex === -1) {
      logger.warn(`Invalid output ID format (missing dot): ${outputId}`)
      return outputId
    }

    const blockName = outputId.substring(0, dotIndex)
    const path = outputId.substring(dotIndex + 1)

    const normalizedBlockName = normalizeName(blockName)
    const block = Object.values(blocks).find((b: any) => {
      return normalizeName(b.name || '') === normalizedBlockName
    })

    if (!block) {
      logger.warn(`Block not found for name: ${blockName} (from output ID: ${outputId})`)
      return outputId
    }

    const resolvedId = `${block.id}_${path}`
    logger.debug(`Resolved output ID: ${outputId} -> ${resolvedId}`)
    return resolvedId
  })
}

type AsyncExecutionParams = {
  requestId: string
  workflowId: string
  userId: string
  workspaceId: string
  input: any
  triggerType: CoreTriggerType
  executionId: string
  callChain?: string[]
}

async function handleAsyncExecution(params: AsyncExecutionParams): Promise<NextResponse> {
  const { requestId, workflowId, userId, workspaceId, input, triggerType, executionId, callChain } =
    params
  const asyncLogger = logger.withMetadata({
    requestId,
    workflowId,
    workspaceId,
    userId,
    executionId,
  })

  const correlation = {
    executionId,
    requestId,
    source: 'workflow' as const,
    workflowId,
    triggerType,
  }

  const payload: WorkflowExecutionPayload = {
    workflowId,
    userId,
    workspaceId,
    input,
    triggerType,
    executionId,
    requestId,
    correlation,
    callChain,
  }

  try {
    const useBullMQ = shouldUseBullMQ()
    const jobQueue = useBullMQ ? null : await getJobQueue()
    const jobId = useBullMQ
      ? await enqueueWorkspaceDispatch({
          id: executionId,
          workspaceId,
          lane: 'runtime',
          queueName: 'workflow-execution',
          bullmqJobName: 'workflow-execution',
          bullmqPayload: createBullMQJobData(payload, {
            workflowId,
            userId,
            correlation,
          }),
          metadata: {
            workflowId,
            userId,
            correlation,
          },
        })
      : await jobQueue!.enqueue('workflow-execution', payload, {
          metadata: { workflowId, workspaceId, userId, correlation },
        })

    asyncLogger.info('Queued async workflow execution', { jobId })

    if (shouldExecuteInline() && jobQueue) {
      const inlineJobQueue = jobQueue
      void (async () => {
        try {
          await inlineJobQueue.startJob(jobId)
          const output = await executeWorkflowJob(payload)
          await inlineJobQueue.completeJob(jobId, output)
        } catch (error) {
          const errorMessage = error instanceof Error ? error.message : String(error)
          asyncLogger.error('Async workflow execution failed', {
            jobId,
            error: errorMessage,
          })
          try {
            await inlineJobQueue.markJobFailed(jobId, errorMessage)
          } catch (markFailedError) {
            asyncLogger.error('Failed to mark job as failed', {
              jobId,
              error:
                markFailedError instanceof Error
                  ? markFailedError.message
                  : String(markFailedError),
            })
          }
        }
      })()
    }

    return NextResponse.json(
      {
        success: true,
        async: true,
        jobId,
        executionId,
        message: 'Workflow execution queued',
        statusUrl: `${getBaseUrl()}/api/jobs/${jobId}`,
      },
      { status: 202 }
    )
  } catch (error: any) {
    if (error instanceof DispatchQueueFullError) {
      return NextResponse.json(
        {
          error: 'Service temporarily at capacity',
          message: error.message,
          retryAfterSeconds: 10,
        },
        { status: 503, headers: { 'Retry-After': '10' } }
      )
    }

    asyncLogger.error('Failed to queue async execution', error)
    return NextResponse.json(
      { error: `Failed to queue async execution: ${error.message}` },
      { status: 500 }
    )
  }
}

async function enqueueDirectWorkflowExecution(
  payload: QueuedWorkflowExecutionPayload,
  priority: number,
  lane: WorkspaceDispatchLane
) {
  return enqueueWorkspaceDispatch({
    id: payload.metadata.executionId,
    workspaceId: payload.metadata.workspaceId,
    lane,
    queueName: 'workflow-execution',
    bullmqJobName: DIRECT_WORKFLOW_JOB_NAME,
    bullmqPayload: createBullMQJobData(payload, {
      workflowId: payload.metadata.workflowId,
      userId: payload.metadata.userId,
      correlation: payload.metadata.correlation,
    }),
    metadata: {
      workflowId: payload.metadata.workflowId,
      userId: payload.metadata.userId,
      correlation: payload.metadata.correlation,
    },
    priority,
  })
}

/**
 * POST /api/workflows/[id]/execute
 *
 * Unified server-side workflow execution endpoint.
 * Supports both SSE streaming (for interactive/manual runs) and direct JSON responses (for background jobs).
 */
export async function POST(req: NextRequest, { params }: { params: Promise<{ id: string }> }) {
  const isSessionRequest = req.headers.has('cookie') && !hasExternalApiCredentials(req.headers)
  if (isSessionRequest) {
    return handleExecutePost(req, params)
  }

  const ticket = tryAdmit()
  if (!ticket) {
    return admissionRejectedResponse()
  }

  try {
    return await handleExecutePost(req, params)
  } finally {
    ticket.release()
  }
}

async function handleExecutePost(
  req: NextRequest,
  params: Promise<{ id: string }>
): Promise<NextResponse | Response> {
  const requestId = generateRequestId()
  const { id: workflowId } = await params
  let reqLogger = logger.withMetadata({ requestId, workflowId })

  const incomingCallChain = parseCallChain(req.headers.get(SIM_VIA_HEADER))
  const callChainError = validateCallChain(incomingCallChain)
  if (callChainError) {
    reqLogger.warn(`Call chain rejected: ${callChainError}`)
    return NextResponse.json({ error: callChainError }, { status: 409 })
  }
  const callChain = buildNextCallChain(incomingCallChain, workflowId)

  try {
    const auth = await checkHybridAuth(req, { requireWorkflowId: false })

    let userId: string
    let isPublicApiAccess = false

    if (!auth.success || !auth.userId) {
      const hasExplicitCredentials =
        req.headers.has('x-api-key') || req.headers.get('authorization')?.startsWith('Bearer ')
      if (hasExplicitCredentials) {
        return NextResponse.json({ error: auth.error || 'Unauthorized' }, { status: 401 })
      }

      const { db: dbClient, workflow: workflowTable } = await import('@sim/db')
      const { eq } = await import('drizzle-orm')
      const [wf] = await dbClient
        .select({
          isPublicApi: workflowTable.isPublicApi,
          isDeployed: workflowTable.isDeployed,
          userId: workflowTable.userId,
        })
        .from(workflowTable)
        .where(eq(workflowTable.id, workflowId))
        .limit(1)

      if (!wf?.isPublicApi || !wf.isDeployed) {
        return NextResponse.json({ error: auth.error || 'Unauthorized' }, { status: 401 })
      }

      const { isPublicApiDisabled } = await import('@/lib/core/config/feature-flags')
      if (isPublicApiDisabled) {
        return NextResponse.json({ error: auth.error || 'Unauthorized' }, { status: 401 })
      }

      const { getUserPermissionConfig } = await import('@/ee/access-control/utils/permission-check')
      const ownerConfig = await getUserPermissionConfig(wf.userId)
      if (ownerConfig?.disablePublicApi) {
        return NextResponse.json({ error: auth.error || 'Unauthorized' }, { status: 401 })
      }

      userId = wf.userId
      isPublicApiAccess = true
    } else {
      userId = auth.userId
    }

    let body: any = {}
    try {
      const text = await req.text()
      if (text) {
        body = JSON.parse(text)
      }
    } catch (error) {
      reqLogger.warn('Failed to parse request body, using defaults')
    }

    const validation = ExecuteWorkflowSchema.safeParse(body)
    if (!validation.success) {
      reqLogger.warn('Invalid request body:', validation.error.errors)
      return NextResponse.json(
        {
          error: 'Invalid request body',
          details: validation.error.errors.map((e) => ({
            path: e.path.join('.'),
            message: e.message,
          })),
        },
        { status: 400 }
      )
    }

    const defaultTriggerType =
      isPublicApiAccess || auth.authType === AuthType.API_KEY ? 'api' : 'manual'

    const {
      selectedOutputs,
      triggerType = defaultTriggerType,
      stream: streamParam,
      useDraftState,
      input: validatedInput,
      isClientSession = false,
      includeFileBase64,
      base64MaxBytes,
      workflowStateOverride,
      stopAfterBlockId,
      runFromBlock: rawRunFromBlock,
    } = validation.data

    if (isPublicApiAccess && isClientSession) {
      return NextResponse.json(
        { error: 'Public API callers cannot set isClientSession' },
        { status: 400 }
      )
    }

    if (auth.authType === 'api_key') {
      if (isClientSession) {
        return NextResponse.json(
          { error: 'API key callers cannot set isClientSession' },
          { status: 400 }
        )
      }

      if (workflowStateOverride) {
        return NextResponse.json(
          { error: 'API key callers cannot provide workflowStateOverride' },
          { status: 400 }
        )
      }

      if (useDraftState) {
        return NextResponse.json(
          { error: 'API key callers cannot execute draft workflow state' },
          { status: 400 }
        )
      }
    }

    // Resolve runFromBlock snapshot from executionId if needed
    let resolvedRunFromBlock:
      | { startBlockId: string; sourceSnapshot: SerializableExecutionState }
      | undefined
    if (rawRunFromBlock) {
      if (rawRunFromBlock.sourceSnapshot && auth.authType === 'api_key') {
        return NextResponse.json(
          { error: 'API key callers cannot provide runFromBlock.sourceSnapshot' },
          { status: 400 }
        )
      }

      if (rawRunFromBlock.executionId && (auth.authType === 'api_key' || isPublicApiAccess)) {
        return NextResponse.json(
          { error: 'External callers cannot resume from stored execution snapshots' },
          { status: 400 }
        )
      }

      if (rawRunFromBlock.sourceSnapshot && !isPublicApiAccess) {
        // Public API callers cannot inject arbitrary block state via sourceSnapshot.
        // They must use executionId to resume from a server-stored execution state.
        resolvedRunFromBlock = {
          startBlockId: rawRunFromBlock.startBlockId,
          sourceSnapshot: rawRunFromBlock.sourceSnapshot as SerializableExecutionState,
        }
      } else if (rawRunFromBlock.executionId) {
        const { getExecutionStateForWorkflow, getLatestExecutionState } = await import(
          '@/lib/workflows/executor/execution-state'
        )
        const snapshot =
          rawRunFromBlock.executionId === 'latest'
            ? await getLatestExecutionState(workflowId)
            : await getExecutionStateForWorkflow(rawRunFromBlock.executionId, workflowId)
        if (!snapshot) {
          return NextResponse.json(
            {
              error: `No execution state found for ${rawRunFromBlock.executionId === 'latest' ? 'workflow' : `execution ${rawRunFromBlock.executionId}`}. Run the full workflow first.`,
            },
            { status: 400 }
          )
        }
        resolvedRunFromBlock = {
          startBlockId: rawRunFromBlock.startBlockId,
          sourceSnapshot: snapshot,
        }
      } else {
        return NextResponse.json(
          { error: 'runFromBlock requires either sourceSnapshot or executionId' },
          { status: 400 }
        )
      }
    }

    // For API key and internal JWT auth, the entire body is the input (except for our control fields)
    // For session auth, the input is explicitly provided in the input field
    const input =
      isPublicApiAccess ||
      auth.authType === AuthType.API_KEY ||
      auth.authType === AuthType.INTERNAL_JWT
        ? (() => {
            const {
              selectedOutputs,
              triggerType,
              stream,
              useDraftState,
              includeFileBase64,
              base64MaxBytes,
              workflowStateOverride,
              stopAfterBlockId: _stopAfterBlockId,
              runFromBlock: _runFromBlock,
              workflowId: _workflowId, // Also exclude workflowId used for internal JWT auth
              ...rest
            } = body
            return Object.keys(rest).length > 0 ? rest : validatedInput
          })()
        : validatedInput

    // Public API callers must not inject arbitrary workflow state overrides (code injection risk).
    // stopAfterBlockId and runFromBlock are safe — they control execution flow within the deployed state.
    const sanitizedWorkflowStateOverride = isPublicApiAccess ? undefined : workflowStateOverride

    // Public API callers always execute the deployed state, never the draft.
    const shouldUseDraftState = isPublicApiAccess
      ? false
      : (useDraftState ?? auth.authType === AuthType.SESSION)
    const streamHeader = req.headers.get('X-Stream-Response') === 'true'
    const enableSSE = streamHeader || streamParam === true
    const executionModeHeader = req.headers.get('X-Execution-Mode')
    const isAsyncMode = executionModeHeader === 'async'
    const requiresWriteExecutionAccess = Boolean(
      useDraftState || workflowStateOverride || rawRunFromBlock
    )

    if (
      isAsyncMode &&
      (body.useDraftState !== undefined ||
        body.workflowStateOverride !== undefined ||
        body.runFromBlock !== undefined ||
        body.stopAfterBlockId !== undefined ||
        body.selectedOutputs?.length ||
        body.includeFileBase64 !== undefined ||
        body.base64MaxBytes !== undefined)
    ) {
      return NextResponse.json(
        { error: 'Async execution does not support draft or override execution controls' },
        { status: 400 }
      )
    }

    const executionId = generateId()
    reqLogger = reqLogger.withMetadata({ userId, executionId })

    reqLogger.info('Starting server-side execution', {
      hasInput: !!input,
      triggerType,
      authType: auth.authType,
      streamParam,
      streamHeader,
      enableSSE,
      isAsyncMode,
    })
    let loggingTriggerType: CoreTriggerType = 'manual'
    if (CORE_TRIGGER_TYPES.includes(triggerType as CoreTriggerType)) {
      loggingTriggerType = triggerType as CoreTriggerType
    }
    const loggingSession = new LoggingSession(
      workflowId,
      executionId,
      loggingTriggerType,
      requestId
    )

    // Client-side sessions and personal API keys bill/permission-check the
    // authenticated user, not the workspace billed account.
    const useAuthenticatedUserAsActor =
      isClientSession || (auth.authType === AuthType.API_KEY && auth.apiKeyType === 'personal')

    // Authorization fetches the full workflow record and checks workspace permissions.
    // Run it first so we can pass the record to preprocessing (eliminates a duplicate DB query).
    const workflowAuthorization = await authorizeWorkflowByWorkspacePermission({
      workflowId,
      userId,
      action: requiresWriteExecutionAccess ? 'write' : 'read',
    })
    if (!workflowAuthorization.allowed) {
      return NextResponse.json(
        { error: workflowAuthorization.message || 'Access denied' },
        { status: workflowAuthorization.status }
      )
    }

    // Pass the pre-fetched workflow record to skip the redundant Step 1 DB query in preprocessing.
    const preprocessResult = await preprocessExecution({
      workflowId,
      userId,
      triggerType: loggingTriggerType,
      executionId,
      requestId,
      checkDeployment: !shouldUseDraftState,
      loggingSession,
      useDraftState: shouldUseDraftState,
      useAuthenticatedUserAsActor,
      workflowRecord: workflowAuthorization.workflow ?? undefined,
    })

    if (!preprocessResult.success) {
      return NextResponse.json(
        { error: preprocessResult.error!.message },
        { status: preprocessResult.error!.statusCode }
      )
    }

    const actorUserId = preprocessResult.actorUserId!
    const workflow = preprocessResult.workflowRecord!

    if (!workflow.workspaceId) {
      reqLogger.error('Workflow has no workspaceId')
      return NextResponse.json({ error: 'Workflow has no associated workspace' }, { status: 500 })
    }
    const workspaceId = workflow.workspaceId
    reqLogger = reqLogger.withMetadata({ workspaceId, userId: actorUserId })

    if (auth.apiKeyType === 'workspace' && auth.workspaceId !== workspaceId) {
      return NextResponse.json(
        { error: 'API key is not authorized for this workspace' },
        { status: 403 }
      )
    }

    reqLogger.info('Preprocessing passed')

    if (isAsyncMode) {
      return handleAsyncExecution({
        requestId,
        workflowId,
        userId: actorUserId,
        workspaceId,
        input,
        triggerType: loggingTriggerType,
        executionId,
        callChain,
      })
    }

    let cachedWorkflowData: {
      blocks: Record<string, any>
      edges: any[]
      loops: Record<string, any>
      parallels: Record<string, any>
      deploymentVersionId?: string
      variables?: Record<string, any>
    } | null = null

    let processedInput = input
    try {
      const workflowData = shouldUseDraftState
        ? await loadWorkflowFromNormalizedTables(workflowId)
        : await loadDeployedWorkflowState(workflowId, workspaceId)

      if (workflowData) {
        const deployedVariables =
          !shouldUseDraftState && 'variables' in workflowData
            ? (workflowData as any).variables
            : undefined

        cachedWorkflowData = {
          blocks: workflowData.blocks,
          edges: workflowData.edges,
          loops: workflowData.loops || {},
          parallels: workflowData.parallels || {},
          deploymentVersionId:
            !shouldUseDraftState && 'deploymentVersionId' in workflowData
              ? (workflowData.deploymentVersionId as string)
              : undefined,
          variables: deployedVariables,
        }

        const serializedWorkflow = new Serializer().serializeWorkflow(
          workflowData.blocks,
          workflowData.edges,
          workflowData.loops,
          workflowData.parallels,
          false
        )

        const executionContext = {
          workspaceId,
          workflowId,
          executionId,
        }

        processedInput = await processInputFileFields(
          input,
          serializedWorkflow.blocks,
          executionContext,
          requestId,
          actorUserId
        )
      }
    } catch (fileError) {
      reqLogger.error('Failed to process input file fields:', fileError)

      await loggingSession.safeStart({
        userId: actorUserId,
        workspaceId,
        variables: {},
      })

      await loggingSession.safeCompleteWithError({
        error: {
          message: `File processing failed: ${fileError instanceof Error ? fileError.message : 'Unable to process input files'}`,
          stackTrace: fileError instanceof Error ? fileError.stack : undefined,
        },
        traceSpans: [],
      })

      return NextResponse.json(
        {
          error: `File processing failed: ${fileError instanceof Error ? fileError.message : 'Unable to process input files'}`,
        },
        { status: 400 }
      )
    }

    const effectiveWorkflowStateOverride =
      sanitizedWorkflowStateOverride || cachedWorkflowData || undefined

    if (!enableSSE) {
      reqLogger.info('Using non-SSE execution (direct JSON response)')
      const metadata: ExecutionMetadata = {
        requestId,
        executionId,
        workflowId,
        workspaceId,
        userId: actorUserId,
        sessionUserId: isClientSession ? userId : undefined,
        workflowUserId: workflow.userId,
        triggerType,
        useDraftState: shouldUseDraftState,
        startTime: new Date().toISOString(),
        isClientSession,
        enforceCredentialAccess: useAuthenticatedUserAsActor,
        workflowStateOverride: effectiveWorkflowStateOverride,
        callChain,
      }

      const executionVariables = cachedWorkflowData?.variables ?? workflow.variables ?? {}

      if (shouldUseBullMQ() && !INLINE_TRIGGER_TYPES.has(triggerType)) {
        try {
          const dispatchJobId = await enqueueDirectWorkflowExecution(
            {
              workflow,
              metadata,
              input: processedInput,
              variables: executionVariables,
              selectedOutputs,
              includeFileBase64,
              base64MaxBytes,
              stopAfterBlockId,
              timeoutMs: preprocessResult.executionTimeout?.sync,
              runFromBlock: resolvedRunFromBlock,
            },
            5,
            'interactive'
          )

          const resultRecord = await waitForDispatchJob(
            dispatchJobId,
            (preprocessResult.executionTimeout?.sync ?? 300000) + 30000
          )

          if (resultRecord.status === 'failed') {
            return NextResponse.json(
              {
                success: false,
                executionId,
                error: resultRecord.error ?? 'Workflow execution failed',
              },
              { status: 500 }
            )
          }

          const result = resultRecord.output as QueuedWorkflowExecutionResult

          const resultForResponseBlock = {
            success: result.success,
            logs: result.logs,
            output: result.output,
          }

          if (
            auth.authType !== AuthType.INTERNAL_JWT &&
            workflowHasResponseBlock(resultForResponseBlock)
          ) {
            return createHttpResponseFromBlock(resultForResponseBlock)
          }

          return NextResponse.json(
            {
              success: result.success,
              executionId,
              output: result.output,
              error: result.error,
              metadata: result.metadata,
            },
            { status: result.statusCode ?? 200 }
          )
        } catch (error: unknown) {
          if (error instanceof DispatchQueueFullError) {
            return NextResponse.json(
              {
                error: 'Service temporarily at capacity',
                message: error.message,
                retryAfterSeconds: 10,
              },
              { status: 503, headers: { 'Retry-After': '10' } }
            )
          }

          const errorMessage = error instanceof Error ? error.message : 'Unknown error'

          reqLogger.error(`Queued non-SSE execution failed: ${errorMessage}`)

          return NextResponse.json(
            {
              success: false,
              error: errorMessage,
            },
            { status: 500 }
          )
        }
      }

      const timeoutController = createTimeoutAbortController(
        preprocessResult.executionTimeout?.sync
      )

      try {
        const snapshot = new ExecutionSnapshot(
          metadata,
          workflow,
          processedInput,
          executionVariables,
          selectedOutputs
        )

        const result = await executeWorkflowCore({
          snapshot,
          callbacks: {},
          loggingSession,
          includeFileBase64,
          base64MaxBytes,
          stopAfterBlockId,
          runFromBlock: resolvedRunFromBlock,
          abortSignal: timeoutController.signal,
        })

        await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession })

        if (
          result.status === 'cancelled' &&
          timeoutController.isTimedOut() &&
          timeoutController.timeoutMs
        ) {
          const timeoutErrorMessage = getTimeoutErrorMessage(null, timeoutController.timeoutMs)
          reqLogger.info('Non-SSE execution timed out', {
            timeoutMs: timeoutController.timeoutMs,
          })
          await loggingSession.markAsFailed(timeoutErrorMessage)

          return NextResponse.json(
            {
              success: false,
              output: result.output,
              error: timeoutErrorMessage,
              metadata: result.metadata
                ? {
                    duration: result.metadata.duration,
                    startTime: result.metadata.startTime,
                    endTime: result.metadata.endTime,
                  }
                : undefined,
            },
            { status: 408 }
          )
        }

        const outputWithBase64 = includeFileBase64
          ? ((await hydrateUserFilesWithBase64(result.output, {
              requestId,
              executionId,
              maxBytes: base64MaxBytes,
            })) as NormalizedBlockOutput)
          : result.output

        const resultWithBase64 = { ...result, output: outputWithBase64 }

        if (auth.authType !== AuthType.INTERNAL_JWT && workflowHasResponseBlock(resultWithBase64)) {
          return createHttpResponseFromBlock(resultWithBase64)
        }

        const filteredResult = {
          success: result.success,
          executionId,
          output: outputWithBase64,
          error: result.error,
          metadata: result.metadata
            ? {
                duration: result.metadata.duration,
                startTime: result.metadata.startTime,
                endTime: result.metadata.endTime,
              }
            : undefined,
        }

        return NextResponse.json(filteredResult)
      } catch (error: unknown) {
        const errorMessage = error instanceof Error ? error.message : 'Unknown error'

        reqLogger.error(`Non-SSE execution failed: ${errorMessage}`)

        const executionResult = hasExecutionResult(error) ? error.executionResult : undefined

        return NextResponse.json(
          {
            success: false,
            output: executionResult?.output,
            error: executionResult?.error || errorMessage || 'Execution failed',
            metadata: executionResult?.metadata
              ? {
                  duration: executionResult.metadata.duration,
                  startTime: executionResult.metadata.startTime,
                  endTime: executionResult.metadata.endTime,
                }
              : undefined,
          },
          { status: 500 }
        )
      } finally {
        timeoutController.cleanup()
        if (executionId) {
          void cleanupExecutionBase64Cache(executionId).catch((error) => {
            reqLogger.error('Failed to cleanup base64 cache', { error })
          })
        }
      }
    }

    if (shouldUseDraftState) {
      const shouldDispatchViaQueue = shouldUseBullMQ() && !INLINE_TRIGGER_TYPES.has(triggerType)
      if (shouldDispatchViaQueue) {
        const metadata: ExecutionMetadata = {
          requestId,
          executionId,
          workflowId,
          workspaceId,
          userId: actorUserId,
          sessionUserId: isClientSession ? userId : undefined,
          workflowUserId: workflow.userId,
          triggerType,
          useDraftState: shouldUseDraftState,
          startTime: new Date().toISOString(),
          isClientSession,
          enforceCredentialAccess: useAuthenticatedUserAsActor,
          workflowStateOverride: effectiveWorkflowStateOverride,
          callChain,
        }

        const executionVariables = cachedWorkflowData?.variables ?? workflow.variables ?? {}

        await enqueueDirectWorkflowExecution(
          {
            workflow,
            metadata,
            input: processedInput,
            variables: executionVariables,
            selectedOutputs,
            includeFileBase64,
            base64MaxBytes,
            stopAfterBlockId,
            timeoutMs: preprocessResult.executionTimeout?.sync,
            runFromBlock: resolvedRunFromBlock,
            streamEvents: true,
          },
          1,
          'interactive'
        )

        return new NextResponse(createBufferedExecutionStream(executionId), {
          headers: {
            ...SSE_HEADERS,
            'X-Execution-Id': executionId,
          },
        })
      }

      reqLogger.info('Using SSE console log streaming (manual execution)')
    } else {
      reqLogger.info('Using streaming API response')

      const resolvedSelectedOutputs = resolveOutputIds(
        selectedOutputs,
        cachedWorkflowData?.blocks || {}
      )
      const streamVariables = cachedWorkflowData?.variables ?? (workflow as any).variables
      const stream = await createStreamingResponse({
        requestId,
        workflow: {
          id: workflow.id,
          userId: actorUserId,
          workspaceId,
          isDeployed: workflow.isDeployed,
          variables: streamVariables,
        },
        input: processedInput,
        executingUserId: actorUserId,
        streamConfig: {
          selectedOutputs: resolvedSelectedOutputs,
          isSecureMode: false,
          workflowTriggerType: triggerType === 'chat' ? 'chat' : 'api',
          includeFileBase64,
          base64MaxBytes,
          timeoutMs: preprocessResult.executionTimeout?.sync,
        },
        executionId,
      })

      return new NextResponse(stream, {
        status: 200,
        headers: SSE_HEADERS,
      })
    }

    const encoder = new TextEncoder()
    const timeoutController = createTimeoutAbortController(preprocessResult.executionTimeout?.sync)
    let isStreamClosed = false
    let isManualAbortRegistered = false

    const eventWriter = createExecutionEventWriter(executionId)
    setExecutionMeta(executionId, {
      status: 'active',
      userId: actorUserId,
      workflowId,
    }).catch(() => {})

    const stream = new ReadableStream<Uint8Array>({
      async start(controller) {
        let finalMetaStatus: 'complete' | 'error' | 'cancelled' | null = null

        registerManualExecutionAborter(executionId, timeoutController.abort)
        isManualAbortRegistered = true

        let localEventSeq = 0
        const sendEvent = (event: ExecutionEvent) => {
          const isBuffered = event.type !== 'stream:chunk' && event.type !== 'stream:done'
          if (isBuffered) {
            localEventSeq++
            event.eventId = localEventSeq
          }
          if (!isStreamClosed) {
            try {
              controller.enqueue(encodeSSEEvent(event))
            } catch {
              isStreamClosed = true
            }
          }
          if (isBuffered) {
            eventWriter.write(event).catch(() => {})
          }
        }

        try {
          const startTime = new Date()

          sendEvent({
            type: 'execution:started',
            timestamp: startTime.toISOString(),
            executionId,
            workflowId,
            data: {
              startTime: startTime.toISOString(),
            },
          })

          const onBlockStart = async (
            blockId: string,
            blockName: string,
            blockType: string,
            executionOrder: number,
            iterationContext?: IterationContext,
            childWorkflowContext?: ChildWorkflowContext
          ) => {
            reqLogger.info('onBlockStart called', { blockId, blockName, blockType })
            sendEvent({
              type: 'block:started',
              timestamp: new Date().toISOString(),
              executionId,
              workflowId,
              data: {
                blockId,
                blockName,
                blockType,
                executionOrder,
                ...(iterationContext && {
                  iterationCurrent: iterationContext.iterationCurrent,
                  iterationTotal: iterationContext.iterationTotal,
                  iterationType: iterationContext.iterationType,
                  iterationContainerId: iterationContext.iterationContainerId,
                  ...(iterationContext.parentIterations?.length && {
                    parentIterations: iterationContext.parentIterations,
                  }),
                }),
                ...(childWorkflowContext && {
                  childWorkflowBlockId: childWorkflowContext.parentBlockId,
                  childWorkflowName: childWorkflowContext.workflowName,
                }),
              },
            })
          }

          const onBlockComplete = async (
            blockId: string,
            blockName: string,
            blockType: string,
            callbackData: any,
            iterationContext?: IterationContext,
            childWorkflowContext?: ChildWorkflowContext
          ) => {
            const hasError = callbackData.output?.error
            const childWorkflowData = childWorkflowContext
              ? {
                  childWorkflowBlockId: childWorkflowContext.parentBlockId,
                  childWorkflowName: childWorkflowContext.workflowName,
                }
              : {}

            const instanceData = callbackData.childWorkflowInstanceId
              ? { childWorkflowInstanceId: callbackData.childWorkflowInstanceId }
              : {}

            if (hasError) {
              reqLogger.info('onBlockComplete (error) called', {
                blockId,
                blockName,
                blockType,
                error: callbackData.output.error,
              })
              sendEvent({
                type: 'block:error',
                timestamp: new Date().toISOString(),
                executionId,
                workflowId,
                data: {
                  blockId,
                  blockName,
                  blockType,
                  input: callbackData.input,
                  error: callbackData.output.error,
                  durationMs: callbackData.executionTime || 0,
                  startedAt: callbackData.startedAt,
                  executionOrder: callbackData.executionOrder,
                  endedAt: callbackData.endedAt,
                  ...(iterationContext && {
                    iterationCurrent: iterationContext.iterationCurrent,
                    iterationTotal: iterationContext.iterationTotal,
                    iterationType: iterationContext.iterationType,
                    iterationContainerId: iterationContext.iterationContainerId,
                    ...(iterationContext.parentIterations?.length && {
                      parentIterations: iterationContext.parentIterations,
                    }),
                  }),
                  ...childWorkflowData,
                  ...instanceData,
                },
              })
            } else {
              reqLogger.info('onBlockComplete called', {
                blockId,
                blockName,
                blockType,
              })
              sendEvent({
                type: 'block:completed',
                timestamp: new Date().toISOString(),
                executionId,
                workflowId,
                data: {
                  blockId,
                  blockName,
                  blockType,
                  input: callbackData.input,
                  output: callbackData.output,
                  durationMs: callbackData.executionTime || 0,
                  startedAt: callbackData.startedAt,
                  executionOrder: callbackData.executionOrder,
                  endedAt: callbackData.endedAt,
                  ...(iterationContext && {
                    iterationCurrent: iterationContext.iterationCurrent,
                    iterationTotal: iterationContext.iterationTotal,
                    iterationType: iterationContext.iterationType,
                    iterationContainerId: iterationContext.iterationContainerId,
                    ...(iterationContext.parentIterations?.length && {
                      parentIterations: iterationContext.parentIterations,
                    }),
                  }),
                  ...childWorkflowData,
                  ...instanceData,
                },
              })
            }
          }

          const onStream = async (streamingExec: StreamingExecution) => {
            const blockId = (streamingExec.execution as any).blockId

            const reader = streamingExec.stream.getReader()
            const decoder = new TextDecoder()

            try {
              while (true) {
                const { done, value } = await reader.read()
                if (done) break

                const chunk = decoder.decode(value, { stream: true })
                sendEvent({
                  type: 'stream:chunk',
                  timestamp: new Date().toISOString(),
                  executionId,
                  workflowId,
                  data: { blockId, chunk },
                })
              }

              sendEvent({
                type: 'stream:done',
                timestamp: new Date().toISOString(),
                executionId,
                workflowId,
                data: { blockId },
              })
            } catch (error) {
              reqLogger.error('Error streaming block content:', error)
            } finally {
              try {
                await reader.cancel().catch(() => {})
              } catch {}
            }
          }

          const metadata: ExecutionMetadata = {
            requestId,
            executionId,
            workflowId,
            workspaceId,
            userId: actorUserId,
            sessionUserId: isClientSession ? userId : undefined,
            workflowUserId: workflow.userId,
            triggerType,
            useDraftState: shouldUseDraftState,
            startTime: new Date().toISOString(),
            isClientSession,
            enforceCredentialAccess: useAuthenticatedUserAsActor,
            workflowStateOverride: effectiveWorkflowStateOverride,
            callChain,
          }

          const sseExecutionVariables = cachedWorkflowData?.variables ?? workflow.variables ?? {}

          const snapshot = new ExecutionSnapshot(
            metadata,
            workflow,
            processedInput,
            sseExecutionVariables,
            selectedOutputs
          )

          const onChildWorkflowInstanceReady = (
            blockId: string,
            childWorkflowInstanceId: string,
            iterationContext?: IterationContext,
            executionOrder?: number
          ) => {
            sendEvent({
              type: 'block:childWorkflowStarted',
              timestamp: new Date().toISOString(),
              executionId,
              workflowId,
              data: {
                blockId,
                childWorkflowInstanceId,
                ...(iterationContext && {
                  iterationCurrent: iterationContext.iterationCurrent,
                  iterationContainerId: iterationContext.iterationContainerId,
                }),
                ...(executionOrder !== undefined && { executionOrder }),
              },
            })
          }

          const result = await executeWorkflowCore({
            snapshot,
            callbacks: {
              onBlockStart,
              onBlockComplete,
              onStream,
              onChildWorkflowInstanceReady,
            },
            loggingSession,
            abortSignal: timeoutController.signal,
            includeFileBase64,
            base64MaxBytes,
            stopAfterBlockId,
            runFromBlock: resolvedRunFromBlock,
          })

          await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession })

          if (result.status === 'cancelled') {
            if (timeoutController.isTimedOut() && timeoutController.timeoutMs) {
              const timeoutErrorMessage = getTimeoutErrorMessage(null, timeoutController.timeoutMs)
              reqLogger.info('Workflow execution timed out', {
                timeoutMs: timeoutController.timeoutMs,
              })

              await loggingSession.markAsFailed(timeoutErrorMessage)

              sendEvent({
                type: 'execution:error',
                timestamp: new Date().toISOString(),
                executionId,
                workflowId,
                data: {
                  error: timeoutErrorMessage,
                  duration: result.metadata?.duration || 0,
                },
              })
              finalMetaStatus = 'error'
            } else {
              reqLogger.info('Workflow execution was cancelled')

              sendEvent({
                type: 'execution:cancelled',
                timestamp: new Date().toISOString(),
                executionId,
                workflowId,
                data: {
                  duration: result.metadata?.duration || 0,
                },
              })
              finalMetaStatus = 'cancelled'
            }
            return
          }

          const sseOutput = includeFileBase64
            ? await hydrateUserFilesWithBase64(result.output, {
                requestId,
                executionId,
                maxBytes: base64MaxBytes,
              })
            : result.output

          if (result.status === 'paused') {
            sendEvent({
              type: 'execution:paused',
              timestamp: new Date().toISOString(),
              executionId,
              workflowId,
              data: {
                output: sseOutput,
                duration: result.metadata?.duration || 0,
                startTime: result.metadata?.startTime || startTime.toISOString(),
                endTime: result.metadata?.endTime || new Date().toISOString(),
              },
            })
          } else {
            sendEvent({
              type: 'execution:completed',
              timestamp: new Date().toISOString(),
              executionId,
              workflowId,
              data: {
                success: result.success,
                output: sseOutput,
                duration: result.metadata?.duration || 0,
                startTime: result.metadata?.startTime || startTime.toISOString(),
                endTime: result.metadata?.endTime || new Date().toISOString(),
              },
            })
          }
          finalMetaStatus = 'complete'
        } catch (error: unknown) {
          const isTimeout = isTimeoutError(error) || timeoutController.isTimedOut()
          const errorMessage = isTimeout
            ? getTimeoutErrorMessage(error, timeoutController.timeoutMs)
            : error instanceof Error
              ? error.message
              : 'Unknown error'

          reqLogger.error(`SSE execution failed: ${errorMessage}`, { isTimeout })

          const executionResult = hasExecutionResult(error) ? error.executionResult : undefined

          sendEvent({
            type: 'execution:error',
            timestamp: new Date().toISOString(),
            executionId,
            workflowId,
            data: {
              error: executionResult?.error || errorMessage,
              duration: executionResult?.metadata?.duration || 0,
            },
          })
          finalMetaStatus = 'error'
        } finally {
          if (isManualAbortRegistered) {
            unregisterManualExecutionAborter(executionId)
            isManualAbortRegistered = false
          }
          try {
            await eventWriter.close()
          } catch (closeError) {
            reqLogger.warn('Failed to close event writer', {
              error: closeError instanceof Error ? closeError.message : String(closeError),
            })
          }
          if (finalMetaStatus) {
            setExecutionMeta(executionId, { status: finalMetaStatus }).catch(() => {})
          }
          timeoutController.cleanup()
          if (executionId) {
            await cleanupExecutionBase64Cache(executionId)
          }
          if (!isStreamClosed) {
            try {
              controller.enqueue(encoder.encode('data: [DONE]\n\n'))
              controller.close()
            } catch {}
          }
        }
      },
      cancel() {
        isStreamClosed = true
        reqLogger.info('Client disconnected from SSE stream')
      },
    })

    return new NextResponse(stream, {
      headers: {
        ...SSE_HEADERS,
        'X-Execution-Id': executionId,
      },
    })
  } catch (error: any) {
    if (error instanceof DispatchQueueFullError) {
      return NextResponse.json(
        {
          error: 'Service temporarily at capacity',
          message: error.message,
          retryAfterSeconds: 10,
        },
        { status: 503, headers: { 'Retry-After': '10' } }
      )
    }

    reqLogger.error('Failed to start workflow execution:', error)
    return NextResponse.json(
      { error: error.message || 'Failed to start workflow execution' },
      { status: 500 }
    )
  }
}
