Counter CRDT

counter.ex

defmodule Counter do
  use GenServer
  @announce_interval 5_000

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

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

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

  def value() do
    GenServer.call(__MODULE__, :value)
  end

  def state() do
    :sys.get_state(__MODULE__)
  end

  def init([]) do
    broadcast({:join, self()})
    Process.send_after(self(), :announce, @announce_interval)

    {:ok, %{incr: %{}, decr: %{}}}
  end

  def handle_call(:value, _from, counter) do
    incr = sum(counter.incr)
    decr = sum(counter.decr)

    {:reply, incr - decr, counter}
  end

  def handle_cast(:increment, counter) do
    value = Map.get(counter.incr, self(), 0)
    counter = put_in(counter.incr[self()], value + 1)
    broadcast({:sync, counter})

    {:noreply, counter}
  end

  def handle_cast(:decrement, counter) do
    value = Map.get(counter.decr, self(), 0)
    counter = put_in(counter.decr[self()], value + 1)
    broadcast({:sync, counter})

    {:noreply, counter}
  end

  def handle_info(:announce, counter) do
    broadcast({:sync, counter})
    Process.send_after(self(), :announce, @announce_interval)
    
    {:noreply, counter}
  end

  def handle_info({:join, pid}, counter) do
    send(pid, {:sync, counter})
    
    {:noreply, counter}
  end

  def handle_info({:sync, updates}, counter) do
    incr = merge(counter.incr, updates.incr)
    decr = merge(counter.decr, updates.decr)

    {:noreply, %{incr: incr, decr: decr}}
  end

  defp sum(counter) do
    counter
    |> Map.values()
    |> Enum.sum()
  end

  defp merge(counter1, counter2) do
    Map.merge(counter1, counter2, fn _key, a, b ->
      max(a, b)
    end)
  end

  defp broadcast(msg) do
    Node.list()
    |> Enum.each(fn node ->
      send({__MODULE__, node}, msg)
    end)
  end
end