squid_mesh

defmodule SquidMesh do
  @moduledoc """
  Public entrypoint for the Squid Mesh runtime.

  The API exposed here stays focused on declarative workflow operations. Host
  applications start, inspect, and later control runs through this surface
  without needing to work directly with persistence internals.
  """

  alias SquidMesh.Config
  alias SquidMesh.ReadModel.Inspection
  alias SquidMesh.ReadModel.Listing
  alias SquidMesh.Runs.GraphInspection
  alias SquidMesh.Runtime.Journal.Cancellation
  alias SquidMesh.Runtime.Journal.Executor
  alias SquidMesh.Runtime.Journal.ManualControl
  alias SquidMesh.Runtime.Journal.Options
  alias SquidMesh.Runtime.Journal.Replay
  alias SquidMesh.Runtime.Journal.Starter
  alias SquidMesh.Runtime.ScheduleIdentity

  @read_models [:read_model]
  @runtimes [:journal]
  @projection_snapshot_options [:queue, :now]
  @projection_list_options [:queue, :now]
  @journal_start_options [:runtime, :journal_storage, :queue, :now, :run_id]
  @journal_control_options [:runtime, :journal_storage, :queue, :now]
  @journal_execute_options [:runtime, :journal_storage, :queue, :owner_id, :lease_for, :now]

  @typedoc """
  Structured validation errors returned by the public read-model APIs.
  """
  @type read_option_error ::
          {:invalid_option,
           {:opts, term()}
           | {:read_model, term()}
           | {:journal_storage, nil}
           | {:run_id, term()}}

  @typedoc """
  Structured validation errors returned by the public start APIs.
  """
  @type start_option_error ::
          {:invalid_option,
           {:opts, term()}
           | {:runtime, term()}
           | {:journal_storage, nil}
           | {:queue, term()}
           | {:now, term()}
           | {:run_id, term()}
           | {:schedule_idempotency_key, term()}}

  @doc """
  Loads Squid Mesh configuration from the application environment with optional
  runtime overrides.
  """
  @spec config(keyword()) :: {:ok, Config.t()} | {:error, Config.config_error()}
  defdelegate config(overrides \\ []), to: Config, as: :load

  @doc """
  Loads Squid Mesh configuration and raises if required keys are missing.
  """
  @spec config!(keyword()) :: Config.t()
  defdelegate config!(overrides \\ []), to: Config, as: :load!

  @doc """
  Starts a new workflow run with the given payload through the workflow's
  default trigger.

  The Jido journal runtime returns a projection-backed
  `SquidMesh.ReadModel.Inspection.Snapshot`.
  """
  @spec start_run(module(), map()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error, Config.config_error()}
          | {:error, start_option_error()}
          | {:error, Starter.start_error()}
          | {:error, {:dispatch_failed, term()}}
  def start_run(workflow, payload) when is_map(payload) do
    start_run(workflow, payload, [])
  end

  @spec start_run(module(), map(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error, Config.config_error()}
          | {:error, {:invalid_option, atom()}}
          | {:error, start_option_error()}
          | {:error, Starter.start_error()}
          | {:error, {:dispatch_failed, term()}}
  def start_run(workflow, payload, overrides) when is_map(payload) and is_list(overrides) do
    with :ok <- reject_public_start_options(overrides),
         {:ok, :journal} <- runtime(overrides) do
      start_default_run_with_runtime(:journal, workflow, payload, overrides)
    end
  end

  def start_run(_workflow, _payload, overrides) when is_list(overrides) do
    {:error, {:invalid_payload, :expected_map}}
  end

  @doc """
  Starts a new workflow run through a named trigger with the given payload.
  """
  @spec start_run(module(), atom(), map()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error, Config.config_error()}
          | {:error, start_option_error()}
          | {:error, Starter.start_error()}
          | {:error, {:dispatch_failed, term()}}
  def start_run(workflow, trigger_name, payload)
      when is_atom(trigger_name) and is_map(payload) do
    start_run(workflow, trigger_name, payload, [])
  end

  @doc """
  Starts a named trigger while applying runtime configuration overrides.

  Overrides are intended for host-app test and integration boundaries. Runtime
  context injection is kept out of this public API so scheduled starts and other
  internal callers can keep their idempotency metadata isolated.

  The Jido journal runtime is the default and infers Ecto-backed journal storage
  from the configured repo. Pass `journal_storage:` only when a host, test, or
  integration boundary needs a non-default storage adapter. Journal execution
  supports normal action steps, immediate built-in `:log` steps, built-in
  `:wait` steps in transition and dependency workflows, and manual `:pause` or
  `:approval` boundaries.
  """
  @spec start_run(module(), atom(), map(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error, Config.config_error()}
          | {:error, {:invalid_option, atom()}}
          | {:error, start_option_error()}
          | {:error, Starter.start_error()}
          | {:error, {:dispatch_failed, term()}}
  def start_run(workflow, trigger_name, payload, overrides)
      when is_atom(trigger_name) and is_map(payload) and is_list(overrides) do
    with :ok <- reject_public_start_options(overrides),
         {:ok, :journal} <- runtime(overrides) do
      start_triggered_run_with_runtime(:journal, workflow, trigger_name, payload, overrides)
    end
  end

  @doc false
  @spec start_run_with_initial_context(module(), atom(), map(), map(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:ok, {:duplicate_schedule_start, SquidMesh.ReadModel.Inspection.Snapshot.t()}}
          | {:error, Config.config_error()}
          | {:error, start_option_error()}
          | {:error, Starter.start_error()}
          | {:error, {:dispatch_failed, term()}}
  def start_run_with_initial_context(workflow, trigger_name, payload, initial_context, overrides)
      when is_atom(trigger_name) and is_map(payload) and is_map(initial_context) and
             is_list(overrides) do
    with :ok <- reject_public_start_options(overrides),
         {:ok, :journal} <- runtime(overrides) do
      start_initial_context_run_with_runtime(
        :journal,
        workflow,
        trigger_name,
        payload,
        initial_context,
        overrides
      )
    else
      {:error, reason} -> {:error, reason}
    end
  end

  @doc """
  Fetches one workflow run by id.

  The selected read model comes from host configuration unless overridden. The
  read model rebuilds a projection-backed snapshot from durable journal entries.
  """
  @spec inspect_run(String.t(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | read_option_error()
             | Config.config_error()
             | Inspection.snapshot_error()}
  def inspect_run(run_id, overrides \\ []) do
    with {:ok, :read_model} <- read_model(overrides) do
      inspect_projected_run(run_id, overrides)
    end
  end

  @doc """
  Fetches one workflow run as graph-oriented inspection data.

  The graph projection preserves `inspect_run/2` as the factual run snapshot and
  derives nodes and edges from that same durable state. The selected read model
  comes from host configuration unless overridden.
  """
  @spec inspect_run_graph(String.t(), keyword()) ::
          {:ok, GraphInspection.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | read_option_error()
             | Config.config_error()
             | Inspection.snapshot_error()}
  def inspect_run_graph(run_id, overrides \\ []) do
    with {:ok, :read_model} <- read_model(overrides),
         {:ok, inspection} <- inspect_graph_source(run_id, :read_model, overrides) do
      {:ok, graph_inspection(inspection, :read_model, overrides)}
    end
  end

  @doc """
  Explains the current runtime state of one workflow run.

  The result is structured diagnostic data for host apps, CLIs, and dashboards.
  Use `inspect_run/2` for the factual run snapshot and `explain_run/2` when an
  operator-facing surface needs the reason, evidence, and valid next actions for
  the run's current state.

  The selected read model comes from host configuration unless overridden. The
  read model derives diagnostics from durable journal projections.
  """
  @spec explain_run(String.t(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Explanation.Diagnostic.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | read_option_error()
             | Config.config_error()
             | SquidMesh.ReadModel.Explanation.explanation_error()}
  def explain_run(run_id, overrides \\ []) do
    with {:ok, :read_model} <- read_model(overrides) do
      explain_projected_run(run_id, overrides)
    end
  end

  @doc """
  Executes the next visible workflow attempt through the selected runtime.

  The default journal runtime claims one visible Jido journal-backed attempt,
  runs its declared step, and appends durable attempt completion or failure
  facts. Pass `journal_storage:` only when overriding the inferred Ecto storage
  boundary.
  """
  @spec execute_next(keyword()) :: Executor.execute_result()
  def execute_next(overrides \\ [])

  def execute_next(overrides) when is_list(overrides) do
    with :ok <- public_execute_options(overrides),
         {:ok, :journal} <- runtime(overrides) do
      Executor.execute_next(journal_execute_options(overrides))
    end
  end

  def execute_next(overrides), do: Executor.execute_next(overrides)

  defp public_execute_options(opts) do
    cond do
      not Keyword.keyword?(opts) ->
        :ok

      unsupported = Enum.find(Keyword.keys(opts), &(&1 not in public_execute_option_keys())) ->
        {:error, {:invalid_option, {:option, unsupported}}}

      true ->
        :ok
    end
  end

  defp public_execute_option_keys do
    [:runtime, :journal_storage, :queue, :owner_id, :lease_for, :now]
  end

  @doc """
  Lists workflow runs with optional filters.

  Journal-backed runtime calls return redacted listing summaries. Use
  `inspect_run/2` or `inspect_run_graph/2` with a run id when callers need
  detailed runtime state for one run.
  """
  @spec list_runs([Listing.list_filter()], keyword()) ::
          {:ok, [Listing.Summary.t()]}
          | {:error, Config.config_error() | Listing.list_error()}
  def list_runs(filters \\ [], overrides \\ []) do
    with {:ok, :journal} <- runtime(overrides) do
      list_runs_with_runtime(:journal, filters, overrides)
    end
  end

  @doc """
  Requests cancellation for an eligible workflow run.
  """
  @spec cancel_run(Ecto.UUID.t(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | Config.config_error()
             | Cancellation.cancel_error()}
  def cancel_run(run_id, overrides \\ []) do
    with {:ok, :journal} <- runtime(overrides) do
      cancel_run_with_runtime(:journal, run_id, overrides)
    end
  end

  @doc """
  Resumes a run that is intentionally paused for manual intervention.

  This arity uses the configured runtime. By default, it resolves an inspectable
  journal pause boundary using the inferred Ecto journal storage.
  """
  @spec unblock_run(Ecto.UUID.t()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | Config.config_error()
             | term()}
  def unblock_run(run_id), do: unblock_run(run_id, %{}, [])

  @doc """
  Resumes a paused run with either configuration overrides or manual action
  attributes.

  Pass a keyword list to override runtime configuration, or a map to provide
  manual action attributes. Attribute maps are validated against the paused
  step's manual action contract before the runtime appends resume events or
  dispatches successor work.
  """
  @spec unblock_run(Ecto.UUID.t(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | Config.config_error()
             | term()}
  def unblock_run(run_id, overrides) when is_list(overrides) do
    unblock_run(run_id, %{}, overrides)
  end

  @spec unblock_run(Ecto.UUID.t(), map()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | Config.config_error()
             | term()}
  def unblock_run(run_id, attrs) when is_map(attrs) do
    unblock_run(run_id, attrs, [])
  end

  @doc """
  Resumes a paused run with manual action attributes and configuration overrides.

  By default, this resolves an inspectable journal pause boundary and persists
  the manual action attributes in journal resolution metadata. Pass
  `journal_storage:` only when overriding the inferred Ecto storage boundary.
  """
  @spec unblock_run(Ecto.UUID.t(), map(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | Config.config_error()
             | term()}
  def unblock_run(run_id, attrs, overrides) when is_map(attrs) and is_list(overrides) do
    with {:ok, :journal} <- runtime(overrides) do
      ManualControl.resume(run_id, attrs, journal_control_options(overrides))
    end
  end

  @doc """
  Approves a paused approval step and resumes the run through its success path.

  By default, this resolves an inspectable journal approval boundary and
  persists the decision as journal facts. Pass `journal_storage:` only when
  overriding the inferred Ecto storage boundary.
  """
  @spec approve_run(Ecto.UUID.t(), map(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | Config.config_error()
             | term()}
  def approve_run(run_id, attrs, overrides \\ []) when is_map(attrs) and is_list(overrides) do
    with {:ok, :journal} <- runtime(overrides) do
      ManualControl.approve(run_id, attrs, journal_control_options(overrides))
    end
  end

  @doc """
  Rejects a paused approval step and resumes the run through its rejection path.

  By default, this resolves an inspectable journal approval boundary and
  persists the decision as journal facts. Pass `journal_storage:` only when
  overriding the inferred Ecto storage boundary.
  """
  @spec reject_run(Ecto.UUID.t(), map(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | Config.config_error()
             | term()}
  def reject_run(run_id, attrs, overrides \\ []) when is_map(attrs) and is_list(overrides) do
    with {:ok, :journal} <- runtime(overrides) do
      ManualControl.reject(run_id, attrs, journal_control_options(overrides))
    end
  end

  @doc """
  Creates a new run from a prior run and links it to the original run.

  Replays are blocked by default once the source run completed an irreversible
  or non-compensatable step. Pass `allow_irreversible: true` only after an
  operator has reviewed the side effect and accepted re-execution.
  """
  @spec replay_run(Ecto.UUID.t(), keyword()) ::
          {:ok, SquidMesh.ReadModel.Inspection.Snapshot.t()}
          | {:error,
             :not_found
             | :invalid_run_id
             | Config.config_error()
             | Replay.replay_error()}
          | {:error, {:dispatch_failed, term()}}
  def replay_run(run_id, overrides \\ []) do
    {replay_opts, config_overrides} = Keyword.split(overrides, [:allow_irreversible])

    with {:ok, :journal} <- runtime(config_overrides),
         {:ok, run} <- replay_run_with_runtime(:journal, run_id, replay_opts, config_overrides) do
      {:ok, run}
    else
      {:error, reason} = error when reason in [:not_found, :invalid_run_id] ->
        error

      {:error, {:unsafe_replay, _details} = reason} ->
        {:error, reason}

      {:error, {:invalid_run, _changeset} = reason} ->
        {:error, reason}

      {:error, {:invalid_option, _details} = reason} ->
        {:error, reason}

      {:error, {:incompatible_workflow_definition, _operation} = reason} ->
        {:error, reason}

      {:error, {:invalid_replay_source, _details} = reason} ->
        {:error, reason}

      {:error, %_struct{} = reason} ->
        {:error, {:dispatch_failed, reason}}

      {:error, reason} ->
        {:error, {:dispatch_failed, reason}}
    end
  end

  defp replay_run_with_runtime(:journal, run_id, replay_opts, config_overrides) do
    Replay.replay(run_id, replay_opts, journal_control_options(config_overrides))
  end

  defp start_default_run_with_runtime(:journal, workflow, payload, overrides) do
    Starter.start_run(workflow, nil, payload, journal_start_options(overrides))
  end

  defp start_triggered_run_with_runtime(:journal, workflow, trigger_name, payload, overrides) do
    Starter.start_run(workflow, trigger_name, payload, journal_start_options(overrides))
  end

  defp start_initial_context_run_with_runtime(
         :journal,
         workflow,
         trigger_name,
         payload,
         initial_context,
         overrides
       ) do
    with {:ok, opts} <-
           journal_initial_context_start_options(
             workflow,
             trigger_name,
             initial_context,
             overrides
           ) do
      Starter.start_run(workflow, trigger_name, payload, opts)
    end
  end

  defp list_runs_with_runtime(:journal, filters, overrides) do
    with {:ok, storage} <- journal_storage(overrides) do
      Listing.list(storage, filters, journal_list_options(overrides))
    end
  end

  defp cancel_run_with_runtime(:journal, run_id, overrides) do
    Cancellation.cancel(run_id, journal_control_options(overrides))
  end

  defp journal_initial_context_start_options(workflow, trigger_name, initial_context, overrides) do
    opts =
      overrides
      |> journal_start_options()
      |> Keyword.put(:initial_context, initial_context)

    with {:ok, idempotency_key} <- schedule_idempotency_key(initial_context) do
      case idempotency_key do
        nil ->
          {:ok, opts}

        idempotency_key ->
          opts =
            opts
            |> Keyword.put(:run_id, schedule_run_id(workflow, trigger_name, idempotency_key))
            |> Keyword.put(:duplicate_schedule_start, true)

          {:ok, opts}
      end
    end
  end

  defp schedule_idempotency_key(context) when is_map(context) do
    context
    |> schedule_context()
    |> schedule_value(:idempotency_key)
    |> validate_schedule_idempotency_key()
  end

  defp validate_schedule_idempotency_key(nil), do: {:ok, nil}

  defp validate_schedule_idempotency_key(key) when is_binary(key), do: {:ok, key}

  defp validate_schedule_idempotency_key(_key) do
    {:error, {:invalid_option, {:schedule_idempotency_key, :invalid}}}
  end

  defp schedule_context(context) do
    case Map.fetch(context, :schedule) do
      {:ok, schedule} -> schedule
      :error -> Map.get(context, "schedule", %{})
    end
  end

  defp schedule_value(schedule, key) when is_map(schedule) do
    case Map.fetch(schedule, key) do
      {:ok, value} -> value
      :error -> Map.get(schedule, Atom.to_string(key))
    end
  end

  defp schedule_value(_schedule, _key), do: nil

  defp schedule_run_id(workflow, trigger_name, idempotency_key) do
    workflow_name = SquidMesh.Workflow.Definition.serialize_workflow(workflow)
    trigger = SquidMesh.Workflow.Definition.serialize_trigger(trigger_name)

    {:ok, run_id} = ScheduleIdentity.run_id(workflow_name, trigger, idempotency_key)
    run_id
  end

  defp reject_public_start_options(overrides) do
    cond do
      Keyword.has_key?(overrides, :context) ->
        {:error, {:invalid_option, :context}}

      Keyword.has_key?(overrides, :initial_context) ->
        {:error, {:invalid_option, :initial_context}}

      true ->
        :ok
    end
  end

  defp inspect_projected_run(run_id, overrides) when is_binary(run_id) do
    with {:ok, storage} <- journal_storage(overrides) do
      Inspection.snapshot(storage, run_id, projected_snapshot_options(overrides))
    end
  end

  defp inspect_projected_run(_run_id, _overrides) do
    {:error, {:invalid_option, {:run_id, :invalid}}}
  end

  defp inspect_graph_source(run_id, :read_model, overrides) do
    inspect_projected_run(run_id, overrides)
  end

  defp graph_inspection(%Inspection.Snapshot{} = snapshot, read_model, overrides) do
    GraphInspection.from_snapshot(snapshot, graph_inspection_options(read_model, overrides))
  end

  defp graph_inspection_options(read_model, overrides) do
    [
      source: read_model,
      include_details: Keyword.get(overrides, :include_history, false)
    ]
  end

  defp explain_projected_run(run_id, overrides) do
    with {:ok, storage} <- journal_storage(overrides) do
      SquidMesh.ReadModel.Explanation.explain(
        storage,
        run_id,
        projected_snapshot_options(overrides)
      )
    end
  end

  defp read_model(overrides) when is_list(overrides) do
    with :ok <- validate_keyword_options(overrides) do
      configured_read_model(overrides)
    end
  end

  defp read_model(_overrides), do: {:error, {:invalid_option, {:opts, :invalid}}}

  defp runtime(overrides) when is_list(overrides) do
    with :ok <- validate_keyword_options(overrides) do
      configured_runtime(overrides)
    end
  end

  defp runtime(_overrides), do: {:error, {:invalid_option, {:opts, :invalid}}}

  defp validate_keyword_options(overrides) do
    if Keyword.keyword?(overrides) do
      :ok
    else
      {:error, {:invalid_option, {:opts, :invalid}}}
    end
  end

  defp configured_read_model(overrides) do
    case Keyword.fetch(overrides, :read_model) do
      {:ok, read_model} when read_model in @read_models ->
        {:ok, read_model}

      {:ok, _read_model} ->
        {:error, {:invalid_option, {:read_model, :invalid}}}

      :error ->
        load_configured_read_model(overrides)
    end
  end

  defp load_configured_read_model(overrides) do
    case Config.load(config_routing_overrides(overrides)) do
      {:ok, %Config{read_model: read_model}} -> {:ok, read_model}
      {:error, _reason} = error -> error
    end
  end

  defp configured_runtime(overrides) do
    case Keyword.fetch(overrides, :runtime) do
      {:ok, runtime} when runtime in @runtimes ->
        {:ok, runtime}

      {:ok, _runtime} ->
        {:error, {:invalid_option, {:runtime, :invalid}}}

      :error ->
        load_configured_runtime(overrides)
    end
  end

  defp load_configured_runtime(overrides) do
    case Config.load(config_routing_overrides(overrides)) do
      {:ok, %Config{runtime: runtime}} -> {:ok, runtime}
      {:error, _reason} = error -> error
    end
  end

  defp journal_storage(overrides) do
    case Keyword.fetch(overrides, :journal_storage) do
      {:ok, storage} ->
        Options.storage(storage)

      :error ->
        case Config.load(config_routing_overrides(overrides)) do
          {:ok, %Config{} = config} -> Options.storage(config.journal_storage)
          {:error, {:missing_config, [:journal_storage]}} -> Options.storage(nil)
          {:error, _reason} = error -> error
        end
    end
  end

  defp projected_snapshot_options(overrides) do
    configured_journal_options(overrides, @projection_snapshot_options)
  end

  defp journal_start_options(overrides) do
    configured_journal_options(overrides, @journal_start_options)
  end

  defp journal_control_options(overrides) do
    configured_journal_options(overrides, @journal_control_options)
  end

  defp journal_execute_options(overrides) do
    configured_journal_options(overrides, @journal_execute_options)
  end

  defp journal_list_options(overrides) do
    configured_journal_options(overrides, @projection_list_options)
  end

  defp configured_journal_options(overrides, keys) do
    case load_config_for_journal_options(overrides) do
      {:ok, %Config{} = config} ->
        [
          runtime: :journal,
          journal_storage: config.journal_storage,
          queue: config.queue
        ]
        |> Keyword.merge(Keyword.take(overrides, keys))
        |> Keyword.take(keys)

      {:error, _reason} ->
        Keyword.take(overrides, keys)
    end
  end

  defp load_config_for_journal_options(overrides) do
    case Config.load(config_routing_overrides(overrides)) do
      {:ok, %Config{} = config} ->
        {:ok, config}

      {:error, _reason} ->
        overrides
        |> Keyword.delete(:journal_storage)
        |> config_routing_overrides()
        |> Config.load()
    end
  end

  defp config_routing_overrides(overrides) do
    Keyword.take(overrides, [
      :repo,
      :runtime,
      :read_model,
      :journal_storage
    ])
  end
end