journey

defmodule Journey.Tools do
  @moduledoc """
  This module contains utility functions for the Journey library.
  """

  require Logger

  import Ecto.Query

  alias Journey.Graph
  alias Journey.Helpers.GrabBag
  alias Journey.Persistence.Schema.Execution.Computation
  alias Journey.Persistence.Schema.Execution.Value

  @doc """
  Shows the status of upstream dependencies for a computation node.

  Lists each dependency with a checkmark (✅) if satisfied or a stop sign (🛑) if not.
  Useful for debugging to see which dependencies are met and which are still blocking.

  ## Parameters
  - `execution_id` - The ID of the execution to analyze
  - `computation_node_name` - The atom name of the computation node to check

  ## Returns
  A string showing the readiness status with checkmarks for met conditions and
  stop signs for unmet conditions.

  ## Example

      iex> import Journey.Node
      iex> graph = Journey.new_graph("what_am_i_waiting_for test graph Elixir.Journey.Tools", "v1.0.0", [
      ...>   input(:name),
      ...>   input(:title),
      ...>   compute(:greeting, [:name, :title], fn %{name: name, title: title} ->
      ...>     {:ok, "Hello, \#{title} \#{name}!"}
      ...>   end)
      ...> ])
      iex> execution = Journey.start(graph)
      iex> Journey.Tools.what_am_i_waiting_for(execution.id, :greeting) |> IO.puts()
      🛑 :name | &is_set/1
      🛑 :title | &is_set/1
      :ok
      iex> execution = Journey.set(execution, :name, "Alice")
      iex> Journey.Tools.what_am_i_waiting_for(execution.id, :greeting) |> IO.puts()
      ✅ :name | &is_set/1 | rev 1
      🛑 :title | &is_set/1
      :ok
      iex> execution = Journey.set(execution, :title, "Dr.")
      iex> {:ok, _greeting_value, _revision} = Journey.get(execution, :greeting, wait: :newer)
      iex> Journey.Tools.what_am_i_waiting_for(execution.id, :greeting) |> IO.puts()
      ✅ :name | &is_set/1 | rev 1
      ✅ :title | &is_set/1 | rev 2
      :ok
  """
  def what_am_i_waiting_for(execution_id, computation_node_name)
      when is_binary(execution_id) and is_atom(computation_node_name) do
    execution = Journey.load(execution_id)
    graph = Journey.Graph.Catalog.fetch(execution.graph_name, execution.graph_version)

    case Graph.find_node_by_name(graph, computation_node_name) do
      nil ->
        "Node :#{computation_node_name} not found in graph"

      graph_node ->
        gated_by = Map.get(graph_node, :gated_by)

        readiness =
          Journey.Node.UpstreamDependencies.Computations.evaluate_computation_for_readiness(
            execution.values,
            gated_by
          )

        # Use flat format for this debugging function
        Enum.map_join(readiness.conditions_met, "", fn condition_map = %{upstream_node: v, f_condition: f} ->
          node_display = format_node_name_with_context(v.node_name, condition_map)
          "✅ #{node_display} | #{f_name(f)} | rev #{v.ex_revision}\n"
        end) <>
          Enum.map_join(readiness.conditions_not_met, "\n", fn condition_map = %{upstream_node: v, f_condition: f} ->
            node_display = format_node_name_with_context(v.node_name, condition_map)
            "🛑 #{node_display} | #{f_name(f)}"
          end)
    end
  end

  @doc """
  Returns the current state of a computation node.

  Returns the state of the most recent computation attempt for the given node.
  If no computation has been attempted yet, returns `:not_set`.
  For input nodes (non-compute nodes), returns `:not_compute_node`.

  ## Parameters
  - `execution_id` - The ID of the execution to check
  - `node_name` - The atom name of the node to check

  ## Returns
  - `:not_set` - No computation has been attempted yet
  - `:computing` - Currently computing
  - `:success` - Computation completed successfully
  - `:failed` - Computation failed
  - `:abandoned` - Computation was abandoned
  - `:cancelled` - Computation was cancelled
  - `:not_compute_node` - The node is an input node, not a computation

  ## Examples

      iex> import Journey.Node
      iex> graph = Journey.new_graph("computation_state doctest graph", "v1.0.0", [
      ...>   input(:value),
      ...>   compute(:double, [:value], fn %{value: v} -> {:ok, v * 2} end)
      ...> ])
      iex> execution = Journey.start(graph)
      iex> Journey.Tools.computation_state(execution.id, :double)
      :not_set
      iex> Journey.Tools.computation_state(execution.id, :value)
      :not_compute_node
      iex> execution = Journey.set(execution, :value, 5)
      iex> {:ok, _result, _revision} = Journey.get(execution, :double, wait: :newer)
      iex> Journey.Tools.computation_state(execution.id, :double)
      :success
  """
  def computation_state(execution_id, node_name)
      when is_binary(execution_id) and is_atom(node_name) do
    case Journey.load(execution_id, include_archived: true) do
      nil ->
        raise ArgumentError, "Execution '#{execution_id}' not found"

      execution ->
        graph = Journey.Graph.Catalog.fetch(execution.graph_name, execution.graph_version)
        do_computation_state(graph, execution, node_name)
    end
  end

  defp do_computation_state(graph, execution, node_name) do
    case Graph.find_node_by_name(graph, node_name) do
      nil ->
        :node_not_found

      %{type: :input} ->
        :not_compute_node

      _ ->
        Journey.Executions.find_computations_by_node_name(execution, node_name)
        |> computation_state_impl()
    end
  end

  defp computation_state_impl(computations) when computations == [], do: :not_set

  defp computation_state_impl(computations) when is_list(computations) do
    computations
    |> Enum.max_by(fn c -> {c.ex_revision_at_completion || -1, c.id} end)
    |> Map.get(:state)
  end

  @doc """
  Converts a computation state atom to human-readable text with a visual symbol.

  Returns a formatted string with an appropriate symbol and the state atom
  for each computation state, following the pattern used in other Journey
  text formatting functions.

  ## Parameters
  - `state` - The computation state atom returned by `computation_state/2`

  ## Returns
  A string with symbol and the state atom.

  ## State Representations
  - `:not_set` - "⬜ :not_set (not yet attempted)"
  - `:computing` - "⏳ :computing"
  - `:success` - "✅ :success"
  - `:failed` - "❌ :failed"
  - `:abandoned` - "❓ :abandoned"
  - `:cancelled` - "🛑 :cancelled"
  - `:not_compute_node` - "📝 :not_compute_node"

  ## Examples

      iex> Journey.Tools.computation_state_to_text(:success)
      "✅ :success"

      iex> Journey.Tools.computation_state_to_text(:computing)
      "⏳ :computing"

      iex> Journey.Tools.computation_state_to_text(:not_set)
      "⬜ :not_set (not yet attempted)"
  """
  def computation_state_to_text(:not_set), do: "⬜ :not_set (not yet attempted)"
  def computation_state_to_text(:computing), do: "⏳ :computing"
  def computation_state_to_text(:success), do: "✅ :success"
  def computation_state_to_text(:failed), do: "❌ :failed"
  def computation_state_to_text(:abandoned), do: "❓ :abandoned"
  def computation_state_to_text(:cancelled), do: "🛑 :cancelled"
  def computation_state_to_text(:not_compute_node), do: "📝 :not_compute_node"
  def computation_state_to_text(:node_not_found), do: "❓ :node_not_found (node not in current graph)"
  def computation_state_to_text(other) when is_atom(other), do: "? :#{other}"

  @doc """
  Shows the status and dependencies for a single computation node.

  Provides a focused view of one specific computation node's status and dependencies,
  similar to the computation sections in Journey.Tools.introspect/1 but for just one node.

  ## Parameters
  - `execution_id` - The ID of the execution to analyze
  - `node_name` - The atom name of the computation node to check

  ## Returns
  A string showing the node's current status and dependencies.

  For completed computations, shows the result with inputs used:

      :send_follow_up (CMPTA5MDJHVXRMG54150EGX): ✅ :success | :compute | rev 4
      inputs used:
         :user_applied (rev 0)
         :card_mailed (rev 0)

  For outstanding computations, shows the dependency tree:

      :send_weekly_reminder (CMPTA5MDJHVXRMG54150EGX): ⬜ :not_set (not yet attempted) | :compute
           :and
            ├─ 🛑 :subscribe_weekly | &true?/1
            ├─ 🛑 :weekly_reminder_schedule | &provided?/1
            └─ ✅ :email_address | &provided?/1 | rev 2

  For input nodes (non-compute nodes), returns an appropriate message.

  ## Examples

      iex> import Journey.Node
      iex> graph = Journey.new_graph("computation_status_as_text doctest", "v1.0.0", [
      ...>   input(:value),
      ...>   compute(:double, [:value], fn %{value: v} -> {:ok, v * 2} end)
      ...> ])
      iex> execution = Journey.start(graph)
      iex> Journey.Tools.computation_status_as_text(execution.id, :double)
      ":double: ⬜ :not_set (not yet attempted) | :compute\\n       ✅ :value | &is_set/1"

      iex> import Journey.Node
      iex> graph = Journey.new_graph("computation_status_as_text completed doctest", "v1.0.0", [
      ...>   input(:value),
      ...>   compute(:triple, [:value], fn %{value: v} -> {:ok, v * 3} end)
      ...> ])
      iex> execution = Journey.start(graph)
      iex> execution = Journey.set(execution, :value, 5)
      iex> {:ok, _result, _revision} = Journey.get(execution, :triple, wait: :newer)
      iex> result = Journey.Tools.computation_status_as_text(execution.id, :triple)
      iex> result =~ ":triple"
      true
      iex> result =~ "✅ :success"
      true
      iex> result =~ "inputs used"
      true
  """
  def computation_status_as_text(execution_id, node_name)
      when is_binary(execution_id) and is_atom(node_name) do
    case Journey.load(execution_id) do
      nil ->
        raise ArgumentError, "Execution '#{execution_id}' not found"

      execution ->
        do_computation_status_as_text(execution, execution_id, node_name)
    end
  end

  defp do_computation_status_as_text(execution, execution_id, node_name) do
    graph = Journey.Graph.Catalog.fetch(execution.graph_name, execution.graph_version)

    case Graph.find_node_by_name(graph, node_name) do
      nil ->
        "Node :#{node_name} not found in graph"

      graph_node ->
        case graph_node.type do
          :input ->
            ":#{node_name}: 📝 :not_compute_node (input nodes do not compute)"

          _ ->
            state = computation_state(execution_id, node_name)
            format_single_computation_status(execution, graph, graph_node, state)
        end
    end
  end

  defp format_single_computation_status(execution, graph, graph_node, state) do
    case state do
      state when state in [:success, :failed, :abandoned, :cancelled, :computing] ->
        format_completed_computation_for_node(execution, graph_node, state)

      :not_set ->
        format_outstanding_computation_status(execution, graph, graph_node)
    end
  end

  defp format_completed_computation_for_node(execution, graph_node, state) do
    node_name = graph_node.name
    computations = Journey.Executions.find_computations_by_node_name(execution, node_name)

    if Enum.empty?(computations) do
      ":#{node_name}: #{computation_state_to_text(state)} | #{inspect(graph_node.type)}"
    else
      most_recent_computation =
        computations
        |> Enum.max_by(fn c -> {c.ex_revision_at_completion || -1, c.id} end)

      format_completed_computation_status(most_recent_computation, execution, graph_node)
    end
  end

  defp format_completed_computation_status(
         %{
           id: id,
           node_name: node_name,
           state: state,
           computation_type: computation_type,
           computed_with: computed_with,
           ex_revision_at_completion: ex_revision_at_completion
         },
         execution,
         graph_node
       ) do
    header =
      ":#{node_name} (#{id}): #{computation_state_to_text(state)} | #{inspect(computation_type)} | rev #{ex_revision_at_completion}\n"

    # For failed computations with no computed_with data, show the dependency tree
    if state == :failed and empty_computed_with?(computed_with) do
      header <> format_failed_computation_dependencies(execution, graph_node)
    else
      # For successful computations or failed with computed_with data, show what was used
      header <> format_inputs_used(computed_with)
    end
  end

  defp empty_computed_with?(nil), do: true
  defp empty_computed_with?([]), do: true
  defp empty_computed_with?(inputs) when inputs == %{}, do: true
  defp empty_computed_with?(_), do: false

  defp format_failed_computation_dependencies(execution, graph_node) do
    gated_by = Map.get(graph_node, :gated_by)

    if gated_by do
      readiness =
        Journey.Node.UpstreamDependencies.Computations.evaluate_computation_for_readiness(
          execution.values,
          gated_by
        )

      format_condition_tree(readiness.structure, "    ")
    else
      "inputs used:\n   <none>"
    end
  end

  defp format_inputs_used(computed_with) do
    "inputs used:\n" <>
      if empty_computed_with?(computed_with) do
        "   <none>"
      else
        Enum.map_join(computed_with, "\n", fn {node_name, revision} ->
          "   #{inspect(node_name)} (rev #{revision})"
        end)
      end
  end

  defp format_outstanding_computation_status(execution, _graph, graph_node) do
    node_name = graph_node.name
    gated_by = Map.get(graph_node, :gated_by)

    readiness =
      Journey.Node.UpstreamDependencies.Computations.evaluate_computation_for_readiness(
        execution.values,
        gated_by
      )

    # Find the computation ID for this outstanding computation
    computation_id =
      execution.computations
      |> Enum.find(fn c -> c.node_name == node_name and c.state == :not_set end)
      |> case do
        nil -> "NO_COMPUTATION_ID"
        computation -> computation.id
      end

    header =
      ":#{node_name} (#{computation_id}): #{computation_state_to_text(:not_set)} | #{inspect(graph_node.type)}\n"

    formatted_conditions = format_condition_tree(readiness.structure, "    ")

    header <> formatted_conditions
  end

  @doc false
  def outstanding_computations(execution_id) when is_binary(execution_id) do
    execution = Journey.load(execution_id)
    graph = Journey.Graph.Catalog.fetch(execution.graph_name, execution.graph_version)

    all_candidates_for_computation =
      from(c in Computation,
        where:
          c.execution_id == ^execution_id and
            c.state == ^:not_set and
            c.computation_type in [
              ^:compute,
              ^:schedule_once,
              ^:tick_once,
              ^:schedule_recurring,
              ^:tick_recurring,
              ^:archive
            ],
        lock: "FOR UPDATE"
      )
      |> Journey.Repo.all()
      |> Journey.Executions.convert_values_to_atoms(:node_name)

    all_value_nodes =
      from(v in Value, where: v.execution_id == ^execution_id)
      |> Journey.Repo.all()
      |> Enum.map(fn %Value{node_name: node_name} = n -> %Value{n | node_name: String.to_atom(node_name)} end)
      |> Enum.map(fn v -> de_ecto(v) end)

    all_candidates_for_computation
    |> Enum.map(fn computation_candidate -> de_ecto(computation_candidate) end)
    |> Enum.flat_map(fn computation_candidate ->
      case Graph.find_node_by_name(graph, computation_candidate.node_name) do
        nil ->
          Logger.warning(
            "Skipping orphaned computation node :#{computation_candidate.node_name} " <>
              "in execution #{execution_id} — node not found in current graph definition"
          )

          []

        graph_node ->
          [
            Journey.Node.UpstreamDependencies.Computations.evaluate_computation_for_readiness(
              all_value_nodes,
              Map.get(graph_node, :gated_by)
            )
            |> Map.put(:computation, computation_candidate)
          ]
      end
    end)
  end

  defp de_ecto(ecto_struct) do
    ecto_struct
    |> Map.drop([:__meta__, :__struct__, :execution_id, :execution])
  end

  @doc false
  def set_computed_node_value(execution_id, computation_node_name, value)
      when is_binary(execution_id) and is_atom(computation_node_name) do
    {execution, _changed_keys} =
      execution_id
      |> Journey.load()
      |> Journey.Executions.set_value(computation_node_name, value)

    execution
  end

  @doc false
  def advance(execution_id) when is_binary(execution_id) do
    execution_id
    |> Journey.load()
    |> Journey.Scheduler.advance()
  end

  @doc """
  Generates structured data about an execution's current state.

  Returns a map containing:
  - Execution metadata (ID, graph, timestamps, duration, revision, archived status)
  - Values categorized as set/not_set with their details
  - Computations categorized as completed/outstanding with dependency info

  ## Example

      iex> Journey.Tools.introspect("EXEC07B2H0H7J1LTAE0VJDAL")
      %{
        execution_id: "EXEC07B2H0H7J1LTAE0VJDAL",
        graph_name: "g1",
        graph_version: "v1",
        archived_at: nil,
        created_at: 1723656196,
        updated_at: 1723656210,
        duration_seconds: 14,
        revision: 7,
        values: %{
          set: [...],
          not_set: [...]
        },
        computations: %{
          completed: [...],
          outstanding: [...]
        }
      }

  ## Parameters
  - `execution_id` - The ID of the execution to analyze

  ## Returns
  A structured map with execution state data.

  Use `introspect/1` to get execution summary as text.
  """
  def summarize_as_data(execution_id) when is_binary(execution_id) do
    execution =
      case Journey.load(execution_id, include_archived: true) do
        nil ->
          raise ArgumentError, "Execution '#{execution_id}' not found"

        exec ->
          exec
      end

    graph = Journey.Graph.Catalog.fetch(execution.graph_name, execution.graph_version)

    if graph == nil do
      raise "Graph '#{execution.graph_name}' not found in catalog."
    end

    set_values = execution.values |> Enum.filter(fn v -> v.set_time != nil end) |> Enum.sort_by(& &1.set_time, :desc)
    not_set_values = execution.values |> Enum.filter(fn v -> v.set_time == nil end)

    computations_completed =
      execution.computations
      |> Enum.filter(fn c -> c.ex_revision_at_completion != nil end)
      |> Enum.sort_by(& &1.ex_revision_at_completion, :desc)

    computations_outstanding =
      execution.computations
      |> Enum.filter(fn c -> c.ex_revision_at_completion == nil end)
      |> Enum.sort_by(& &1.node_name, :desc)

    %{
      execution_id: execution_id,
      graph_name: execution.graph_name,
      graph_version: execution.graph_version,
      archived_at: execution.archived_at,
      created_at: execution.inserted_at,
      updated_at: execution.updated_at,
      duration_seconds: execution.updated_at - execution.inserted_at,
      revision: execution.revision,
      values: %{
        set: set_values,
        not_set: not_set_values
      },
      computations: %{
        completed: computations_completed,
        outstanding: computations_outstanding
      },
      graph: graph
    }
  end

  @doc """
  Introspects an execution's current state with a human-readable text summary.

  This is the primary debugging and introspection tool for Journey executions,
  providing a comprehensive snapshot of values, computations, and dependencies.

  ## Example

      iex> Journey.Tools.introspect("EXEC07B2H0H7J1LTAE0VJDAL") |> IO.puts()
      Execution summary:
      - ID: 'EXEC07B2H0H7J1LTAE0VJDAL'
      - Graph: 'g1' | 'v1'
      ...
      :ok

  ## Parameters
  - `execution_or_id` - Either the ID of the execution to analyze, or a
    `%Journey.Persistence.Schema.Execution{}` struct (its `:id` will be used).

  ## Returns
  A formatted string with the complete execution state summary.

  Use `summarize_as_data/1` to get execution summary as structured data.
  """
  def introspect(%Journey.Persistence.Schema.Execution{id: id}), do: introspect(id)

  def introspect(execution_id) when is_binary(execution_id) do
    execution_id
    |> summarize_as_data()
    |> convert_summary_data_to_text()
  end

  @doc """
  Generates a human-readable text summary of an execution's current state.

  **This function is deprecated.** Use `introspect/1` instead.

  ## Parameters
  - `execution_id` - The ID of the execution to analyze

  ## Returns
  A formatted string with the complete execution state summary.

  Use `summarize_as_data/1` to get execution summary as data.
  """
  @deprecated "Use introspect/1 instead"
  def summarize_as_text(execution_id) when is_binary(execution_id) do
    introspect(execution_id)
  end

  @doc """
  Generates a human-readable text summary of an execution's current state.

  **This function is deprecated.** Use `introspect/1` instead.

  ## Parameters
  - `execution_id` - The ID of the execution to analyze

  ## Returns
  A formatted string with the complete execution state summary.
  """
  @deprecated "Use introspect/1 instead"
  def summarize(execution_id) when is_binary(execution_id) do
    introspect(execution_id)
  end

  defp convert_summary_data_to_text(summary_data) when is_map(summary_data) do
    %{
      execution_id: execution_id,
      graph_name: graph_name,
      graph_version: graph_version,
      archived_at: archived_at,
      created_at: created_at,
      updated_at: updated_at,
      duration_seconds: duration_seconds,
      revision: revision,
      values: %{set: set_values, not_set: not_set_values},
      computations: %{completed: computations_completed, outstanding: computations_outstanding},
      graph: graph
    } = summary_data

    archived_at_text =
      case archived_at do
        nil -> "not archived"
        _ -> DateTime.from_unix!(archived_at)
      end

    now = System.system_time(:second)

    """
    Execution summary:
    - ID: '#{execution_id}'
    - Graph: '#{graph_name}' | '#{graph_version}'
    - Archived at: #{archived_at_text}
    - Created at: #{DateTime.from_unix!(created_at)} UTC | #{now - created_at} seconds ago
    - Last updated at: #{DateTime.from_unix!(updated_at)} UTC | #{now - updated_at} seconds ago
    - Duration: #{GrabBag.delimit_integer(duration_seconds)} seconds
    - Revision: #{revision}
    - # of Values: #{Enum.count(set_values)} (set) / #{Enum.count(set_values) + Enum.count(not_set_values)} (total)
    - # of Computations: #{Enum.count(computations_completed) + Enum.count(computations_outstanding)}

    Values:
    - Set:
    """ <>
      (set_values
       |> Enum.sort_by(fn %{ex_revision: ex_revision, node_name: node_name} ->
         {-ex_revision, node_name}
       end)
       |> Enum.map_join("\n", fn %{
                                   node_type: node_type,
                                   node_name: node_name,
                                   set_time: set_time,
                                   node_value: node_value,
                                   ex_revision: ex_revision
                                 } ->
         verb = if node_type == :input, do: "set", else: "computed"
         formatted_value = format_node_value(node_name, node_value)

         "  - #{node_name}: '#{formatted_value}' | #{inspect(node_type)}\n" <>
           "    #{verb} at #{DateTime.from_unix!(set_time)} | rev: #{ex_revision}\n"
       end)) <>
      """
      \n
      - Not set:
      """ <>
      (not_set_values
       |> Enum.sort_by(fn %{node_name: node_name} -> node_name end)
       |> Enum.map_join("\n", fn %{node_type: node_type, node_name: node_name} ->
         "  - #{node_name}: <unk> | #{inspect(node_type)}"
       end)) <>
      list_computations(graph, set_values ++ not_set_values, computations_completed, computations_outstanding)
  end

  @doc """
  Generates a Mermaid diagram representation of a Journey graph.

  Converts a graph into Mermaid syntax for visualization. By default returns only
  the flow diagram without timestamp.

  ## Quick Example

  ```elixir
  # Just the flow
  mermaid = Journey.Tools.generate_mermaid_graph(graph)

  # Include timestamp
  mermaid = Journey.Tools.generate_mermaid_graph(graph,
    include_timestamp: true
  )
  ```

  ## Options
  * `:include_timestamp` - Include generation timestamp (default: `false`)
  """
  def generate_mermaid_graph(graph, opts \\ []) do
    opts_schema = [
      include_legend: [is: :boolean],
      include_timestamp: [is: :boolean]
    ]

    KeywordValidator.validate!(opts, opts_schema)

    if Keyword.get(opts, :include_legend) do
      Logger.warning(
        "include_legend option is deprecated for generate_mermaid_graph/2 and will be removed in a future version"
      )
    end

    mermaid_opts =
      opts
      |> Keyword.take([:include_timestamp])
      |> Enum.map(fn
        {:include_timestamp, value} -> {:timestamp, value}
      end)

    JourneyMermaidConverter.compose_mermaid(graph, mermaid_opts)
  end

  @doc """
  Generates a Mermaid diagram of an execution with runtime status emojis overlaid on each node.

  Similar to `generate_mermaid_graph/2`, but takes an execution ID and shows the current
  state of each node using status emojis:

  - ✅ Success / Set (input with value, or compute with `:success`)
  - ⏳ Computing (currently executing)
  - 🚫 Blocked (upstream dependencies not satisfied)
  - ⬜ Not yet set / picked up (input without value, or unblocked compute awaiting pickup)
  - ❌ Failed
  - ❓ Abandoned
  - 🛑 Cancelled

  ## Example

      iex> Journey.Tools.generate_mermaid_execution("EXEC07B2H0H7J1LTAE0VJDAL") |> IO.puts()
      graph TD
          %% Graph
          subgraph Graph["🧩 'my_graph', version v1, EXEC07B2H0H7J1LTAE0VJDAL"]
              ...
          end
      ...
      :ok

  ## Parameters
  - `execution_id` - The ID of the execution to visualize

  ## Options
  * `:include_legend` - Include node type and status legend (default: `false`)
  * `:include_timestamp` - Include generation timestamp (default: `false`)
  """
  def generate_mermaid_execution(execution_id, opts \\ []) when is_binary(execution_id) do
    opts_schema = [
      include_legend: [is: :boolean],
      include_timestamp: [is: :boolean]
    ]

    KeywordValidator.validate!(opts, opts_schema)

    execution =
      case Journey.load(execution_id) do
        nil -> raise ArgumentError, "Execution '#{execution_id}' not found"
        exec -> exec
      end

    graph = Journey.Graph.Catalog.fetch(execution.graph_name, execution.graph_version)

    if graph == nil do
      raise "Graph '#{execution.graph_name}' not found in catalog."
    end

    node_statuses = build_node_status_map(graph, execution)

    mermaid_opts =
      opts
      |> Keyword.take([:include_legend, :include_timestamp])
      |> Enum.map(fn
        {:include_legend, value} -> {:legend, value}
        {:include_timestamp, value} -> {:timestamp, value}
      end)

    JourneyMermaidConverter.compose_mermaid_execution(graph, execution.id, node_statuses, mermaid_opts)
  end

  defp build_node_status_map(graph, execution) do
    graph.nodes
    |> Enum.map(fn node -> {node.name, node_status_emoji(node, execution)} end)
    |> Map.new()
  end

  defp node_status_emoji(%Graph.Input{name: name}, execution) do
    value = Enum.find(execution.values, fn v -> v.node_name == name end)

    if value != nil and value.set_time != nil do
      "✅"
    else
      "⬜"
    end
  end

  defp node_status_emoji(%Graph.Step{name: name, gated_by: gated_by}, execution) do
    state = most_recent_computation_state(execution, name)

    state =
      if state == :success do
        value = Enum.find(execution.values, fn v -> v.node_name == name end)
        value_is_set = value != nil and value.set_time != nil

        readiness =
          Journey.Node.UpstreamDependencies.Computations.evaluate_computation_for_readiness(
            execution.values,
            gated_by
          )

        if value_is_set and readiness.ready?, do: :success, else: :not_set
      else
        state
      end

    computation_state_to_emoji(state, execution.values, gated_by)
  end

  defp most_recent_computation_state(execution, node_name) do
    execution.computations
    |> Enum.filter(fn c -> c.node_name == node_name end)
    |> case do
      [] -> :no_computations
      comps -> comps |> Enum.max_by(fn c -> {c.ex_revision_at_completion || -1, c.id} end) |> Map.get(:state)
    end
  end

  defp computation_state_to_emoji(:success, _values, _gated_by), do: "✅"
  defp computation_state_to_emoji(:computing, _values, _gated_by), do: "⏳"
  defp computation_state_to_emoji(:failed, _values, _gated_by), do: "❌"
  defp computation_state_to_emoji(:abandoned, _values, _gated_by), do: "❓"
  defp computation_state_to_emoji(:cancelled, _values, _gated_by), do: "🛑"

  defp computation_state_to_emoji(not_started, values, gated_by)
       when not_started in [:not_set, :no_computations] do
    readiness =
      Journey.Node.UpstreamDependencies.Computations.evaluate_computation_for_readiness(
        values,
        gated_by
      )

    if readiness.ready?, do: "⬜", else: "🚫"
  end

  defp f_name(fun) when is_function(fun) do
    fi =
      fun
      |> :erlang.fun_info()

    "&#{fi[:name]}/#{fi[:arity]}"
  end

  defp format_condition_tree(%{type: :or, children: children}, indent) do
    "#{indent}:or\n" <>
      format_children_with_connectors(children, indent)
  end

  defp format_condition_tree(%{type: :and, children: [single_child]}, indent) do
    format_condition_tree(single_child, indent)
  end

  defp format_condition_tree(%{type: :and, children: children}, indent) do
    "#{indent}:and\n" <>
      format_children_with_connectors(children, indent)
  end

  defp format_condition_tree(%{type: :not, child: %{type: :leaf, met?: met?, condition: condition}}, indent) do
    status = if met?, do: "✅", else: "🛑"
    node_display = format_node_name_with_context(condition.upstream_node.node_name, condition)
    revision_info = if met?, do: " | rev #{condition.upstream_node.ex_revision}", else: ""

    "#{indent}#{status} :not(#{node_display}) | #{f_name(condition.f_condition)}#{revision_info}"
  end

  defp format_condition_tree(%{type: :not, child: child}, indent) do
    "#{indent}:not\n" <> format_condition_tree(child, indent <> " └─ ")
  end

  defp format_condition_tree(%{type: :leaf, met?: met?, condition: condition}, indent) do
    status = if met?, do: "✅", else: "🛑"
    node_display = format_node_name_with_context(condition.upstream_node.node_name, condition)
    revision_info = if met?, do: " | rev #{condition.upstream_node.ex_revision}", else: ""

    "#{indent}#{status} #{node_display} | #{f_name(condition.f_condition)}#{revision_info}"
  end

  defp format_child_with_connector(child, child_indent, continuation_indent) do
    case child.type do
      type when type in [:and, :or] ->
        "#{child_indent}#{type}\n" <> format_children_with_connectors(child.children, continuation_indent)

      :not ->
        case child.child.type do
          :leaf -> format_condition_tree(child, child_indent)
          _ -> "#{child_indent}:not\n" <> format_condition_tree(child.child, continuation_indent <> " └─ ")
        end

      :leaf ->
        format_condition_tree(child, child_indent)
    end
  end

  defp format_children_with_connectors(children, base_indent) do
    children
    |> Enum.with_index()
    |> Enum.map_join("\n", fn {child, index} ->
      is_last = index == length(children) - 1
      connector = if is_last, do: " └─ ", else: " ├─ "
      child_indent = base_indent <> connector
      continuation_indent = base_indent <> if(is_last, do: "    ", else: " │  ")

      format_child_with_connector(child, child_indent, continuation_indent)
    end)
  end

  defp list_computations(graph, values, computations_completed, computations_outstanding) do
    {abandoned, completed_non_abandoned} =
      Enum.split_with(computations_completed, fn comp -> comp.state == :abandoned end)

    """
      \n
    Computations:
    - Completed:
    """ <>
      Enum.map_join(completed_non_abandoned, "\n\n", fn comp -> completed_computation(comp, graph) end) <>
      if Enum.empty?(abandoned) do
        ""
      else
        """
        \n
        - Abandoned:
        """ <>
          Enum.map_join(abandoned, "\n\n", fn comp -> completed_computation(comp, graph) end)
      end <>
      """
      \n
      - Outstanding:
      """ <>
      Enum.map_join(computations_outstanding, "\n", fn oc -> outstanding_computation(graph, values, oc, true) end)
  end

  defp completed_computation(
         %{
           id: id,
           node_name: node_name,
           state: state,
           computation_type: computation_type,
           computed_with: computed_with,
           ex_revision_at_completion: ex_revision_at_completion,
           start_time: start_time,
           completion_time: completion_time,
           error_details: error_details,
           loop_iteration: loop_iteration,
           loop_state: loop_state
         },
         graph
       ) do
    timing_line = format_computation_timing(start_time, completion_time)
    max_iter = max_iterations_for(graph, node_name)

    inputs_block =
      "    inputs used:\n" <>
        if empty_computed_with?(computed_with) do
          "       <none>"
        else
          Enum.map_join(computed_with, "\n", fn
            {node_name, revision} ->
              "       #{inspect(node_name)} (rev #{revision})"
          end)
        end

    "  - :#{node_name} (#{id}): #{computation_state_to_text(state)} | #{inspect(computation_type)}#{loop_segment(computation_type, loop_iteration, loop_state, max_iter)} | rev #{ex_revision_at_completion}\n" <>
      timing_line <>
      inputs_block <>
      format_error_details(state, error_details)
  end

  # For :loop computation rows, surface loop_iteration (with the configured cap, when
  # known) and the carried disposition. Without this, a cap-failure row (state: :success,
  # loop_state.disposition: "cont_no_fallback") is indistinguishable in the rendered
  # output from a terminal-:ok row, even though the loop's value is unset.
  #
  # max_iterations may be nil for orphaned rows whose graph node was removed via schema
  # evolution; in that case the cap is omitted from the rendering.
  defp loop_segment(:loop, loop_iteration, loop_state, max_iterations) do
    iter_part =
      case max_iterations do
        n when is_integer(n) -> " | iter #{loop_iteration} of #{n}"
        _ -> " | iter #{loop_iteration}"
      end

    disp_part =
      case loop_state do
        %{"disposition" => disposition} -> " | disposition: #{disposition}"
        _ -> ""
      end

    iter_part <> disp_part
  end

  defp loop_segment(_other_type, _loop_iteration, _loop_state, _max_iterations), do: ""

  defp max_iterations_for(graph, node_name) do
    node_name_atom = if is_atom(node_name), do: node_name, else: String.to_atom(node_name)

    case Graph.find_node_by_name(graph, node_name_atom) do
      nil -> nil
      graph_node -> Map.get(graph_node, :max_iterations)
    end
  end

  @error_preview_max_length 500

  defp format_error_details(state, error_details)
       when state in [:failed, :abandoned] and is_binary(error_details) do
    one_line = error_details |> String.replace(~r/\s+/, " ") |> String.trim()

    shown =
      if String.length(one_line) > @error_preview_max_length do
        String.slice(one_line, 0, @error_preview_max_length) <> "..."
      else
        one_line
      end

    "\n    error: #{shown}"
  end

  defp format_error_details(_state, _error_details), do: ""

  defp outstanding_computation(
         graph,
         values,
         %{
           node_name: node_name,
           state: state,
           computation_type: computation_type,
           start_time: start_time,
           loop_iteration: loop_iteration
         },
         with_header?
       ) do
    case Graph.find_node_by_name(graph, node_name) do
      nil ->
        if with_header? do
          "  - #{node_name}: (node not found in current graph definition)\n"
        else
          ""
        end

      graph_node ->
        gated_by = Map.get(graph_node, :gated_by)

        readiness =
          Journey.Node.UpstreamDependencies.Computations.evaluate_computation_for_readiness(
            values,
            gated_by
          )

        header =
          if with_header? do
            # In-flight rows have loop_state: nil (it's only set at iteration completion
            # by record_loop_result), so passing nil here is correct — only the iter segment
            # appears for outstanding :loop rows.
            max_iter = Map.get(graph_node, :max_iterations)

            "  - #{node_name}: #{computation_state_to_text(state)} | #{inspect(computation_type)}#{loop_segment(computation_type, loop_iteration, nil, max_iter)}\n"
          else
            ""
          end

        timing_line = format_outstanding_timing(state, start_time)

        formatted_conditions = format_condition_tree(readiness.structure, "       ")

        header <> timing_line <> formatted_conditions
    end
  end

  defp format_computation_timing(start_time, completion_time)
       when is_integer(start_time) and is_integer(completion_time) do
    duration = completion_time - start_time

    "    started: #{DateTime.from_unix!(start_time)} | completed: #{DateTime.from_unix!(completion_time)} (#{format_duration_seconds(duration)})\n"
  end

  defp format_computation_timing(_start_time, _completion_time), do: ""

  defp format_outstanding_timing(:computing, start_time) when is_integer(start_time) do
    now = System.system_time(:second)
    elapsed = now - start_time
    "    started: #{DateTime.from_unix!(start_time)} | running for #{format_duration_seconds(elapsed)}\n"
  end

  defp format_outstanding_timing(_state, _start_time), do: ""

  defp format_duration_seconds(seconds) when seconds < 60, do: "#{seconds}s"

  defp format_duration_seconds(seconds) when seconds < 3600 do
    minutes = div(seconds, 60)
    remaining = rem(seconds, 60)
    "#{minutes}m #{remaining}s"
  end

  defp format_duration_seconds(seconds) do
    hours = div(seconds, 3600)
    remaining = rem(seconds, 3600)
    minutes = div(remaining, 60)
    secs = rem(remaining, 60)
    "#{hours}h #{minutes}m #{secs}s"
  end

  # Helper function to format node values appropriately
  defp format_node_value(node_name, node_value) do
    case node_name do
      :execution_id -> node_value
      :last_updated_at -> node_value
      _ -> inspect(node_value)
    end
  end

  @doc """
  Retries a failed computation.

  This function enables retrying computations that have exhausted their max_retries
  by making their previous attempts "stale" through upstream revision changes, then
  creating a new computation for the scheduler to pick up.

  ## Parameters
  - `execution_id` - The ID of the execution containing the failed computation
  - `computation_node_name` - The atom name of the computation node to retry

  ## Returns
  The updated execution struct

  ## Example
      iex> Journey.Tools.retry_computation("EXEC123", :email_horoscope)
      %Journey.Persistence.Schema.Execution{...}

  ## How It Works
  1. Finds upstream dependencies that are currently satisfied
  3. Creates a new :not_set computation for the scheduler to pick up
  4. Previous failed attempts become "stale" in the retry counting logic
  5. The scheduler can now execute the new computation attempt
  """
  def retry_computation(execution_id, computation_node_name)
      when is_binary(execution_id) and is_atom(computation_node_name) do
    execution = Journey.load(execution_id)
    graph = Journey.Graph.Catalog.fetch(execution.graph_name, execution.graph_version)

    case Journey.Graph.find_node_by_name(graph, computation_node_name) do
      nil ->
        {:error, :node_not_found}

      graph_node ->
        # Create a new computation for the scheduler to pick up
        Journey.Repo.transaction(fn repo ->
          %Journey.Persistence.Schema.Execution.Computation{
            execution_id: execution_id,
            node_name: Atom.to_string(computation_node_name),
            computation_type: graph_node.type,
            state: :not_set
          }
          |> repo.insert!()
        end)

        advance(execution_id)
    end
  end

  @doc """
  Manually abandons a computation that is currently in the `:computing` state.

  This is useful when you want to immediately abandon a stuck or problematic computation
  without waiting for its deadline to expire. The behavior matches the automatic
  abandonment sweeper: it will schedule a retry if the node hasn't exhausted its
  `max_retries`, then mark the computation as `:abandoned`.

  ## Parameters
  - `computation_id` - The ID of the computation to abandon (starts with "CMP")

  ## Returns
  - `{:ok, computation}` - The updated computation struct
  - `{:error, :not_found}` - Computation with the given ID doesn't exist
  - `{:error, :invalid_state, state}` - Computation is not in `:computing` state

  ## Example

      # Abandon a stuck computation
      {:ok, computation} = Journey.Tools.abandon_computation("CMP123ABC")

      # The computation is now marked as :abandoned
      computation.state
      #=> :abandoned

  ## How It Works
  1. Looks up the computation by ID
  2. Validates the computation is in `:computing` state
  3. Schedules a retry if `max_retries` hasn't been exhausted
  4. Marks the computation as `:abandoned` with current timestamp
  5. Kicks the execution to trigger downstream processing
  """
  def abandon_computation(computation_id) when is_binary(computation_id) do
    case Journey.Repo.get(Computation, computation_id) do
      nil ->
        {:error, :not_found}

      computation ->
        do_abandon_computation(computation)
    end
  end

  defp do_abandon_computation(%Computation{state: :computing} = computation) do
    computation = Journey.Executions.computation_db_to_atoms(computation)

    {:ok, updated_computation} =
      Journey.Repo.transaction(fn repo ->
        # Schedule a retry if max_retries hasn't been exhausted (matches auto-abandonment behavior)
        Journey.Scheduler.Retry.maybe_schedule_a_retry(computation, repo)

        # Mark the computation as abandoned
        computation
        |> Ecto.Changeset.change(%{
          state: :abandoned,
          completion_time: System.system_time(:second)
        })
        |> repo.update!()
      end)

    # Kick the execution to trigger downstream processing
    Journey.kick(updated_computation.execution_id)

    {:ok, updated_computation}
  end

  defp do_abandon_computation(%Computation{state: state}) do
    {:error, :invalid_state, state}
  end

  # Helper function to format node names with conditional context
  defp format_node_name_with_context(node_name, condition_map) do
    case Map.get(condition_map, :condition_context, :direct) do
      :negated -> "not(#{inspect(node_name)})"
      :direct -> inspect(node_name)
    end
  end
end