ram

%% ==========================================================================================================
%% Ram - An in-memory distributed KV store for Erlang and Elixir.
%%
%% The MIT License (MIT)
%%
%% Copyright (c) 2021 Roberto Ostinelli <roberto@ostinelli.net>.
%%
%% Permission is hereby granted, free of charge, to any person obtaining a copy
%% of this software and associated documentation files (the "Software"), to deal
%% in the Software without restriction, including without limitation the rights
%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%% copies of the Software, and to permit persons to whom the Software is
%% furnished to do so, subject to the following conditions:
%%
%% The above copyright notice and this permission notice shall be included in
%% all copies or substantial portions of the Software.
%%
%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
%% THE SOFTWARE.
%% ==========================================================================================================
-module(ram_kv).
-behaviour(gen_server).

%% API
-export([start_link/0]).
-export([get/1]).
-export([put/3]).
-export([delete/1]).

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

%% records
-record(state, {
    sync_requested = false :: boolean()
}).

%% includes
-include("ram.hrl").

%% ===================================================================
%% API
%% ===================================================================
-spec start_link() -> {ok, pid()} | {error, term()}.
start_link() ->
    Options = [],
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], Options).

-spec get(Key :: term()) -> Value :: term().
get(Key) ->
    global:trans({{?MODULE, Key}, self()},
        fun() ->
            case ets:lookup(?TABLE, Key) of
                [] -> {error, undefined};
                [{Key, Value, Version}] -> {ok, Value, Version}
            end
        end).

-spec put(Key :: term(), Value :: term(), Version :: term()) -> ok | {error, Reason :: term()}.
put(Key, Value, Version) ->
    global:trans({{?MODULE, Key}, self()},
        fun() ->
            gen_server:call(?MODULE, {put, Key, Value, Version})
        end).

-spec delete(Key :: term()) -> ok.
delete(Key) ->
    global:trans({{?MODULE, Key}, self()},
        fun() ->
            gen_server:call(?MODULE, {delete, Key})
        end).

%% ===================================================================
%% Callbacks
%% ===================================================================

