eventstore

defmodule EventStore.AdvisoryLocks do
  @moduledoc false

  # PostgreSQL provides a means for creating locks that have application-defined
  # meanings. Advisory locks are faster, avoid table bloat, and are
  # automatically cleaned up by the server at the end of the session.

  use GenServer

  defmodule State do
    @moduledoc false
    defstruct conn: nil, state: :connected, locks: %{}
  end

  defmodule Lock do
    @moduledoc false
    defstruct key: nil, opts: nil
  end

  alias EventStore.AdvisoryLocks.{Lock, State}
  alias EventStore.Storage

  def start_link(conn) do
    GenServer.start_link(__MODULE__, %State{conn: conn}, name: __MODULE__)
  end

  def init(%State{} = state) do
    {:ok, state}
  end

  @doc """
  Attempt to obtain an advisory lock.

     - `key` - an application specific integer to acquire a lock on.
     - `opts` an optional keyword list:
        - `lock_released` - a 0-arity function called when the lock is released
          (usually due to a lost database connection).
        - `lock_reacquired` - a 0-arity function called when the lock has been
          successfully reacquired.

  Returns `:ok` when lock successfully acquired, or
  `{:error, :lock_already_taken}` if the lock cannot be acquired immediately.

  """
  @spec try_advisory_lock(key :: non_neg_integer(), opts :: []) ::
          :ok | {:error, :lock_already_taken} | {:error, term}
  def try_advisory_lock(key, opts \\ []) when is_integer(key) do
    GenServer.call(__MODULE__, {:try_advisory_lock, key, self(), opts})
  end

  def disconnect do
    GenServer.cast(__MODULE__, :disconnect)
  end

  def reconnect do
    GenServer.cast(__MODULE__, :reconnect)
  end

  def handle_call({:try_advisory_lock, key, pid, opts}, _from, %State{} = state) do
    %State{conn: conn} = state

    case Storage.Lock.try_acquire_exclusive_lock(conn, key) do
      :ok ->
        {:reply, :ok, monitor_lock(key, pid, opts, state)}

      reply ->
        {:reply, reply, state}
    end
  end

  def handle_cast(:disconnect, %State{locks: locks} = state) do
    for {_ref, %Lock{opts: opts}} <- locks do
      :ok = notify(:lock_released, opts)
    end

    {:noreply, %State{state | state: :disconnected}}
  end

  def handle_cast(:reconnect, %State{} = state) do
    %State{conn: conn, locks: locks} = state

    for {_ref, %Lock{key: key, opts: opts}} <- locks do
      with :ok <- Storage.Lock.try_acquire_exclusive_lock(conn, key),
           :ok <- notify(:lock_reacquired, opts) do
        :ok
      else
        {:error, :lock_already_taken} -> :ok
      end
    end

    {:noreply, %State{state | state: :connected}}
  end

  defp notify(notification, opts) do
    case Keyword.get(opts, notification) do
      fun when is_function(fun, 0) ->
        apply(fun, [])

      _ ->
        :ok
    end
  end

  defp monitor_lock(key, pid, opts, %State{locks: locks} = state) do
    ref = Process.monitor(pid)

    %State{state | locks: Map.put(locks, ref, %Lock{key: key, opts: opts})}
  end

  def handle_info({:DOWN, ref, :process, _pid, _reason}, %State{locks: locks} = state) do
    state =
      case Map.get(locks, ref) do
        nil ->
          state

        %Lock{key: key} ->
          :ok = release_lock(key, state)

          %State{state | locks: Map.delete(locks, ref)}
      end

    {:noreply, state}
  end

  defp release_lock(key, %State{conn: conn, state: state}) do
    case state do
      :connected -> Storage.Lock.unlock(conn, key)
      _ -> :ok
    end
  end
end