import { createLogger } from '@sim/logger'
import { isLargeValueRef } from '@/lib/execution/payloads/large-value-ref'
import { EDGE } from '@/executor/constants'
import type { DAG, DAGNode } from '@/executor/dag/builder'
import type { BlockExecutor } from '@/executor/execution/block-executor'
import type { BlockStateController } from '@/executor/execution/types'
import type { LoopOrchestrator } from '@/executor/orchestrators/loop'
import type { ParallelOrchestrator } from '@/executor/orchestrators/parallel'
import type { ExecutionContext, NormalizedBlockOutput } from '@/executor/types'
import { extractBaseBlockId } from '@/executor/utils/subflow-utils'

const logger = createLogger('NodeExecutionOrchestrator')

function getResultCount(value: unknown): number {
  if (isLargeValueRef(value)) {
    const preview = value.preview
    if (
      preview &&
      typeof preview === 'object' &&
      typeof (preview as Record<string, unknown>).length === 'number'
    ) {
      return (preview as { length: number }).length
    }
  }
  return Array.isArray(value) ? value.length : 0
}

export interface NodeExecutionResult {
  nodeId: string
  output: NormalizedBlockOutput
  isFinalOutput: boolean
}

export class NodeExecutionOrchestrator {
  constructor(
    private dag: DAG,
    private state: BlockStateController,
    private blockExecutor: BlockExecutor,
    private loopOrchestrator: LoopOrchestrator,
    private parallelOrchestrator: ParallelOrchestrator
  ) {}

  async executeNode(ctx: ExecutionContext, nodeId: string): Promise<NodeExecutionResult> {
    const node = this.dag.nodes.get(nodeId)
    if (!node) {
      throw new Error(`Node not found in DAG: ${nodeId}`)
    }

    if (ctx.runFromBlockContext && !ctx.runFromBlockContext.dirtySet.has(nodeId)) {
      const cachedOutput = this.state.getBlockOutput(nodeId) || {}
      logger.debug('Skipping non-dirty block in run-from-block mode', { nodeId })
      return {
        nodeId,
        output: cachedOutput,
        isFinalOutput: false,
      }
    }

    const isDirtyBlock = ctx.runFromBlockContext?.dirtySet.has(nodeId) ?? false
    if (!isDirtyBlock && this.state.hasExecuted(nodeId)) {
      const output = this.state.getBlockOutput(nodeId) || {}
      return {
        nodeId,
        output,
        isFinalOutput: false,
      }
    }

    const loopId = node.metadata.loopId
    if (loopId && !this.loopOrchestrator.getLoopScope(ctx, loopId)) {
      await this.loopOrchestrator.initializeLoopScope(ctx, loopId)
    }

    const parallelId = node.metadata.parallelId
    if (parallelId && !this.parallelOrchestrator.getParallelScope(ctx, parallelId)) {
      await this.parallelOrchestrator.initializeParallelScope(ctx, parallelId)
    }

    if (node.metadata.isSentinel) {
      const output = await this.handleSentinel(ctx, node)
      const isFinalOutput = node.outgoingEdges.size === 0
      return {
        nodeId,
        output,
        isFinalOutput,
      }
    }

    const output = await this.blockExecutor.execute(ctx, node, node.block)
    const isFinalOutput = node.outgoingEdges.size === 0
    return {
      nodeId,
      output,
      isFinalOutput,
    }
  }

  private async handleSentinel(
    ctx: ExecutionContext,
    node: DAGNode
  ): Promise<NormalizedBlockOutput> {
    const sentinelType = node.metadata.sentinelType
    const loopId = node.metadata.loopId
    const parallelId = node.metadata.parallelId
    const isParallelSentinel = node.metadata.isParallelSentinel

    if (isParallelSentinel) {
      return await this.handleParallelSentinel(ctx, node, sentinelType, parallelId)
    }

    switch (sentinelType) {
      case 'start': {
        if (loopId) {
          const shouldExecute = await this.loopOrchestrator.evaluateInitialCondition(ctx, loopId)
          if (!shouldExecute) {
            logger.info('Loop initial condition false, skipping loop body', { loopId })
            return {
              sentinelStart: true,
              shouldExit: true,
              selectedRoute: EDGE.LOOP_EXIT,
            }
          }
        }
        return { sentinelStart: true }
      }

      case 'end': {
        if (!loopId) {
          logger.warn('Sentinel end called without loopId')
          return { shouldExit: true, selectedRoute: EDGE.LOOP_EXIT }
        }

        const continuationResult = await this.loopOrchestrator.evaluateLoopContinuation(ctx, loopId)

        if (continuationResult.shouldContinue) {
          return {
            shouldContinue: true,
            shouldExit: false,
            selectedRoute: continuationResult.selectedRoute,
          }
        }

        return {
          results: continuationResult.aggregatedResults || [],
          shouldContinue: false,
          shouldExit: true,
          selectedRoute: continuationResult.selectedRoute,
          totalIterations:
            continuationResult.totalIterations ??
            getResultCount(continuationResult.aggregatedResults),
        }
      }

      default:
        logger.warn('Unknown sentinel type', { sentinelType })
        return {}
    }
  }

