csh2fjhyll_kvasir_agent

defmodule Kvasir.Agent.Cache.ETS do
  @behaviour Kvasir.Agent.Cache
  @storage_table __MODULE__

  @impl Kvasir.Agent.Cache
  def init(_agent, _partition, _) do
    ensure_storage_table_created()
  end

  @impl Kvasir.Agent.Cache
  def cache(agent, _partition, id), do: {:ok, {agent, id}}

  @impl Kvasir.Agent.Cache
  def track_command(cache) do
    ensure_storage_table_created()

    case load(cache) do
      {:ok, offset, data} ->
        :ets.insert(@storage_table, {cache, true, data, offset})

      :no_previous_state ->
        :ets.insert(@storage_table, {cache, true})

      err ->
        err
    end
  end

  @impl Kvasir.Agent.Cache
  def stream(agent) do
    ensure_storage_table_created()
    |> :ets.tab2list()
    |> Stream.filter(&(elem(elem(&1, 0), 0) == agent))
    |> Stream.map(fn
      {{_, id}, true} -> {id, :corrupted_state}
      {{_, id}, true, _, _} -> {id, :corrupted_state}
      {c = {_, k}, false, state, offset} -> {k, offset, state, c}
    end)
  end

  @impl Kvasir.Agent.Cache
  def save(cache, data, offset) do
    ensure_storage_table_created()

    :ets.insert(@storage_table, {cache, false, data, offset})
    :ok
  end

  @impl Kvasir.Agent.Cache
  def load(cache) do
    ensure_storage_table_created()

    case :ets.lookup(@storage_table, cache) do
      [] ->
        :no_previous_state

      [{_, true}] ->
        {:error, :corrupted_state}

      [{_, processing, data, offset}] ->
        if processing do
          {:error, :corrupted_state}
        else
          {:ok, offset, data}
        end

      _ ->
        {:ok, Kvasir.Offset.create(), nil}
    end
  end

  @impl Kvasir.Agent.Cache
  def delete(cache) do
    ensure_storage_table_created()
    :ets.delete(@storage_table, cache)

    :ok
  end

  @spec ensure_storage_table_created :: :ok
  defp ensure_storage_table_created do
    case :ets.info(@storage_table) do
      :undefined -> :ets.new(@storage_table, [:set, :public, :named_table])
      _ -> @storage_table
    end
  end
end