avrora

defmodule Avrora.Codec.ObjectContainerFile do
  @dialyzer {:no_opaque, [extract_json_schema: 1, extract_schema: 1]}

  @moduledoc """
  An Avro encoder/decoder working with an Object Container File formatted Avro messages.

  It works with a binary format, which includes a required schema inside the message.
  See more about [Object Container File](https://avro.apache.org/docs/1.8.1/spec.html#Object+Container+Files).
  """

  @behaviour Avrora.Codec
  @magic_bytes <<"Obj", 1>>
  @meta_schema_key "avro.schema"

  require Logger
  alias Avrora.AvroDecoderOptions
  alias Avrora.Codec
  alias Avrora.Config
  alias Avrora.Resolver
  alias Avrora.Schema.Encoder, as: SchemaEncoder

  @impl true
  def is_decodable(payload) when is_binary(payload) do
    case payload do
      <<@magic_bytes, _::binary>> -> true
      _ -> false
    end
  end

  @impl true
  def extract_schema(payload) when is_binary(payload) do
    with {:ok, {headers, {_, _, _, _, _, _, full_name, _} = erlavro, _}} <- do_decode(payload),
         {:ok, nil} <- memory_storage().get(full_name),
         {:ok, json} <- extract_json_schema(headers),
         {:ok, schema} <- SchemaEncoder.from_erlavro(erlavro, json: json) do
      memory_storage().put(full_name, schema)
    end
  end

  @impl true
  def decode(payload) when is_binary(payload) do
    with {:ok, {_, _, decoded}} <- do_decode(payload), do: {:ok, decoded}
  end

  @impl true
  def decode(payload, schema: _schema) when is_binary(payload) do
    Logger.warn("message already contains embeded schema, given schema will be ignored")
    decode(payload)
  end

  @impl true
  def encode(payload, schema: schema) when is_binary(payload) or is_map(payload) do
    with {:ok, schema} <- resolve(schema),
         {:ok, body} <- Codec.Plain.encode(payload, schema: schema),
         {:ok, schema} <- SchemaEncoder.to_erlavro(schema) do
      do_encode(body, schema)
    end
  end

  defp resolve(schema) do
    cond do
      is_binary(schema.full_name) && is_reference(schema.lookup_table) -> {:ok, schema}
      is_binary(schema.full_name) -> Resolver.resolve(schema.full_name)
      true -> {:error, :unusable_schema}
    end
  end

  defp do_decode(payload) do
    {:ok, :avro_ocf.decode_binary(payload, AvroDecoderOptions.options())}
  rescue
    MatchError -> {:error, :schema_mismatch}
    error -> {:error, error}
  end

  defp do_encode(payload, schema) do
    encoded =
      schema
      |> :avro_ocf.make_header()
      |> :avro_ocf.make_ocf(List.wrap(payload))
      |> :erlang.list_to_binary()

    {:ok, encoded}
  rescue
    error -> {:error, error}
  end

  defp extract_json_schema(headers) do
    with {_, _, meta, _} <- headers,
         {@meta_schema_key, json} <- Enum.find(meta, fn {key, _} -> key == @meta_schema_key end) do
      {:ok, json}
    end
  end

  defp memory_storage, do: Config.self().memory_storage()
end