globals

defmodule Globals do
  use GenServer
  require Logger

  @moduledoc """
  Globals is a module that provides ETS backed global varaibles with PubSub capabilities.

  How this is different from other PubSub implementations:

  - Support PubSub patterns and value storage together
  - keys can be any() term such as atoms, strings, tuples, etc.
  - Application starts `Globals` automatically no additional Supervision tree entry required
  - Support for `depends_on` for variable dependencies
  - `await` and `update_await` for awaiting async updates
  """

  defstruct [
    # subscribers is a map of {key => {pid => monitor_ref}}
    subscribers: %{},
    # refreshers is a map of {key => fun}
    refreshers: %{},
    # dependencies is a map of {key => [key]}
    dependencies: %{},
    # monitors is a map of {pid => {monitor_ref, [key]}}
    monitors: %{},
    # waiting is a map of {key => [{generation, pid, timer}]}
    waiting: %{},
    # Generation tracks update_await() call origin times. So that a update_await() call
    # can only be responded to using an update that has be started at or after the update_await() call,
    # but not from an previous update that is still running.
    generation: 0
  ]

  @timeout 30_000

  def start_link(_args) do
    GenServer.start_link(__MODULE__, [], name: __MODULE__, hibernate_after: 5000)
  end

  @impl true
  def init([]) do
    :ets.new(__MODULE__, [:public, :named_table, read_concurrency: true])
    {:ok, %Globals{}}
  end

  @doc """
  get returns the value of the key

  - key is the name of the key and can be any() term
  - default is the default value to return if the key is not found

  Example:

  iex> Globals.set(:key11, "value")
  iex> Globals.get(:key11)
  "value"

  iex> Globals.get(:key12)
  nil

  iex> Globals.get(:key13, "default")
  "default"
  """
  def get(key, default \\ nil) do
    case :ets.lookup(__MODULE__, key) do
      [] -> default
      [{_, value}] -> value
    end
  end

  @doc """
  get_lazy returns the value of the key or calls the function to get the value if the key is not found

  Example:

  iex> Globals.get_lazy(:key14, fn -> "value" end)
  "value"
  iex> Globals.set(:key14, false)
  false
  iex> Globals.get_lazy(:key14, fn -> "value" end)
  false
  iex> Globals.set(:key14, nil)
  nil
  iex> Globals.get_lazy(:key14, fn -> "value" end)
  "value"
  """
  def get_lazy(key, fun) do
    case get(key) do
      nil -> fun.()
      value -> value
    end
  end

  @doc """
    get_g returns the value of the key and the generation

    The generation is a monotonically increasing number that is incremented every time a value is updated.
    If g has changed then there was a change to the value.
  """
  def get_g(key, default \\ nil) do
    call({:get_g, key, default})
  end

  def update_await(key, opts \\ []) do
    timeout = Keyword.get(opts, :timeout)
    call({:update_await, key}, timeout)
  end

  @doc """
  await is similiar to get() but will block for timeout until another process
  defines the key or the timeout expires.

  Example:

  iex> Globals.set(:key, "value")
  iex> Globals.await(:key)
  "value"

  iex> Globals.update(:key2, fn ->
  ...>   Process.sleep(100)
  ...>   "value2"
  ...> end)
  iex> Globals.await(:key2)
  "value2"
  """
  def await(key, timeout \\ :infinity) do
    get_lazy(key, fn -> call({:await, key}, timeout) end)
  end

  @doc """
    pop returns the value of the key and deletes the key

    Example:

    iex> Globals.set(:key, "value")
    iex> Globals.pop(:key)
    "value"
    iex> Globals.get(:key)
    nil
  """
  def pop(key, default \\ nil) do
    case get(key) do
      nil ->
        default

      value ->
        :ets.delete(__MODULE__, key)
        GenServer.cast(__MODULE__, :pop)
        value
    end
  end

  @doc """
    incr increments the value of the key and returns the previous value

    Example:

    iex> Globals.incr(:key)
    0
    iex> Globals.incr(:key)
    1
    iex> Globals.get(:key)
    2
  """
  def incr(key, default \\ 0) when is_number(default) do
    call({:incr, key, default})
  end

  @doc """
    put_new sets the new value and returns true if the key was not already set

    Example:

    iex> Globals.put_new(:key5, "value")
    true

    iex> Globals.put_new(:key6, "value1")
    true
    iex> Globals.put_new(:key6, "value2")
    false
    iex> Globals.get(:key6)
    "value1"
  """
  def put_new(key, value) do
    if :ets.insert_new(__MODULE__, {key, value}) do
      GenServer.cast(__MODULE__, {:put, key, nil, value, nil})
      true
    else
      false
    end
  end

  @doc """
    put sets the new value and returns the OLD value

    Example:

    iex> Globals.set(:key, "value")
    iex> Globals.put(:key, "new_value")
    "value"
  """
  def put(key, value, src_gen \\ nil) do
    old = get(key, nil)
    :ets.insert(__MODULE__, {key, value})
    GenServer.cast(__MODULE__, {:put, key, old, value, src_gen})
    old
  end

  @doc """
    pushes a new value to all subscribers without setting it
    if a value is already set, it is deleted

    returns the value

    Example:

    iex> Globals.set(:key, "value")
    iex> Globals.push(:key, "new_value")
    "new_value"
    iex> Globals.get(:key)
    nil
  """
  def push(key, value) do
    call({:push, key, value})
  end

  @doc """
    set sets the new value and returns the NEW value

    Example:

    iex> Globals.set(:key, "old_value")
    "old_value"
    iex> Globals.set(:key, "new_value")
    "new_value"
  """
  def set(key, value) do
    put(key, value)
    value
  end

  @doc """
    update triggers an update to the key with the registered function in a background process

    Example:

    iex> Globals.register(:key, fn -> "new_value" end)
    iex> Globals.update_await(:key)
    iex> Globals.get(:key)
    "new_value"
  """
  def update(key) do
    Debouncer.immediate(
      {__MODULE__, key},
      fn ->
        {fun, gen} = call({:update, key})
        if fun != nil, do: do_update(key, fun, gen)
      end,
      update_timeout()
    )
  end

  @doc """
    update updates the key with the given function in a background process

    Example:

    iex> Globals.update(:key23, fn -> "new_value" end)
    iex> Process.sleep(100)
    iex> Globals.get(:key23)
    "new_value"
  """
  def update(key, fun) do
    Debouncer.immediate(
      {__MODULE__, key},
      fn -> do_update(key, fun) end,
      update_timeout()
    )
  end

  @doc """
    Update_timeout returns the debounce timeout for the update function. When
    calling `Globals.update()` or `Globals.update_await()` the function will be debounced for the timeout.

    Example:

    iex> Globals.update_timeout()
    1000
  """
  def update_timeout() do
    :persistent_term.get(:globals_update_timeout, 1000)
  end

  @doc """
    Set_update_timeout sets the debounce timeout for the update function.

    Example:

    iex> Globals.set_update_timeout(2000)
    :ok
  """
  def set_update_timeout(timeout) when is_integer(timeout) do
    :persistent_term.put(:globals_update_timeout, timeout)
  end

  defp do_update(key, fun, gen \\ nil) do
    {old, other_gen} = get_g(key)
    gen = gen || other_gen

    new =
      case fun do
        {module, fun, args} -> apply(module, fun, args)
        no_args when is_function(fun, 0) -> no_args.()
        one_arg when is_function(fun, 1) -> one_arg.(old)
      end

    old2 = put(key, new, gen)

    if old2 != old do
      Logger.warning("update conflict in #{inspect(key)}")
    end

    new
  end

  @doc """
    Register a key and function to be called when the key is updated.
    If the function is not provided, the key must be a tuple of {module, function, args}.

    - key is the name of the key and can be any() term
    - fun is the function to call to get the value, can be a tuple of {module, function, args}, a function of 0 arguments, or a function of 1 argument (the old value)
    - opts is a keyword list of options
    - opts.depends_on is a list of keys that the key depends on
    - opts.changes is a list of keys that the key changes
    - opts.await is a boolean to wait for the key to be updated

    Example:

    # Registering key with a function
    iex> Globals.register(:one, fn -> "one" end)
    iex> Globals.register(:time, fn -> System.os_time() end)
    iex> Globals.register(:time, {System, :os_time, []})

    # Registering where key == {module, function, args} and acts as update function as well:
    iex> Globals.register({System, :os_time, []})

    # Depends on and changes:
    iex> Globals.register(:a, fn -> "a" end)
    iex> Globals.register(:b, fn -> Globals.await(:a) <> "b" end)
    iex> Globals.await(:a)
    "a"
    iex> Globals.await(:b)
    "ab"
  """
  def register(key, fun \\ nil, opts \\ []) do
    {opts, fun} =
      if opts == [] and is_list(fun) do
        {fun, nil}
      else
        {opts, fun}
      end

    fun =
      if fun == nil do
        if not is_tuple(key) and tuple_size(key) != 3 do
          raise "Need to provide a function for key: #{inspect(key)} or key must be a tuple of {module, function, args}"
        end

        key
      else
        fun
      end

    if !call({:register, key, fun, opts}) do
      update(key, fun)
    end

    if Keyword.get(opts, :await, false) do
      await(key)
    end
  end

  @doc """
    subscribe subscribes to the given keys

    - key(s) can be a single key or a list of keys
    - whom is the pid that will receive the updates. defaults to self()

    The `whom` pid will receives updates in the form of `{:update, key, value}`

    Example:

    iex> Globals.subscribe(:key21)
    :ok
    iex> Globals.subscribe([:key22, :key23])
    :ok
    iex> Globals.push(:key21, "value")
    "value"
    iex> receive do
    ...> {:update, :key21, "value"} -> :ok
    ...> end
    :ok
  """
  def subscribe(keys, whom \\ self()) do
    List.wrap(keys)
    |> Enum.each(fn key -> GenServer.cast(__MODULE__, {:subscribe, whom, key}) end)

    :ok
  end

  @doc """
    subscribed? returns true if the given key is subscribed to by the given whom

    - key is the name of the key and can be any() term
    - whom is the pid that will receive the updates

    Example:

    iex> Globals.subscribe(:key24)
    iex> Globals.subscribed?(:key24)
    true
    iex> Globals.subscribed?(:key25)
    false
  """
  def subscribed?(key, whom \\ self()) do
    call({:subscribed?, key, whom})
  end

  @doc """
    unsubscribe unsubscribes from the given keys

    - key(s) can be a single key or a list of keys
    - whom is the pid that will receive the updates
  """
  def unsubscribe(keys, whom \\ self()) do
    call({:unsubscribe, List.wrap(keys), whom})
  end

  defp do_unsubscribe(g = %Globals{subscribers: subs, monitors: mons}, :all, whom) do
    {ref, remove_keys} = Map.get(mons, whom)
    Process.demonitor(ref)
    mons = Map.delete(mons, whom)

    subs = do_unsubscribe_key_subs(subs, Enum.to_list(remove_keys), whom)
    %Globals{g | subscribers: subs, monitors: mons}
  end

  defp do_unsubscribe(g = %Globals{subscribers: subs, monitors: mons}, remove_keys, whom) do
    {ref, keys} = Map.get(mons, whom)
    new_keys = Enum.reduce(remove_keys, keys, &MapSet.delete(&2, &1))

    mons =
      if MapSet.size(new_keys) == 0 do
        Process.demonitor(ref)
        Map.delete(mons, whom)
      else
        Map.put(mons, whom, new_keys)
      end

    subs = do_unsubscribe_key_subs(subs, Enum.to_list(remove_keys), whom)
    %Globals{g | subscribers: subs, monitors: mons}
  end

  defp do_unsubscribe_key_subs(subs, [], _whom) do
    subs
  end

  defp do_unsubscribe_key_subs(subs, [key | rest], whom) do
    {_ref, key_subs} = Map.get(subs, key, %{}) |> Map.pop(whom)

    if map_size(key_subs) == 0 do
      Map.delete(subs, key)
    else
      Map.put(subs, key, key_subs)
    end
    |> do_unsubscribe_key_subs(rest, whom)
  end

  @impl true
  def handle_cast({:put, key, old, value, src_gen}, g = %Globals{generation: generation}) do
    g = %Globals{g | generation: generation + 1}
    {:noreply, publish(g, key, value, old, src_gen)}
  end

  def handle_cast(:pop, g = %Globals{generation: generation}) do
    {:noreply, %Globals{g | generation: generation + 1}}
  end

  def handle_cast({:subscribe, whom, key}, g = %Globals{subscribers: subs, monitors: mons})
      when is_pid(whom) do
    {ref, keys} =
      case Map.get(mons, whom) do
        nil -> {Process.monitor(whom), MapSet.new([key])}
        {ref, keys} -> {ref, MapSet.put(keys, key)}
      end

    mons = Map.put(mons, whom, {ref, keys})

    key_subs =
      Map.get(subs, key, %{})
      |> Map.put_new(whom, ref)

    subs = Map.put(subs, key, key_subs)
    {:noreply, %Globals{g | subscribers: subs, monitors: mons}}
  end

  def handle_cast({:subscribe, mfa, key}, g = %Globals{subscribers: subs}) do
    key_subs =
      Map.get(subs, key, %{})
      |> Map.put_new(mfa, mfa)

    subs = Map.put(subs, key, key_subs)
    {:noreply, %Globals{g | subscribers: subs}}
  end

  @impl true
  def handle_call({:unsubscribe, keys, whom}, _from, g = %Globals{}) do
    {:reply, :ok, do_unsubscribe(g, keys, whom)}
  end

  def handle_call({:subscribed?, key, whom}, _from, g = %Globals{}) do
    {:reply, Map.has_key?(Map.get(g.subscribers, key, %{}), whom), g}
  end

  def handle_call(
        {:register, key, fun, opts},
        _from,
        g = %Globals{refreshers: refs, dependencies: deps}
      ) do
    depends_on = List.wrap(Keyword.get(opts, :depends_on, []))
    changes = List.wrap(Keyword.get(opts, :changes, []))

    deps =
      Enum.reduce(depends_on, deps, fn dep, deps ->
        Map.put(deps, dep, Map.get(deps, dep, []) ++ [key])
      end)

    deps =
      Enum.reduce(changes, deps, fn dst, deps ->
        Map.put(deps, key, Map.get(deps, key, []) ++ [dst])
      end)

    {:reply, Map.has_key?(refs, key),
     %Globals{g | dependencies: deps, refreshers: Map.put(refs, key, fun)}}
  end

  def handle_call(
        {:await, key},
        from,
        g = %Globals{waiting: waiting, refreshers: refs}
      ) do
    case get(key) do
      nil ->
        timer = Process.send_after(self(), {:timeout, key, from}, @timeout - 2000)

        waiting =
          Map.update(waiting, key, [{nil, from, timer}], fn pids ->
            [{nil, from, timer} | pids]
          end)

        if Map.has_key?(refs, key) do
          update(key)
        else
          Logger.info("Awaiting undefined key: #{inspect(key)}")
        end

        {:noreply, %Globals{g | waiting: waiting}}

      value ->
        {:reply, value, g}
    end
  end

  def handle_call({:incr, key, default}, _from, g = %Globals{generation: generation}) do
    prev_value =
      case get(key, default) do
        prev_value when is_number(prev_value) ->
          prev_value

        _other ->
          default
      end

    value = prev_value + 1
    :ets.insert(__MODULE__, {key, value})
    g = %Globals{g | generation: generation + 1}
    {:reply, prev_value, publish(g, key, value, nil, nil)}
  end

  def handle_call({:push, key, value}, _from, g = %Globals{generation: generation}) do
    :ets.delete(__MODULE__, key)
    g = %Globals{g | generation: generation + 1}
    {:reply, value, publish(g, key, value, nil, nil)}
  end

  def handle_call({:get_g, key, default}, _from, g = %Globals{generation: generation}) do
    value = get(key, default)
    {:reply, {value, generation}, g}
  end

  def handle_call({:update_await, key}, from, g = %Globals{waiting: waiting, generation: gen}) do
    timer = Process.send_after(self(), {:timeout, key, from}, @timeout - 2000)

    waiting =
      Map.update(waiting, key, [{gen, from, timer}], fn pids -> [{gen, from, timer} | pids] end)

    update(key)
    {:noreply, %Globals{g | waiting: waiting}}
  end

  def handle_call({:update, key}, _from, g = %Globals{refreshers: refs, generation: gen}) do
    {:reply, {Map.get(refs, key), gen}, g}
  end

  @impl true
  def handle_info({:DOWN, _ref, :process, pid, _reason}, g = %Globals{}) do
    {:noreply, do_unsubscribe(g, :all, pid)}
  end

  def handle_info({:timeout, key, from}, g = %Globals{waiting: waiting}) do
    froms = Map.get(waiting, key, [])
    zombie? = Enum.all?(froms, fn {_gen, f, _timer} -> f != from end)

    # Spawn a new process to handle the timeout reporting
    # this protects Globals from waiting for the worker() and stracktrace() calls
    spawn(fn ->
      pid = Debouncer.worker({__MODULE__, key})

      if pid == nil do
        Logger.error("Timeout waiting for #{inspect(key)} (zombie=#{zombie?}) - No worker found")
      else
        Logger.error("Timeout waiting for #{inspect(key)} (zombie=#{zombie?}) - \
          Worker stuck in: #{inspect(Profiler.stacktrace(pid))}")
      end
    end)

    {:noreply, g}
  end

  defp call(what, timeout \\ @timeout) do
    GenServer.call(__MODULE__, what, timeout || @timeout)
  end

  defp publish(
         g = %Globals{subscribers: subs, waiting: waiting, dependencies: deps},
         key,
         value,
         old,
         src_gen
       ) do
    if old == nil or value != old do
      pids = Map.keys(Map.get(subs, key, %{}))

      Debouncer.immediate(
        {__MODULE__, :publish, key, value},
        fn ->
          {pids, mfas} = Enum.split_with(pids, &is_pid/1)

          for pid <- pids do
            send(pid, {:update, key, value})
          end

          for fun <- mfas do
            case fun do
              {module, fun, args} -> apply(module, fun, args)
              no_args when is_function(fun, 0) -> no_args.()
              one_arg when is_function(fun, 1) -> one_arg.(value)
              two_arg when is_function(fun, 2) -> two_arg.(key, value)
            end
          end
        end,
        500
      )

      for dep_key <- Map.get(deps, key, []) do
        update(dep_key)
      end
    end

    list =
      Map.get(waiting, key, [])
      |> Enum.reject(fn {gen, from, timer} ->
        if src_gen == nil or gen == nil or gen <= src_gen do
          Process.cancel_timer(timer)
          GenServer.reply(from, value)
          true
        end
      end)

    waiting = if list == [], do: Map.delete(waiting, key), else: Map.put(waiting, key, list)
    %Globals{g | waiting: waiting}
  end

  @doc """
    Locked locks the given lockname and calls the given function with the lock held
    until the function returns.

    This is useful for ensuring that only one process can update the key at a time.

    - lockname is the name of the lock and can be any() term

    Example:

    iex> Globals.locked(:lockname, fn -> "new_value" end)
    "new_value"
  """
  def locked(lockname, fun) do
    :global.trans({lockname, self()}, fun)
  end

  @doc """
    caches the given key with the given function. If timeout is not provided, the value is cached for 5 seconds.

    - key is the name of the key and can be any() term
    - fun is the function to call to get the value
    - timeout is the timeout in milliseconds or :infinity to cache forever

    Example:

    iex> Globals.cache(:cache_key, fn -> "new_value" end, 1000)
    "new_value"
    iex> Globals.get(:cache_key)
    "new_value"
    iex> Process.sleep(2000)
    :ok
    iex> Globals.get(:cache_key)
    nil
  """
  def cache(key, fun, timeout \\ 5_000) do
    value =
      get_lazy(key, fn ->
        locked({__MODULE__, :cache, key}, fn ->
          get(key) || set(key, fun.())
        end)
      end)

    if timeout != :infinity do
      Debouncer.apply(
        {__MODULE__, :cache, key},
        fn ->
          pop(key)
        end,
        timeout
      )
    end

    value
  end
end