memstore

%% Copyright (c) 2017-present. Benoit Chesneau
%%
%% This source code is licensed under the MIT license found in the
%% LICENSE file in the root directory of this source tree.

-module(memstore_versions).
-author("benoitc").

%% API
-export([
  start_link/2
]).

-export([
  last_sequence/1
]).

%% gen_server callbacks
-export([
  init/1,
  handle_call/3,
  handle_cast/2,
  handle_info/2,
  terminate/2,
  code_change/3
]).

-include_lib("stdlib/include/ms_transform.hrl").


start_link(LocalName, Name) ->
  gen_server:start_link({local, LocalName}, ?MODULE, [Name], []).

init([Name]) ->
  {ok, #{  name => Name}}.

handle_call(new_snapshot, _From, State = #{ name := Name }) ->
  %% create a new snapshot
  Snapshot = do_new_snapshot(Name),
  {reply, Snapshot, State};

handle_call({release_snapshot, Snapshot}, _From, State = #{ name := Name }) ->
  _ = do_release_snapshot(Snapshot, Name),
  {reply, ok, State};

handle_call(get_last_seq, _From, State = #{ name := Name }) ->
  {reply, last_sequence(Name), State};

handle_call(_Msg, _From, State) ->
  {reply, bad_call, State}.

handle_cast(_Msg, State) ->
  {noreply, State}.

handle_info(_Info, State) ->
  {noteply, State}.


terminate(_Reason, _Config) -> ok.

code_change(_Vsn, State, _Extra) -> {ok, State}.


%% ----------------------
%% internals

last_sequence(Name) ->
  ets:update_counter(Name, last_seq, {2, 0}).

incr_ref(Name, Seq) ->
  try ets:update_counter(Name, {lock, Seq}, {2, 1})
  catch
    error:badarg ->
      true = ets:insert(Name, {{lock, Seq}, 0}),
      0
  end.

decr_ref(Name, Seq) ->
  try decr_ref1(Name, Seq)
  catch
    error:badarg ->
      0
  end.

decr_ref1(Name, Seq) ->
  case ets:update_counter(Name, {lock, Seq}, {2, -1}) of
    0 ->
      ets:delete(Name, {lock, Seq}),
      0;
    Count ->
      Count
  end.

do_new_snapshot(Name) ->
  Seq = last_sequence(Name),
  SnapshotId = erlang:make_ref(),
  ets:insert(Name, {{snapshot, SnapshotId}, Seq}),
  %% increase the number of sequences owners
  _ = incr_ref(Name, Seq),
  #{ name => Name, snapshot_id => SnapshotId, seq => Seq}.

do_release_snapshot(#{ snapshot_id := SnapshotId}, Name) ->
  case ets:take(Name, {snapshot, SnapshotId}) of
    [] -> ok;
    [{{snapshot, SnapshotId}, Seq}] ->
      _ = decr_ref(Name, Seq),
      ok
  end.