gen_consumer

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

  defmodule State do
    @moduledoc false
    defstruct [:topics, :group, :consumer, :callback, :conn, :last_ids, :max_batch_size]
  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)

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

    last_ids = for _topic <- topics, do: nil
    state =
      %State{
        max_batch_size: max_batch_size,
        topics: topics,
        group: group,
        consumer: consumer,
        callback: callback,
        conn: conn,
        last_ids: last_ids
      }

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

  def handle_info(:block, state) do
    streams = state.topics
    command =
      if state.group do
        ids = for _stream <- streams, do: ">"
        (["XREADGROUP", "GROUP", state.group, state.consumer, "BLOCK", 1000, "COUNT", state.max_batch_size, "STREAMS"] ++ streams ++ ids)
      else
        (["XREAD", "BLOCK", 1000, "COUNT", state.max_batch_size, "STREAMS"] ++ streams ++ state.last_ids)
      end

    case Redix.command(state.conn, command, timeout: 60_000) do
      {:ok, matched_streams} when is_list(matched_streams) ->
        last_ids =
          for [_stream, events] <- matched_streams do
            events =
              for [id, ["value", value]] <- events do
                {id, value}
              end
            for {_id, event} <- events, do: state.callback.(event)
            {last_id, _} = List.last(events)
            last_id
          end
        Process.send_after self(), :block, 1000
        state = %{state | last_ids: last_ids}
        {:noreply, state, 60_000}
      {:error, error} ->
        Logger.error("Redix command failed with exception: #{inspect error}")
        Process.send_after self(), :block, 1000
        {:noreply, state, 60_000}
      _ ->
        Process.send_after self(), :block, 1000
        {:noreply, state, 60_000}
    end
  rescue
    exception ->
      Logger.error(inspect(exception))
      {:stop, :shutdown, state}
  end

  defp create_group_if_not_existant(state) do
    # Logger.info "Ensuring group '#{state.group}' exists..."
    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)
          #   |> IO.inspect
          # if !exists? do
          # Logger.info "Group '#{state.group}' does not exist. Creating group..."
          command = ["XGROUP", "CREATE", stream, state.group, "$", "MKSTREAM"]
          result =
            try do
              Redix.command(state.conn, command)
            rescue
              exception ->
                {:error, exception}
            end
          case result do
            {:error, %Redix.Error{message: "BUSYGROUP Consumer Group name already exists"}} ->
              :ok
            _ ->
              :ok
          end
          # case Redix.command(state.conn, command) do
          #   {:error, %Redix.Error{message: "ERR no such key"}} ->
          #     Logger.info "Stream '#{stream}' does not exist. Creating stream...."
          #     {:ok, _} = Redix.command(state.conn, ["XADD", stream, "*", "", ""])
          #     :ok
          #   {:error, %Redix.Error{message: "BUSYGROUP Consumer Group name already exists"}} ->
          #     :ok
          #   {:error, error} ->
          #     raise error
          #   {:ok, _} ->
          #     :ok
          #   # end
          # end
        # rescue
        #   _exception ->
        #     nil
        # end
      end
    end
  end

end