squid_mesh

defmodule SquidMesh.Runtime.Journal.Starter do
  @moduledoc """
  Journal-backed workflow start boundary for the Jido-native runtime.

  This module resolves the public Squid Mesh workflow contract, plans initial
  runnables through Runic, appends durable run and run-index facts to
  `Jido.Storage`, then uses `WorkflowAgent` and `DispatchAgent` as rebuildable
  coordinators to schedule dispatch attempts from the journal.

  The journal runtime can execute normal action steps, immediate built-in `:log`
  steps, built-in `:wait` steps in transition and dependency workflows, and
  manual `:pause` or `:approval` boundaries. Manual boundaries persist
  inspectable intervention state and can be resumed or reviewed through journal
  manual controls. Callers enter this path explicitly with `runtime: :journal`,
  `journal_storage:`, and optional queue or clock overrides. No Jido primitive
  is required in workflow authoring.
  """

  alias SquidMesh.ReadModel.Inspection
  alias SquidMesh.Runtime.AgentRecovery
  alias SquidMesh.Runtime.DispatchAgent
  alias SquidMesh.Runtime.DispatchProtocol
  alias SquidMesh.Runtime.Journal
  alias SquidMesh.Runtime.Journal.Options
  alias SquidMesh.Runtime.RunCatalogProjection
  alias SquidMesh.Runtime.RunIndexProjection
  alias SquidMesh.Runtime.WorkflowAgent
  alias SquidMesh.Runtime.WorkflowAgent.Projection
  alias SquidMesh.Workflow.Definition
  alias SquidMesh.Workflow.RunicPlanner

  @dispatch_schedule_retries 25

  @type start_error ::
          {:invalid_payload, Definition.payload_error_details()}
          | Definition.load_error()
          | Definition.trigger_error()
          | {:invalid_option,
             {:journal_storage, nil} | {:now, term()} | {:queue, term()} | {:run_id, term()}}
          | term()

  @doc """
  Starts a workflow run by appending Jido journal facts and scheduling dispatch.

  The returned snapshot is rebuilt from the same journal-backed read model used
  by `SquidMesh.inspect_run/2` with `read_model: :read_model`.
  """
  @spec start_run(module(), atom() | nil, map(), keyword()) ::
          {:ok, Inspection.Snapshot.t()}
          | {:ok, {:duplicate_schedule_start, Inspection.Snapshot.t()}}
          | {:error, start_error()}
  def start_run(workflow, trigger_name, payload, opts)
      when is_atom(workflow) and is_map(payload) and is_list(opts) do
    with {:ok, storage} <- journal_storage(opts),
         {:ok, queue} <- queue(opts),
         {:ok, now} <- now(opts),
         {:ok, definition} <- Definition.load(workflow),
         {:ok, trigger} <- trigger(definition, trigger_name),
         {:ok, resolved_payload} <- Definition.resolve_payload(trigger, payload),
         {:ok, planner} <- RunicPlanner.new(workflow),
         {:ok, _planned, runnables} <- RunicPlanner.plan(planner, resolved_payload),
         {:ok, run_id} <- run_id(opts),
         {:ok, journal_runnables} <- journal_runnables(definition, run_id, queue, runnables, now),
         {:ok, start_state} <-
           ensure_run_started(
             storage,
             %{
               workflow: workflow,
               definition: definition,
               trigger: trigger,
               input: resolved_payload,
               run_id: run_id
             },
             journal_runnables,
             now,
             opts
           ) do
      complete_started_run(storage, workflow, run_id, queue, now, start_state, opts)
    end
  end

  defp trigger(definition, nil),
    do: Definition.trigger(definition, Definition.default_trigger(definition))

  defp trigger(definition, trigger_name), do: Definition.trigger(definition, trigger_name)

  defp ensure_run_started(
         storage,
         %{
           workflow: workflow,
           definition: definition,
           trigger: trigger,
           input: input,
           run_id: run_id
         },
         runnables,
         %DateTime{} = now,
         opts
       ) do
    expected_fingerprint = Definition.fingerprint(definition)

    with {:ok, run_started} <-
           DispatchProtocol.new_entry(:run_started, %{
             run_id: run_id,
             workflow: Definition.serialize_workflow(workflow),
             trigger: Atom.to_string(Map.fetch!(trigger, :name)),
             input: input,
             context: initial_context(opts),
             replayed_from_run_id: replayed_from_run_id(opts),
             definition_fingerprint: expected_fingerprint,
             occurred_at: now
           }),
         {:ok, runnables_planned} <-
           DispatchProtocol.new_entry(:runnables_planned, %{
             run_id: run_id,
             runnables: runnables,
             occurred_at: now
           }),
         {:ok, _run_thread} <-
           Journal.append_entries(storage, [run_started, runnables_planned], expected_rev: 0) do
      {:ok, :created}
    else
      {:error, :conflict} ->
        rebuild_existing_start(storage, workflow, run_id, runnables, expected_fingerprint, opts)

      {:error, _reason} = error ->
        error
    end
  end

  defp ensure_run_queued(storage, run_id, queue, %DateTime{} = now) do
    ensure_run_queued(storage, run_id, queue, now, @dispatch_schedule_retries)
  end

  defp ensure_run_queued(_storage, _run_id, _queue, %DateTime{}, 0), do: {:error, :conflict}

  defp ensure_run_queued(storage, run_id, queue, %DateTime{} = now, retries_left) do
    with {:ok, dispatch_agent} <- DispatchAgent.rebuild(storage, queue) do
      case DispatchAgent.ensure_run_queued(storage, dispatch_agent, run_id, now: now) do
        {:ok, _queued} ->
          :ok

        {:error, :conflict} ->
          ensure_run_queued(storage, run_id, queue, now, retries_left - 1)

        {:error, _reason} = error ->
          error
      end
    end
  end

  defp journal_runnables(definition, run_id, queue, runnables, %DateTime{} = now) do
    result =
      Enum.reduce_while(runnables, {:ok, []}, fn runnable, {:ok, acc} ->
        case journal_runnable(definition, run_id, queue, runnable, now) do
          {:ok, journal_runnable} -> {:cont, {:ok, [journal_runnable | acc]}}
          {:error, _reason} = error -> {:halt, error}
        end
      end)

    case result do
      {:ok, journal_runnables} -> {:ok, Enum.reverse(journal_runnables)}
      {:error, _reason} = error -> error
    end
  end

  defp journal_runnable(
         definition,
         run_id,
         queue,
         %{step: step, input: input},
         %DateTime{} = now
       )
       when is_atom(step) do
    attempt_number = 1
    step_name = Definition.serialize_step(step)
    runnable_key = "#{run_id}:#{step_name}:#{attempt_number}"

    with {:ok, recovery} <- replay_recovery_policy(definition, step) do
      {:ok,
       %{
         run_id: run_id,
         runnable_key: runnable_key,
         idempotency_key: runnable_key,
         attempt_number: attempt_number,
         queue: queue,
         step: step_name,
         input: input || %{},
         recovery: recovery,
         visible_at: now
       }}
    end
  end

  defp journal_runnable(_definition, _run_id, _queue, runnable, %DateTime{}) do
    {:error, {:invalid_runnable, runnable}}
  end

  defp replay_recovery_policy(definition, step) do
    with {:ok, recovery} <- Definition.step_recovery_policy(definition, step) do
      {:ok,
       %{
         "irreversible?" => recovery.irreversible?,
         "compensatable?" => recovery.compensatable?,
         "replay" => Atom.to_string(recovery.replay),
         "recovery" => Atom.to_string(recovery.recovery)
       }}
    end
  end

  defp put_checkpoints(storage, workflow_agent, dispatch_agent, %DateTime{} = now) do
    _checkpoint_result = WorkflowAgent.put_checkpoint(storage, workflow_agent, updated_at: now)
    _checkpoint_result = DispatchAgent.put_checkpoint(storage, dispatch_agent, updated_at: now)

    :ok
  end

  defp ensure_run_indexed(storage, workflow, run_id, queue, %DateTime{} = now) do
    workflow = Definition.serialize_workflow(workflow)

    with :ok <- ensure_workflow_run_indexed(storage, workflow, run_id, queue, now) do
      ensure_global_run_cataloged(storage, workflow, run_id, queue, now)
    end
  end

  defp ensure_workflow_run_indexed(storage, workflow, run_id, queue, %DateTime{} = now) do
    with {:ok, entry} <-
           DispatchProtocol.new_entry(:run_indexed, %{
             run_id: run_id,
             workflow: workflow,
             queue: queue,
             occurred_at: now
           }) do
      ensure_run_index_entry(storage, workflow, run_id, entry, @dispatch_schedule_retries)
    end
  end

  defp ensure_global_run_cataloged(storage, workflow, run_id, queue, %DateTime{} = now) do
    with {:ok, entry} <-
           DispatchProtocol.new_entry(:run_cataloged, %{
             run_id: run_id,
             workflow: workflow,
             queue: queue,
             occurred_at: now
           }) do
      ensure_run_cataloged(storage, run_id, entry, @dispatch_schedule_retries)
    end
  end

  defp ensure_run_index_entry(storage, workflow, run_id, entry, retries_left) do
    case load_run_index(storage, workflow) do
      {:ok, %{rev: rev, projection: projection}} ->
        append_missing_run_index(storage, workflow, run_id, entry, rev, projection, retries_left)

      {:error, _reason} = error ->
        error
    end
  end

  defp append_missing_run_index(storage, workflow, run_id, entry, rev, projection, retries_left) do
    case run_index_state(projection, run_id, workflow, entry.data.queue) do
      :matching ->
        :ok

      :missing ->
        append_run_index_entry(storage, workflow, run_id, entry, rev, retries_left)

      {:conflicting, _summary} ->
        {:error, {:conflicting_run_index, run_id}}
    end
  end

  defp append_run_index_entry(storage, workflow, run_id, entry, rev, retries_left) do
    case Journal.append_entries(storage, [entry], expected_rev: rev) do
      {:ok, _thread} ->
        :ok

      {:error, :conflict} when retries_left > 0 ->
        ensure_run_index_entry(storage, workflow, run_id, entry, retries_left - 1)

      {:error, _reason} = error ->
        error
    end
  end

  defp load_run_index(storage, workflow) do
    case Journal.load_thread(storage, {:run_index, workflow}) do
      {:ok, %{rev: rev, entries: entries}} ->
        projection =
          workflow
          |> RunIndexProjection.new()
          |> RunIndexProjection.replay(entries)

        {:ok, %{rev: rev, projection: projection}}

      {:error, :not_found} ->
        {:ok, %{rev: 0, projection: RunIndexProjection.new(workflow)}}

      {:error, _reason} = error ->
        error
    end
  end

  defp run_index_state(%RunIndexProjection{} = projection, run_id, workflow, queue) do
    projection
    |> RunIndexProjection.runs()
    |> Enum.find(&(&1.run_id == run_id))
    |> case do
      nil ->
        :missing

      %{workflow: ^workflow, queue: ^queue} ->
        :matching

      summary ->
        {:conflicting, summary}
    end
  end

  defp ensure_run_cataloged(storage, run_id, entry, retries_left) do
    case load_run_catalog(storage) do
      {:ok, %{rev: rev, projection: projection}} ->
        append_missing_run_catalog(storage, run_id, entry, rev, projection, retries_left)

      {:error, _reason} = error ->
        error
    end
  end

  defp append_missing_run_catalog(storage, run_id, entry, rev, projection, retries_left) do
    case run_catalog_state(projection, run_id, entry.data.workflow, entry.data.queue) do
      :matching ->
        :ok

      :missing ->
        append_run_catalog_entry(storage, run_id, entry, rev, retries_left)

      {:conflicting, _summary} ->
        {:error, {:conflicting_run_catalog, run_id}}
    end
  end

  defp append_run_catalog_entry(storage, run_id, entry, rev, retries_left) do
    case Journal.append_entries(storage, [entry], expected_rev: rev) do
      {:ok, _thread} ->
        :ok

      {:error, :conflict} when retries_left > 0 ->
        ensure_run_cataloged(storage, run_id, entry, retries_left - 1)

      {:error, _reason} = error ->
        error
    end
  end

  defp load_run_catalog(storage) do
    case Journal.load_thread(storage, {:run_catalog, "all"}) do
      {:ok, %{rev: rev, entries: entries}} ->
        {:ok, %{rev: rev, projection: RunCatalogProjection.rebuild(entries)}}

      {:error, :not_found} ->
        {:ok, %{rev: 0, projection: RunCatalogProjection.new()}}

      {:error, _reason} = error ->
        error
    end
  end

  defp run_catalog_state(%RunCatalogProjection{} = projection, run_id, workflow, queue) do
    projection
    |> RunCatalogProjection.runs()
    |> Enum.find(&(&1.run_id == run_id))
    |> case do
      nil ->
        :missing

      %{workflow: ^workflow, queue: ^queue} ->
        :matching

      summary ->
        {:conflicting, summary}
    end
  end

  defp complete_started_run(
         storage,
         workflow,
         run_id,
         queue,
         %DateTime{} = now,
         start_state,
         opts
       ) do
    queue = completion_queue(start_state, queue)

    case do_complete_started_run(storage, workflow, run_id, queue, now) do
      {:ok, %Inspection.Snapshot{} = snapshot} ->
        duplicate_start_result(snapshot, start_state, opts)

      {:error, reason} ->
        {:error, {:journal_start_committed, run_id, reason}}
    end
  end

  defp completion_queue({:existing_schedule_duplicate, queue}, _queue), do: queue
  defp completion_queue(_start_state, queue), do: queue

  defp duplicate_start_result(
         %Inspection.Snapshot{} = snapshot,
         {:existing_schedule_duplicate, _queue},
         _opts
       ) do
    {:ok, {:duplicate_schedule_start, snapshot}}
  end

  defp duplicate_start_result(%Inspection.Snapshot{} = snapshot, :existing, opts) do
    if Keyword.get(opts, :duplicate_schedule_start, false) do
      {:ok, {:duplicate_schedule_start, snapshot}}
    else
      {:ok, snapshot}
    end
  end

  defp duplicate_start_result(%Inspection.Snapshot{} = snapshot, _start_state, _opts) do
    {:ok, snapshot}
  end

  defp do_complete_started_run(storage, workflow, run_id, queue, %DateTime{} = now) do
    with :ok <- ensure_run_indexed(storage, workflow, run_id, queue, now),
         :ok <- ensure_run_queued(storage, run_id, queue, now),
         {:ok, %{workflow_agent: workflow_agent, dispatch_agent: dispatch_agent}} <-
           recover_or_continue(storage, run_id, queue, now),
         :ok <- put_checkpoints(storage, workflow_agent, dispatch_agent, now) do
      Inspection.snapshot(storage, run_id, queue: queue, now: now)
    end
  end

  defp rebuild_existing_start(
         storage,
         workflow,
         run_id,
         expected_runnables,
         expected_fingerprint,
         opts
       ) do
    with {:ok, workflow_agent} <- WorkflowAgent.rebuild(storage, run_id),
         {:ok, existing_fingerprint} <- persisted_definition_fingerprint(storage, run_id) do
      mode = existing_start_mode(opts)

      validate_existing_start_mode(
        mode,
        workflow_agent,
        workflow,
        expected_runnables,
        expected_fingerprint,
        existing_fingerprint,
        opts
      )
    end
  end

  defp existing_start_mode(opts) do
    if Keyword.get(opts, :duplicate_schedule_start, false) do
      :schedule_duplicate
    else
      :strict
    end
  end

  defp validate_existing_start_mode(
         :schedule_duplicate,
         workflow_agent,
         workflow,
         _expected_runnables,
         _expected_fingerprint,
         _existing_fingerprint,
         opts
       ) do
    validate_existing_schedule_start(workflow_agent, workflow, opts)
  end

  defp validate_existing_start_mode(
         :strict,
         workflow_agent,
         workflow,
         expected_runnables,
         expected_fingerprint,
         existing_fingerprint,
         _opts
       ) do
    with :ok <-
           validate_existing_start(
             workflow_agent,
             workflow,
             expected_runnables,
             expected_fingerprint,
             existing_fingerprint
           ) do
      {:ok, :existing}
    end
  end

  defp validate_existing_schedule_start(workflow_agent, workflow, opts) do
    existing_workflow = workflow_agent.state.workflow
    expected_workflow = Definition.serialize_workflow(workflow)
    expected_idempotency_key = schedule_idempotency_key(initial_context(opts))
    existing_idempotency_key = existing_schedule_idempotency_key(workflow_agent)

    cond do
      existing_workflow != expected_workflow ->
        {:error, :conflict}

      is_nil(expected_idempotency_key) ->
        {:error, :conflict}

      existing_idempotency_key != expected_idempotency_key ->
        {:error, :conflict}

      true ->
        existing_queue(workflow_agent)
    end
  end

  defp existing_schedule_idempotency_key(%{
         state: %{projection: %Projection{context: context}}
       })
       when is_map(context) do
    context
    |> schedule_context()
    |> schedule_value(:idempotency_key)
  end

  defp existing_schedule_idempotency_key(_workflow_agent), do: nil

  defp schedule_idempotency_key(context) when is_map(context) do
    context
    |> schedule_context()
    |> schedule_value(:idempotency_key)
  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 existing_queue(workflow_agent) do
    workflow_agent
    |> WorkflowAgent.planned_runnables()
    |> Enum.find_value(&runnable_value(&1, :queue))
    |> case do
      queue when is_binary(queue) -> {:ok, {:existing_schedule_duplicate, queue}}
      _missing_queue -> {:error, :conflict}
    end
  end

  defp validate_existing_start(
         workflow_agent,
         workflow,
         expected_runnables,
         expected_fingerprint,
         existing_fingerprint
       ) do
    existing_workflow = workflow_agent.state.workflow
    expected_workflow = Definition.serialize_workflow(workflow)
    existing_runnables = WorkflowAgent.planned_runnables(workflow_agent)

    cond do
      existing_workflow != expected_workflow ->
        {:error, :conflict}

      not equivalent_definition_fingerprint?(existing_fingerprint, expected_fingerprint) ->
        {:error, :conflict}

      equivalent_runnables?(existing_runnables, expected_runnables) ->
        :ok

      true ->
        {:error, :conflict}
    end
  end

  defp persisted_definition_fingerprint(storage, run_id) do
    with {:ok, %{entries: entries}} <- Journal.load_thread(storage, {:run, run_id}) do
      fingerprint =
        Enum.find_value(entries, fn
          %{type: :run_started, data: data} -> Map.get(data, :definition_fingerprint)
          _entry -> nil
        end)

      {:ok, fingerprint}
    end
  end

  defp equivalent_definition_fingerprint?(existing_fingerprint, expected_fingerprint),
    do: existing_fingerprint == expected_fingerprint

  defp equivalent_runnables?(left, right) when length(left) == length(right) do
    left =
      left
      |> Enum.map(&stable_runnable/1)
      |> Enum.sort()

    right =
      right
      |> Enum.map(&stable_runnable/1)
      |> Enum.sort()

    left == right
  end

  defp equivalent_runnables?(_left, _right), do: false

  defp stable_runnable(runnable) when is_map(runnable) do
    %{
      run_id: runnable_value(runnable, :run_id),
      runnable_key: runnable_value(runnable, :runnable_key),
      idempotency_key: runnable_value(runnable, :idempotency_key),
      attempt_number: runnable_value(runnable, :attempt_number),
      queue: runnable_value(runnable, :queue),
      step: runnable_value(runnable, :step),
      input: runnable_value(runnable, :input)
    }
  end

  defp recover_or_continue(storage, run_id, queue, %DateTime{} = now) do
    case recover(storage, run_id, queue, now, @dispatch_schedule_retries) do
      {:ok, recovery} ->
        {:ok, recovery}

      {:error, :conflict} ->
        with {:ok, workflow_agent} <- WorkflowAgent.rebuild(storage, run_id),
             {:ok, dispatch_agent} <- DispatchAgent.rebuild(storage, queue) do
          {:ok,
           %{
             workflow_agent: workflow_agent,
             dispatch_agent: dispatch_agent,
             scheduled_runnables: [],
             applied_attempts: []
           }}
        end

      {:error, _reason} = error ->
        error
    end
  end

  defp recover(storage, run_id, queue, now, retries_left) do
    case AgentRecovery.recover(storage, run_id, queue, now: now) do
      {:ok, recovery} ->
        {:ok, recovery}

      {:error, :conflict} when retries_left > 0 ->
        recover(storage, run_id, queue, now, retries_left - 1)

      {:error, _reason} = error ->
        error
    end
  end

  defp journal_storage(opts) do
    opts
    |> Keyword.get(:journal_storage)
    |> Options.storage()
  end

  defp queue(opts) do
    opts
    |> Keyword.get(:queue, "default")
    |> Options.queue()
  end

  defp now(opts) do
    case Keyword.get(opts, :now, DateTime.utc_now()) do
      %DateTime{} = now -> {:ok, now}
      _invalid -> {:error, {:invalid_option, {:now, :invalid}}}
    end
  end

  defp run_id(opts) do
    case Keyword.get(opts, :run_id, Ecto.UUID.generate()) do
      run_id -> Options.uuid(run_id)
    end
  end

  defp initial_context(opts) do
    opts
    |> Keyword.get(:initial_context, %{})
    |> pick_reserved_context()
  end

  defp pick_reserved_context(context) when is_map(context) do
    context
    |> Map.take([:schedule, "schedule"])
    |> normalize_schedule_context()
  end

  defp pick_reserved_context(_context), do: %{}

  defp normalize_schedule_context(%{schedule: nil}), do: %{}

  defp normalize_schedule_context(%{schedule: schedule}) do
    %{schedule: schedule}
  end

  defp normalize_schedule_context(%{"schedule" => nil}), do: %{}

  defp normalize_schedule_context(%{"schedule" => schedule}) do
    %{schedule: schedule}
  end

  defp normalize_schedule_context(_context), do: %{}

  defp replayed_from_run_id(opts) do
    case Keyword.get(opts, :replayed_from_run_id) do
      nil -> nil
      run_id -> run_id
    end
  end

  defp runnable_value(runnable, key) when is_map(runnable) and is_atom(key) do
    Map.get(runnable, key) || Map.get(runnable, Atom.to_string(key))
  end
end