eidetic_projector

defmodule Eidetic.Projector.Consumer.Kaffe do
  @moduledoc """
  Kaffe Integration with Eidetic Projector

  This module can be set as the Kaffe event handler and it will convert the
  payload received into the Eidetic Projector `Bundle` format and handle
  the event.

  This integration will only work with a Projector running under the default
  name; it will not work if you've opted to override its name.

  This is a Kaffe payload, as documented in their README:

  ```elixir
  # https://github.com/spreedly/kaffe
  %{
    attributes: 0,
    crc: 1914336469,
    key: "kafka message key",
    magic_byte: 0,
    offset: 41,
    partition: 17,
    topic: "some-kafka-topic",
    value: "the actual kafka message value is here",
    ts: 1234567890123, # timestamp in milliseconds
    ts_type: :append  # timestamp type: :undefined | :create | :append
  }
  ```
  """
  alias Eidetic.Projector
  alias Eidetic.Projector.Bundle
  alias Eidetic.Projector.Bundle.ConsumerMeta
  require Logger

  @spec handle_message(map()) :: :ok | :error

  def handle_message(message) when is_map(message) do
    _ = Logger.debug(fn -> "#{__MODULE__} received a message" end)

    consumer_meta = %ConsumerMeta{
      topic: message.topic,
      partition: message.partition,
      offset: message.offset
    }

    _ = Logger.debug(fn -> "ConsumerMeta:\n\n#{inspect(consumer_meta)}" end)

    {status, result} = Projector.consume(consumer_meta, message.value)

    _ = Logger.debug(fn -> "Received status: #{status}" end)
    _ = Logger.debug(fn -> "Received result: #{inspect(result)}" end)

    case status do
      :ok -> :ok
      :skip -> :ok
      :error -> :error
    end
  end
end