<?php

declare(strict_types=1);

namespace Drupal\flowdrop_workflow_executor\Plugin\FlowDropNodeProcessor;

use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\flowdrop\Attribute\FlowDropNodeProcessor;
use Drupal\flowdrop\DTO\ConfigEdit\ConfigEdit;
use Drupal\flowdrop\DTO\ConfigEdit\ExternalEditLink;
use Drupal\flowdrop\DTO\ExecutionContextDTO;
use Drupal\flowdrop\DTO\NodeMetadata\NodePort;
use Drupal\flowdrop\DTO\ParameterBagInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\AbstractFlowDropNodeProcessor;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\CancellableNodeProcessorInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\ConfigEditProviderInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\ExecutionContextAwareInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\ResumableNodeProcessorInterface;
use Drupal\flowdrop_interrupt\DTO\ExecutionStamp;
use Drupal\flowdrop_interrupt\Message\LinkedInterruptMessage;
use Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface;
use Drupal\flowdrop_interrupt\Enum\InterruptType;
use Drupal\flowdrop_interrupt\Exception\InterruptRequiredException;
use Drupal\flowdrop_interrupt\Service\InterruptManagerInterface;
use Drupal\flowdrop_interrupt\Service\InterruptMessageBusInterface;
use Drupal\flowdrop_workflow_executor\Message\PipelineCompletionSentinelMessage;
use Drupal\flowdrop_orchestration\DTO\ResolvedInterruptInterface;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;
use Drupal\flowdrop_workflow\Service\WorkflowSchemaResolverInterface;
use Drupal\flowdrop_workflow_executor\Enum\WorkflowExecutionMode;
use Drupal\flowdrop_workflow_executor\Enum\WorkflowExecutionStatus;
use Drupal\flowdrop_workflow_executor\Plugin\Deriver\WorkflowNodeDeriver;
use Drupal\flowdrop_workflow_executor\Service\WorkflowExecutionServiceInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Typed workflow-as-node derivative plugin.
 *
 * Each FlowDropWorkflow config entity surfaces as a `workflow:<id>`
 * derivative. The plugin's parameter and output schemas come straight from
 * the target workflow's declared FlowDrop Schema. Execution delegates to
 * WorkflowExecutionServiceInterface — the same code path WorkflowExecutor
 * uses — so recursion guard, circular-reference detection, and queue
 * dispatch are shared.
 *
 * For runtime/late-binding dispatch (selecting the workflow at runtime),
 * use the existing WorkflowExecutor plugin instead.
 *
 * @api
 */
#[FlowDropNodeProcessor(
  id: 'flowdrop_workflow',
  label: new TranslatableMarkup('Run Workflow'),
  description: 'Run another FlowDrop workflow with typed inputs/outputs',
  deriver: WorkflowNodeDeriver::class,
)]
final class WorkflowNode extends AbstractFlowDropNodeProcessor implements ExecutionContextAwareInterface, CancellableNodeProcessorInterface, ResumableNodeProcessorInterface, ConfigEditProviderInterface {

  /**
   * The current execution context, if any.
   */
  private ?ExecutionContextDTO $executionContext = NULL;

