avro_schema

defmodule AvroSchema do
  @moduledoc File.read!("README.md")

  use GenServer

  require Logger

  @app :avro_schema
  @dets_table @app
  @ets_table __MODULE__

  # Public API

  # Types
  @typedoc "Integer ID returned by Schema Registry"
  @type regid() :: pos_integer()

  @typedoc "Subject in Schema Registry / Avro name"
  @type subject() :: binary()

  @typedoc "Fingerprint, normally CRC-64-AVRO but could be e.g. MD5"
  @type fp() :: binary()

  @type decoded() :: map | [{binary, term}]

  @typedoc "Cache key"
  @type ref() :: regid() | {subject(), fp()}

  @typedoc "Cache value"
  @type cache_value :: list({ref, :avro.avro_type()}) | {{binary(), binary()}, integer()}

  @typedoc "Tesla client"
  @type client() :: Tesla.Client.t()

  @doc """
  Tag Avro binary data with schema that created it.

  Adds a tag to the front of data indicating the schema that was used
  to encode it.

  Uses
  [Confluent wire format](https://docs.confluent.io/current/schema-registry/serializer-formatter.html#wire-format)
  for integer registry IDs and
  [Avro single object encoding](https://avro.apache.org/docs/1.8.2/spec.html#single_object_encoding_spec)
  for fingerprints.

  This function matches schema IDs as integers and encodes them using Confluent
  format, and fingerprints as binary and encodes them as Avro.

  Strictly speaking, however, fingerprints are integers, so make sure that you
  convert them to binary before calling this function.

  Note that this function returns an iolist for efficiency, not a binary.

  ## Examples

      iex> schema_json = "{\"name\":\"test\",\"type\":\"record\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}"
      iex> fp = AvroSchema.fingerprint_schema(schema_json)
      iex> AvroSchema.tag("hello", fp)
      [<<195, 1>>, <<172, 194, 58, 14, 16, 237, 158, 12>>, "hello"]

  """
  @spec tag(iodata, regid | fp) :: iolist
  def tag(bin, fp) when is_binary(fp) do
    [<<0xC3, 0x01>>, fp, bin]
    # [<<0xC3, 0x01, fp::unsigned-little-64>>, bin]
  end

  def tag(bin, regid) when is_integer(regid) do
    [<<0, regid::unsigned-big-32>>, bin]
  end

  @doc """
  Split schema tag from tagged Avro binary data.

  Supports [Confluent wire format](https://docs.confluent.io/current/schema-registry/serializer-formatter.html#wire-format)
  for integer registry IDs and [Avro single object encoding](https://avro.apache.org/docs/1.8.2/spec.html#single_object_encoding_spec)
  for fingerprints.

  ## Examples

    iex> tagged_avro = IO.iodata_to_binary([<<195, 1>>, <<172, 194, 58, 14, 16, 237, 158, 12>>, "hello"])
    iex> AvroSchema.untag(tagged_avro)
    {:ok, {{:avro, <<172, 194, 58, 14, 16, 237, 158, 12>>}, "hello"}}

    iex> tagged_confluent = IO.iodata_to_binary([<<0, 0, 0, 0, 7>>, "hello"])
    iex> AvroSchema.untag(tagged_confluent)
    {:ok, {{:confluent, 7}, "hello"}}
  """
  @spec untag(iodata) ::
          {:ok, {{:confluent, regid}, binary}}
          | {:ok, {{:avro, fp}, binary}}
          | {:error, :unknown_tag}
  def untag(<<0, regid::unsigned-big-32, data::binary>>) do
    {:ok, {{:confluent, regid}, data}}
  end

  def untag(<<0xC3, 0x01, fp::bytes-size(8), data::binary>>) do
    {:ok, {{:avro, fp}, data}}
  end

  def untag(_) do
    {:error, :unknown_tag}
  end

  @doc """
  Get schema for schema reference.

  This tries to read the schema from the cache. If not found, it makes a call
  to the Schema Registry.

  This is typically called by a Kafka consumer to find the schema which was
  used to encode data based on the tag.

  This call has the overhead of an ETS lookup and potentially a GenServer call
  to fetch the Avro schema via HTTP. If you need maximum performance, keep the
  result and reuse it for future requests with the same reference.
  """
  @spec get_schema(ref) :: {:ok, :avro.avro_type()} | {:error, term}
  def get_schema(ref) do
    case cache_lookup(ref) do
      nil ->
        GenServer.call(__MODULE__, {:get_schema, ref})

      value ->
        {:ok, value}
    end
  end

  @doc """
  Cache schema locally.

  Inserts the schema in the local cache under one or more references.

  `get_schema/1` will then return the schema without needing to communicate
  with the Schema Registry.

  ## Examples

      iex> schema_json = "{\"name\":\"test\",\"type\":\"record\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}"
      iex> {:ok, schema} = AvroSchema.parse_schema(schema_json)
      iex> full_name = AvroSchema.full_name(schema_json)
      iex> fp = AvroSchema.fingerprint_schema(schema_json)
      iex> ref = {full_name, fp}
      iex> :ok = AvroSchema.cache_schema(ref, schema)
      :ok
  """
  @spec cache_schema(ref | list(ref), binary | :avro.avro_type(), boolean) :: :ok | {:error, term}
  def cache_schema(refs, schema, persistent \\ false)

  def cache_schema(refs, schema, persistent) when is_list(refs) do
    case process_schema(schema) do
      {:ok, value} ->
        objects = Enum.map(refs, &{&1, value})
        cache_insert(objects, persistent)

      {:error, reason} ->
        {:error, reason}
    end
  end

  def cache_schema(ref, schema, persistent), do: cache_schema([ref], schema, persistent)

  # TODO
  @doc """
  Register schema in Confluent Schema Registry.

  The subject is a unique name to register the schema, often the full name from the Avro schema.
  See the [standard strategies](https://docs.confluent.io/current/schema-registry/serializer-formatter.html#subject-name-strategy)
  used by Confluent in their Kafka libraries.

  It is safe to register the same schema multiple times, it will always return
  the same ID.
  """
  @spec register_schema(subject, binary) :: {:ok, regid} | {:error, term}
  def register_schema(subject, schema) when is_binary(schema) do
    case cache_lookup({subject, schema}) do
      nil ->
        GenServer.call(__MODULE__, {:register_schema, subject, schema})

      value ->
        {:ok, value}
    end
  end

  @doc """
  Cache registration.

  Inserts the schema in the local cache.

  `register_schema/1` will then return the id without needing to communicate
  with the Schema Registry.
  """
  @spec cache_registration(binary(), binary(), integer(), boolean()) :: :ok | :error
  def cache_registration(subject, schema, regid, persistent \\ false) when is_binary(schema) do
    cache_insert({{subject, schema}, regid}, persistent)
  end

  @doc """
  Make Avro decoder for schema.

  Creates a function which decodes a Avro encoded binary data to a map. By default,
  a `:hook` option is provided that will convert all `:null` values to `nil`.
  """
  @spec make_decoder(binary | :avro.avro_type(), Keyword.t()) :: {:ok, fun} | {:error, term}
  def make_decoder(schema, decoder_opts \\ [record_type: :map, map_type: :map])

  def make_decoder(schema_json, decoder_opts) when is_binary(schema_json) do
    case parse_schema(schema_json) do
      {:ok, schema} ->
        make_decoder(schema, decoder_opts)

      error ->
        error
    end
  end

  def make_decoder(schema, decoder_opts) do
    decoder_opts = Keyword.put_new(decoder_opts, :hook, &decoder_hook/4)
    {:ok, :avro.make_simple_decoder(schema, decoder_opts)}
  end

  @doc """
  Get decoder function for registration id.

  Convenience function, calls `get_schema/1` on the id, then `make_decoder/2`.
  """
  @spec get_decoder(ref | {:confluent, regid}, Keyword.t()) :: {:ok, fun} | {:error, term}
  def get_decoder(ref, decoder_opts \\ [record_type: :map, map_type: :map])
  def get_decoder({:confluent, regid}, decoder_opts), do: get_decoder(regid, decoder_opts)

  def get_decoder(ref, decoder_opts) do
    case get_schema(ref) do
      {:ok, schema} ->
        make_decoder(schema, decoder_opts)

      error ->
        error
    end
  end

  defp decoder_hook(type, _sub_info, data, decode_fn) do
    case :avro.get_type_name(type) do
      "null" -> {nil, data}
      _ -> decode_fn.(data)
    end
  end

  # TODO
  @doc """
  Make Avro encoder for schema.

  Creates a function which encodes Avro terms to binary.
  """
  @spec make_encoder(binary | :avro.avro_type(), Keyword.t()) :: {:ok, fun} | {:error, term}
  def make_encoder(schema_json, encoder_opts \\ [])

  def make_encoder(schema_json, encoder_opts) when is_binary(schema_json) do
    case parse_schema(schema_json) do
      {:ok, schema} ->
        make_encoder(schema, encoder_opts)

      error ->
        error
    end
  end

  def make_encoder(schema, encoder_opts) do
    {:ok, :avro.make_simple_encoder(schema, encoder_opts)}
  end

  @doc """
  Get encoder function for registration id.

  Convenience function, calls `get_schema/1` on the id, then `make_encoder/2`.
  """
  @spec get_encoder(ref, Keyword.t()) :: {:ok, fun} | {:error, term}
  def get_encoder(ref, encoder_opts \\ []) do
    case get_schema(ref) do
      {:ok, schema} ->
        make_encoder(schema, encoder_opts)

      error ->
        error
    end
  end

  # @spec decode(binary) :: {:ok, map | [{binary, term}]} | {:error, term}
  # def decode(tagged_bin) do
  #   with {:ok, {{:confluent, regid}, data}} <- untag(tagged_bin),
  #        {:ok, schema} <- get_schema(regid)
  #   do
  #     {:ok, decoder.(data)}
  #   else
  #     error -> error
  #   end
  # end

  @doc "Decode binary Avro data."
  @spec decode(binary, fun) :: {:ok, decoded()} | {:error, term()}
  def decode(bin, decoder) do
    try do
      {:ok, decoder.(bin)}
    rescue
      error -> {:error, error}
    end
  end

  @doc "Decode binary Avro data, raises if there is a decoding error"
  @spec decode!(binary, fun) :: decoded()
  def decode!(bin, decoder) do
    decoder.(bin)
  end

  @doc "Encode Avro data to binary."
  @spec encode(map | [{binary, term}], fun) :: {:ok, binary} | {:error, term()}
  def encode(data, encoder) do
    try do
      {:ok, encoder.(data)}
    rescue
      error -> {:error, error}
    end
  end

  @doc "Encode Avro data to binary, raises if there is an encoding error"
  @spec encode!(map | [{binary, term}], fun) :: binary
  def encode!(data, encoder) do
    encoder.(data)
  end

  @doc """
  Create fingerprint of schema JSON.

  Ensures that schema is in standard form, then generates an
  CRC-64-AVRO fingerprint on it using `create_fingerprint/1`.

  ## Examples

      iex> schema_json = "{\"name\":\"test\",\"type\":\"record\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}"
      iex> AvroSchema.fingerprint_schema(schema_json)
      <<172, 194, 58, 14, 16, 237, 158, 12>>

  """
  @spec fingerprint_schema(binary) :: fp
  def fingerprint_schema(schema) do
    schema
    |> canonicalize_schema()
    |> normalize_json()
    |> create_fingerprint()
  end

  @doc """
  Create CRC-64-AVRO fingerprint hash for Avro schema JSON.

  See [CRC-64-AVRO](https://avro.apache.org/docs/1.8.2/spec.html#schema_fingerprints)

  ## Examples

      iex> schema_json = "{\"name\":\"test\",\"type\":\"record\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}"
      iex> AvroSchema.fingerprint_schema(schema_json)
      <<172, 194, 58, 14, 16, 237, 158, 12>>
  """
  @spec create_fingerprint(binary) :: fp
  def create_fingerprint(binary) do
    <<:avro.crc64_fingerprint(binary)::little-size(64)>>
  end

  @doc """
  Ensure Avro schema is in Parsing Canonical Form.

  Converts schema into [Parsing Canonical Form](https://avro.apache.org/docs/current/spec.html#Parsing+Canonical+Form+for+Schemas)
  by decoding it and re-encoding it.
  """
  @spec canonicalize_schema(binary) :: binary
  def canonicalize_schema(schema) when is_binary(schema) do
    schema = :avro.decode_schema(schema)
    :avro.encode_schema(schema, canon: true)
  end

  @doc """
  Normalize JSON by decoding and re-encoding it.

  This reduces irrelevant differences such as whitespace which may affect
  fingerprinting.
  """
  @spec normalize_json(binary) :: binary
  def normalize_json(json) do
    parsed_json = :jsone.decode(json, object_format: :tuple)
    :jsone.encode(parsed_json)
  end

  @doc "Encode parsed schema as JSON."
  @spec encode_schema(:avro.avro_type(), Keyword.t()) :: binary
  def encode_schema(schema, opts \\ []) do
    :avro.encode_schema(schema, opts)
  end

  @doc "Make registration subject from name + fingerprint."
  @spec make_fp_subject({binary, fp}) :: binary
  def make_fp_subject({name, fp}) when is_binary(fp), do: "#{name}-#{to_hex(fp)}"

  @spec make_fp_subject(binary, fp) :: binary
  def make_fp_subject(name, fp) when is_binary(fp), do: make_fp_subject({name, fp})

  @doc "
  Get full name field from schema.

  This can be used as the Schema Registry subject.
  "
  @spec full_name(:avro.avro_type() | binary) :: binary
  def full_name(schema_json) when is_binary(schema_json) do
    {:ok, schema} = parse_schema(schema_json)
    full_name(schema)
  end

  def full_name({:avro_record_type, _, _, _, _, _, full_name, _}) do
    to_string(full_name)
  end

  @doc """
  Parse schema into Avro library internal form.

  ## Examples

    iex> schema_json = "{\"name\":\"test\",\"type\":\"record\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}"
    iex> {:ok, schema} = AvroSchema.parse_schema(schema_json)
    {:ok,
     {:avro_record_type, "test", "", "", [],
      [
        {:avro_record_field, "field1", "", {:avro_primitive_type, "string", []},
          :undefined, :ascending, []},
  """
  @spec parse_schema(binary) :: {:ok, :avro.avro_type()} | {:error, term}
  def parse_schema(json) do
    avro_type = :avro.decode_schema(json)
    lkup = :avro.make_lkup_fun(avro_type)
    {:ok, :avro.expand_type_bloated(avro_type, lkup)}
  rescue
    # e in RuntimeError -> {:error, e}
    e in ArgumentError ->
      {:error, e.message}
      # e in ErlangError -> {:error, e.original}
  end

  @doc """
  Convert `DateTime` to Avro integer timestamp with ms precision.

  ## Examples

      iex> datetime = DateTime.utc_now()
      ~U[2019-11-08 09:09:01.055742Z]

      iex> timestamp = AvroSchema.to_timestamp(datetime)
      1573204141055742
  """
  @spec to_timestamp(DateTime.t()) :: non_neg_integer
  def to_timestamp(date_time) do
    DateTime.to_unix(date_time, :microsecond)
  end

  @doc """
  Convert Avro integer timestamp to `DateTime`.

      iex> timestamp = 1573204141055742
      iex> datetime = AvroSchema.to_datetime(timestamp)
      ~U[2019-11-08 09:09:01.055742Z]
  """
  def to_datetime(timestamp) do
    {:ok, datetime} = DateTime.from_unix(timestamp, :microsecond)
    datetime
  end

  @doc """
  List schema files in directory.

  Lists files in a directory matching an extension, default `avsc`.
  """
  @spec get_schema_files(Path.t()) :: {:ok, list(Path.t())} | {:error, File.posix()}
  def get_schema_files(dir, ext \\ "avsc") do
    {:ok, re} = Regex.compile("#{ext}$")

    case File.ls(dir) do
      {:ok, files} ->
        paths =
          files
          |> Enum.filter(&Regex.match?(re, &1))
          |> Enum.map(&Path.join(dir, &1))

        {:ok, paths}

      error ->
        error
    end
  end

  @doc """
  Cache schema files with fingerprints.

  Loads a schema file from disk, and parses it to get the full name.

  Generates fingerprints for the schema using the raw file bytes,
  [Parsing Canonical Form](https://avro.apache.org/docs/1.8.2/spec.html#Parsing+Canonical+Form+for+Schemas),
  and normalized JSON for the canonical form (whitespace stripped).
  This improves interop with fingerprints generated by other programs,
  avoiding insignificant differences.

  Also accepts a map with aliases that the schema should be registered
  under, e.g. with a different subject name or legacy fingerprint.
  Map key is the full name, and value is a list of fingerprints or
  `{name, fingerprint}` tuples.

  Deduplicates the fingerprints and aliases, then calls `cache_schema/3`.
  """
  @spec cache_schema_file(Path.t(), map) :: {:ok, list({binary, binary})} | {:error, term}
  def cache_schema_file(path, subject_aliases \\ %{}) do
    {:ok, json} = File.read(path)
    {:ok, schema} = AvroSchema.parse_schema(json)
    full_name = AvroSchema.full_name(schema)

    # Logger.info("full_name #{full_name}")

    # Put schema in Parsing Canonical Form
    canon_json = AvroSchema.canonicalize_schema(json)
    normal_json = AvroSchema.normalize_json(canon_json)

    primary_ref = {full_name, AvroSchema.create_fingerprint(normal_json)}

    intermediate_refs =
      for bin <- [json, canon_json] do
        {full_name, AvroSchema.create_fingerprint(bin)}
      end

    aliases = subject_aliases[full_name] || []

    alias_refs =
      Enum.map(aliases, fn
        fp when is_binary(fp) -> {full_name, fp}
        {_name, _fp} = a -> a
      end)

    refs = Enum.uniq([primary_ref] ++ intermediate_refs ++ alias_refs)

    # for {subject, fp} <- refs do
    #   fp_hex = Base.encode16(fp, case: :lower)
    #   Logger.debug("Registering #{subject} #{fp_hex}")
    # end
    case AvroSchema.cache_schema(refs, normal_json) do
      :ok ->
        {:ok, refs}

      error ->
        error
    end
  end

  @doc "Convert binary fingerprint to hex"
  @spec to_hex(fp) :: binary
  def to_hex(fp), do: Base.encode16(fp, case: :lower)

  # GenServer callbacks

  @doc "Start cache GenServer."
  @spec start_link(list, list) :: {:ok, pid} | {:error, any}
  def start_link(args, opts \\ []) do
    GenServer.start_link(__MODULE__, args, Keyword.merge([name: __MODULE__], opts))
  end

  @doc "Stop cache GenServer"
  @spec stop() :: :ok
  def stop() do
    GenServer.call(__MODULE__, :stop)
  end

  @impl GenServer
  def init(args) do
    # Logger.info("init: #{inspect args}")
    cache_dir = args[:cache_dir]
    persistent = if cache_dir, do: true, else: false
    client = ConfluentSchemaRegistry.client(args[:client] || [])

    # Logger.info("Starting with refresh cycle #{refresh_cycle}")

    @ets_table = :ets.new(@ets_table, [:named_table, :set, :public, {:read_concurrency, true}])

    load_cache(cache_dir)

    state = %{
      persistent: persistent,
      client: client
    }

    {:ok, state}
  end

  defp load_cache(nil), do: :ok

  defp load_cache(cache_dir) do
    path = to_charlist(Path.join(cache_dir, "#{@dets_table}.dets"))

    case :dets.open_file(@dets_table, file: path) do
      {:error, reason} ->
        Logger.error("Error opening DETS file #{path}: #{inspect(reason)}")

      {:ok, ref} ->
        # Logger.debug("DETS info #{inspect ref}: #{inspect :dets.info(ref)}")
        case :dets.to_ets(ref, @ets_table) do
          {:error, reason} ->
            Logger.error("Error loading data from DETS table #{path}: #{inspect(reason)}")

          _ ->
            # Logger.debug("Initialized ETS cache from DETS table #{path}")
            :ok
        end
    end

    :ok
  end

  @impl GenServer
  def handle_call({:get_schema, ref}, _from, state) do
    client = state[:client]
    persistent = state[:persistent]
    response = cache_apply(ref, &do_get_schema/2, [client | [ref]], persistent)
    {:reply, response, state}
  end

  def handle_call({:register_schema, subject, schema}, _from, state) do
    client = state[:client]
    persistent = state[:persistent]

    response =
      cache_apply(
        {subject, schema},
        &do_register_schema/3,
        [client | [subject, schema]],
        persistent
      )

    {:reply, response, state}
  end

  @impl GenServer
  def terminate(_reason, %{persistent: true}) do
    case :dets.close(@dets_table) do
      {:error, message} ->
        Logger.error("Error closing DETS table #{@dets_table}: #{inspect(message)}")

      :ok ->
        :ok
    end
  end

  def terminate(_reason, _state), do: :ok

  # Get schema from registry, parse and create encoder and decoder
  @spec do_get_schema(client, ref) :: {:ok, :avro.avro_type()} | {:error, term}
  defp do_get_schema(client, ref) when is_integer(ref) do
    case ConfluentSchemaRegistry.get_schema(client, ref) do
      {:ok, schema} ->
        process_schema(schema)

      {:error, code, reason} ->
        {:error, {:confluent_schema_registry, code, reason}}
    end
  end

  defp do_get_schema(client, {_name, _fp} = ref) do
    subject = make_fp_subject(ref)
    # Fingerprint is unique, so version is always 1
    version = 1

    case ConfluentSchemaRegistry.get_schema(client, subject, version) do
      {:ok, %{"schema" => schema}} ->
        process_schema(schema)

      {:error, code, reason} ->
        {:error, {:confluent_schema_registry, code, reason}}

      error ->
        error
    end
  end

  # Convert schema before storing in the cache
  @spec process_schema(binary | :avro.avro_type()) :: {:ok, :avro.avro_type()} | {:error, term}
  defp process_schema(schema) when is_binary(schema), do: parse_schema(schema)
  defp process_schema(schema), do: {:ok, schema}

  # @doc "Register schema"
  @spec do_register_schema(client, subject, binary | :avro.avro_type()) ::
          {:ok, regid} | {:error, term}
  defp do_register_schema(client, subject, schema) when is_binary(schema) do
    case ConfluentSchemaRegistry.register_schema(client, subject, schema) do
      {:ok, regid} ->
        {:ok, regid}

      {:error, code, reason} ->
        {:error, {:confluent_schema_registry, code, reason}}
    end
  end

  # Internal, may be public so that they can be used by tests

  @doc false
  # Look up value in cache by key
  @spec cache_lookup(term) :: term | nil
  def cache_lookup(key) do
    try do
      case :ets.lookup(@ets_table, key) do
        [{^key, value}] -> value
        [] -> nil
      end
    catch
      # This may fail if clients make calls before the table is created
      :error, :badarg ->
        Logger.error("badarg")
        nil
    end
  end

  # Run function and cache results if successful
  @spec cache_apply(term, fun, list(any), boolean) :: any
  defp cache_apply(key, fun, args, persistent) do
    case cache_lookup(key) do
      nil ->
        # Logger.debug("cache_apply: #{inspect key} not found")
        case apply(fun, args) do
          {:ok, result} ->
            cache_insert([{key, result}], persistent)
            {:ok, result}

          error ->
            error
        end

      value ->
        # Logger.debug("cache_apply: #{inspect key} #{inspect value}")
        {:ok, value}
    end
  end

  # Insert into ETS and optionally DETS
  @spec cache_insert(cache_value(), boolean) :: :ok | :error

  # defp cache_insert(objects, persistent?) do
  #   ets_insert(objects)

  #   if persistent? do
  #     dets_insert(objects)
  #   end
  # end

  defp cache_insert(objects, persistent)

  defp cache_insert(objects, true) do
    ets_insert(objects)
    dets_insert(objects)
  end

  defp cache_insert(objects, _) do
    ets_insert(objects)
  end

  # Put value in ETS cache
  @spec ets_insert(cache_value()) :: :ok | :error
  defp ets_insert(objects) do
    # Logger.debug("ETS insert #{inspect key} = #{inspect value}")
    # Logger.debug("ETS insert #{inspect key}")

    try do
      :ets.insert(@ets_table, objects)
      :ok
    catch
      # This may fail if clients make calls before the table is created
      :error, :badarg ->
        :error
    end
  end

  # Put value in DETS cache
  @spec dets_insert(cache_value()) :: :ok | :error
  defp dets_insert(objects) do
    # Logger.debug("DETS insert #{inspect key} = #{inspect value}")
    # Logger.debug("DETS insert #{inspect key}")

    try do
      # DETS insert succeeds even if there is no table open :-/
      case :dets.insert(@dets_table, objects) do
        :ok ->
          :ok

        {:error, reason} ->
          Logger.warn(
            "Could not insert into DETS table #{@dets_table} #{inspect(reason)} #{
              inspect(objects)
            }"
          )

          :error
      end
    catch
      :error, :badarg ->
        Logger.warn("Could not insert into DETS table #{@dets_table} badarg #{inspect(objects)}")
        :error
    end
  end

  @doc false
  # Dump cache
  def dump do
    :ets.foldl(fn entry, acc -> [entry | acc] end, [], @ets_table)
  end
end