kafka_ex_helpers

defmodule KafkaExHelpers do
  require Logger

  @moduledoc """
  KafkaExHelpers provides helper functions to simplify producing, consuming and
  interacting with kafka topics
  """

  @doc ~S"""
  validates a string as a uuid

  ## Examples
  ```
  iex> KafkaExHelpers.validate_uuid("33c5aa7e-3f35-47bc-883b-33ea0ace89f0")
  {:ok, "33c5aa7e-3f35-47bc-883b-33ea0ace89f0"}

  iex> KafkaExHelpers.validate_uuid("some_string")
  {:error, :invalid_uuid}

  ```
  """
  @spec validate_uuid(String.t) :: {:ok, String.t} | {:error, :invalid_uuid}
  def validate_uuid(uuid) do
    [
      {:uuid, valid_uuid},
      {:binary, _},
      {:type, :default},
      {:version, 4},
      {:variant, :rfc4122}
    ] = UUID.info!(uuid)
    {:ok, valid_uuid}
  rescue
    _ -> {:error, :invalid_uuid}
  end

  @doc ~S"""
  generate a utc standardized timestamp
  """
  @spec now() :: DateTime.t
  def now do
   DateTime.utc_now
  end

  @doc ~S"""
  formate a timestamp to a iso formated datetime

  Examples
  ```
  iex> timestamp = "2017-03-19T02:48:15.814147Z"
  iex> {:ok, time, _} = DateTime.from_iso8601(timestamp)
  iex> KafkaExHelpers.format_time(time)
  "2017-03-19T02:48:15.814147Z"
  ```
  """
  @spec format_time(DateTime.t) :: String.t
  def format_time(datetime) do
     DateTime.to_iso8601(datetime)
  end

  @doc ~S"""
  genreates an iso8601 now timestamp
  """
  def now_string do
    format_time(now())
  end

  @doc ~S"""
  creates a json payload from an incoming map
  """
  @spec encode_payload_request(map) :: String.t
  def encode_payload_request(normalized_payload) do
    Poison.encode!(normalized_payload)
  end

  @doc ~S"""
  selects a random value from a set of partions

  Examples
  ```
  iex> partitions = [1]
  iex> KafkaExHelpers.select_random_partition(partitions)
  1
  ```
  """
  @spec select_random_partition([integer]) :: integer
  def select_random_partition(partitions) do
    Enum.random(partitions)
  end

  @doc ~S"""
  Normalizes an outgoing payload

  Examples
  ```
  iex> payload = %{hello: :world}
  iex> time = "2017-03-19T02:48:15.814147Z"
  iex> uuid = "33c5aa7e-3f35-47bc-883b-33ea0ace89f0"
  iex> KafkaExHelpers.payload_normalizer(payload, time, uuid)
  %{meta: %{
    requested_at: "2017-03-19T02:48:15.814147Z",
    transaction_id: "33c5aa7e-3f35-47bc-883b-33ea0ace89f0"},
  request_body: %{hello: :world}}
  ```
  """
  @spec payload_normalizer(map, String.t, String.t) :: map
  def payload_normalizer(payload, timestamp, uuid) do
    %{
      meta: %{
        requested_at: timestamp,
        transaction_id: uuid
      },
      request_body: payload
    }
  end

  @doc ~S"""
  fetches uniq worker ids from a collection of kafka worker module names via
  kafka_meta/0
  """
  @spec fetch_handler_ids([reference]) :: [atom]
  def fetch_handler_ids(handlers) do
    handlers
      |> Enum.map(&fetch_handler_id/1)
      |> Enum.uniq
  end

  @doc ~S"""
  collects normalized payloads of uniq consumers, handlers and expected batch
  sizes
  """
  @spec fetch_worker_ids([reference]) :: atom | [atom]
  def fetch_worker_ids(topics) do
    topics
      |> Enum.map(
        fn(w) -> fetch_worker_payloads(
          w[:consume],
          w[:handler],
          w[:batch_size],
          w[:options]
        ) end)
      |> Enum.uniq()
  end

  @doc ~S"""
  starts a collection of consumer workers defaulting to KafkaEx.fetch/3 for
  the message handler
  """
  @spec start_consumers([String.t], fun) :: {:ok, [pid]}
  def start_consumers(topics, handler \\ &KafkaEx.fetch/3) do
    worker_ids = fetch_worker_ids(topics)
    pids = worker_ids
      |> Enum.map(fn(w) ->
        Logger.info(fn -> "Starting consumer for #{w.topic}" end)
        get_linksp(w[:partitions], w, handler)
      end)
    {:ok, pids}
  end

  @spec fetch_handler_id(module) :: atom
  defp fetch_handler_id(worker) do
     worker.kafka_meta[:worker_id]
  end

  @spec fetch_worker_payloads(module, fun, integer, map()) ::
    %{
      topic: String.t, partitions: list(any), handler: fun,
      batch_size: integer, worker_id: atom, options: map()
    }
  defp fetch_worker_payloads(consumer, handler, batch_size, opts) do
    %{
      topic: consumer.kafka_meta[:topic],
      partitions: consumer.kafka_meta[:partitions],
      handler: handler,
      batch_size: batch_size,
      worker_id: consumer.kafka_meta[:worker_id],
      options: opts
    }
  end

  @doc ~S"""
  Reindexes payload by provided key `:request_id` by default

  Examples
  ```
  iex> request_id = "33c5aa7e-3f35-47bc-883b-33ea0ace89f0"
  iex> payload = %{request_id: request_id, other_key: "foo"}
  iex> KafkaExHelpers.index_payload_by_request_id(payload)
  %{
    "33c5aa7e-3f35-47bc-883b-33ea0ace89f0": [
      %{request_id: "33c5aa7e-3f35-47bc-883b-33ea0ace89f0",
      other_key: "foo"}
    ]
  }
  ```
  """
  def index_payload_by_request_id(payload, key \\ :request_id) do
    {:ok, request_id} = extract_request_idp(payload, key)
    {:ok, valid_uuid} = validate_uuid(request_id)
    %{
      "#{valid_uuid}": [payload]
    }
  end

  @doc ~S"""
   encode_and_publish_message publishes messages using the defined format provided to work with the helper modules. it encodes a Map into a json string and persists it to kafka

   The default action for publishing a message is `&KafkaEx.produce/4`, `&MockProducer.produce/4` is used for demonstration only

  ## Examples

    ```
    iex> defmodule MockTopic do
    ...>   def kafka_meta do
    ...>     %{worker_id: :mock_handler, partitions: [0], topic: 'Mock.V1'}
    ...>   end
    ...> end
    ...> defmodule MockProducer do
    ...>   def produce(_topic, _partition, _payload, worker_name: _worker_id) do
    ...>     :ok
    ...>   end
    ...> end
    ...> payload = %{hello: "world"}
    ...> reference_id = payload.hello
    ...> KafkaExHelpers.encode_and_publish_message(payload, reference_id, MockTopic.kafka_meta, &MockProducer.produce/4)
    :ok
    ```
  """
  def encode_and_publish_message(
    payload,
    reference_id,
    reference_meta,
    producer \\ &KafkaEx.produce/4) do

    string_payload = Poison.encode!(payload)
    partition = select_partition_by_event_id(
      reference_id,
      reference_meta.partitions)
    producer.(
      reference_meta.topic,
      partition,
      string_payload,
      worker_name: reference_meta.worker_id)
  end

  @doc ~S"""
   `select_partition_by_event_id/2` deterministically selects a partition given the value of a provided string event_id. the string is converted to a char_list and the last value is selected.

  ## Examples

    if there is a single partition it will always be selected

    ```
    iex> KafkaExHelpers.select_partition_by_event_id("hello_world", [0])
    0
    ```

    a more complex string will always result in the same partition being selected

    ```
    iex> partitions = [0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18]
    ...> reference_id = "64c0cc33-b37a-4768-8463-4715b4dbadd5"
    ...> KafkaExHelpers.select_partition_by_event_id(reference_id, partitions)
    16
    ```
  """
  @spec select_partition_by_event_id(String.t, [integer()]) :: integer()
  def select_partition_by_event_id(_, [first_partition]) do
    first_partition
  end

  def select_partition_by_event_id(event_id, partitions) do
    partition_count = Enum.count(partitions)
    {val, _} = event_id
      |> Integer.parse(32)
    partition_idx = rem(val, partition_count)
    {partition, _} = List.pop_at(partitions, partition_idx)
    partition
  end

  defp extract_request_idp(payload, key) do
    case payload do
      %{^key => id} -> {:ok, id}
      _anything    -> {:error, :malfomred_request}
    end
  end

  def looping_fetch_messages(partition, w, handler) do
    worker_msgs = handler.(w[:topic], partition, [worker_name: w[:worker_id]])
    handle_messagesp(worker_msgs, w)
    looping_fetch_messages(partition, w, handler)
  end

  @doc ~S"""
   create_workers spawns a set of `KafkaEx` workers based on an incoming array of modules.

   The default action for spawning works is `&KafkaEx.create_worker/2`, `&MockHandler.publish/2` is used for demonstration

  ## Examples

    ```
    iex> defmodule MockCreate do
    ...>   def kafka_meta do
    ...>     %{worker_id: :mock_handler, partitions: [0], topic: 'Mock.V1'}
    ...>   end
    ...>   def publish(_worker_id, _options) do
    ...>     true
    ...>   end
    ...> end
    ...> KafkaExHelpers.create_workers([MockCreate], &MockCreate.publish/2)
    [true]
    ```

  """
  @spec create_workers([module], function) :: [pid]
  def create_workers(workers, handler \\ &KafkaEx.create_worker/2) do
  handlers = fetch_handler_ids(workers)
  handlers
    |> Enum.map(fn(w) ->
      handler.(w, [consumer_group: "#{w}"]) end)
end

  defp handle_messagesp(:topic_not_found, _), do: :no_topic

  defp handle_messagesp(worker_msgs, w) do
    messages = worker_msgs
      |> Enum.flat_map(fn(p) -> p.partitions end)
      |> Enum.flat_map(fn(p) -> p.message_set end)
    process_batchp(messages, w[:handler], w[:batch_size], w[:options])
  end

  defp get_linksp(partitions, worker_id, handler) do
    partitions
      |> Enum.map(fn(partition) ->
        Logger.info(fn -> "Spawning Link for #{partition}" end)
        spawn_link(fn ->
          looping_fetch_messages(partition, worker_id, handler)
        end)
      end)
  end

  defp process_batchp([], _, _, _) do
    :timer.sleep(1000)
    :empty_messages
  end

  defp process_batchp(kafka_messages, handler, batch_size, options) do
    messages = kafka_messages
      |> Flow.from_enumerable(max_demand: batch_size)
      |> Flow.map(fn(m) -> m.value end)
      |> Enum.to_list
    handler.(messages, options)
    :processed_batch
  end

end