  public function __construct(
    array $configuration,
    $plugin_id,
    $plugin_definition,
    private readonly EntityTypeManagerInterface $entityTypeManager,
    private readonly WorkflowExecutionServiceInterface $workflowExecutionService,
    private readonly WorkflowSchemaResolverInterface $schemaResolver,
    private readonly ?InterruptManagerInterface $interruptManager = NULL,
    private readonly ?InterruptMessageBusInterface $messageBus = NULL,
  ) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
  }

  /**
   * {@inheritdoc}
   */
  public static function create(
    ContainerInterface $container,
    array $configuration,
    $plugin_id,
    $plugin_definition,
  ): static {
    return new static(
      $configuration,
      $plugin_id,
      $plugin_definition,
      $container->get('entity_type.manager'),
      $container->get('flowdrop_workflow_executor.service'),
      $container->get('flowdrop_workflow.schema_resolver'),
      $container->has('flowdrop_interrupt.manager') ? $container->get('flowdrop_interrupt.manager') : NULL,
      $container->has('flowdrop_interrupt.message_bus') ? $container->get('flowdrop_interrupt.message_bus') : NULL,
    );
  }

  /**
   * {@inheritdoc}
   */
  public function setExecutionContext(ExecutionContextDTO $context): void {
    $this->executionContext = $context;
  }

  /**
   * {@inheritdoc}
   *
   * Surface an "Edit Workflow" link that opens the referenced sub-workflow in
   * the editor. Unlike WorkflowExecutor — which selects its target at config
   * time and templates the URL from `config.workflow_id` — each WorkflowNode
   * derivative targets a fixed workflow baked into the plugin definition by
   * WorkflowNodeDeriver, so the URL is resolved literally here with no
   * parameter mapping.
   */
  public function getConfigEdit(): ConfigEdit {
    $definition = $this->getPluginDefinition();
    $workflowId = is_array($definition) ? ($definition['workflow_id'] ?? '') : '';
    if (!is_string($workflowId) || $workflowId === '') {
      // No resolvable target (should not happen for a real derivative) — omit
      // the link rather than emit a broken URL.
      return new ConfigEdit();
    }

    return new ConfigEdit(
      externalEditLink: new ExternalEditLink(
        url: sprintf('/admin/flowdrop/workflows/%s/editor', $workflowId),
        label: 'Edit Workflow',
        icon: 'mdi:sitemap-outline',
        description: 'Open the referenced workflow in the editor',
        openInNewTab: TRUE,
      ),
    );
  }

  /**
   * {@inheritdoc}
   */
  public function getParameterSchema(): array {
    $workflow = $this->loadWorkflow();
    $schema = $workflow !== NULL
      ? ($workflow->getParameterSchema() ?? ['type' => 'object', 'properties' => []])
      : ['type' => 'object', 'properties' => []];
    // Declare __interrupt_id__ so the runtime's parameter resolver passes
    // it through on resume invocations. Without this entry the resolver
    // strips it (only declared params are forwarded), and both the
    // ResumableNodeProcessor dispatch and the inline resume guard below
    // would receive NULL — causing the node to re-dispatch the sub-workflow
    // instead of returning cached outputs.
    if (!isset($schema['properties']) || !is_array($schema['properties'])) {
      $schema['properties'] = [];
    }
    $schema['properties']['__interrupt_id__'] = [
      'type' => 'string',
      'description' => 'Internal: Interrupt ID for resume operations',
      'connectable' => FALSE,
      'configurable' => FALSE,
    ];
    return $schema;
  }

  /**
   * {@inheritdoc}
   */
  public function getOutputSchema(): array {
    $workflow = $this->loadWorkflow();
    if ($workflow === NULL) {
      return ['type' => 'object', 'properties' => []];
    }
    return $workflow->getOutputSchema()
      ?? ['type' => 'object', 'properties' => []];
  }

  /**
   * {@inheritdoc}
   *
   * Resume hook invoked when a previously raised WorkflowExecution interrupt
   * (sentinel-linked, async-dispatched, or sync→HITL-bubbled) has resolved.
   * The cascade subscriber stores the sub-pipeline's outputs in the
   * interrupt's response_data; this method returns those outputs as the
   * node's outputs so downstream nodes see the same result they would have
   * had the sub-workflow run synchronously to completion.
   *
   * When the cached outputs are missing or empty (sub failed, cancelled,
   * or wrote no outputs), return an empty array — downstream gateways will
   * see "no inputs from this branch" and behave as if the sub produced
   * nothing.
   */
  public function resume(
    ResolvedInterruptInterface $interrupt,
    ParameterBagInterface $params,
  ): array {
    $response = $interrupt->getResponseData();
    if (!is_array($response)) {
      return [];
    }
    $outputs = $response['outputs'] ?? NULL;
    if (!is_array($outputs)) {
      return [];
    }
    // Translate the sub-workflow's node-keyed outputs to its declared
    // output-port keys so this node's downstream consumers see the public
    // contract (e.g. `text` / `result`) instead of the sub's internal node
    // ids (`text_output.3` / `prompt_template.1`). Falls back to the raw
    // node-keyed outputs when the workflow isn't loadable.
    $workflow = $this->loadWorkflow();
    if ($workflow !== NULL) {
      return $this->schemaResolver->extractOutputs($workflow, $outputs);
    }
    return $outputs;
  }

  /**
   * {@inheritdoc}
   */
  public function process(ParameterBagInterface $params): array {
    $workflow = $this->loadWorkflow();
    if ($workflow === NULL) {
      throw new \RuntimeException(sprintf(
        'WorkflowNode: target workflow not found for plugin %s',
        $this->getPluginId(),
      ));
    }

    // Resume invocations land in resume() via ResumableNodeProcessorInterface
    // — the runtime's NodeRuntimeService::resolveResumeInterrupt() detects
    // the resolved interrupt and dispatches there. process() runs for the
    // initial invocation and for pending interrupts (no resolved entity to
    // hand off to resume() yet).
    //
    // Translate the named inputs the canvas passes into node-keyed
    // initialData via the schema resolver.
    $named = [];
    foreach ($workflow->getInputPorts() as $entry) {
      if (!is_array($entry)) {
        continue;
      }
      $name = NodePort::fromArray($entry)->name;
      if ($name !== '' && $params->has($name)) {
        $named[$name] = $params->get($name);
      }
    }
    $initialData = $this->schemaResolver->resolveInputs($workflow, $named);

    // Plumb parent context so the service's chain tracking + recursion
    // guard sees the outer workflow.
    $parentContext = $params->getArray('__session_metadata', [])['workflow_executor_parent_context'] ?? [];
    if (!is_array($parentContext)) {
      $parentContext = [];
    }

    // Stamp the parent's session id so WorkflowExecutionService reuses
    // it for the sub-pipeline (the sub's messages and HITL interrupts
    // land in the parent's playground stream). Sourced from the parent
    // pipeline's typed session_id column added in Phase A. We piggyback
    // the hierarchy and lineage reads onto the same load — no extra
    // storage access.
    if (
      empty($parentContext['session_id'])
      || empty($parentContext['hierarchy'])
      || empty($parentContext['parent_pipeline_id'])
      || empty($parentContext['root_pipeline_id'])
    ) {
      $parentPipelineId = $this->executionContext?->getPipelineId();
      if ($parentPipelineId !== NULL && $parentPipelineId !== '') {
        // The spawning pipeline id is known without a load.
        if (empty($parentContext['parent_pipeline_id'])) {
          $parentContext['parent_pipeline_id'] = (string) $parentPipelineId;
        }
        $parentPipeline = $this->entityTypeManager
          ->getStorage('flowdrop_pipeline')
          ->load($parentPipelineId);
        if ($parentPipeline instanceof FlowDropPipelineInterface) {
          if (empty($parentContext['session_id'])) {
            $parentSessionId = $parentPipeline->getSessionId();
            if (is_string($parentSessionId) && $parentSessionId !== '') {
              $parentContext['session_id'] = $parentSessionId;
            }
          }
          if (empty($parentContext['hierarchy'])) {
            $parentHierarchy = $parentPipeline->getExecutionContext()['playground']['hierarchy'] ?? NULL;
            if (is_array($parentHierarchy) && $parentHierarchy !== []) {
              $parentContext['hierarchy'] = $parentHierarchy;
            }
          }
          if (empty($parentContext['root_pipeline_id'])) {
            // The sub-workflow's root is the parent's root. The parent always
            // has a stored root (self-root on insert), but fall back to the
            // parent's own id defensively so a sub-pipeline never inherits a
            // NULL root.
            $parentRoot = $parentPipeline->getRootPipelineId();
            $parentContext['root_pipeline_id'] = ($parentRoot !== NULL && $parentRoot !== '')
              ? $parentRoot
              : (string) $parentPipelineId;
          }
        }
      }
    }

    $modeValue = $params->getString('__execution_mode__', WorkflowExecutionMode::Job->value);
    $mode = WorkflowExecutionMode::tryFrom($modeValue) ?? WorkflowExecutionMode::Job;

    if ($mode === WorkflowExecutionMode::JobAsync && $this->interruptManager !== NULL) {
      return $this->dispatchAsync($workflow, $initialData, $parentContext);
    }

    // Synchronous path (default).
    $result = $this->workflowExecutionService->execute(
      $workflow,
      $initialData,
      WorkflowExecutionMode::Job,
      300,
      2000,
      $parentContext,
    );

    if ($result->getStatus() === WorkflowExecutionStatus::Failed) {
      throw new \RuntimeException(sprintf(
        'Sub-workflow %s failed: %s',
        (string) $workflow->id(),
        $result->getErrorMessage(),
      ));
    }

    if ($result->getStatus() === WorkflowExecutionStatus::Paused) {
      // Sub-workflow stopped in a non-terminal state mid-sync (today:
      // a HITL interrupt). Convert into a linked interrupt so the
      // parent pipeline also pauses — the same sentinel + cascade
      // machinery used by dispatchAsync handles the resume path.
      // Without this, the parent would either spin-poll until timeout
      // (and fail) or report success on stale outputs.
      if ($this->interruptManager === NULL || $this->messageBus === NULL) {
        throw new \RuntimeException(sprintf(
          'Sub-workflow %s paused, but the flowdrop_interrupt module is '
          . 'not available to bubble the pause up to the parent.',
          (string) $workflow->id(),
        ));
      }
      // Sub-pipeline id is what the cascade subscriber matches on now —
      // resilient to whether the sub-workflow used its own session or
      // shared the parent's.
      $subPipelineId = $result->getPipelineId();
      if ($subPipelineId === '') {
        throw new \RuntimeException(sprintf(
          'Sub-workflow %s paused but did not report its pipeline id.',
          (string) $workflow->id(),
        ));
      }
      $this->raiseLinkedInterruptForSubPipeline(
        $workflow,
        $subPipelineId,
        $result->getSessionId(),
        'Sub-workflow paused, awaiting completion',
      );
      // raiseLinkedInterruptForSubPipeline() always throws.
    }

    if (!$result->isSuccess()) {
      throw new \RuntimeException(sprintf(
        'Sub-workflow %s did not complete successfully (status: %s)',
        (string) $workflow->id(),
        $result->getStatus()->value,
      ));
    }

    return $this->schemaResolver->extractOutputs($workflow, $result->getOutputs());
  }

  /**
   * Dispatches the sub-workflow asynchronously and throws an interrupt.
   *
   * Idempotency: if a queue retry re-enters this method before the
   * orchestrator persisted the previous attempt's interrupt-id on the job,
   * we'd otherwise spawn a duplicate sub-pipeline + sentinel pair. Before
   * doing anything destructive, look for an existing pending
   * WorkflowExecution interrupt belonging to this parent node and reuse it.
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The sub-workflow to dispatch.
   * @param array<string, mixed> $initialData
   *   Initial input data for the sub-workflow.
   * @param array<string, mixed> $parentContext
   *   Resolved parent context to forward.
   *
   * @return array<string, mixed>
   *   The outputs collected from the sub-workflow result.
   */
  private function dispatchAsync(
    WorkflowDefinitionInterface $workflow,
    array $initialData,
    array $parentContext,
  ): array {
    $parentPipelineId = $this->executionContext?->getPipelineId();
    $parentNodeId = $this->executionContext?->getMetadata()['node_id'] ?? '';
    if (
      $this->interruptManager !== NULL
      && $parentPipelineId !== NULL
      && $parentPipelineId !== ''
      && $parentNodeId !== ''
    ) {
      foreach ($this->interruptManager->getPendingInterruptsForPipeline($parentPipelineId) as $existing) {
        if (
          $existing->getType() === InterruptType::WorkflowExecution
          && $existing->getNodeId() === $parentNodeId
        ) {
          // Found a pending dispatch from a prior attempt — propagate it
          // and skip the redundant sub-execute. The sentinel created with
          // it (linked via linked_interrupt_id) still cascades normally
          // when the sub-pipeline finishes.
          throw new InterruptRequiredException(
            $existing,
            $this->buildExecutionContext(),
            'Sub-workflow dispatch already pending — reusing existing interrupt.',
          );
        }
      }
    }

    $result = $this->workflowExecutionService->execute(
      $workflow,
      $initialData,
      WorkflowExecutionMode::JobAsync,
      300,
      2000,
      $parentContext,
    );

    if (!$result->isSuccess()) {
      throw new \RuntimeException(sprintf(
        'Sub-workflow %s async dispatch failed: %s',
        (string) $workflow->id(),
        $result->getErrorMessage(),
      ));
    }

    $sessionId = $result->getSessionId();
    if ($sessionId === '') {
      throw new \RuntimeException('Sub-workflow dispatch returned empty session ID.');
    }

    if ($this->messageBus === NULL) {
      throw new \RuntimeException(
        'WorkflowNode async dispatch requires flowdrop_interrupt.message_bus '
        . '— the interrupt module must be enabled.',
      );
    }

    // Async dispatch returns a session id but no pipeline id (the queue
    // worker creates the pipeline later). For the sentinel target, fall
    // back to the session's id — the completion event correlation works
    // via the playground session for async-dispatched runs.
    // @todo when async dispatch pre-creates the pipeline, switch this
    // to the sub-pipeline id like the sync path.
    $subPipelineId = $result->getPipelineId() !== '' ? $result->getPipelineId() : $sessionId;
    $this->raiseLinkedInterruptForSubPipeline(
      $workflow,
      $subPipelineId,
      $sessionId,
      'Sub-workflow dispatched, awaiting completion',
    );
  }

  /**
   * Creates the sentinel + linked interrupt pair and pauses the parent.
   *
   * The sub-pipeline's completion resolves a sentinel interrupt. The
   * parent's WorkflowExecution interrupt links to that sentinel via
   * `linked_interrupt_id` so the generic cascade subscriber resolves it
   * without a per-type lookup.
   *
   * Used by both async dispatch and sync→HITL bubble-up so the resume
   * mechanics are identical regardless of how the parent paused.
   *
   * The sentinel matches on the sub-pipeline id (carried in
   * `PipelineCompletedEvent::executionId`) — so the same primitive works
   * whether the sub-pipeline runs in its own session or shares the
   * parent's. `subSessionId` is retained in context for back-compat with
   * older interrupt entries; cascade lookup itself no longer reads it.
   *
   * @throws \Drupal\flowdrop_interrupt\Exception\InterruptRequiredException
   *   Always — this function never returns normally.
   */
  private function raiseLinkedInterruptForSubPipeline(
    WorkflowDefinitionInterface $workflow,
    string $subPipelineId,
    string $subSessionId,
    string $reason,
  ): never {
    assert($this->messageBus !== NULL);

    $context = $this->buildExecutionContext();
    $context['sub_pipeline_id'] = $subPipelineId;
    $context['sub_session_id'] = $subSessionId;
    $context['target_workflow_id'] = (string) $workflow->id();

    $parentWorkflowId = $this->executionContext?->getWorkflowId() ?? '';
    $parentNodeIdForStamp = $this->executionContext?->getMetadata()['node_id'] ?? 'workflow';

    $sentinel = $this->messageBus->dispatch(
      new PipelineCompletionSentinelMessage(
        message: sprintf('Awaiting sub-workflow %s completion', (string) $workflow->id()),
        subPipelineId: $subPipelineId,
        context: [
          'sub_pipeline_id' => $subPipelineId,
          'sub_session_id' => $subSessionId,
        ],
      ),
      new ExecutionStamp(
        workflowId: $parentWorkflowId,
        nodeId: $parentNodeIdForStamp,
      ),
    );

    $interrupt = $this->messageBus->dispatch(
      new LinkedInterruptMessage(
        type: InterruptType::WorkflowExecution,
        message: sprintf('Sub-workflow %s running asynchronously', (string) $workflow->id()),
        linkedInterruptId: (string) ($sentinel->uuid() ?? ''),
        context: $context,
      ),
      new ExecutionStamp(
        workflowId: $parentWorkflowId,
        nodeId: $parentNodeIdForStamp,
        pipelineId: $this->executionContext?->getPipelineId(),
      ),
    );

    throw new InterruptRequiredException(
      $interrupt,
      $context,
      $reason,
    );
  }

  /**
   * Builds the execution context payload for interrupts.
   *
   * @return array<string, mixed>
   *   The execution context.
   */
  private function buildExecutionContext(): array {
    $context = [];
    if ($this->executionContext !== NULL) {
      $context['workflow_id'] = $this->executionContext->getWorkflowId();
      $context['pipeline_id'] = $this->executionContext->getPipelineId();
      $context['metadata'] = $this->executionContext->getMetadata();
    }
    return $context;
  }

  /**
   * Resolves the target workflow entity from the derivative definition.
   *
   * `workflow_id` is set by WorkflowNodeDeriver on every derivative — we
   * rely on it exclusively rather than parsing the plugin ID, which is
   * fragile under provider namespacing.
   */
  private function loadWorkflow(): ?WorkflowDefinitionInterface {
    $definition = $this->getPluginDefinition();
    $workflowId = is_array($definition) ? ($definition['workflow_id'] ?? '') : '';
    if ($workflowId === '') {
      return NULL;
    }
    $entity = $this->entityTypeManager
      ->getStorage('flowdrop_workflow')
      ->load($workflowId);
    return $entity instanceof WorkflowDefinitionInterface ? $entity : NULL;
  }

  /**
   * {@inheritdoc}
   *
   * Cascade: when the parent pipeline is being cancelled, this hook
   * propagates the Cancel signal to the in-flight sub-workflow.
   * Prefers `sub_pipeline_id` from the interrupt context (stamped by
   * `raiseLinkedInterruptForSubPipeline()`) so the signal targets the
   * specific sub-pipeline; falls back to `sub_session_id` for older
   * interrupt entries that pre-date the sub-pipeline-id plumbing.
   *
   * Best-effort by contract — failures are absorbed by the runtime's
   * dispatcher rather than blocking the parent's teardown.
   */
  public function cancel(
    ResolvedInterruptInterface $interrupt,
    ParameterBagInterface $params,
  ): void {
    // The interrupt entity carries the sub-pipeline / sub-session ids in
    // its context. We need a concrete entity (not just the
    // ResolvedInterruptInterface) to access getContext(); the orchestration
    // interface intentionally doesn't expose it.
    if (!$interrupt instanceof FlowDropInterruptInterface) {
      return;
    }
    $context = $interrupt->getContext();
    $subPipelineId = $context['sub_pipeline_id'] ?? NULL;
    if (is_string($subPipelineId) && $subPipelineId !== '') {
      $this->workflowExecutionService->cancelPipeline(
        $subPipelineId,
        $interrupt->getReason() ?? 'Parent workflow cancelled',
        $interrupt->getInitiatorUid(),
      );
      return;
    }
    // Legacy fallback: pre-Phase-B interrupts only stamped session id.
    $subSessionId = $context['sub_session_id'] ?? NULL;
    if (is_string($subSessionId) && $subSessionId !== '') {
      $this->workflowExecutionService->cancelExecution(
        $subSessionId,
        $interrupt->getReason() ?? 'Parent workflow cancelled',
        $interrupt->getInitiatorUid(),
      );
    }
  }

}
