rodar

defmodule Rodar do
  @moduledoc """
  Main dispatcher for the Rodar Workflow execution engine.

  Routes BPMN elements to their handler modules based on element type. Each node
  in a parsed BPMN process is represented as a `{:bpmn_node_type, %{...}}` tuple,
  and the dispatcher resolves which module handles it.

  ## Execution Modes

  - `execute/2` — Simple dispatch, returns the handler result directly.
  - `execute/3` — Token-aware dispatch via `Rodar.Token`, records execution
    history in `Rodar.Context`, notifies `Rodar.Hooks`, and emits
    `Rodar.Telemetry` events.

  ## Token Flow

  `release_token/2` passes a token to the next node by ID. `release_token/3`
  forks child tokens for parallel branches (e.g., from a parallel gateway).

  ## Execution History Classification

  `execute/3` records each node's completion result in the execution history.
  A node that calls `release_token` is classified as `:ok` regardless of the
  downstream result, because calling `release_token` means the node itself
  completed its work successfully. Only nodes that return directly without
  releasing (e.g., a user task returning `{:manual, _}`) are classified by
  their own return value.

  ## Return Values

  All handlers return one of:

  - `{:ok, context}` — Node completed successfully.
  - `{:error, message}` — Execution error with a description.
  - `{:manual, context}` — Process paused at an external activity (user task, receive task, etc.).
  - `{:fatal, reason}` — Unrecoverable error.
  - `{:not_implemented}` — Element type has no handler implementation.
  """

  require Logger

  alias Rodar.Activity.MultiInstance
  alias Rodar.Activity.Subprocess
  alias Rodar.Activity.Subprocess.Embedded, as: SubprocessEmbedded
  alias Rodar.Activity.Subprocess.Event, as: SubprocessEvent
  alias Rodar.Activity.Subprocess.Transaction, as: SubprocessTransaction
  alias Rodar.Activity.Task.BusinessRule
  alias Rodar.Activity.Task.Manual
  alias Rodar.Activity.Task.Receive, as: TaskReceive
  alias Rodar.Activity.Task.Script
  alias Rodar.Activity.Task.Send, as: TaskSend
  alias Rodar.Activity.Task.Service
  alias Rodar.Activity.Task.User
  alias Rodar.Compensation
  alias Rodar.Context
  alias Rodar.Event.Boundary
  alias Rodar.Event.End
  alias Rodar.Event.Intermediate
  alias Rodar.Event.Intermediate.Catch, as: IntermediateCatch
  alias Rodar.Event.Intermediate.Throw, as: IntermediateThrow
  alias Rodar.Event.Start
  alias Rodar.Gateway.Complex
  alias Rodar.Gateway.Exclusive
  alias Rodar.Gateway.Exclusive.Event, as: ExclusiveEvent
  alias Rodar.Gateway.Inclusive
  alias Rodar.Gateway.Parallel
  alias Rodar.Hooks
  alias Rodar.SequenceFlow
  alias Rodar.TaskRegistry
  alias Rodar.Telemetry
  alias Rodar.Token

  @typedoc "A BPMN element represented as a tagged tuple with a map of attributes"
  @type element :: {atom(), map()}

  @typedoc "A BPMN execution context (GenServer pid)"
  @type context :: pid()

  @typedoc "Result of executing a BPMN element"
  @type result ::
          {:ok, context()}
          | {:error, String.t()}
          | {:manual, any()}
          | {:fatal, any()}
          | {:not_implemented}
          | {false}

  @doc """
  Parse a string representation of a process into an executable process representation
  """
  @spec parse(any()) :: {:ok, map()}
  def parse(_process) do
    {:ok, %{"start_node_id" => {:bpmn_event_start, %{}}}}
  end

  @doc """
  Get a node from a process by target id
  """
  @spec next(String.t(), map()) :: element() | nil
  def next(target, process) do
    Map.get(process, target)
  end

  @doc """
  Release token to another target node
  """
  @spec release_token(String.t() | [String.t()], context()) :: result()
  def release_token(targets, context) when is_list(targets) do
    mark_token_released(context)

    targets
    |> Task.async_stream(&release_token(&1, context))
    |> Enum.reduce({:ok, context}, &reduce_result/2)
  end

  def release_token(target, context) do
    mark_token_released(context)
    process = Context.get(context, :process)
    next = next(target, process)

    case next do
      nil -> {:error, "Unable to find node '#{target}'"}
      _ -> execute(next, context)
    end
  end

  @doc """
  Release token to another target node, threading a `Rodar.Token` through execution.

  When `targets` is a list (parallel fork), creates child tokens via `Rodar.Token.fork/1`
  for each branch.
  """
  @spec release_token(String.t() | [String.t()], context(), Rodar.Token.t()) :: result()
  def release_token(targets, context, %Token{} = token) when is_list(targets) do
    mark_token_released(context)

    targets
    |> Task.async_stream(fn target ->
      child_token = Token.fork(token)
      release_token(target, context, child_token)
    end)
    |> Enum.reduce({:ok, context}, &reduce_result/2)
  end

  def release_token(target, context, %Token{} = token) do
    mark_token_released(context)
    process = Context.get(context, :process)
    next = next(target, process)

    case next do
      nil -> {:error, "Unable to find node '#{target}'"}
      _ -> execute(next, context, token)
    end
  end

  @doc """
  Execute a node in the process
  """
  @spec execute(element(), context()) :: result()
  def execute(elem, context) do
    token = Token.new()
    execute(elem, context, token)
  end

  @doc """
  Execute a node in the process with token tracking.

  Updates the token's `current_node` before dispatching to the handler,
  stores the token on the context, and records execution history.
  """
  @spec execute(element(), context(), Token.t()) :: result()
  def execute({type, %{id: id} = _attrs} = elem, context, %Token{} = token) do
    token = %{token | current_node: id}
    Context.put_meta(context, :current_token, token)
    Context.put_meta(context, {:_token_released, token.id}, false)

    Context.record_visit(context, %{
      node_id: id,
      token_id: token.id,
      node_type: type,
      timestamp: System.monotonic_time(:millisecond)
    })

    Logger.metadata(
      rodar_node_id: id,
      rodar_node_type: type,
      rodar_token_id: token.id
    )

    span_metadata = %{node_id: id, node_type: type, token_id: token.id}

    Hooks.notify(context, :before_node, %{node_id: id, node_type: type, token: token})

    if activity_type?(type), do: pre_register_compensation(context, id)

    result = Telemetry.node_span(span_metadata, fn -> maybe_multi_instance(elem, context) end)

    Hooks.notify(context, :after_node, %{
      node_id: id,
      node_type: type,
      token: token,
      result: result
    })

    token_was_released = Context.get_meta(context, {:_token_released, token.id})
    Context.put_meta(context, {:_token_released, token.id}, nil)

    result_type =
      if token_was_released do
        :ok
      else
        classify_result(result, context, id)
      end

    Context.record_completion(context, id, token.id, result_type)

    if result_type != :ok and activity_type?(type) do
      Compensation.remove_handlers(context, id)
    end

    result
  end

  # Elements without :id (e.g., bare sequence flows in some paths) skip history recording
  def execute(elem, context, %Token{} = token) do
    Context.put_meta(context, :current_token, token)
    dispatch(elem, context)
  end

  defp maybe_multi_instance(
         {_type, %{loop_characteristics: %{type: :multi_instance}}} = elem,
         context
       ) do
    MultiInstance.execute(elem, context, &dispatch/2)
  end

  defp maybe_multi_instance(elem, context), do: dispatch(elem, context)

  defp dispatch({:bpmn_event_start, _} = elem, context),
    do: Start.token_in(elem, context)

  defp dispatch({:bpmn_event_end, _} = elem, context),
    do: End.token_in(elem, context)

  defp dispatch({:bpmn_event_intermediate, _} = elem, context),
    do: Intermediate.token_in(elem, context)

  defp dispatch({:bpmn_event_intermediate_throw, _} = elem, context),
    do: IntermediateThrow.token_in(elem, context)

  defp dispatch({:bpmn_event_intermediate_catch, _} = elem, context),
    do: IntermediateCatch.token_in(elem, context)

  defp dispatch({:bpmn_event_boundary, _} = elem, context),
    do: Boundary.token_in(elem, context)

  defp dispatch({:bpmn_activity_task_user, _} = elem, context),
    do: User.token_in(elem, context)

  defp dispatch({:bpmn_activity_task_script, _} = elem, context),
    do: Script.token_in(elem, context)

  defp dispatch({:bpmn_activity_task_service, _} = elem, context),
    do: Service.token_in(elem, context)

  defp dispatch({:bpmn_activity_task_business_rule, _} = elem, context),
    do: BusinessRule.token_in(elem, context)

  defp dispatch({:bpmn_activity_task_manual, _} = elem, context),
    do: Manual.token_in(elem, context)

  defp dispatch({:bpmn_activity_task_send, _} = elem, context),
    do: TaskSend.token_in(elem, context)

  defp dispatch({:bpmn_activity_task_receive, _} = elem, context),
    do: TaskReceive.token_in(elem, context)

  defp dispatch({:bpmn_activity_subprocess, _} = elem, context),
    do: Subprocess.token_in(elem, context)

  defp dispatch({:bpmn_activity_subprocess_embeded, _} = elem, context),
    do: SubprocessEmbedded.token_in(elem, context)

  defp dispatch({:bpmn_activity_subprocess_event, _} = elem, context),
    do: SubprocessEvent.token_in(elem, context)

  defp dispatch({:bpmn_activity_subprocess_transaction, _} = elem, context),
    do: SubprocessTransaction.token_in(elem, context)

  defp dispatch({:bpmn_gateway_exclusive, _} = elem, context),
    do: Exclusive.token_in(elem, context)

  defp dispatch({:bpmn_gateway_exclusive_event, _} = elem, context),
    do: ExclusiveEvent.token_in(elem, context)

  defp dispatch({:bpmn_gateway_parallel, _} = elem, context) do
    from_flow = pop_from_flow()
    Parallel.token_in(elem, context, from_flow)
  end

  defp dispatch({:bpmn_gateway_inclusive, _} = elem, context) do
    from_flow = pop_from_flow()
    Inclusive.token_in(elem, context, from_flow)
  end

  defp dispatch({:bpmn_gateway_complex, _} = elem, context) do
    from_flow = pop_from_flow()
    Complex.token_in(elem, context, from_flow)
  end

  defp dispatch({:bpmn_sequence_flow, _} = elem, context),
    do: SequenceFlow.token_in(elem, context)

  defp dispatch({type, %{id: id}} = elem, context) do
    case TaskRegistry.lookup(id) do
      {:ok, handler} ->
        handler.token_in(elem, context)

      :error ->
        case TaskRegistry.lookup(type) do
          {:ok, handler} -> handler.token_in(elem, context)
          :error -> {:not_implemented}
        end
    end
  end

  defp dispatch(_elem, _context), do: {:not_implemented}

  @activity_types [
    :bpmn_activity_task_user,
    :bpmn_activity_task_script,
    :bpmn_activity_task_service,
    :bpmn_activity_task_business_rule,
    :bpmn_activity_task_manual,
    :bpmn_activity_task_send,
    :bpmn_activity_task_receive,
    :bpmn_activity_subprocess,
    :bpmn_activity_subprocess_embeded,
    :bpmn_activity_subprocess_event,
    :bpmn_activity_subprocess_transaction
  ]

  defp classify_result({:ok, _}, _context, _id), do: :ok
  defp classify_result({:manual, _}, _context, _id), do: :manual
  defp classify_result({:fatal, _}, _context, _id), do: :fatal
  defp classify_result({:not_implemented}, _context, _id), do: :not_implemented

  defp classify_result({:error, reason}, context, id) do
    Hooks.notify(context, :on_error, %{node_id: id, error: reason})
    :error
  end

  defp classify_result(_, _context, _id), do: :unknown

  defp activity_type?(type), do: type in @activity_types

  defp pop_from_flow do
    case Process.delete(:_rodar_from_flow) do
      :undefined -> nil
      value -> value
    end
  end

  defp mark_token_released(context) do
    case Context.get_meta(context, :current_token) do
      %Token{id: token_id} ->
        Context.put_meta(context, {:_token_released, token_id}, true)

      _ ->
        :ok
    end
  end

  defp pre_register_compensation(context, activity_id) do
    process = Context.get(context, :process)

    process
    |> find_compensation_boundaries(activity_id)
    |> Enum.each(fn {outgoing, _attrs} ->
      handler_id = find_handler_target(outgoing, process)
      if handler_id, do: Compensation.register_handler(context, activity_id, handler_id)
    end)
  end

  defp find_compensation_boundaries(process, activity_id) do
    Enum.flat_map(process, fn
      {_id, {:bpmn_event_boundary, %{attachedToRef: ^activity_id, outgoing: outgoing} = attrs}} ->
        if has_compensate_definition?(attrs), do: [{outgoing, attrs}], else: []

      _ ->
        []
    end)
  end

  defp has_compensate_definition?(attrs) do
    match?({:bpmn_event_definition_compensate, _}, Map.get(attrs, :compensateEventDefinition))
  end

  defp find_handler_target([flow_id | _], process) do
    case Map.get(process, flow_id) do
      {:bpmn_sequence_flow, %{targetRef: target}} -> target
      _ -> nil
    end
  end

  defp find_handler_target(_, _), do: nil

  defp reduce_result({:ok, {:ok, _} = result}, {:ok, _}), do: result
  defp reduce_result({:ok, {:error, _} = result}, {:ok, _}), do: result
  defp reduce_result({:ok, {:error, _}}, {:error, _} = acc), do: acc
  defp reduce_result({:ok, {:fatal, _} = result}, _), do: result
  defp reduce_result({:ok, {:not_implemented} = result}, _), do: result
  defp reduce_result({:ok, {:manual, _} = result}, {:ok, _}), do: result
  defp reduce_result({:ok, {:manual, _}}, acc), do: acc
end