cache_worker

defmodule CacheWorker do
  @moduledoc String.trim_leading(
               Regex.replace(
                 ~r/```(elixir|json)(\n|.*)```/Us,
                 File.read!("README.md"),
                 fn _, _, code -> Regex.replace(~r/^/m, code, "    ") end
               ),
               "# CacheWorker\n\n"
             )

  use GenServer
  alias CacheWorker.{Bucket, Config}
  require Logger

  @type bucket :: atom
  @type key :: term
  @type value :: term
  @type opts :: keyword
  @type init_return ::
          {:ok, map} | :ok | {:warn, String.t()} | {:stop, String.t()}
  @type load_return :: {:ok, value} | {:error, String.t()}
  @type fetch_return :: {:ok, value} | {:error, String.t()} | :no_bucket
  @type fetch_no_save_return ::
          {:ok, value, boolean} | {:error, String.t()} | :no_bucket

  @doc "Returns the child_spec which should be used by the supervisor"
  @callback child_spec(keyword) :: Supervisor.child_spec()

  @doc "Give a CacheWorker the opportunity to settle in"
  @callback init(Config.t()) :: init_return

  @doc "Do the work to procure the value for a given key."
  @callback load(key) :: load_return

  @doc "Invoked on the `:refresh_interval` when the cache should be refreshed"
  @callback full_refresh :: :ok

  @callback fetch(CacheWorker.key(), CacheWorker.opts()) ::
              CacheWorker.fetch_return()

  @callback fetch_no_save(CacheWorker.key(), CacheWorker.opts()) ::
              CacheWorker.fetch_no_save_return()

  @callback refresh_filter(key, load_return) :: load_return

  defmacro __using__(use_opts) do
    # Compile-time check with friendly reminder
    is_atom(Keyword.get(use_opts, :bucket, "not atom")) ||
      raise ":bucket option must be defined directly in the use options!"

    bucket = Keyword.get(use_opts, :bucket)

    quote do
      alias CacheWorker.Bucket

      @behaviour CacheWorker

      @doc false
      @impl true
      @spec child_spec(keyword) :: Supervisor.child_spec()
      def child_spec(opts) do
        CacheWorker.child_spec(__MODULE__, unquote(use_opts), opts)
      end

      @doc "Get the bucket name for this module."
      @spec bucket :: CacheWorker.bucket()
      def bucket, do: unquote(Keyword.get(use_opts, :bucket))

      @doc "Returns the `%CacheWorker.Config{}`"
      @spec config :: Config.t() | no_return
      def config do
        case Bucket.get(unquote(config_table(bucket)), nil) do
          %Config{} = config ->
            config

          nil ->
            raise """
            No config available for #{unquote(bucket)} bucket. \
            Did you forget to start its CacheWorker?\
            """
        end
      end

      @doc "Returns a particular value from the config"
      @spec config(CacheWorker.key(), term) :: term | no_return
      def config(key, default \\ nil) do
        Map.get(config(), key, default)
      end

      @doc """
      Initialize a CacheWorker.

      If a cache file is not loaded on startup, this callback will be invoked
      with the CacheWorker's `%Config{}`.

      This function should return `{:ok, map}` if a key/val set should be
      used to seed the cache, `:ok` if not, `{:warn, reason}` if a warning
      should be logged, and `{:stop, reason}` if initialization should be
      halted.
      """
      @impl true
      @spec init(%Config{}) :: CacheWorker.init_return()
      def init(%Config{} = _config), do: :ok

      @doc "Call `&load/1` for each key in the cache. Returns `:ok`"
      @impl true
      @spec full_refresh :: :ok
      def full_refresh do
        CacheWorker.full_refresh(__MODULE__)
      end

      @doc """
      For the background full refresh, values returned from `&load/1` will be
      passed through this function to allow for any particular logic that
      isn't needed for normal fetches.
      """
      @impl true
      @spec refresh_filter(CacheWorker.key(), CacheWorker.load_return()) ::
              CacheWorker.load_return()
      def refresh_filter(_key, load_return), do: load_return

      @doc "Fetches the value for a specific key in the bucket."
      @impl true
      @spec fetch(CacheWorker.key(), CacheWorker.opts()) ::
              CacheWorker.fetch_return()
      def fetch(key, opts \\ []), do: CacheWorker.fetch(__MODULE__, key, opts)

      @doc """
      Fetches the value for a specific key in the bucket with `:skip_save`
      set to true. On success, the `:ok` tuple has a third element as a
      boolean which will be true if `&load/1` was called.
      """
      @impl true
      @spec fetch_no_save(CacheWorker.key(), CacheWorker.opts()) ::
              CacheWorker.fetch_no_save_return()
      def fetch_no_save(key, opts \\ []),
        do: CacheWorker.fetch_no_save(__MODULE__, key, opts)

      @doc "Gets the value for a specific key in the bucket."
      @spec get(CacheWorker.key(), CacheWorker.opts()) :: term
      def get(key, opts \\ []),
        do: CacheWorker.get(__MODULE__, key, opts)

      @doc "Set a key/val in the bucket directly, avoiding `&load/1`"
      @spec direct_get(CacheWorker.key()) :: term
      def direct_get(key), do: CacheWorker.direct_get(unquote(bucket), key)

      @doc "Get a key/val from the bucket directly, avoiding `&load/1`"
      @spec direct_set(CacheWorker.key(), CacheWorker.value()) ::
              :ok | :no_bucket
      def direct_set(key, val),
        do: CacheWorker.direct_set(unquote(bucket), key, val)

      @doc "Gets a list of all keys in a given bucket."
      @spec keys :: [CacheWorker.key()] | :no_bucket
      def keys, do: Bucket.keys(unquote(bucket))

      defoverridable init: 1, full_refresh: 0, refresh_filter: 2
    end
  end

  @doc false
  def child_spec(mod, use_opts, opts) do
    opts = Keyword.merge(use_opts, opts)
    %{id: mod, start: {__MODULE__, :start_link, [mod, opts]}}
  end

  @doc false
  def start_link(mod, opts) do
    GenServer.start_link(__MODULE__, {mod, opts}, name: mod)
  end

  @doc false
  def init({mod, use_opts}) do
    %{bucket: bucket} = config = Config.normalize!(mod, use_opts)

    # Skip init if we load a cache file
    bucket
    |> maybe_load_file(config.file)
    |> if do
      :ok
    else
      :ok = Bucket.ensure_new(bucket)

      {config.mod, :init, [config]}
      |> invoke_carefully()
      |> handle_init_ret(config)
    end
    |> case do
      :ok ->
        c_table = config_table(bucket)
        :ok = Bucket.ensure_new(c_table)
        :ok = Bucket.set(c_table, nil, config)

        schedule_full_refresh(config.refresh_interval)

        Logger.debug(fn ->
          l = config |> Map.from_struct() |> Enum.into([])
          "#{bucket}: Initialized CacheWorker: #{inspect(l)}"
        end)

        {:ok, nil}

      {:stop, msg} ->
        Bucket.delete(bucket)
        {:stop, msg}
    end
  end

  @doc "Handle the signal to refresh the cache"
  def handle_info(:full_refresh, _) do
    this_worker_module().full_refresh()
    {:noreply, nil}
  end

  @doc "Fetch a value from the CacheWorker"
  @spec fetch(module, key, opts) :: fetch_return
  def fetch(mod, key, opts \\ []) do
    with {:ok, value, _load_called?} <- do_fetch(mod.config(), key, opts) do
      {:ok, value}
    end
  end

  @doc "Fetch a value from the CacheWorker, but dont save"
  @spec fetch_no_save(module, key, opts) :: fetch_no_save_return
  def fetch_no_save(mod, key, opts \\ []) do
    opts = Keyword.merge([skip_save?: true], opts)
    do_fetch(mod.config(), key, opts)
  end

  @doc "Get a value from the CacheWorker"
  @spec get(module, key, opts) :: value | nil
  def get(mod, key, opts \\ []) do
    case fetch(mod, key, opts) do
      {:ok, val} -> val
      {:error, _} -> nil
    end
  end

  @doc "Handle the refreshing of all keys for a given worker module"
  @spec full_refresh(module) :: :ok | no_return
  def full_refresh(mod) do
    %{bucket: bucket} = config = mod.config()

    {:ok, keys} = Bucket.keys(bucket)

    Logger.info("refreshing #{inspect(config)}, #{inspect(keys)}")

    refresh_for_keys(config, keys)

    file_dump(bucket, config.file)

    schedule_full_refresh(config.refresh_interval)

    :ok
  end

  @doc """
  Get a value directly out of the cache (bucket), without calling `&load/1`
  """
  @spec direct_get(bucket, key) :: term
  def direct_get(bucket, key) do
    Bucket.get(bucket, key)
  end

  @doc "Set a value directly into the cache (bucket)"
  @spec direct_set(bucket, key, value) :: :ok | :no_bucket
  def direct_set(bucket, key, val) do
    Bucket.set(bucket, key, val)
  end

  @doc "Delete the in-memory tables related to the given bucket."
  @spec delete_tables(bucket) :: :ok | :no_table
  def delete_tables(bucket) do
    with :ok <- Bucket.delete(bucket) do
      Bucket.delete(config_table(bucket))
    end
  end

  @doc "Get the name of a bucket's config table by its name"
  def config_table(bucket), do: String.to_atom("#{bucket}_config")

  defp do_fetch(%{cache_enabled: false} = config, key, opts) do
    run_load(config, key, opts)
  end

  defp do_fetch(%{bucket: bucket} = config, key, opts) do
    with :undefined <- Bucket.fetch(bucket, key),
         {:ok, val, load_called?} <- run_load(config, key, opts) do
      file_dump(bucket, config.file)
      {:ok, val, load_called?}
    else
      {:ok, val} -> {:ok, val, false}
      other -> other
    end
  end

  defp invoke_carefully({mod, fun, args}) do
    apply(mod, fun, args)
  catch
    type, error -> {:caught, type, error, __STACKTRACE__}
  end

  @spec handle_init_ret(init_return | {:caught, atom, map}, Config.t()) ::
          :ok | {:stop, String.t()}
  defp handle_init_ret(init_ret, config) do
    case init_ret do
      {:ok, map} when is_map(map) ->
        store_map_into_cache(config, map)
        :ok

      :ok ->
        :ok

      {:stop, msg} ->
        {:stop, msg}

      {:error, msg} ->
        Logger.warn(fn -> "#{config.mod}.init Error: #{msg}" end)
        :ok

      {:caught, type, error, stacktrace} ->
        Logger.warn(fn ->
          """
          #{config.mod}.init error:
          #{Exception.format(type, error, stacktrace)}
          """
        end)

        :ok

      wat ->
        Logger.warn(fn ->
          "Unrecognized `&init/1` return from #{config.mod}: #{inspect(wat)}"
        end)

        :ok
    end
  end

  # Optimizes & executes calling the `&load/1` function for the given keys.
  # Blocks until refresh is finished.
  @spec refresh_for_keys(Config.t(), [key]) :: :ok
  defp refresh_for_keys(config, keys) do
    keys
    |> Enum.map(&spawn_refresher(&1, config))
    |> Enum.each(fn pid ->
      receive do
        ^pid -> nil
      end
    end)
  end

  defp spawn_refresher(key, config) do
    parent = self()
    opts = [filter_fn: &config.mod.refresh_filter/2]

    spawn(fn ->
      try do
        {:ok, _, _} = run_load(config, key, opts)
      rescue
        e ->
          Logger.error("""
          Error processing key #{inspect(key)}: #{inspect(e)}\
          """)
      end

      send(parent, self())
    end)
  end

  # Refresh and return a particular value; log errors
  @spec run_load(Config.t(), key, opts) ::
          {:ok, value, boolean} | {:error, String.t()}
  defp run_load(%{mod: mod, bucket: bucket} = config, key, opts) do
    filter_fn = Keyword.get(opts, :filter_fn, fn _k, v -> v end)

    case invoke_carefully({mod, :load, [key]}) do
      {:ok, val} ->
        Logger.debug(fn ->
          ins = inspect(val, limit: 2, printable_limit: 100)
          "Loaded #{bucket}[#{inspect(key)}]: #{ins}"
        end)

        unless Keyword.get(opts, :skip_save?), do: save_value(config, key, val)
        {:ok, val, true}

      {:ok, val, map} when is_map(map) ->
        unless Keyword.get(opts, :skip_save?),
          do: store_map_into_cache(config, map)

        {:ok, val, true}

      {:error, error} ->
        msg = "#{mod}.load(#{inspect(key)}) error: #{error}"
        Logger.error(msg)
        {:error, msg}

      {:caught, type, error, stacktrace} ->
        msg = """
        #{mod}.load(#{inspect(key)}) error:
        #{Exception.format(type, error, stacktrace)}
        """

        Logger.warn(msg)

        {:error, msg}

      woah ->
        msg = """
        Something invalid was returned from \
        #{mod}.load(#{inspect(key)}): #{inspect(woah)}\
        """

        Logger.warn(msg)

        {:error, msg}
    end
    |> (fn ret ->
          filter_fn.(key, ret)
        end).()
  end

  # Save a value to the bucket; log errors
  @spec save_value(module, key, value) :: :ok
  defp save_value(%{bucket: bucket}, key, val) do
    :ok = Bucket.set(bucket, key, val)
  end

  # Load an entire map into the cache
  @spec store_map_into_cache(Config.t(), map) :: :ok
  defp store_map_into_cache(%{bucket: bucket} = config, map) do
    Enum.each(Map.keys(map), fn key ->
      save_value(config, key, map[key])
    end)

    file_dump(bucket, config.file)

    Logger.debug(fn ->
      {:ok, keys} = Bucket.keys(bucket)
      "#{bucket}: Loaded cache with #{length(keys)} keys"
    end)

    :ok
  end

  # Dump an entire cache to disk log errors
  @spec file_dump(bucket, String.t()) :: :ok
  defp file_dump(_bucket, nil), do: :ok

  defp file_dump(bucket, file) do
    case Bucket.dump(bucket, file) do
      :ok ->
        Logger.debug(fn ->
          {:ok, keys} = Bucket.keys(bucket)
          "#{bucket}: Saved bucket to disk with #{length(keys)} keys"
        end)

        :ok

      {:error, msg} ->
        Logger.error("#{bucket}: Failed to dump bucket: #{inspect(msg)}")
        :ok
    end
  end

  @spec maybe_load_file(bucket, String.t() | nil) :: boolean
  defp maybe_load_file(bucket, file) when byte_size(file) > 0 do
    case Bucket.load(bucket, file) do
      :ok ->
        Logger.debug(fn ->
          {:ok, keys} = Bucket.keys(bucket)
          "#{bucket}: Loaded cache file with #{length(keys)} keys"
        end)

        true

      {:error, msg} ->
        Logger.warn("""
        #{bucket}: Failed loading cache file: #{file}: #{inspect(msg)}\
        """)

        false
    end
  end

  defp maybe_load_file(_, _), do: false

  # Get the module name for which this instance of CacheWorker is running
  @spec this_worker_module :: module
  defp this_worker_module do
    self() |> Process.info() |> Keyword.get(:registered_name)
  end

  defp schedule_full_refresh(interval)
       when is_number(interval) and interval > 0 do
    Process.send_after(self(), :full_refresh, Kernel.trunc(interval * 1_000))
  end

  defp schedule_full_refresh(_), do: nil
end