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

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

  ## Examples

  Start reading the page creation feed, and expose as a GenStage.stream:

  ```elixir
  Wiki.EventStreams.start_link(streams: "page-create")
  Wiki.EventStreams.stream()
  |> Stream.take(6)
  |> Enum.to_list
  ```

  Combine multiple feeds,

  ```elixir
  Wiki.EventStreams.start_link(streams: ["revision-create", "revision-score"])
  Wiki.EventStreams.stream()
  |> Stream.take(6)
  |> Enum.to_list
  ```

  ## TODO

  * Currently only a single supervisor tree is supported, so calling applications can only read from one stream.
  * 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, name: __MODULE__)
      # 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/v2/stream/"

    @doc false
    @spec child_spec(EventStreams.client_options()) :: map
    def child_spec(opts \\ []) do
      adapter = opts[:adapter]
      endpoint = opts[:endpoint] || @default_endpoint
      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
  end

  defmodule RelaySupervisor do
    @moduledoc false

    # TODO: Should this logic be moved to init/1?
    @doc """
    ## Arguments

    * `opts` - Client keyword options
    """
    @spec start_link(Wiki.EventStreams.client_options()) :: GenServer.on_start()
    def start_link(opts) do
      {:ok, _} = Supervisor.start_link([{Relay, opts}], strategy: :one_for_one)
    end
  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()]

  # TODO: Is only run as a genserver from tests, so this should go away or be
  # replaced by a supervisor.
  use GenServer

  @impl true
  @spec init(term) :: {:ok, any}
  def init(_) do
    {:ok, {}}
  end

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

  ## Arguments

  - `options` - Keyword list,
    - `{:adapter, module}` - Override HTTPoison 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
    RelaySupervisor.start_link(args)
  end

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