cmdc_gateway

defmodule CMDCGateway.WorkflowReplay do
  @moduledoc """
  Workflow RunEvent 只读 replay 接口。

  Gateway 不拥有 RunStore,也不启动 workflow executor。事件来源必须由宿主
  应用通过 `config :cmdc_gateway, :workflow_event_source, ...` 显式配置。
  """

  alias CMDCGateway.WorkflowEventAdapter

  @default_limit 50
  @max_limit 500

  @type replay_result :: %{
          runId: String.t(),
          events: [map()],
          page: map()
        }

  @doc """
  从服务端配置的 event source 读取 run event ledger 并投影为 Gateway 事件。
  """
  @spec replay(String.t(), map() | keyword()) :: {:ok, replay_result()} | {:error, term()}
  def replay(run_id, params \\ %{}) when is_binary(run_id) do
    opts = replay_opts(params)

    with {:ok, source} <- event_source(),
         {:ok, events} <- list_events(source, run_id, opts) do
      translated =
        events
        |> Enum.map(&WorkflowEventAdapter.to_gateway_event(&1, run_id: run_id))
        |> Enum.flat_map(fn
          {:ok, event} -> [event]
          :skip -> []
        end)

      {:ok,
       %{
         runId: run_id,
         events: translated,
         page: page_info(translated, opts)
       }}
    end
  end

  @doc "把 HTTP query params 转为 event source 支持的 replay opts。"
  @spec replay_opts(map() | keyword()) :: keyword()
  def replay_opts(params) do
    []
    |> put_opt(:limit, bounded_limit(value(params, "limit", :limit)))
    |> put_opt(
      :after_id,
      value(params, "afterId", :after_id) || value(params, "after_id", :after_id)
    )
    |> put_opt(
      :after_seq,
      non_neg_integer(
        value(params, "afterSeq", :after_seq) || value(params, "after_seq", :after_seq)
      )
    )
    |> put_opt(:type, value(params, "type", :type))
    |> put_opt(:node_id, value(params, "nodeId", :node_id) || value(params, "node_id", :node_id))
  end

  # ==========================================================================
  # Event source
  # ==========================================================================

  defp event_source do
    case Application.get_env(:cmdc_gateway, :workflow_event_source) do
      nil -> {:error, :not_configured}
      source -> {:ok, source}
    end
  end

  defp list_events(module, run_id, opts) when is_atom(module) do
    cond do
      Code.ensure_loaded?(module) and function_exported?(module, :list_events, 2) ->
        module.list_events(run_id, opts)

      Code.ensure_loaded?(module) and function_exported?(module, :events, 2) ->
        module.events(run_id, opts)

      true ->
        {:error, {:invalid_event_source, module}}
    end
  end

  defp list_events({module, :list_events}, run_id, opts), do: list_events(module, run_id, opts)
  defp list_events({module, :events}, run_id, opts), do: list_events(module, run_id, opts)

  defp list_events({module, function, static_opts}, run_id, opts)
       when function in [:list_events, :events] and is_list(static_opts) do
    list_events({module, function}, run_id, Keyword.merge(static_opts, opts))
  end

  defp list_events(_source, _run_id, _opts), do: {:error, :invalid_event_source}

  # ==========================================================================
  # Helpers
  # ==========================================================================

  defp page_info(events, opts) do
    last = List.last(events)

    %{
      limit: Keyword.fetch!(opts, :limit),
      count: length(events),
      nextCursor: last && Map.get(last, :event_id),
      nextSeq: last && seq_from_event_id(Map.get(last, :event_id))
    }
    |> drop_empty_values()
  end

  defp put_opt(opts, _key, nil), do: opts
  defp put_opt(opts, key, value), do: Keyword.put(opts, key, value)

  defp value(params, camel, snake) when is_map(params) do
    Map.get(params, camel) || Map.get(params, Atom.to_string(snake)) || Map.get(params, snake)
  end

  defp value(params, _camel, snake) when is_list(params), do: Keyword.get(params, snake)

  defp bounded_limit(nil), do: @default_limit

  defp bounded_limit(value) do
    value
    |> positive_integer()
    |> case do
      nil -> @default_limit
      int -> min(int, @max_limit)
    end
  end

  defp positive_integer(value) when is_integer(value) and value > 0, do: value

  defp positive_integer(value) when is_binary(value) do
    case Integer.parse(value) do
      {int, ""} when int > 0 -> int
      _ -> nil
    end
  end

  defp positive_integer(_value), do: nil

  defp non_neg_integer(nil), do: nil
  defp non_neg_integer(value) when is_integer(value) and value >= 0, do: value

  defp non_neg_integer(value) when is_binary(value) do
    case Integer.parse(value) do
      {int, ""} when int >= 0 -> int
      _ -> nil
    end
  end

  defp non_neg_integer(_value), do: nil

  defp seq_from_event_id(nil), do: nil
  defp seq_from_event_id(seq) when is_integer(seq), do: seq

  defp seq_from_event_id(event_id) when is_binary(event_id) do
    case Regex.run(~r/:evt:(\d+)$/, event_id) do
      [_, seq] -> String.to_integer(seq)
      _ -> nil
    end
  end

  defp drop_empty_values(%{} = map) do
    Map.reject(map, fn
      {_key, nil} -> true
      {_key, ""} -> true
      {_key, []} -> true
      {_key, value} when is_map(value) -> map_size(value) == 0
      _ -> false
    end)
  end
end