  This Source Code Form is subject to the terms of the Mozilla Public
  License , v. 2.0 . If a copy of the MPL was not distributed with this
  file , You can obtain one at /.
%%
  Copyright ( c ) 2017 - 2020 VMware , Inc. or its affiliates .   All rights reserved .
%%
%% @hidden
-module(ra_log_meta).
-behaviour(gen_batch_server).

-export([start_link/1,
         init/1,
         handle_batch/2,
         terminate/2,
         format_status/1,
         store/3,
         store_sync/3,
         delete/1,
         delete_sync/1,
         fetch/2,
         fetch/3
        ]).

-include("ra.hrl").

%% centralised meta data storage server for ra servers.

-type key() :: current_term | voted_for | last_applied.
-type value() :: non_neg_integer() | atom() | {atom(), atom()}.

-define(TBL_NAME, ?MODULE).
-define(TIMEOUT, 30000).
-define(SYNC_INTERVAL, 5).

-record(?MODULE, {ref :: reference()}).

-opaque state() :: #?MODULE{}.

-export_type([state/0]).

-spec start_link(Config :: map()) ->
    {ok, pid()} | {error, {already_started, pid()}}.
start_link(Config) ->
    gen_batch_server:start_link({local, ?MODULE}, ?MODULE, Config, []).