%% ----------------------------------------------------------------------------------------------------------
%% Init
%% ----------------------------------------------------------------------------------------------------------
-spec init([]) ->
    {ok, #state{}} |
    {ok, #state{}, Timeout :: non_neg_integer()} |
    ignore |
    {stop, Reason :: term()}.
init([]) ->
    %% monitor nodes
    ok = net_kernel:monitor_nodes(true),
    %% empty local database
    true = ets:delete_all_objects(?TABLE),
    %% init
    {ok, #state{}}.

%% ----------------------------------------------------------------------------------------------------------
%% Call messages
%% ----------------------------------------------------------------------------------------------------------
-spec handle_call(Request :: term(), From :: term(), #state{}) ->
    {reply, Reply :: term(), #state{}} |
    {reply, Reply :: term(), #state{}, Timeout :: non_neg_integer()} |
    {noreply, #state{}} |
    {noreply, #state{}, Timeout :: non_neg_integer()} |
    {stop, Reason :: term(), Reply :: term(), #state{}} |
    {stop, Reason :: term(), #state{}}.
handle_call({put, Key, Value, Version}, From, State) ->
    VersionMatch = case ets:lookup(?TABLE, Key) of
        [] ->
            case Version of
                undefined -> ok;
                _ -> {error, deleted}
            end;
        [{_, _, Version}] ->
            ok;

        _ ->
            {error, outdated}
    end,
    case VersionMatch of
        ok ->
            spawn(fun() ->
                Nodes = nodes(),
                Version1 = generate_id(),
                %% send
                lists:foreach(fun(RemoteNode) ->
                    {?MODULE, RemoteNode} ! {'1.0', self(), put, Key, Value, Version1}
                end, Nodes),
                %% wait for confirmation
                receive_put_ack(Key, Value, Version1, Nodes),
                %% TODO: rollback on timeout / errors
                %% insert
                true = ets:insert(?TABLE, {Key, Value, Version1}),
                %% reply
                gen_server:reply(From, {ok, Version1})
            end),
            %% return
            {noreply, State};

        {error, Reason} ->
            {reply, {error, Reason}, State}
    end;

handle_call({delete, Key}, From, State) ->
    case ets:lookup(?TABLE, Key) of
        [] ->
            {reply, {error, undefined}, State};

        [{_, _, _}] ->
            spawn(fun() ->
                Nodes = nodes(),
                %% send
                lists:foreach(fun(RemoteNode) ->
                    {?MODULE, RemoteNode} ! {'1.0', self(), delete, Key}
                end, Nodes),
                %% wait for confirmation
                receive_delete_ack(Key, Nodes),
                %% TODO: rollback on timeout / errors
                %% delete
                true = ets:delete(?TABLE, Key),
                %% reply
                gen_server:reply(From, ok)
            end),
            %% return
            {noreply, State}
    end;

handle_call(Request, From, State) ->
    error_logger:warning_msg("RAM[~s] Received from ~p an unknown call message: ~p", [node(), From, Request]),
    {reply, undefined, State}.

%% ----------------------------------------------------------------------------------------------------------
%% Cast messages
%% ----------------------------------------------------------------------------------------------------------
-spec handle_cast(Msg :: term(), #state{}) ->
    {noreply, #state{}} |
    {noreply, #state{}, Timeout :: non_neg_integer()} |
    {stop, Reason :: term(), #state{}}.
handle_cast(Msg, State) ->
    error_logger:warning_msg("RAM[~s] Received an unknown cast message: ~p", [node(), Msg]),
    {noreply, State}.

%% ----------------------------------------------------------------------------------------------------------
%% All non Call / Cast messages
%% ----------------------------------------------------------------------------------------------------------
-spec handle_info(Info :: term(), #state{}) ->
    {noreply, #state{}} |
    {noreply, #state{}, Timeout :: non_neg_integer()} |
    {stop, Reason :: term(), #state{}}.
handle_info({'1.0', RemotePid, put, Key, Value, Version}, State) ->
    %% insert
    true = ets:insert(?TABLE, {Key, Value, Version}),
    %% reply
    RemotePid ! {self(), put, Key, Value, Version},
    %% return
    {noreply, State};

handle_info({'1.0', RemotePid, delete, Key}, State) ->
    %% delete
    true = ets:delete(?TABLE, Key),
    %% reply
    RemotePid ! {self(), delete, Key},
    %% return
    {noreply, State};

handle_info({nodedown, RemoteNode}, State) ->
    error_logger:info_msg("RAM[~s] Node ~s left the cluster", [node(), RemoteNode]),
    {noreply, State};

handle_info({nodeup, RemoteNode}, State) ->
    error_logger:info_msg("RAM[~s] Node ~s joined the cluster", [node(), RemoteNode]),
    %% send syn
    {?MODULE, RemoteNode} ! {'1.0', self(), syn},
    %% return
    {noreply, State};

handle_info({'1.0', RemotePid, syn}, State) ->
    error_logger:info_msg("RAM[~s] Received SYN from node ~s", [node(), node(RemotePid)]),
    %% reply
    RemotePid ! {'1.0', self(), ack},
    %% return
    {noreply, State};

handle_info({'1.0', RemotePid, ack}, #state{sync_requested = false} = State) ->
    error_logger:info_msg("RAM[~s] Received ACK from node ~s, sending SYNC_REQ", [node(), node(RemotePid)]),
    %% request data
    RemotePid ! {'1.0', self(), sync_req},
    %% return
    {noreply, State#state{sync_requested = true}};

handle_info({'1.0', RemotePid, sync_req}, State) ->
    error_logger:info_msg("RAM[~s] Received SYNC_REQ from node ~s", [node(), node(RemotePid)]),
    %% send local data
    LocalData = ets:tab2list(?TABLE),
    RemotePid ! {'1.0', self(), sync, LocalData},
    %% return
    {noreply, State};

handle_info({'1.0', RemotePid, sync, RemoteData}, State) ->
    error_logger:info_msg("RAM[~s] Received SYNC (~w entries) from node ~s", [node(), length(RemoteData), node(RemotePid)]),
    %% store data
    merge(RemoteData),
    %% return
    {noreply, State};

handle_info(Info, State) ->
    error_logger:warning_msg("RAM[~s] Received an unknown info message: ~p", [node(), Info]),
    {noreply, State}.

%% ----------------------------------------------------------------------------------------------------------
%% Terminate
%% ----------------------------------------------------------------------------------------------------------
-spec terminate(Reason :: term(), #state{}) -> terminated.
terminate(Reason, _State) ->
    error_logger:info_msg("RAM[~s] Terminating with reason: ~p", [node(), Reason]),
    %% return
    terminated.

%% ----------------------------------------------------------------------------------------------------------
%% Convert process state when code is changed.
%% ----------------------------------------------------------------------------------------------------------
-spec code_change(OldVsn :: term(), #state{}, Extra :: term()) -> {ok, #state{}}.
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% ===================================================================
%% Internal
%% ===================================================================
-spec receive_put_ack(Key :: term(), Value :: term(), Version :: term(), Nodes :: [node()]) -> ok.
receive_put_ack(_Key, _Value, _Version, []) -> ok;
receive_put_ack(Key, Value, Version, Nodes) ->
    receive
        {Pid, put, Key, Value, Version} ->
            receive_put_ack(Key, Value, Version, lists:delete(node(Pid), Nodes))
    end.

-spec receive_delete_ack(Key :: term(), Nodes :: [node()]) -> ok.
receive_delete_ack(_Key, []) -> ok;
receive_delete_ack(Key, Nodes) ->
    receive
        {Pid, delete, Key} ->
            receive_delete_ack(Key, lists:delete(node(Pid), Nodes))
    end.

-spec generate_id() -> binary().
generate_id() ->
    binary:encode_hex(crypto:hash(sha256, erlang:term_to_binary({node(), erlang:system_time()}))).

-spec merge(RemoteData :: [ram_entry()]) -> any().
merge(RemoteData) ->
    %% TODO: loop for conflicts
    true = ets:insert(?TABLE, RemoteData).