vela_cache

defmodule Vela.Cache do
  @moduledoc """
  Public API for Vela caches.

  ## Usage

      defmodule MyApp.Cache do
        use Vela.Cache,
          backend: Vela.Backend.ETS,
          default_ttl: :timer.minutes(5)
      end

      # Add to your supervision tree:
      children = [MyApp.Cache]

      # Use it:
      MyApp.Cache.put(:greeting, "hello")
      MyApp.Cache.get(:greeting)   # => {:ok, "hello"}
  """

  alias Vela.Cache.Entry

  # ---- The `use` macro ----

  defmacro __using__(opts) do
    quote do
      @vela_opts unquote(opts)

      def child_spec(runtime_opts \\ []) do
        opts = Keyword.merge(@vela_opts, runtime_opts)
        config = Vela.Cache.Config.new(__MODULE__, opts)

        %{
          id: __MODULE__,
          start: {Vela.Cache, :start_link, [config]},
          type: :supervisor
        }
      end

      def start_link(runtime_opts \\ []) do
        opts = Keyword.merge(@vela_opts, runtime_opts)
        config = Vela.Cache.Config.new(__MODULE__, opts)
        Vela.Cache.start_link(config)
      end

      def get(key, opts \\ []), do: Vela.Cache.get(__MODULE__, key, opts)
      def get!(key, opts \\ []), do: Vela.Cache.get!(__MODULE__, key, opts)
      def put(key, value, opts \\ []), do: Vela.Cache.put(__MODULE__, key, value, opts)
      def delete(key, opts \\ []), do: Vela.Cache.delete(__MODULE__, key, opts)
      def exists?(key, opts \\ []), do: Vela.Cache.exists?(__MODULE__, key, opts)
      def flush(opts \\ []), do: Vela.Cache.flush(__MODULE__, opts)
      def size, do: Vela.Cache.size(__MODULE__)

      def get_or_fetch(key, fetch_fn, opts \\ []),
        do: Vela.Cache.get_or_fetch(__MODULE__, key, fetch_fn, opts)

      def stats, do: Vela.Cache.stats(__MODULE__)
      def invalidate_tag(tag, opts \\ []), do: Vela.Cache.invalidate_tag(__MODULE__, tag, opts)
    end
  end

  # ---- Lifecycle ----

  def start_link(%Vela.Cache.Config{} = config) do
    DynamicSupervisor.start_child(
      Vela.Cache.DynamicSupervisor,
      {Vela.Cache.Supervisor, config: config}
    )
  end

  # ---- Core API ----

  @spec get(atom(), term(), keyword()) :: {:ok, term()} | {:error, :not_found}
  def get(cache, key, _opts \\ []) do
    config = get_config(cache)
    ts = :persistent_term.get({:vela_topology_state, cache})

    start = System.monotonic_time()

    result =
      if config.near_cache do
        case Vela.NearCache.get(cache, key) do
          {:ok, entry} -> {:ok, entry.value}
          :miss -> get_from_topology(config, ts, cache, key)
        end
      else
        get_from_topology(config, ts, cache, key)
      end

    duration = System.monotonic_time() - start

    if config.stats_enabled do
      case result do
        {:ok, _} -> Vela.Stats.Collector.incr_hits(cache)
        {:error, _} -> Vela.Stats.Collector.incr_misses(cache)
      end
    end

    :telemetry.execute(
      config.telemetry_prefix ++ [:get, :stop],
      %{duration: duration},
      %{cache: cache, key: key, result: if(match?({:ok, _}, result), do: :hit, else: :miss)}
    )

    result
  end

  defp get_from_topology(config, ts, cache, key) do
    case config.topology.get(ts, key) do
      {:ok, entry} ->
        if Entry.expired?(entry) do
          config.topology.delete(ts, key)
          {:error, :not_found}
        else
          maybe_promote_to_l1(config, cache, entry)
          {:ok, entry.value}
        end

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

  defp maybe_promote_to_l1(%{near_cache: true} = config, cache, entry) do
    Vela.NearCache.put(cache, entry, config.near_cache_l1_ttl)
  end

  defp maybe_promote_to_l1(_config, _cache, _entry), do: :ok

  @spec get!(atom(), term(), keyword()) :: term()
  def get!(cache, key, opts \\ []) do
    case get(cache, key, opts) do
      {:ok, value} -> value
      {:error, :not_found} -> raise KeyError, key: key, term: cache
    end
  end

  @spec put(atom(), term(), term(), keyword()) :: :ok | {:error, term()}
  def put(cache, key, value, opts \\ []) do
    config = get_config(cache)
    ts = :persistent_term.get({:vela_topology_state, cache})

    ttl = Keyword.get(opts, :ttl, config.default_ttl)
    entry = Entry.new(key, value, Keyword.put(opts, :ttl, ttl))

    start = System.monotonic_time()

    result =
      case config.topology.put(ts, entry) do
        {:ok, _new_ts} ->
          if config.near_cache do
            Vela.NearCache.put(cache, entry, config.near_cache_l1_ttl)
          end

          :ok

        error ->
          error
      end

    duration = System.monotonic_time() - start

    if config.stats_enabled, do: Vela.Stats.Collector.incr_writes(cache)

    :telemetry.execute(
      config.telemetry_prefix ++ [:put, :stop],
      %{duration: duration},
      %{cache: cache, key: key, ttl: ttl}
    )

    result
  end

  @spec delete(atom(), term(), keyword()) :: :ok
  def delete(cache, key, _opts \\ []) do
    config = get_config(cache)
    ts = :persistent_term.get({:vela_topology_state, cache})
    {:ok, _new_ts} = config.topology.delete(ts, key)

    if config.near_cache, do: Vela.NearCache.delete(cache, key)
    if config.stats_enabled, do: Vela.Stats.Collector.incr_deletes(cache)

    :ok
  end

  @spec exists?(atom(), term(), keyword()) :: boolean()
  def exists?(cache, key, opts \\ []) do
    match?({:ok, _}, get(cache, key, opts))
  end

  @spec flush(atom(), keyword()) :: :ok
  def flush(cache, _opts \\ []) do
    config = get_config(cache)
    ts = :persistent_term.get({:vela_topology_state, cache})
    {:ok, _new_ts} = config.topology.flush(ts)

    if config.near_cache, do: Vela.NearCache.flush(cache)

    :ok
  end

  @spec size(atom()) :: non_neg_integer()
  def size(cache) do
    config = get_config(cache)
    ts = :persistent_term.get({:vela_topology_state, cache})
    config.topology.size(ts)
  end

  @spec stats(atom()) :: map()
  def stats(cache) do
    Vela.Stats.Collector.stats(cache)
  end

  @spec invalidate_tag(atom(), atom(), keyword()) :: {:ok, non_neg_integer()}
  def invalidate_tag(cache, tag, _opts \\ []) do
    config = get_config(cache)
    ts = :persistent_term.get({:vela_topology_state, cache})

    {:ok, count, _new_ts} = config.topology.invalidate_tag(ts, tag)

    if config.near_cache, do: Vela.NearCache.invalidate_tag(cache, tag)

    :telemetry.execute(
      config.telemetry_prefix ++ [:invalidate_tag, :stop],
      %{count: count},
      %{cache: cache, tag: tag}
    )

    {:ok, count}
  end

  # ---- Stampede-protected fetch ----

  @spec get_or_fetch(atom(), term(), (term() -> {:ok, term()} | {:error, term()}), keyword()) ::
          {:ok, term()} | {:error, term()}
  def get_or_fetch(cache, key, fetch_fn, opts \\ []) do
    case get(cache, key, opts) do
      {:ok, value} ->
        {:ok, value}

      {:error, :not_found} ->
        config = get_config(cache)

        if config.stampede_protection do
          fetch_with_lock(cache, key, fetch_fn, opts)
        else
          fetch_and_store(cache, key, fetch_fn, opts)
        end
    end
  end

  defp fetch_with_lock(cache, key, fetch_fn, opts) do
    case Vela.Lock.Manager.acquire(cache, key) do
      :granted ->
        # Re-check cache — another process may have populated it while we waited
        result =
          case get(cache, key, opts) do
            {:ok, value} ->
              {:ok, value}

            {:error, :not_found} ->
              fetch_and_store(cache, key, fetch_fn, opts)
          end

        Vela.Lock.Manager.release(cache, key)
        result

      {:error, :timeout} ->
        # Waited too long — fetch directly as fallback
        fetch_and_store(cache, key, fetch_fn, opts)
    end
  end

  defp fetch_and_store(cache, key, fetch_fn, opts) do
    start = System.monotonic_time()

    result =
      case fetch_fn.(key) do
        {:ok, value} ->
          put(cache, key, value, opts)
          {:ok, value}

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

    duration = System.monotonic_time() - start
    config = get_config(cache)

    :telemetry.execute(
      config.telemetry_prefix ++ [:fetch, :stop],
      %{duration: duration},
      %{cache: cache, key: key, result: if(match?({:ok, _}, result), do: :fetched, else: :error)}
    )

    result
  end

  # ---- Helpers ----

  defp get_config(cache) do
    :persistent_term.get({:vela_config, cache})
  end
end