squid_mesh

defmodule SquidMesh.Workflow.Validation do
  @moduledoc """
  Compile-time validation and normalization for workflow modules.

  This module keeps contract enforcement in one place so the DSL in
  `SquidMesh.Workflow` can remain compact and declarative.
  """

  @terminal_transitions [:complete]
  @supported_transition_outcomes [:ok, :error]
  @supported_transition_recovery_markers [:compensation, :undo]
  @supported_transaction_boundaries [:repo]
  @allowed_trigger_types [:manual, :cron]
  @allowed_cron_idempotency_strategies [:return_existing_run, :skip_duplicate]
  @built_in_step_kinds [:wait, :log, :pause, :approval]
  @log_levels [:debug, :info, :warning, :error]

  alias SquidMesh.Workflow.InputMapping

  @doc """
  Validates a compiled workflow definition and raises a compile error when the
  declaration is invalid.
  """
  @spec validate!(map(), Macro.Env.t()) :: :ok
  def validate!(definition, env) do
    case validation_errors(definition) do
      [] ->
        :ok

      errors ->
        description =
          Enum.join(["workflow validation failed:" | Enum.map(errors, &"- #{&1}")], "\n")

        raise CompileError,
          file: env.file,
          line: env.line,
          description: description
    end
  end

  @doc """
  Returns the workflow entry steps or raises when the workflow declaration does
  not define a valid entry set.
  """
  @spec entry_steps!(map(), Macro.Env.t()) :: [atom()]
  def entry_steps!(definition, env) do
    case entry_steps(definition) do
      [] ->
        description =
          if dependency_mode?(definition.steps) do
            "workflow validation failed:\n- dependency-based workflow must define at least one root step"
          else
            "workflow validation failed:\n- workflow must define exactly one entry step"
          end

        raise CompileError,
          file: env.file,
          line: env.line,
          description: description

      [entry_step] ->
        [entry_step]

      entry_steps ->
        if dependency_mode?(definition.steps) do
          entry_steps
        else
          raise CompileError,
            file: env.file,
            line: env.line,
            description:
              "workflow validation failed:\n- workflow must define exactly one entry step"
        end
    end
  end

  @doc """
  Returns the single workflow entry step for transition-based workflows.

  Dependency-based workflows return `nil` because they may declare multiple root
  steps instead of one singular entry step.
  """
  @spec entry_step!(map(), Macro.Env.t()) :: atom() | nil
  def entry_step!(definition, env) do
    if dependency_mode?(definition.steps) do
      nil
    else
      definition
      |> entry_steps!(env)
      |> List.first()
    end
  end

  @doc """
  Returns the first step to schedule for runtime dispatch.
  """
  @spec initial_step!(map(), Macro.Env.t()) :: atom()
  def initial_step!(definition, env) do
    definition
    |> entry_steps!(env)
    |> List.first()
  end

  @doc """
  Converts trigger declarations into the normalized runtime trigger shape.
  """
  @spec normalize_triggers!(map()) :: [map()]
  def normalize_triggers!(definition) do
    Enum.map(definition.triggers, fn trigger ->
      definition_entry = List.first(trigger.definitions)

      %{
        name: trigger.name,
        type: definition_entry.type,
        config: definition_entry.config,
        payload: trigger.payload
      }
    end)
  end

  @doc """
  Returns the canonical workflow payload contract derived from the trigger set.
  """
  @spec workflow_payload!([map()]) :: [map()]
  def workflow_payload!(triggers) when is_list(triggers) do
    triggers
    |> Enum.flat_map(& &1.payload)
    |> Enum.uniq_by(& &1.name)
  end

  def workflow_payload!([]), do: []

  @doc """
  Derives workflow retry declarations from per-step retry configuration.
  """
  @spec derive_retries([map()]) :: [map()]
  def derive_retries(steps) do
    Enum.flat_map(steps, fn step ->
      case Keyword.get(step.opts, :retry) do
        nil ->
          []

        opts when is_list(opts) ->
          [%{step: step.name, opts: opts}]

        opts ->
          [%{step: step.name, opts: opts}]
      end
    end)
  end

  defp validation_errors(definition) do
    step_names = Enum.map(definition.steps, & &1.name)
    payload_fields = workflow_payload_fields(definition)

    []
    |> validate_triggers(definition.triggers)
    |> validate_payload_defaults(payload_fields)
    |> require_steps(step_names)
    |> validate_built_in_steps(definition.steps, definition.transitions)
    |> validate_step_mappings(definition.steps)
    |> validate_step_recovery_markers(definition.steps)
    |> validate_unique_step_names(step_names)
    |> validate_dependency_graph(definition.steps, step_names)
    |> validate_transitions(definition.transitions, step_names)
    |> validate_dependency_transitions(definition.steps, definition.transitions)
    |> validate_retries(definition.retries, step_names)
  end

  defp validate_dependency_graph(errors, steps, step_names) do
    errors
    |> validate_step_dependencies(steps, step_names)
    |> validate_dependency_cycles(steps)
  end

  defp validate_step_dependencies(errors, steps, step_names) do
    Enum.reduce(steps, errors, fn %{name: name, opts: opts}, acc ->
      case dependency_list(opts) do
        {:ok, dependencies} ->
          acc
          |> validate_known_dependencies(name, dependencies, step_names)
          |> validate_self_dependency(name, dependencies)

        :absent ->
          acc

        :error ->
          ["step #{inspect(name)} defines an invalid :after dependency list" | acc]
      end
    end)
  end

  defp validate_known_dependencies(errors, step_name, dependencies, step_names) do
    Enum.reduce(dependencies, errors, fn dependency, acc ->
      if dependency in step_names do
        acc
      else
        ["step #{inspect(step_name)} depends on unknown step #{inspect(dependency)}" | acc]
      end
    end)
  end

  defp validate_self_dependency(errors, step_name, dependencies) do
    if step_name in dependencies do
      ["step #{inspect(step_name)} cannot depend on itself" | errors]
    else
      errors
    end
  end

  defp validate_dependency_cycles(errors, steps) do
    if dependency_mode?(steps) and not dependency_graph_acyclic?(steps) do
      ["workflow dependency graph must be acyclic" | errors]
    else
      errors
    end
  end

  defp validate_triggers(errors, triggers) do
    errors
    |> validate_trigger_count(triggers)
    |> validate_unique_trigger_names(triggers)
    |> validate_trigger_payload_conflicts(triggers)
    |> validate_trigger_definitions(triggers)
  end

  defp validate_trigger_count(errors, []), do: ["at least one trigger is required" | errors]
  defp validate_trigger_count(errors, _triggers), do: errors

  defp validate_unique_trigger_names(errors, triggers) do
    duplicates =
      triggers
      |> Enum.map(& &1.name)
      |> Enum.frequencies()
      |> Enum.filter(fn {_name, count} -> count > 1 end)
      |> Enum.map_join(", ", fn {name, _count} -> inspect(name) end)

    case duplicates do
      "" -> errors
      names -> ["duplicate trigger names: #{names}" | errors]
    end
  end

  defp validate_trigger_payload_conflicts(errors, triggers) do
    triggers
    |> Enum.flat_map(& &1.payload)
    |> Enum.group_by(& &1.name)
    |> Enum.reduce(errors, fn {name, fields}, acc ->
      fields
      |> Enum.map(& &1.type)
      |> Enum.uniq()
      |> case do
        [_single_type] ->
          acc

        types ->
          [
            "payload field #{inspect(name)} defines conflicting types across triggers: #{inspect(types)}"
            | acc
          ]
      end
    end)
  end

  defp validate_trigger_definitions(errors, triggers) do
    Enum.reduce(triggers, errors, fn trigger, acc ->
      acc
      |> validate_trigger_type_count(trigger)
      |> validate_trigger_type_allowed(trigger)
      |> validate_trigger_config(trigger)
    end)
  end

  defp validate_trigger_type_count(errors, %{definitions: [_single_definition]}) do
    errors
  end

  defp validate_trigger_type_count(errors, %{name: name}) do
    ["trigger #{inspect(name)} must define exactly one type" | errors]
  end

  defp validate_trigger_type_allowed(errors, %{definitions: [%{type: type}]})
       when type in @allowed_trigger_types do
    errors
  end

  defp validate_trigger_type_allowed(errors, %{name: name, definitions: [%{type: type}]}) do
    ["trigger #{inspect(name)} defines unsupported type #{inspect(type)}" | errors]
  end

  defp validate_trigger_type_allowed(errors, _trigger), do: errors

  defp validate_trigger_config(errors, %{definitions: [%{type: :manual}]}) do
    errors
  end

  defp validate_trigger_config(errors, %{
         name: name,
         definitions: [%{type: :cron, config: config}]
       }) do
    expression = Map.get(config, :expression)
    timezone = Map.get(config, :timezone)

    errors
    |> validate_cron_schedule(name, expression, timezone)
    |> validate_cron_idempotency(name, Map.get(config, :idempotency))
  end

  defp validate_trigger_config(errors, _trigger), do: errors

  defp validate_cron_schedule(errors, _name, expression, timezone)
       when is_binary(expression) and expression != "" and is_binary(timezone) and timezone != "" do
    errors
  end

  defp validate_cron_schedule(errors, name, _expression, _timezone) do
    ["trigger #{inspect(name)} must define a cron expression and timezone" | errors]
  end

  defp validate_cron_idempotency(errors, _name, nil), do: errors

  defp validate_cron_idempotency(errors, _name, strategy)
       when strategy in @allowed_cron_idempotency_strategies do
    errors
  end

  defp validate_cron_idempotency(errors, name, strategy) do
    [
      "trigger #{inspect(name)} defines invalid cron idempotency strategy #{inspect(strategy)}"
      | errors
    ]
  end

  defp validate_payload_defaults(errors, payload_fields) do
    Enum.reduce(payload_fields, errors, &validate_payload_default/2)
  end

  defp validate_payload_default(field, acc) do
    case Keyword.fetch(field.opts, :default) do
      {:ok, default} -> maybe_invalid_payload_default(field, default, acc)
      :error -> acc
    end
  end

  defp maybe_invalid_payload_default(field, default, acc) do
    if valid_payload_default?(field.type, default) do
      acc
    else
      [
        "payload field #{inspect(field.name)} defines an invalid default for type #{inspect(field.type)}"
        | acc
      ]
    end
  end

  defp require_steps(errors, []), do: ["at least one step is required" | errors]
  defp require_steps(errors, _step_names), do: errors

  defp validate_built_in_steps(errors, steps, transitions) do
    errors =
      Enum.reduce(steps, errors, fn step, acc ->
        validate_built_in_step(acc, step)
      end)

    errors
    |> validate_dependency_manual_step_kind(steps, :pause)
    |> validate_dependency_manual_step_kind(steps, :approval)
    |> validate_manual_step_transition_conditions(steps, transitions)
    |> validate_approval_transitions(steps, transitions)
  end

  defp validate_built_in_step(errors, %{module: kind} = step) when kind in @built_in_step_kinds do
    errors =
      if Keyword.has_key?(step.opts, :transaction) do
        ["built-in step #{inspect(step.name)} cannot declare a :transaction boundary" | errors]
      else
        errors
      end

    case kind do
      :wait -> validate_wait_step(errors, step)
      :log -> validate_log_step(errors, step)
      :pause -> errors
      :approval -> errors
    end
  end

  defp validate_built_in_step(errors, _step), do: errors

  defp validate_dependency_manual_step_kind(errors, steps, kind) do
    if dependency_mode?(steps) and Enum.any?(steps, &(&1.module == kind)) do
      ["dependency-based workflows cannot declare built-in #{inspect(kind)} steps" | errors]
    else
      errors
    end
  end

  defp validate_manual_step_transition_conditions(errors, steps, transitions) do
    manual_steps =
      steps
      |> Enum.filter(&(&1.module in [:pause, :approval]))
      |> MapSet.new(& &1.name)

    Enum.reduce(transitions, errors, fn transition, acc ->
      if Map.has_key?(transition, :condition) and MapSet.member?(manual_steps, transition.from) do
        [
          "transition from built-in manual step #{inspect(transition.from)} cannot define a condition"
          | acc
        ]
      else
        acc
      end
    end)
  end

  defp validate_approval_transitions(errors, steps, transitions) do
    Enum.reduce(steps, errors, fn
      %{name: name, module: :approval}, acc ->
        if has_transition?(transitions, name, :ok) and has_transition?(transitions, name, :error) do
          acc
        else
          ["approval step #{inspect(name)} must define both :ok and :error transitions" | acc]
        end

      _step, acc ->
        acc
    end)
  end

  defp validate_step_mappings(errors, steps) do
    Enum.reduce(steps, errors, fn %{name: name, opts: opts}, acc ->
      acc
      |> validate_step_input_mapping(name, opts)
      |> validate_step_output_mapping(name, opts)
      |> validate_step_transaction_boundary(name, opts)
    end)
  end

  defp validate_step_input_mapping(errors, name, opts) do
    case Keyword.get(opts, :input) do
      nil ->
        errors

      input_mapping ->
        if InputMapping.valid?(input_mapping) do
          errors
        else
          ["step #{inspect(name)} defines an invalid :input mapping" | errors]
        end
    end
  end

  defp validate_step_output_mapping(errors, name, opts) do
    case Keyword.get(opts, :output) do
      nil ->
        errors

      output_mapping when is_atom(output_mapping) ->
        errors

      _other ->
        ["step #{inspect(name)} defines an invalid :output mapping" | errors]
    end
  end

  defp validate_step_transaction_boundary(errors, name, opts) do
    case Keyword.fetch(opts, :transaction) do
      {:ok, boundary} when boundary in @supported_transaction_boundaries ->
        errors

      {:ok, _boundary} ->
        ["step #{inspect(name)} defines an invalid :transaction boundary" | errors]

      :error ->
        errors
    end
  end

  defp validate_step_recovery_markers(errors, steps) do
    Enum.reduce(steps, errors, fn %{name: name, opts: opts}, acc ->
      acc
      |> validate_boolean_step_option(name, opts, :irreversible)
      |> validate_boolean_step_option(name, opts, :compensatable)
      |> validate_step_compensation_callback(name, opts)
      |> validate_recovery_marker_conflict(name, opts)
      |> validate_compensation_marker_conflict(name, opts)
    end)
  end

  defp validate_boolean_step_option(errors, name, opts, option) do
    case Keyword.fetch(opts, option) do
      {:ok, value} when is_boolean(value) ->
        errors

      {:ok, _value} ->
        ["step #{inspect(name)} defines an invalid #{inspect(option)} marker" | errors]

      :error ->
        errors
    end
  end

  defp validate_recovery_marker_conflict(errors, name, opts) do
    if Keyword.get(opts, :irreversible) == true and Keyword.get(opts, :compensatable) == true do
      ["step #{inspect(name)} cannot be both irreversible and compensatable" | errors]
    else
      errors
    end
  end

  defp validate_step_compensation_callback(errors, name, opts) do
    case Keyword.fetch(opts, :compensate) do
      {:ok, callback} when is_atom(callback) ->
        if module_atom?(callback) and callback not in @built_in_step_kinds do
          errors
        else
          ["step #{inspect(name)} defines an invalid :compensate callback" | errors]
        end

      {:ok, _callback} ->
        ["step #{inspect(name)} defines an invalid :compensate callback" | errors]

      :error ->
        errors
    end
  end

  defp module_atom?(callback) when is_atom(callback) do
    callback
    |> Atom.to_string()
    |> String.starts_with?("Elixir.")
  end

  defp validate_compensation_marker_conflict(errors, name, opts) do
    if Keyword.has_key?(opts, :compensate) and
         (Keyword.get(opts, :irreversible) == true or Keyword.get(opts, :compensatable) == false) do
      [
        "step #{inspect(name)} cannot declare :compensate when it is irreversible or non-compensatable"
        | errors
      ]
    else
      errors
    end
  end

  defp validate_wait_step(errors, %{name: name, opts: opts}) do
    duration = Keyword.get(opts, :duration)

    if is_integer(duration) and duration > 0 do
      errors
    else
      ["built-in step #{inspect(name)} requires a positive :duration option" | errors]
    end
  end

  defp validate_log_step(errors, %{name: name, opts: opts}) do
    errors
    |> validate_log_message(name, opts)
    |> validate_log_level(name, opts)
  end

  defp validate_log_message(errors, name, opts) do
    case Keyword.get(opts, :message) do
      message when is_binary(message) and message != "" ->
        errors

      _other ->
        ["built-in step #{inspect(name)} requires a non-empty :message option" | errors]
    end
  end

  defp validate_log_level(errors, name, opts) do
    case Keyword.get(opts, :level, :info) do
      level when level in @log_levels ->
        errors

      _other ->
        ["built-in step #{inspect(name)} defines unsupported :level" | errors]
    end
  end

  defp validate_unique_step_names(errors, step_names) do
    duplicates =
      step_names
      |> Enum.frequencies()
      |> Enum.filter(fn {_name, count} -> count > 1 end)
      |> Enum.map_join(", ", fn {name, _count} -> inspect(name) end)

    case duplicates do
      "" -> errors
      names -> ["duplicate step names: #{names}" | errors]
    end
  end

  defp validate_transitions(errors, transitions, step_names) do
    errors
    |> validate_duplicate_transitions(transitions)
    |> then(fn acc ->
      Enum.reduce(transitions, acc, fn transition, reduce_acc ->
        reduce_acc
        |> validate_transition_from(transition, step_names)
        |> validate_transition_outcome(transition)
        |> validate_transition_condition(transition)
        |> validate_transition_recovery_marker(transition)
        |> validate_transition_to(transition, step_names)
      end)
    end)
  end

  defp validate_dependency_transitions(errors, steps, transitions) do
    if dependency_mode?(steps) and transitions != [] do
      ["dependency-based workflows cannot declare transitions" | errors]
    else
      errors
    end
  end

  defp validate_transition_from(errors, %{from: from}, step_names) do
    if from in step_names do
      errors
    else
      ["transition references unknown step: #{inspect(from)}" | errors]
    end
  end

  defp validate_transition_outcome(errors, %{from: from, on: outcome}) do
    if outcome in @supported_transition_outcomes do
      errors
    else
      [
        "transition from #{inspect(from)} defines unsupported outcome #{inspect(outcome)}"
        | errors
      ]
    end
  end

  defp validate_transition_recovery_marker(errors, %{recovery: _recovery, from: from, on: on})
       when on != :error do
    [
      "transition from #{inspect(from)} can only define recovery markers for :error outcomes"
      | errors
    ]
  end

  defp validate_transition_recovery_marker(errors, %{recovery: recovery, from: from}) do
    if recovery in @supported_transition_recovery_markers do
      errors
    else
      [
        "transition from #{inspect(from)} defines unsupported recovery marker #{inspect(recovery)}"
        | errors
      ]
    end
  end

  defp validate_transition_recovery_marker(errors, _transition), do: errors

  defp validate_transition_to(errors, %{to: to}, step_names) do
    if to in step_names or to in @terminal_transitions do
      errors
    else
      ["transition targets unknown step: #{inspect(to)}" | errors]
    end
  end

  defp validate_transition_condition(errors, %{condition: condition, from: from}) do
    case SquidMesh.Workflow.TransitionCondition.normalize(condition) do
      {:ok, _condition} ->
        errors

      {:error, :invalid_condition} ->
        ["transition from #{inspect(from)} defines an invalid condition" | errors]
    end
  end

  defp validate_transition_condition(errors, _transition), do: errors

  defp validate_duplicate_transitions(errors, transitions) do
    transitions
    |> Enum.group_by(fn %{from: from, on: outcome} -> {from, outcome} end)
    |> Enum.reduce(errors, fn
      {{from, outcome}, grouped}, acc ->
        validate_transition_group(acc, from, outcome, grouped)
    end)
  end

  defp validate_transition_group(errors, _from, _outcome, [_single_transition]), do: errors

  defp validate_transition_group(errors, from, outcome, transitions) do
    unconditional_count = Enum.count(transitions, &(not Map.has_key?(&1, :condition)))

    cond do
      unconditional_count > 1 ->
        [
          "duplicate transition declared from #{inspect(from)} on outcome #{inspect(outcome)}"
          | errors
        ]

      duplicate_transition_condition?(transitions) ->
        [
          "duplicate transition declared from #{inspect(from)} on outcome #{inspect(outcome)}"
          | errors
        ]

      true ->
        errors
    end
  end

  defp duplicate_transition_condition?(transitions) do
    conditions =
      transitions
      |> Enum.filter(&Map.has_key?(&1, :condition))
      |> Enum.map(&SquidMesh.Workflow.TransitionCondition.serialize(&1.condition))

    Enum.uniq(conditions) != conditions
  end

  defp has_transition?(transitions, from_step, outcome) do
    Enum.any?(transitions, &(&1.from == from_step and &1.on == outcome))
  end

  defp validate_retries(errors, retries, step_names) do
    Enum.reduce(retries, errors, fn retry, acc ->
      acc
      |> validate_retry_step(retry, step_names)
      |> validate_retry_opts(retry)
    end)
  end

  defp validate_retry_step(errors, %{step: step}, step_names) do
    if step in step_names do
      errors
    else
      ["retry references unknown step: #{inspect(step)}" | errors]
    end
  end

  defp validate_retry_opts(errors, %{step: step, opts: opts}) do
    if is_list(opts) do
      max_attempts = Keyword.get(opts, :max_attempts)

      if is_integer(max_attempts) and max_attempts > 0 do
        validate_retry_backoff(errors, step, opts)
      else
        ["retry for #{inspect(step)} must define a positive :max_attempts" | errors]
      end
    else
      ["retry for #{inspect(step)} must define a positive :max_attempts" | errors]
    end
  end

  defp validate_retry_backoff(errors, step, opts) do
    case Keyword.get(opts, :backoff) do
      nil ->
        errors

      backoff when is_list(backoff) ->
        if valid_retry_backoff?(backoff) do
          errors
        else
          ["retry for #{inspect(step)} defines an invalid :backoff option" | errors]
        end

      _other ->
        ["retry for #{inspect(step)} defines an invalid :backoff option" | errors]
    end
  end

  defp valid_retry_backoff?(backoff) do
    case Keyword.get(backoff, :type) do
      :exponential ->
        min_delay = Keyword.get(backoff, :min)
        max_delay = Keyword.get(backoff, :max)

        is_integer(min_delay) and min_delay > 0 and
          is_integer(max_delay) and max_delay >= min_delay

      _other ->
        false
    end
  end

  defp valid_payload_default?(:string, {:today, :iso8601}), do: true
  defp valid_payload_default?(:string, {:now, :iso8601}), do: true
  defp valid_payload_default?(type, default), do: input_matches_type?(default, type)

  defp workflow_payload_fields(%{payload: payload}) when is_list(payload), do: payload

  defp workflow_payload_fields(%{triggers: triggers}) when is_list(triggers) do
    Enum.flat_map(triggers, &Map.get(&1, :payload, []))
  end

  defp workflow_payload_fields(_definition), do: []

  defp entry_steps(definition) do
    if dependency_mode?(definition.steps) do
      incoming_dependencies = dependency_map(definition.steps)

      definition.steps
      |> Enum.map(& &1.name)
      |> Enum.reject(fn step_name ->
        incoming_dependencies
        |> Map.get(step_name, [])
        |> Enum.any?()
      end)
    else
      transition_targets =
        definition.transitions
        |> Enum.map(& &1.to)
        |> MapSet.new()

      definition.steps
      |> Enum.map(& &1.name)
      |> Enum.reject(&MapSet.member?(transition_targets, &1))
    end
  end

  defp dependency_mode?(steps) when is_list(steps) do
    Enum.any?(steps, fn step ->
      case Keyword.get(step.opts, :after) do
        dependencies when is_list(dependencies) -> dependencies != []
        _other -> false
      end
    end)
  end

  defp dependency_list(opts) do
    case Keyword.fetch(opts, :after) do
      {:ok, []} ->
        :error

      {:ok, dependencies} when is_list(dependencies) ->
        if Enum.all?(dependencies, &is_atom/1) do
          {:ok, Enum.uniq(dependencies)}
        else
          :error
        end

      {:ok, _other} ->
        :error

      :error ->
        :absent
    end
  end

  defp dependency_graph_acyclic?(steps) do
    adjacency = dependency_map(steps)

    {result, _state} =
      Enum.reduce_while(
        Map.keys(adjacency),
        {:ok, %{visiting: MapSet.new(), visited: MapSet.new()}},
        fn
          step_name, {:ok, state} ->
            case visit_dependency(step_name, adjacency, state) do
              {:ok, next_state} -> {:cont, {:ok, next_state}}
              {:error, :cycle} -> {:halt, {:error, :cycle}}
            end
        end
      )

    result == :ok
  end

  defp dependency_map(steps) do
    Map.new(steps, fn %{name: name, opts: opts} ->
      explicit_dependencies =
        case dependency_list(opts) do
          {:ok, dependencies} -> dependencies
          _other -> []
        end

      {name, explicit_dependencies}
    end)
  end

  defp visit_dependency(step_name, adjacency, %{visited: visited} = state) do
    cond do
      MapSet.member?(visited, step_name) ->
        {:ok, state}

      MapSet.member?(state.visiting, step_name) ->
        {:error, :cycle}

      true ->
        state = %{state | visiting: MapSet.put(state.visiting, step_name)}

        adjacency
        |> Map.get(step_name, [])
        |> visit_dependencies(adjacency, state)
        |> case do
          {:ok, next_state} ->
            {:ok,
             %{
               next_state
               | visiting: MapSet.delete(next_state.visiting, step_name),
                 visited: MapSet.put(next_state.visited, step_name)
             }}

          {:error, :cycle} ->
            {:error, :cycle}
        end
    end
  end

  defp visit_dependencies(dependencies, adjacency, state) do
    Enum.reduce_while(dependencies, {:ok, state}, fn dependency, {:ok, acc} ->
      case visit_dependency(dependency, adjacency, acc) do
        {:ok, next_acc} -> {:cont, {:ok, next_acc}}
        {:error, :cycle} -> {:halt, {:error, :cycle}}
      end
    end)
  end

  defp input_matches_type?(value, :string), do: is_binary(value)
  defp input_matches_type?(value, :integer), do: is_integer(value)
  defp input_matches_type?(value, :float), do: is_float(value)
  defp input_matches_type?(value, :boolean), do: is_boolean(value)
  defp input_matches_type?(value, :map), do: is_map(value)
  defp input_matches_type?(value, :list), do: is_list(value)
  defp input_matches_type?(value, :atom), do: is_atom(value)
  defp input_matches_type?(_value, _unknown_type), do: true
end