avrora

defmodule Avrora.Encoder do
  @moduledoc """
  Wraps internal codec interface to add syntax sugar which will be exposed to client.
  """

  require Logger
  alias Avrora.Codec
  alias Avrora.Schema
  alias Avrora.Schema.Name

  @doc """
  Extract schema from the binary Avro message.

  ## Examples

      ...> payload = <<0, 0, 0, 0, 8, 72, 48, 48, 48, 48, 48, 48, 48, 48, 45, 48,
      48, 48, 48, 45, 48, 48, 48, 48, 45, 48, 48, 48, 48, 45, 48, 48, 48, 48, 48,
      48, 48, 48, 48, 48, 48, 48, 123, 20, 174, 71, 225, 250, 47, 64>>
      ...> {:ok, schema} = Avrora.Encoder.extract_schema(payload)
      ...> schema.id
      42
      ...> schema.full_name
      "io.confluent.Payment"
  """
  @spec extract_schema(binary()) :: {:ok, Schema.t()} | {:error, term()}
  def extract_schema(payload) when is_binary(payload) do
    codec =
      [Codec.SchemaRegistry, Codec.ObjectContainerFile, Codec.Plain]
      |> Enum.find(& &1.is_decodable(payload))

    codec.extract_schema(payload)
  end

  @doc """
  Decode binary Avro message, loading schema from Schema Registry or Object Container Files.

  ## Examples

      ...> payload = <<0, 0, 0, 0, 8, 72, 48, 48, 48, 48, 48, 48, 48, 48, 45, 48,
      48, 48, 48, 45, 48, 48, 48, 48, 45, 48, 48, 48, 48, 45, 48, 48, 48, 48, 48,
      48, 48, 48, 48, 48, 48, 48, 123, 20, 174, 71, 225, 250, 47, 64>>
      ...> Avrora.Encoder.decode(payload)
      {:ok, %{"id" => "00000000-0000-0000-0000-000000000000", "amount" => 15.99}}
  """
  @spec decode(binary()) :: {:ok, map() | list(map())} | {:error, term()}
  def decode(payload) when is_binary(payload) do
    codec =
      [Codec.SchemaRegistry, Codec.ObjectContainerFile, Codec.Plain]
      |> Enum.find(& &1.is_decodable(payload))

    codec.decode(payload)
  end

  @doc """
  Decode binary Avro message, loading schema from local file or Schema Registry.

  ## Examples

      ...> payload = <<72, 48, 48, 48, 48, 48, 48, 48, 48, 45, 48, 48, 48, 48, 45,
      48, 48, 48, 48, 45, 48, 48, 48, 48, 45, 48, 48, 48, 48, 48, 48, 48, 48, 48,
      48, 48, 48, 123, 20, 174, 71, 225, 250, 47, 64>>
      ...> Avrora.Encoder.decode(payload, schema_name: "io.confluent.Payment")
      {:ok, %{"id" => "00000000-0000-0000-0000-000000000000", "amount" => 15.99}}
  """
  @spec decode(binary(), schema_name: String.t()) :: {:ok, map() | list(map())} | {:error, term()}
  def decode(payload, schema_name: schema_name) when is_binary(payload) do
    with {:ok, schema_name} <- Name.parse(schema_name) do
      unless is_nil(schema_name.version) do
        Logger.warning("decoding with schema version is not supported, `#{schema_name.name}` used instead")
      end

      schema = %Schema{full_name: schema_name.name}

      codec =
        [Codec.SchemaRegistry, Codec.ObjectContainerFile, Codec.Plain]
        |> Enum.find(& &1.is_decodable(payload))

      if codec == Codec.Plain do
        Logger.warning(
          "`Avrora.Encoder.decode/2` with plain format is deprecated, use `Avrora.Encoder.decode_plain/2` instead"
        )
      end

      codec.decode(payload, schema: schema)
    end
  end

  @doc """
  Decode binary Avro message with a :plain format and load schema from local file.

  ## Examples

      ...> payload = <<0, 232, 220, 144, 233, 11, 200, 1>>
      ...> Avrora.Encoder.decode_plain(payload,"io.confluent.NumericTransfer")
      {:ok, %{"link_is_enabled" => false, "updated_at" => 1586632500, "updated_by_id" => 100}
  """
  @spec decode_plain(binary(), schema_name: String.t()) :: {:ok, map()} | {:error, term()}
  def decode_plain(payload, schema_name: schema_name) when is_binary(payload) do
    with {:ok, schema_name} <- Name.parse(schema_name) do
      unless is_nil(schema_name.version) do
        Logger.warning("decoding with schema version is not supported, `#{schema_name.name}` used instead")
      end

      Codec.Plain.decode(payload, schema: %Schema{full_name: schema_name.name})
    end
  end

  @doc """
  Encode message map in Avro format, loading schema from local file or Schema Registry.

  The `:format` argument controls output format:

  * `:plain` - Just return Avro binary data, with no header or embedded schema
  * `:ocf` - Use [Object Container File]https://avro.apache.org/docs/1.8.1/spec.html#Object+Container+Files)
    format, embedding the full schema with the data
  * `:registry` - Write data with Confluent Schema Registry
    [Wire Format](https://docs.confluent.io/current/schema-registry/serializer-formatter.html#wire-format),
    which prefixes the data with the schema id
  * `:guess` - Use `:registry` if possible, otherwise use `:ocf` (default)

  ## Examples

      ...> payload = %{"id" => "00000000-0000-0000-0000-000000000000", "amount" => 15.99}
      ...> Avrora.Encoder.encode(payload, schema_name: "io.confluent.Payment", format: :plain)
      {:ok, <<72, 48, 48, 48, 48, 48, 48, 48, 48, 45, 48, 48, 48, 48, 45,
            48, 48, 48, 48, 45, 48, 48, 48, 48, 45, 48, 48, 48, 48, 48, 48, 48, 48, 48,
            48, 48, 48, 123, 20, 174, 71, 225, 250, 47, 64>>}
  """
  @spec encode(map(), schema_name: String.t(), format: :guess | :registry | :ocf | :plain) ::
          {:ok, binary()} | {:error, term()}
  def encode(payload, schema_name: schema_name) when is_map(payload),
    do: encode(payload, schema_name: schema_name, format: :guess)

  def encode(payload, schema_name: schema_name, format: format) when is_map(payload) do
    with {:ok, schema_name} <- Name.parse(schema_name) do
      if format == :plain do
        Logger.warning(
          "`Avrora.Encoder.encode/2` with `format: :plain` is deprecated, use `Avrora.Encoder.encode_plain/2` instead"
        )
      end

      unless is_nil(schema_name.version) do
        Logger.warning("encoding with schema version is not supported yet, `#{schema_name.name}` used instead")
      end

      schema = %Schema{full_name: schema_name.name}

      case format do
        :guess ->
          with {:error, _} <- Codec.SchemaRegistry.encode(payload, schema: schema),
               do: Codec.ObjectContainerFile.encode(payload, schema: schema)

        :registry ->
          Codec.SchemaRegistry.encode(payload, schema: schema)

        :ocf ->
          Codec.ObjectContainerFile.encode(payload, schema: schema)

        :plain ->
          Codec.Plain.encode(payload, schema: schema)

        _ ->
          {:error, :unknown_format}
      end
    end
  end

  @doc """
  Encode binary Avro message with a :plain format and load schema from local file.

  ## Examples

      ...> payload = %{"link_is_enabled" => false, "updated_at" => 1586632500, "updated_by_id" => 100}
      ...> Avrora.Encoder.encode_plain(payload,"io.confluent.NumericTransfer")
      {:ok, <<0, 232, 220, 144, 233, 11, 200, 1>>}
  """
  @spec encode_plain(map(), schema_name: String.t()) :: {:ok, binary()} | {:error, term()}
  def encode_plain(payload, schema_name: schema_name) when is_map(payload) do
    with {:ok, schema_name} <- Name.parse(schema_name) do
      unless is_nil(schema_name.version) do
        Logger.warning(
          "encoding with schema version is not supported for plain format, `#{schema_name.name}` used instead"
        )
      end

      Codec.Plain.encode(payload, schema: %Schema{full_name: schema_name.name})
    end
  end
end