  private async handleParallelSentinel(
    ctx: ExecutionContext,
    node: DAGNode,
    sentinelType: string | undefined,
    parallelId: string | undefined
  ): Promise<NormalizedBlockOutput> {
    if (!parallelId) {
      logger.warn('Parallel sentinel called without parallelId')
      return {}
    }

    if (sentinelType === 'start') {
      if (!this.parallelOrchestrator.getParallelScope(ctx, parallelId)) {
        const parallelConfig = this.dag.parallelConfigs.get(parallelId)
        if (parallelConfig) {
          await this.parallelOrchestrator.initializeParallelScope(ctx, parallelId)
        }
      }

      const scope = this.parallelOrchestrator.getParallelScope(ctx, parallelId)
      if (scope?.isEmpty) {
        logger.info('Parallel has empty distribution, skipping parallel body', { parallelId })
        return {
          sentinelStart: true,
          shouldExit: true,
          selectedRoute: EDGE.PARALLEL_EXIT,
        }
      }

      return { sentinelStart: true }
    }

    if (sentinelType === 'end') {
      const result = await this.parallelOrchestrator.aggregateParallelResults(ctx, parallelId)
      if (!result.allBranchesComplete) {
        return {
          results: [],
          sentinelEnd: true,
          selectedRoute: EDGE.PARALLEL_CONTINUE,
          totalBranches: result.totalBranches,
        }
      }
      return {
        results: result.results || [],
        sentinelEnd: true,
        selectedRoute: EDGE.PARALLEL_EXIT,
        totalBranches: result.totalBranches,
      }
    }

    logger.warn('Unknown parallel sentinel type', { sentinelType })
    return {}
  }

  async handleNodeCompletion(
    ctx: ExecutionContext,
    nodeId: string,
    output: NormalizedBlockOutput
  ): Promise<void> {
    const node = this.dag.nodes.get(nodeId)
    if (!node) {
      logger.error('Node not found during completion handling', { nodeId })
      return
    }

    const loopId = node.metadata.loopId
    const isParallelBranch = node.metadata.isParallelBranch
    const isSentinel = node.metadata.isSentinel
    if (isSentinel) {
      this.handleRegularNodeCompletion(ctx, node, output)
    } else if (loopId) {
      this.handleLoopNodeCompletion(ctx, node, output, loopId)
    } else if (isParallelBranch) {
      const parallelId = this.findParallelIdForNode(node.id)
      if (parallelId) {
        await this.handleParallelNodeCompletion(ctx, node, output, parallelId)
      } else {
        this.handleRegularNodeCompletion(ctx, node, output)
      }
    } else {
      this.handleRegularNodeCompletion(ctx, node, output)
    }
  }

  private handleLoopNodeCompletion(
    ctx: ExecutionContext,
    node: DAGNode,
    output: NormalizedBlockOutput,
    loopId: string
  ): void {
    this.loopOrchestrator.storeLoopNodeOutput(ctx, loopId, node.id, output)
    this.state.setBlockOutput(node.id, output)
  }

  private async handleParallelNodeCompletion(
    ctx: ExecutionContext,
    node: DAGNode,
    output: NormalizedBlockOutput,
    parallelId: string
  ): Promise<void> {
    const scope = this.parallelOrchestrator.getParallelScope(ctx, parallelId)
    if (!scope) {
      await this.parallelOrchestrator.initializeParallelScope(ctx, parallelId)
    }
    this.parallelOrchestrator.handleParallelBranchCompletion(ctx, parallelId, node.id, output)
    this.state.setBlockOutput(node.id, output)
  }

  private handleRegularNodeCompletion(
    ctx: ExecutionContext,
    node: DAGNode,
    output: NormalizedBlockOutput
  ): void {
    this.state.setBlockOutput(node.id, output)

    if (
      node.metadata.isSentinel &&
      node.metadata.sentinelType === 'end' &&
      output.selectedRoute === 'loop_continue'
    ) {
      const loopId = node.metadata.loopId
      if (loopId) {
        this.loopOrchestrator.clearLoopExecutionState(loopId, ctx)
        this.loopOrchestrator.restoreLoopEdges(loopId)
      }
    }

    if (
      node.metadata.isParallelSentinel &&
      node.metadata.sentinelType === 'end' &&
      output.selectedRoute === EDGE.PARALLEL_CONTINUE
    ) {
      this.state.deleteBlockState(node.id)
    }
  }

  private findParallelIdForNode(nodeId: string): string | undefined {
    const baseId = extractBaseBlockId(nodeId)
    return this.parallelOrchestrator.findParallelIdForNode(baseId)
  }
}
