misha_cafex

defmodule Cafex.Protocol do
  @moduledoc """
  This module provide encode/decode functions for common structures in Kafka protocol.

  And also provide conveniences for implementing API request and the `Cafex.Protocol.Request`.

  ## APIs

    * `Cafex.Protocol.Metadata`
      - `Cafex.Protocol.Metadata.Request`
      - `Cafex.Protocol.Metadata.Response`
    * `Cafex.Protocol.Produce`
      - `Cafex.Protocol.Produce.Request`
      - `Cafex.Protocol.Produce.Response`
    * `Cafex.Protocol.Fetch`
      - `Cafex.Protocol.Fetch.Request`
      - `Cafex.Protocol.Fetch.Response`
    * `Cafex.Protocol.Offset`
      - `Cafex.Protocol.Offset.Request`
      - `Cafex.Protocol.Offset.Response`
    * `Cafex.Protocol.ConsumerMetadata`
      - `Cafex.Protocol.ConsumerMetadata.Request`
      - `Cafex.Protocol.ConsumerMetadata.Response`
    * `Cafex.Protocol.OffsetCommit`
      - `Cafex.Protocol.OffsetCommit.Request`
      - `Cafex.Protocol.OffsetCommit.Response`
    * `Cafex.Protocol.OffsetFetch`
      - `Cafex.Protocol.OffsetFetch.Request`
      - `Cafex.Protocol.OffsetFetch.Response`
  """

  @type api_version :: 0 | 1 | 2
  @type api_key :: 0..16
  @type error :: Cafex.Protocol.Errors.t

  @apis %{
    :produce           => 0,
    :fetch             => 1,
    :offset            => 2,
    :metadata          => 3,
    :offset_commit     => 8,
    :offset_fetch      => 9,
    :group_coordinator => 10,
    :join_group        => 11,
    :heartbeat         => 12,
    :leave_group       => 13,
    :sync_group        => 14,
    :describe_groups   => 15,
    :list_groups       => 16,
  }

  alias Cafex.Protocol.Request

  for {key, value} <- @apis do
    def api_key(unquote(key)), do: unquote(value)
  end

  defmacro __using__(opts) do
    {opts, []} = Code.eval_quoted(opts, [], __CALLER__)
    api = Keyword.get opts, :api

    api_version = Keyword.get opts, :api_version, 0
    mod = __CALLER__.module

    if api == nil do
      raise CompileError, file: __CALLER__.file, line: __CALLER__.line, description: "To use #{inspect __MODULE__}, `api` must be set"
    end

    if ! Map.has_key?(@apis, api) do
      raise CompileError, file: __CALLER__.file, line: __CALLER__.line, description: "Unsupported api: #{api}"
    end

    Module.put_attribute mod, :api, api
    Module.put_attribute mod, :api_version, api_version

    quote do
      import unquote(__MODULE__), only: [defrequest: 0, defrequest: 1, defresponse: 1]
      import Cafex.Protocol.Codec
      @behaviour Cafex.Protocol.Codec
      @before_compile unquote(__MODULE__)

    end
  end

  defmacro __before_compile__(env) do
    # TODO check defrequest and defresponse
    mod = env.module
    request = Module.get_attribute mod, :request
    response = Module.get_attribute mod, :response

    quoted = []

    # ListGroups Request is empty, generate an empty request
    quoted = quoted ++ if request != true do
      [quote do
        unquote(__MODULE__).defrequest
        def encode(_request), do: <<>>
      end]
    else
      []
    end

    if response == nil do
      raise CompileError, file: __CALLER__.file, line: __CALLER__.line, description: "Use #{inspect __MODULE__} must call `defresponse`"
    end

    api         = Module.get_attribute mod, :api
    api_version = Module.get_attribute mod, :api_version

    quoted ++ [quote do
      def has_response?(%__MODULE__.Request{}), do: true
      def decoder(%__MODULE__.Request{}), do: __MODULE__
      def api_key(%__MODULE__.Request{}), do: unquote(__MODULE__).api_key(unquote(api))
      def api_version(%__MODULE__.Request{}), do: unquote(api_version)

      defoverridable [has_response?: 1, api_version: 1, decoder: 1]
    end]
  end

  defmacro defrequest(opts \\ []) do
    block = Keyword.get(opts, :do)
    mod = __CALLER__.module

    impl_protocol = impl_request_protocol(mod)

    quote do
      defmodule Request do
        import unquote(__MODULE__), only: [field: 3, field: 2]

        Module.register_attribute(__MODULE__, :fields, accumulate: true)
        Module.register_attribute(__MODULE__, :struct_fields, accumulate: true)

        unquote(block)
        unquote(impl_protocol)

        Module.eval_quoted __ENV__, [
          Cafex.Protocol.__struct__(@struct_fields),
          Cafex.Protocol.__typespec__(__MODULE__)
        ]
      end
      Module.put_attribute __MODULE__, :request, true
    end
  end

  defmacro defresponse(do: block) do
    quote do
      defmodule Response do
        Module.register_attribute(__MODULE__, :fields, accumulate: true)
        Module.register_attribute(__MODULE__, :struct_fields, accumulate: true)
        import unquote(__MODULE__), only: [field: 3, field: 2]

        unquote(block)

        Module.eval_quoted __ENV__, [
          Cafex.Protocol.__struct__(@struct_fields),
          Cafex.Protocol.__typespec__(__MODULE__)
        ]
      end
      Module.put_attribute __MODULE__, :response, true
    end
  end

  defmacro field(name, opts \\ [], type) do
    type = Macro.escape(type)
    quote do
      Cafex.Protocol.__field__(__MODULE__, unquote(name), unquote(type), unquote(opts))
    end
  end

  defdelegate encode_request(client_id, correlation_id, request), to: Cafex.Protocol.Codec
  defdelegate encode_string(data), to: Cafex.Protocol.Codec
  defdelegate has_response?(request), to: Cafex.Protocol.Request

  @doc false
  def __typespec__(mod) do
    types = Module.get_attribute(mod, :fields)

    {:%, [], [name, {:%{}, [], _}]} = quote do
      %unquote(mod){}
    end

    type_specs = {:%, [], [name, {:%{}, [], types}]}

    quote do
      @type t :: unquote(type_specs)
    end
  end

  @doc false
  def __field__(mod, name, type, opts) do
    default = Keyword.get(opts, :default)
    Module.put_attribute(mod, :fields, {name, type})
    put_struct_field(mod, name, default)
  end

  @doc false
  def __struct__(struct_fields) do
    quote do
      defstruct unquote(Macro.escape(struct_fields))
    end
  end

  defp put_struct_field(mod, name, assoc) do
    fields = Module.get_attribute(mod, :struct_fields)

    if List.keyfind(fields, name, 0) do
      raise ArgumentError, "field #{inspect name} is already set on #{inspect mod}"
    end

    Module.put_attribute(mod, :struct_fields, {name, assoc})
  end

  defp impl_request_protocol(mod) do
    impls = [:api_key, :api_version, :has_response?, :encode, :decoder]
    |> Enum.map(fn func ->
      quote do
        def unquote(func)(req), do: unquote(mod).unquote(func)(req)
      end
    end)

    quote do
      defimpl Cafex.Protocol.Request do
        unquote(impls)
      end
    end
  end
end