defmodule Wiki.EventStreams do
  @moduledoc """
  This module reads from the real-time stream of [server-sent events](https://en.wikipedia.org/wiki/Server-sent_events)
  annotating actions such as editing or patrolling as they happen across Wikimedia projects.

  For more about the public wiki streams and their format, see
  [EventStreams on Wikitech](https://wikitech.wikimedia.org/wiki/EventStreams)

  ## Examples

  Listen for page creation, expose as a GenStage.stream and print a few:

  ```
  {:ok, pid} = Wiki.EventStreams.start_link(streams: "page-create")
  Wiki.EventStreams.stream(pid) |> Stream.take(1) |> Enum.to_list
  ```

  Combine multiple feeds as one stream,

  ```
  {:ok, pid} = Wiki.EventStreams.start_link(streams: ["revision-create", "page-create"])
  Wiki.EventStreams.stream(pid)
  |> Stream.take(6)
  |> Enum.to_list
  ```

  ## TODO

  * Track the restart ID, disconnect from the feed at some maximum queue size.  Reconnect as demand resumes.
  Application-lifetime or permanent storage for the restart ID tracking, for consumers that need an at-least-once
  guarantee.
  """

  defmodule Relay do
    @moduledoc false

    use GenStage

    @type state :: {:queue.queue(), integer}

    @type reply :: {:noreply, [map], state}

    @doc """
    ## Arguments

    * Client configuration.  `:streams` is required.
    """
    @spec start_link(Wiki.EventStreams.client_options()) :: GenServer.on_start()
    def start_link(opts) do
      {:ok, relay_pid} = GenStage.start_link(__MODULE__, opts)
      # FIXME: Rationalize the supervision tree.
      source_opts = Keyword.put_new(opts, :stream_to, relay_pid)

      {:ok, _} =
        Supervisor.start_link([{Wiki.EventStreams.Source, source_opts}], strategy: :one_for_one)

      {:ok, relay_pid}
    end

    @impl true
    @spec init(any) :: {:producer, state}
    def init(_) do
      {:producer, {:queue.new(), 0}}
    end

    @impl true
    @spec handle_info(EventsourceEx.Message.t(), state) :: reply
    def handle_info(message, {queue, pending_demand}) do
      message
      |> decode_message_data()
      |> :queue.in(queue)
      # FIXME: Suppress reply until above min_demand or periodic timeout has elapsed.
      |> dispatch_events(pending_demand)
    end

    @doc """
    Return the requested number of events

    ## Arguments

    - `demand` - Number of new events requested by the consumer.
    - `state` - queue and pending demand
    """
    @impl true
    @spec handle_demand(pos_integer, state) :: reply
    def handle_demand(demand, {queue, pending_demand}) do
      dispatch_events(queue, demand + pending_demand)
    end

    @spec dispatch_events(:queue.queue(), integer) :: reply
    defp dispatch_events(queue, demand) do
      available = min(demand, :queue.len(queue))
      {retrieved, queue1} = :queue.split(available, queue)
      events = :queue.to_list(retrieved) |> Enum.reverse()
      {:noreply, events, {queue1, demand - available}}
    end

    @spec decode_message_data(EventsourceEx.Message.t()) :: map
    defp decode_message_data(message) do
      message.data
      |> Jason.decode!()
    end
  end

  defmodule Source do
    @moduledoc false

    alias Wiki.{EventStreams, Util}

    @default_endpoint "https://stream.wikimedia.org"

    @doc false
    @spec child_spec(EventStreams.client_options()) :: map
    def child_spec(opts \\ []) do
      adapter = opts[:adapter]
      endpoint = (opts[:endpoint] || @default_endpoint) <> "/v2/stream/"
      sink = opts[:stream_to]
      user_agent = opts[:user_agent] || Util.default_user_agent()
      url = endpoint <> normalize_streams(opts[:streams])

      %{
        id: Source,
        # FIXME: nicer if we could get the Relay sibling's specific PID each time,
        #  to allow an app to use multiple stream listeners.
        start: {
          EventsourceEx,
          :new,
          [
            url,
            [
              adapter: adapter,
              headers: [{"user-agent", user_agent}],
              stream_to: sink
            ]
          ]
        }
      }
    end

    @spec normalize_streams(atom | [atom]) :: atom | String.t()
    defp normalize_streams(streams)

    defp normalize_streams(streams) when is_list(streams), do: Enum.join(streams, ",")

    defp normalize_streams(streams) when is_atom(streams) or is_binary(streams), do: streams

    @doc """
    Convenience for sharing the default across submodules.
    """
    @spec get_default_endpoint() :: binary()
    def get_default_endpoint,
      do: @default_endpoint
  end

  @type client_option ::
          {:adapter, module()}
          | {:endpoint, binary()}
          | {:stream_to, GenServer.server()}
          | {:streams, atom | [atom]}
          | {:user_agent, binary()}
  @typedoc """
    * `:adapter` - Override the HTTP adapter.
    * `:endpoint` - Override the default endpoint URL.
    * `:stream_to` - Optional application which will receive the events, otherwise
      they go to the process starting the EventStream.
    * `:streams` - One or more atoms with the stream names to subscribe to.
    * `:user_agent` - Custom user-agent header string
  """
  @type client_options :: [client_option()]

  @doc """
  Start a supervisor tree to receive and relay server-side events.

  ## Arguments

  - `options` - Keyword list,
    - `{:adapter, module}` - Override eventsource_ex default adapter.
    - `{:endpoint, url}` - Override default endpoint.
    - `{:send_to, pid | module}` - Instead of using the built-in streaming relay,
    send the events directly to your own process.
    - `{:streams, atom | [atom]}` - Select which streams to listen to.  An updated list can be
    [found here](https://stream.wikimedia.org/?doc#/Streams).  Required.
  """
  @spec start_link(client_options()) :: GenServer.on_start()
  def start_link(args),
    do: Relay.start_link(args)

  @doc false
  # TODO: since Elixir 1.16 return turn is `Supervisor.module_spec()`
  @spec child_spec(client_options()) :: any()
  def child_spec(args),
    do: {Relay, args}

  @doc """
  Indefinitely capture subscribed events and relay them as a `Stream`.
  """
  @spec stream(pid(), client_options()) :: Enumerable.t()
  def stream(pid, options \\ []),
    do: GenStage.stream([pid], options)

  @doc """
  Return a list of currently available streams.

  Note that the matching is intentionally fragile, on the assumption that the
  caller is interested in uncovering new routes (eg. a future "v3" or new route
  patterns) immediately.

  ## Arguments

  - `options` - Keyword list,
    - `{:adapter, module}` - Override eventsource_ex default adapter.
    - `{:endpoint, url}` - Override default endpoint.
  """
  @spec available_streams(keyword()) :: [{name :: binary(), description :: binary()}]
  def available_streams(opts \\ []) do
    adapter = opts[:adapter]
    endpoint = (opts[:endpoint] || Source.get_default_endpoint()) <> "/?spec"

    {:ok, %{status_code: 200, body: body}} = adapter.get(endpoint)

    Jason.decode!(body)
    |> Map.fetch!("paths")
    |> Map.to_list()
    |> Enum.map(fn {"/v2/stream/" <> name, %{"get" => %{"description" => doc}}} ->
      {name, doc}
    end)
  end
end