-spec init(file:filename()) -> {ok, state()}.
init(Dir) ->
    process_flag(trap_exit, true),
    ok = ra_lib:make_dir(Dir),
    MetaFile = filename:join(Dir, "meta.dets"),
    {ok, Ref} = dets:open_file(?TBL_NAME, [{file, MetaFile},
                                           {auto_save, ?SYNC_INTERVAL}]),
    _ = ets:new(?TBL_NAME, [named_table, public, {read_concurrency, true}]),
    ?TBL_NAME = dets:to_ets(?TBL_NAME, ?TBL_NAME),
    ?INFO("ra: meta data store initialised. ~b record(s) recovered",
          [ets:info(?TBL_NAME, size)]),
    {ok, #?MODULE{ref = Ref}}.

handle_batch(Commands, #?MODULE{ref = Ref} = State) ->
    DoInsert =
        fun (Id, Key, Value, Inserts0) ->
                case Inserts0 of
                    #{Id := Data} ->
                        Inserts0#{Id => update_key(Key, Value, Data)};
                    _ ->
                        case dets:lookup(Ref, Id) of
                            [{Id, T, V, A}] ->
                                Data = {Id, T, V, A},
                                Inserts0#{Id => update_key(Key, Value, Data)};
                            [] ->
                                Data = {Id, undefined, undefined, undefined},
                                Inserts0#{Id => update_key(Key, Value, Data)}
                        end
                end
        end,
    {Inserts, Replies, ShouldSync} =
        lists:foldl(
          fun ({cast, {store, Id, Key, Value}},
               {Inserts0, Replies, DoSync}) ->
                  {DoInsert(Id, Key, Value, Inserts0), Replies, DoSync};
              ({call, From, {store, Id, Key, Value}},
               {Inserts0, Replies, _DoSync}) ->
                  {DoInsert(Id, Key, Value, Inserts0),
                   [{reply, From, ok} | Replies], true};
              ({cast, {delete, Id}},
               {Inserts0, Replies, DoSync}) ->
                  {handle_delete(Id, Ref, Inserts0), Replies, DoSync};
              ({call, From, {delete, Id}},
               {Inserts0, Replies, _DoSync}) ->
                  {handle_delete(Id, Ref, Inserts0),
                   [{reply, From, ok} | Replies], true}
          end, {#{}, [], false}, Commands),
    Objects = maps:values(Inserts),
    ok = dets:insert(?MODULE, Objects),
    true = ets:insert(?MODULE, Objects),
    case ShouldSync of
        true ->
            ok = dets:sync(?MODULE);
        false ->
            ok
    end,
    {ok, Replies, State}.

terminate(_, #?MODULE{ref = Ref}) ->
    ok = dets:sync(?MODULE),
    _ = dets:close(Ref),
    ok.

format_status(State) ->
    State.

%% send a message to the meta data store using cast
-spec store(ra_uid(), key(), value()) -> ok.
store(UId, Key, Value) ->
    gen_batch_server:cast(?MODULE, {store, UId, Key, Value}).

%% waits until batch has been processed and synced.
%% when it returns the store request has been safely flushed to disk
-spec store_sync(ra_uid(), key(), value()) -> ok.
store_sync(UId, Key, Value) ->
    gen_batch_server:call(?MODULE, {store, UId, Key, Value}, ?TIMEOUT).

-spec delete(ra_uid()) -> ok.
delete(UId) ->
    gen_batch_server:cast(?MODULE, {delete, UId}).

-spec delete_sync(ra_uid()) -> ok.
delete_sync(UId) ->
    gen_batch_server:call(?MODULE, {delete, UId}, ?TIMEOUT).

%% READER API

-spec fetch(ra_uid(), key()) -> value() | undefined.
fetch(Id, current_term) ->
    maybe_fetch(Id, 2);
fetch(Id, voted_for) ->
    maybe_fetch(Id, 3);
fetch(Id, last_applied) ->
    maybe_fetch(Id, 4).

-spec fetch(ra_uid(), key(), term()) -> value().
fetch(Id, Key, Default) ->
    case fetch(Id, Key) of
        undefined -> Default;
        Value -> Value
    end.

%%% internal

maybe_fetch(Id, Pos) ->
    try ets:lookup_element(?TBL_NAME, Id, Pos) of
        E -> E
    catch
        _:badarg ->
            undefined
    end.

handle_delete(Id, Ref, Inserts) ->
  _ = dets:delete(Ref, Id),
  _ = ets:delete(?MODULE, Id),
  maps:remove(Id, Inserts).

update_key(current_term, Value, Data) ->
    case element(2, Data) of
        %% current term matches the new value, nothing to do
        Value -> Data;
        %% current term has changed. Clear voted_for field as part of the update.
        %% See rabbitmq/ra#111.
        _     ->
          Data1 = setelement(3, Data, undefined),
          setelement(2, Data1, Value)
    end;
update_key(voted_for, Value, Data) ->
    setelement(3, Data, Value);
update_key(last_applied, Value, Data) ->
    setelement(4, Data, Value).



 @hidden
 centralised meta data storage server for ra servers.
 send a message to the meta data store using cast
 waits until batch has been processed and synced.
 when it returns the store request has been safely flushed to disk
 READER API
 internal
 current term matches the new value, nothing to do
 current term has changed. Clear voted_for field as part of the update.
 See rabbitmq/ra#111.
  This Source Code Form is subject to the terms of the Mozilla Public
  License , v. 2.0 . If a copy of the MPL was not distributed with this
  file , You can obtain one at /.
  Copyright ( c ) 2017 - 2020 VMware , Inc. or its affiliates .   All rights reserved .
-module(ra_log_meta).
-behaviour(gen_batch_server).

-export([start_link/1,
         init/1,
         handle_batch/2,
         terminate/2,
         format_status/1,
         store/3,
         store_sync/3,
         delete/1,
         delete_sync/1,
         fetch/2,
         fetch/3
        ]).

-include("ra.hrl").


-type key() :: current_term | voted_for | last_applied.
-type value() :: non_neg_integer() | atom() | {atom(), atom()}.

-define(TBL_NAME, ?MODULE).
-define(TIMEOUT, 30000).
-define(SYNC_INTERVAL, 5).

-record(?MODULE, {ref :: reference()}).

-opaque state() :: #?MODULE{}.

-export_type([state/0]).

-spec start_link(Config :: map()) ->
    {ok, pid()} | {error, {already_started, pid()}}.
start_link(Config) ->
    gen_batch_server:start_link({local, ?MODULE}, ?MODULE, Config, []).

-spec init(file:filename()) -> {ok, state()}.
init(Dir) ->
    process_flag(trap_exit, true),
    ok = ra_lib:make_dir(Dir),
    MetaFile = filename:join(Dir, "meta.dets"),
    {ok, Ref} = dets:open_file(?TBL_NAME, [{file, MetaFile},
                                           {auto_save, ?SYNC_INTERVAL}]),
    _ = ets:new(?TBL_NAME, [named_table, public, {read_concurrency, true}]),
    ?TBL_NAME = dets:to_ets(?TBL_NAME, ?TBL_NAME),
    ?INFO("ra: meta data store initialised. ~b record(s) recovered",
          [ets:info(?TBL_NAME, size)]),
    {ok, #?MODULE{ref = Ref}}.

handle_batch(Commands, #?MODULE{ref = Ref} = State) ->
    DoInsert =
        fun (Id, Key, Value, Inserts0) ->
                case Inserts0 of
                    #{Id := Data} ->
                        Inserts0#{Id => update_key(Key, Value, Data)};
                    _ ->
                        case dets:lookup(Ref, Id) of
                            [{Id, T, V, A}] ->
                                Data = {Id, T, V, A},
                                Inserts0#{Id => update_key(Key, Value, Data)};
                            [] ->
                                Data = {Id, undefined, undefined, undefined},
                                Inserts0#{Id => update_key(Key, Value, Data)}
                        end
                end
        end,
    {Inserts, Replies, ShouldSync} =
        lists:foldl(
          fun ({cast, {store, Id, Key, Value}},
               {Inserts0, Replies, DoSync}) ->
                  {DoInsert(Id, Key, Value, Inserts0), Replies, DoSync};
              ({call, From, {store, Id, Key, Value}},
               {Inserts0, Replies, _DoSync}) ->
                  {DoInsert(Id, Key, Value, Inserts0),
                   [{reply, From, ok} | Replies], true};
              ({cast, {delete, Id}},
               {Inserts0, Replies, DoSync}) ->
                  {handle_delete(Id, Ref, Inserts0), Replies, DoSync};
              ({call, From, {delete, Id}},
               {Inserts0, Replies, _DoSync}) ->
                  {handle_delete(Id, Ref, Inserts0),
                   [{reply, From, ok} | Replies], true}
          end, {#{}, [], false}, Commands),
    Objects = maps:values(Inserts),
    ok = dets:insert(?MODULE, Objects),
    true = ets:insert(?MODULE, Objects),
    case ShouldSync of
        true ->
            ok = dets:sync(?MODULE);
        false ->
            ok
    end,
    {ok, Replies, State}.

terminate(_, #?MODULE{ref = Ref}) ->
    ok = dets:sync(?MODULE),
    _ = dets:close(Ref),
    ok.

format_status(State) ->
    State.

-spec store(ra_uid(), key(), value()) -> ok.
store(UId, Key, Value) ->
    gen_batch_server:cast(?MODULE, {store, UId, Key, Value}).

-spec store_sync(ra_uid(), key(), value()) -> ok.
store_sync(UId, Key, Value) ->
    gen_batch_server:call(?MODULE, {store, UId, Key, Value}, ?TIMEOUT).

-spec delete(ra_uid()) -> ok.
delete(UId) ->
    gen_batch_server:cast(?MODULE, {delete, UId}).

-spec delete_sync(ra_uid()) -> ok.
delete_sync(UId) ->
    gen_batch_server:call(?MODULE, {delete, UId}, ?TIMEOUT).


-spec fetch(ra_uid(), key()) -> value() | undefined.
fetch(Id, current_term) ->
    maybe_fetch(Id, 2);
fetch(Id, voted_for) ->
    maybe_fetch(Id, 3);
fetch(Id, last_applied) ->
    maybe_fetch(Id, 4).

-spec fetch(ra_uid(), key(), term()) -> value().
fetch(Id, Key, Default) ->
    case fetch(Id, Key) of
        undefined -> Default;
        Value -> Value
    end.


maybe_fetch(Id, Pos) ->
    try ets:lookup_element(?TBL_NAME, Id, Pos) of
        E -> E
    catch
        _:badarg ->
            undefined
    end.

handle_delete(Id, Ref, Inserts) ->
  _ = dets:delete(Ref, Id),
  _ = ets:delete(?MODULE, Id),
  maps:remove(Id, Inserts).

update_key(current_term, Value, Data) ->
    case element(2, Data) of
        Value -> Data;
        _     ->
          Data1 = setelement(3, Data, undefined),
          setelement(2, Data1, Value)
    end;
update_key(voted_for, Value, Data) ->
    setelement(3, Data, Value);
update_key(last_applied, Value, Data) ->
    setelement(4, Data, Value).
