agent_session_manager

# credo:disable-for-this-file Credo.Check.Readability.PreferImplicitTry
defmodule ASM.Extensions.Rendering do
  @moduledoc """
  Public rendering extension API.

  This domain consumes `%ASM.Event{}` streams and dispatches renderer output
  to one or more sinks, keeping output concerns outside the run loop.
  """

  use Boundary,
    deps: [ASM],
    exports: [
      Renderer,
      Sink,
      Renderers.Compact,
      Renderers.Verbose,
      Sinks.TTY,
      Sinks.File,
      Sinks.JSONL,
      Sinks.Callback
    ]

  alias ASM.Error
  alias ASM.Extensions.Rendering.Renderers
  alias ASM.Extensions.Rendering.Sinks

  @type renderer_spec :: {module(), keyword()}
  @type sink_spec :: {module(), keyword()}

  @spec compact_renderer(keyword()) :: renderer_spec()
  def compact_renderer(opts \\ []) when is_list(opts), do: {Renderers.Compact, opts}

  @spec verbose_renderer(keyword()) :: renderer_spec()
  def verbose_renderer(opts \\ []) when is_list(opts), do: {Renderers.Verbose, opts}

  @spec tty_sink(keyword()) :: sink_spec()
  def tty_sink(opts \\ []) when is_list(opts), do: {Sinks.TTY, opts}

  @spec file_sink(keyword()) :: sink_spec()
  def file_sink(opts) when is_list(opts), do: {Sinks.File, opts}

  @spec jsonl_sink(keyword()) :: sink_spec()
  def jsonl_sink(opts) when is_list(opts), do: {Sinks.JSONL, opts}

  @spec callback_sink(keyword()) :: sink_spec()
  def callback_sink(opts \\ []) when is_list(opts), do: {Sinks.Callback, opts}

  @spec stream(Enumerable.t(), keyword()) :: :ok | {:error, Error.t()}
  def stream(event_stream, opts \\ []) when is_list(opts) do
    with {:ok, {renderer_mod, renderer_opts}} <-
           normalize_renderer_spec(Keyword.get(opts, :renderer)),
         {:ok, sink_specs} <- normalize_sink_specs(Keyword.get(opts, :sinks, [])),
         {:ok, renderer_state} <- init_renderer(renderer_mod, renderer_opts),
         {:ok, sink_states} <- init_sinks(sink_specs) do
      run_stream(event_stream, renderer_mod, renderer_state, sink_states)
    else
      {:error, %Error{} = error} ->
        {:error, error}
    end
  end

  defp run_stream(event_stream, renderer_mod, renderer_state, sink_states) do
    case process_events(event_stream, renderer_mod, renderer_state, sink_states) do
      {:ok, renderer_state, sink_states} ->
        finish_stream(renderer_mod, renderer_state, sink_states)

      {:error, %Error{} = error, _renderer_state, sink_states} ->
        finalize_sinks(sink_states, error)
    end
  end

  defp process_events(event_stream, renderer_mod, renderer_state, sink_states) do
    Enum.reduce_while(event_stream, {:ok, renderer_state, sink_states}, fn event,
                                                                           {:ok, acc_renderer,
                                                                            acc_sinks} ->
      case render_and_dispatch_event(renderer_mod, event, acc_renderer, acc_sinks) do
        {:ok, next_renderer, next_sinks} ->
          {:cont, {:ok, next_renderer, next_sinks}}

        {:error, %Error{} = error, next_renderer, next_sinks} ->
          {:halt, {:error, error, next_renderer, next_sinks}}
      end
    end)
  rescue
    error ->
      {:error, runtime_error("rendering stream crashed", error), renderer_state, sink_states}
  catch
    kind, reason ->
      {:error, runtime_error("rendering stream crashed", %{kind: kind, reason: reason}),
       renderer_state, sink_states}
  end

  defp render_and_dispatch_event(renderer_mod, event, renderer_state, sink_states) do
    with {:ok, iodata, next_renderer_state} <-
           call_renderer_render_event(renderer_mod, event, renderer_state),
         {:ok, next_sink_states} <- write_event_to_sinks(sink_states, event, iodata) do
      {:ok, next_renderer_state, next_sink_states}
    else
      {:error, %Error{} = error} ->
        {:error, error, renderer_state, sink_states}

      {:error, %Error{} = error, next_sink_states} ->
        {:error, error, renderer_state, next_sink_states}
    end
  end

  defp finish_stream(renderer_mod, renderer_state, sink_states) do
    with {:ok, finish_iodata, _state} <- call_renderer_finish(renderer_mod, renderer_state),
         {:ok, sink_states} <- write_finish_to_sinks(sink_states, finish_iodata) do
      finalize_sinks(sink_states, nil)
    else
      {:error, %Error{} = error} ->
        finalize_sinks(sink_states, error)

      {:error, %Error{} = error, next_sink_states} ->
        finalize_sinks(next_sink_states, error)
    end
  end

  defp init_renderer(renderer_mod, renderer_opts) do
    try do
      case renderer_mod.init(renderer_opts) do
        {:ok, state} ->
          {:ok, state}

        {:error, %Error{} = error} ->
          {:error, error}

        {:error, reason} ->
          {:error, runtime_error("renderer init failed", reason)}

        other ->
          {:error, invalid_response_error("renderer init", renderer_mod, other)}
      end
    rescue
      error ->
        {:error, runtime_error("renderer init failed", error)}
    catch
      kind, reason ->
        {:error, runtime_error("renderer init failed", %{kind: kind, reason: reason})}
    end
  end

  defp init_sinks(sink_specs), do: init_sinks(sink_specs, [])

  defp init_sinks([], acc), do: {:ok, Enum.reverse(acc)}

  defp init_sinks([{sink_mod, sink_opts} | rest], acc) do
    try do
      case sink_mod.init(sink_opts) do
        {:ok, state} ->
          init_sinks(rest, [{sink_mod, state} | acc])

        {:error, %Error{} = error} ->
          _ = close_sinks_best_effort(acc)
          {:error, error}

        {:error, reason} ->
          _ = close_sinks_best_effort(acc)
          {:error, runtime_error("sink init failed", reason)}

        other ->
          _ = close_sinks_best_effort(acc)
          {:error, invalid_response_error("sink init", sink_mod, other)}
      end
    rescue
      error ->
        _ = close_sinks_best_effort(acc)
        {:error, runtime_error("sink init failed", error)}
    catch
      kind, reason ->
        _ = close_sinks_best_effort(acc)
        {:error, runtime_error("sink init failed", %{kind: kind, reason: reason})}
    end
  end

  defp write_event_to_sinks(sink_states, event, iodata),
    do: write_event_to_sinks(sink_states, event, iodata, [])

  defp write_event_to_sinks([], _event, _iodata, acc), do: {:ok, Enum.reverse(acc)}

  defp write_event_to_sinks([{sink_mod, sink_state} | rest], event, iodata, acc) do
    case call_sink_write_event(sink_mod, event, iodata, sink_state) do
      {:ok, next_sink_state} ->
        write_event_to_sinks(rest, event, iodata, [{sink_mod, next_sink_state} | acc])

      {:error, %Error{} = error, next_sink_state} ->
        {:error, error, Enum.reverse([{sink_mod, next_sink_state} | acc]) ++ rest}
    end
  end

  defp write_finish_to_sinks(sink_states, finish_iodata),
    do: write_finish_to_sinks(sink_states, finish_iodata, [])

  defp write_finish_to_sinks([], _finish_iodata, acc), do: {:ok, Enum.reverse(acc)}

  defp write_finish_to_sinks([{sink_mod, sink_state} | rest], finish_iodata, acc) do
    case call_sink_write(sink_mod, finish_iodata, sink_state) do
      {:ok, next_sink_state} ->
        write_finish_to_sinks(rest, finish_iodata, [{sink_mod, next_sink_state} | acc])

      {:error, %Error{} = error, next_sink_state} ->
        {:error, error, Enum.reverse([{sink_mod, next_sink_state} | acc]) ++ rest}
    end
  end

  defp finalize_sinks(sink_states, primary_error) do
    {sink_states, flush_error} = flush_sinks_best_effort(sink_states)
    close_error = close_sinks_best_effort(sink_states)
    error = first_error(primary_error, flush_error, close_error)

    case error do
      nil -> :ok
      %Error{} = value -> {:error, value}
    end
  end

  defp flush_sinks_best_effort(sink_states) do
    Enum.reduce(sink_states, {[], nil}, fn {sink_mod, sink_state}, {acc, first_error} ->
      case call_sink_flush(sink_mod, sink_state) do
        {:ok, next_sink_state} ->
          {[{sink_mod, next_sink_state} | acc], first_error}

        {:error, %Error{} = error, next_sink_state} ->
          {[{sink_mod, next_sink_state} | acc], first_error || error}
      end
    end)
    |> then(fn {reversed_states, first_error} -> {Enum.reverse(reversed_states), first_error} end)
  end

  defp close_sinks_best_effort(sink_states) do
    sink_states
    |> Enum.reverse()
    |> Enum.reduce(nil, fn {sink_mod, sink_state}, first_error ->
      case call_sink_close(sink_mod, sink_state) do
        :ok -> first_error
        {:error, %Error{} = error} -> first_error || error
      end
    end)
  end

  defp call_renderer_render_event(renderer_mod, event, renderer_state) do
    try do
      case renderer_mod.render_event(event, renderer_state) do
        {:ok, iodata, next_state} ->
          {:ok, iodata, next_state}

        {:error, %Error{} = error} ->
          {:error, error}

        {:error, reason} ->
          {:error, runtime_error("renderer render_event failed", reason)}

        other ->
          {:error, invalid_response_error("renderer render_event", renderer_mod, other)}
      end
    rescue
      error -> {:error, runtime_error("renderer render_event failed", error)}
    catch
      kind, reason ->
        {:error, runtime_error("renderer render_event failed", %{kind: kind, reason: reason})}
    end
  end

  defp call_renderer_finish(renderer_mod, renderer_state) do
    try do
      case renderer_mod.finish(renderer_state) do
        {:ok, iodata, next_state} ->
          {:ok, iodata, next_state}

        {:error, %Error{} = error} ->
          {:error, error}

        {:error, reason} ->
          {:error, runtime_error("renderer finish failed", reason)}

        other ->
          {:error, invalid_response_error("renderer finish", renderer_mod, other)}
      end
    rescue
      error -> {:error, runtime_error("renderer finish failed", error)}
    catch
      kind, reason ->
        {:error, runtime_error("renderer finish failed", %{kind: kind, reason: reason})}
    end
  end

  defp call_sink_write_event(sink_mod, event, iodata, sink_state) do
    try do
      case sink_mod.write_event(event, iodata, sink_state) do
        {:ok, next_state} ->
          {:ok, next_state}

        {:error, %Error{} = error, next_state} ->
          {:error, error, next_state}

        {:error, %Error{} = error} ->
          {:error, error, sink_state}

        {:error, reason, next_state} ->
          {:error, runtime_error("sink write_event failed", reason), next_state}

        {:error, reason} ->
          {:error, runtime_error("sink write_event failed", reason), sink_state}

        other ->
          {:error, invalid_response_error("sink write_event", sink_mod, other), sink_state}
      end
    rescue
      error ->
        {:error, runtime_error("sink write_event failed", error), sink_state}
    catch
      kind, reason ->
        {:error, runtime_error("sink write_event failed", %{kind: kind, reason: reason}),
         sink_state}
    end
  end

  defp call_sink_write(sink_mod, iodata, sink_state) do
    try do
      case sink_mod.write(iodata, sink_state) do
        {:ok, next_state} ->
          {:ok, next_state}

        {:error, %Error{} = error, next_state} ->
          {:error, error, next_state}

        {:error, %Error{} = error} ->
          {:error, error, sink_state}

        {:error, reason, next_state} ->
          {:error, runtime_error("sink write failed", reason), next_state}

        {:error, reason} ->
          {:error, runtime_error("sink write failed", reason), sink_state}

        other ->
          {:error, invalid_response_error("sink write", sink_mod, other), sink_state}
      end
    rescue
      error ->
        {:error, runtime_error("sink write failed", error), sink_state}
    catch
      kind, reason ->
        {:error, runtime_error("sink write failed", %{kind: kind, reason: reason}), sink_state}
    end
  end

  defp call_sink_flush(sink_mod, sink_state) do
    try do
      case sink_mod.flush(sink_state) do
        {:ok, next_state} ->
          {:ok, next_state}

        {:error, %Error{} = error, next_state} ->
          {:error, error, next_state}

        {:error, %Error{} = error} ->
          {:error, error, sink_state}

        {:error, reason, next_state} ->
          {:error, runtime_error("sink flush failed", reason), next_state}

        {:error, reason} ->
          {:error, runtime_error("sink flush failed", reason), sink_state}

        other ->
          {:error, invalid_response_error("sink flush", sink_mod, other), sink_state}
      end
    rescue
      error ->
        {:error, runtime_error("sink flush failed", error), sink_state}
    catch
      kind, reason ->
        {:error, runtime_error("sink flush failed", %{kind: kind, reason: reason}), sink_state}
    end
  end

  defp call_sink_close(sink_mod, sink_state) do
    try do
      case sink_mod.close(sink_state) do
        :ok ->
          :ok

        {:error, %Error{} = error} ->
          {:error, error}

        {:error, reason} ->
          {:error, runtime_error("sink close failed", reason)}

        other ->
          {:error, invalid_response_error("sink close", sink_mod, other)}
      end
    rescue
      error ->
        {:error, runtime_error("sink close failed", error)}
    catch
      kind, reason ->
        {:error, runtime_error("sink close failed", %{kind: kind, reason: reason})}
    end
  end

  defp normalize_renderer_spec(nil), do: {:ok, compact_renderer()}

  defp normalize_renderer_spec({renderer_mod, renderer_opts})
       when is_atom(renderer_mod) and is_list(renderer_opts) do
    with :ok <- ensure_loaded(renderer_mod),
         :ok <- ensure_export(renderer_mod, :init, 1, :renderer),
         :ok <- ensure_export(renderer_mod, :render_event, 2, :renderer),
         :ok <- ensure_export(renderer_mod, :finish, 1, :renderer) do
      {:ok, {renderer_mod, renderer_opts}}
    end
  end

  defp normalize_renderer_spec(other) do
    {:error, config_error("renderer must be {module, keyword}, got: #{inspect(other)}")}
  end

  defp normalize_sink_specs(sink_specs) when is_list(sink_specs) do
    sink_specs
    |> Enum.reduce_while({:ok, []}, fn
      {sink_mod, sink_opts}, {:ok, acc} when is_atom(sink_mod) and is_list(sink_opts) ->
        with :ok <- ensure_loaded(sink_mod),
             :ok <- ensure_export(sink_mod, :init, 1, :sink),
             :ok <- ensure_export(sink_mod, :write, 2, :sink),
             :ok <- ensure_export(sink_mod, :write_event, 3, :sink),
             :ok <- ensure_export(sink_mod, :flush, 1, :sink),
             :ok <- ensure_export(sink_mod, :close, 1, :sink) do
          {:cont, {:ok, [{sink_mod, sink_opts} | acc]}}
        else
          {:error, %Error{} = error} -> {:halt, {:error, error}}
        end

      invalid, _acc ->
        {:halt,
         {:error, config_error("sink must be {module, keyword}, got: #{inspect(invalid)}")}}
    end)
    |> case do
      {:ok, reversed_specs} -> {:ok, Enum.reverse(reversed_specs)}
      {:error, %Error{} = error} -> {:error, error}
    end
  end

  defp normalize_sink_specs(other) do
    {:error, config_error("sinks must be a list, got: #{inspect(other)}")}
  end

  defp ensure_loaded(module) do
    if Code.ensure_loaded?(module) do
      :ok
    else
      {:error, config_error("module is not available: #{inspect(module)}")}
    end
  end

  defp ensure_export(module, function, arity, kind) do
    if function_exported?(module, function, arity) do
      :ok
    else
      {:error, config_error("#{kind} module #{inspect(module)} must export #{function}/#{arity}")}
    end
  end

  defp first_error(nil, nil, nil), do: nil
  defp first_error(%Error{} = error, _other, _third), do: error
  defp first_error(nil, %Error{} = error, _third), do: error
  defp first_error(nil, nil, %Error{} = error), do: error

  defp invalid_response_error(operation, module, response) do
    config_error(
      "#{operation} returned invalid response from #{inspect(module)}: #{inspect(response)}"
    )
  end

  defp config_error(message) do
    Error.new(:config_invalid, :config, message)
  end

  defp runtime_error(message, cause) do
    Error.new(:unknown, :runtime, message, cause: cause)
  end
end