gen_consumer

defmodule GenConsumer.RedisConsumer do
  @moduledoc """
  Consumer based on Redis streams
  """
  use GenConsumer.Adapter

  defmodule State do
    @moduledoc false
    defstruct [:topics, :group, :consumer, :callback, :conn]
  end

  @impl true
  def init(opts) do
    topics = Keyword.fetch!(opts, :topics)
    group = Keyword.fetch!(opts, :group)
    consumer = Keyword.fetch!(opts, :consumer)
    callback = Keyword.fetch!(opts, :callback)

    host = Keyword.get(opts, :host, "localhost")
    port = Keyword.get(opts, :port, 6379)
    {:ok, conn} = Redix.start_link(host: host, port: port)

    state =
      %State{
        topics: topics,
        group: group,
        consumer: consumer,
        callback: callback,
        conn: conn
      }

    create_group_if_not_existant(state)
    Process.send_after self(), :block, 1000
    {:ok, state}
  end

  def handle_info(:block, state) do
    streams = Enum.join(state.topics, ", ")
    command =
      if state.group do
        ["XREADGROUP", "GROUP", state.group, state.consumer, "BLOCK", 0, "COUNT", 10, "STREAMS", streams, ">"]
      else
        ["XREAD", "BLOCK", 0, "COUNT", 10, "STREAMS", streams, state.last_id]
      end

    case Redix.command(state.conn, command, timeout: :infinity) do
      {:ok, [
        [
          _stream,
          events
        ]
      ]} ->
        events =
          for [id, ["value", value]] <- events,
              {:ok, event} = Poison.decode(value) do
            {id, event}
          end
        {id, _} = List.last(events)
        Process.send_after self(), :block, 0
        state = %{state | last_id: id}
        events = for {id, event} <- events, do: state.callback.(event)
        {:noreply, state, :infinity}
      _ ->
        Process.send_after self(), :block, 0
        {:noreply, state, :infinity}
    end
  end

  defp create_group_if_not_existant(state) do
    if state.group do
      for stream <- state.topics do
        try do
          command = ["XINFO", "GROUPS", stream]
          exists? =
            Redix.command!(state.conn, command)
            |> Enum.any?(fn [_, existing, _, _, _, _, _, _] -> existing == state.group end)
          command = ["XGROUP", "CREATE", stream, state.group, "$"]
          !exists? && Redix.command!(state.conn, command)
        rescue
          exception ->
            nil
        end
      end
    end
  end

end