mnkv

%%%=============================================================================
%%%
%%%               |  o __   _|  _  __  |_   _       _ _   (TM)
%%%               |_ | | | (_| (/_ | | |_) (_| |_| | | |
%%%
%%% @copyright (C) 2014, Lindenbaum GmbH
%%%
%%% Permission to use, copy, modify, and/or distribute this software for any
%%% purpose with or without fee is hereby granted, provided that the above
%%% copyright notice and this permission notice appear in all copies.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
%%% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
%%% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
%%% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
%%% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
%%% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
%%% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
%%%
%%% @doc
%%% Provides simple, Mnesia-based, distributed key value tables. When started,
%%% this application distributes Mnesia (and all `mnkv' tables) over all
%%% dynamically connected nodes. The Mnesia cluster can grow and shrink
%%% dynamically.
%%%
%%% All tables created, have key/value semantic (after all its still Mnesia).
%%% A new key-value-table can be created using {@link create/1}. The table will
%%% automatically be replicated to other nodes as new node connections are
%%% detected. Every connected node has read and write access to all Mnesia
%%% tables. If desired, it is possible to use the default Mnesia API to
%%% manipulate `mnkv' tables. However, `mnkv' uses vector clocks that need
%%% to be updated on every write to be able to use automatic netsplit recovery!
%%% Use the `#mnkv{}' record from the `mnkv.hrl' header file to match
%%% `mnkv' table entries.
%%%
%%% Every `mnkv' table uses vector clocks to keep track of the its entries.
%%% In case of new node connections or netsplits, `mnkv' will use these to
%%% merge the island itself without interaction. However, if there are diverged
%%% entries `mnkv' will look for a user defined callback to resolve the
%%% conflict. If no such callback can be found one of the conflicting nodes will
%%% be restarted!
%%%
%%% To be able to use `mnkv' none of the connected nodes is allowed to have
%%% `disk_copies' of its `schema' table, because Mnesia will fail to merge
%%% schemas on disk nodes (which means that it is likely they can't
%%% participate). If you need `disk_copies', you're on your own here. Do not
%%% mess with table replication and mnesia configuration changes yourself!
%%% There's a lot of black magic happening inside Mnesia and `mnkv' will do
%%% the necessary tricks and workarounds for you. At best you should avoid
%%% having tables created from outside `mnkv'. At least do not create tables
%%% with conflicting names.
%%% @end
%%%=============================================================================

-module(mnkv).

-behaviour(application).
-behaviour(supervisor).

%% API
-export([create/1,
         put/2, put/3,
         del/2,
         get/2, get/3,
         match_key/2, match_key/3,
         match/3, match/4,
         update/2, update/3,
         info/0]).

%% Application callbacks
-export([start/2, stop/1]).

%% supervisor callbacks
-export([init/1]).

-type table() :: atom().
-type key()   :: term().
%% Unfortunately, Mnesia is quite picky when it comes to allowed types for
%% keys, e.g. all special atoms of `match_specs' are not allowed and lead to
%% undefined behaviour when used.
-type value() :: term().
%% Unfortunately, Mnesia is quite picky when it comes to allowed types for
%% values, e.g. all special atoms of `match_specs' are not allowed and lead to
%% undefined behaviour when used.
-type version() :: mnkv_vclock:vclock().
%% A type describing the version of a table entry.

-type update_fun() :: fun((key(), {value, value()} | undefined) ->
                                 {value, value()} | term()).
%% The definition for a function passed to {@link update/2,3}. If there is no
%% mapping associated with a key the atom `undefined' is passed to the function,
%% otherwise the value will be provided as the tuple `{value, Value}'. To add
%% a not yet existing or change an existing mapping the function must return a
%% tuple of the similar form. Any other return value will delete the mapping
%% (if any).

-export_type([table/0, key/0, value/0, version/0, update_fun/0]).

-include("mnkv.hrl").

%%%=============================================================================
%%% Behaviour
%%%=============================================================================

-callback handle_conflict(key(), Local :: value(), Remote :: value()) ->
    {value, value()} | delete | term().
%% An optional callback that will be called on the node performing a table
%% merge (usually an arbitrary node) whenever an entry of table cannot be
%% merged automatically. The callback must be implemented in a module with the
%% same name as the respective table name, e.g. to handle conflicts for values
%% in the table `my_table' the module/function `my_table:handle_conflict/3' has
%% to be implemented.
%%
%% The function can resolve conflicts in several ways. It can provide a (new)
%% value for `Key' by returning `{value, Val}', it can delete all associations
%% for `Key' on all nodes by returning `delete' or it can ignore the
%% inconsistency by returning anything else or crash. When ignoring an
%% inconsistency the values for key will depend on the location of retrieval
%% until a new value gets written for `Key'.
%%
%% If an appropriate callback is not provided, the default conflict resolution
%% strategy is to __restart__ one of the conflicting node islands!

%%%=============================================================================
%%% API
%%%=============================================================================

%%------------------------------------------------------------------------------
%% @doc
%% Create a new key value table which will be replicated as RAM copy across all
%% nodes in the cluster. The table will only be created, if not yet existing.
%% This can be called multiple times (even) on the same node.
%%
%% The table will be ready for reads and writes when this function returns.
%% @end
%%------------------------------------------------------------------------------
-spec create(table()) -> ok | {error, term()}.
create(Table) ->
    case mnesia:create_table(Table, ?MNKV_TABLE_OPTS) of
        {atomic, ok} ->
            await_table(Table);
        {aborted, {already_exists, Table}} ->
            await_table(Table);
        {aborted, Reason} ->
            {error, Reason}
    end.

%%------------------------------------------------------------------------------
%% @doc
%% Puts a key value pair into a table returning the previous mappings. The
%% previous mapping will be overridden if existed.
%% @end
%%------------------------------------------------------------------------------
-spec put(table(), key(), value()) ->
                 {ok, [{key(), value()}]} | {error, term()}.
put(Table, Key, Value) -> ?MODULE:put(Table, [{Key, Value}]).

%%------------------------------------------------------------------------------
%% @doc
%% Puts multiple key value pairs into a table returning the previous mappings.
%% All previous mappings will be overridden.
%% @end
%%------------------------------------------------------------------------------
-spec put(table(), [{key(), value()}]) ->
                 {ok, [{key(), value()}]} | {error, term()}.
put(_Table, []) ->
    {ok, []};
put(Table, KeyValues) when is_list(KeyValues) ->
    do(fun() -> strip_l(r_and_w(Table, KeyValues)) end).

%%------------------------------------------------------------------------------
%% @doc
%% Deletes all values for the given key or list of keys from a table. Previous
%% values for the keys will be returned.
%% @end
%%------------------------------------------------------------------------------
-spec del(table(), key() | [key()]) ->
                 {ok, [{key(), value()}]} | {error, term()}.
del(_Table, [])       -> {ok, []};
del(Table, KeyOrKeys) -> do(fun() -> strip_l(r_and_d(Table, KeyOrKeys)) end).

%%------------------------------------------------------------------------------
%% @doc
%% Similar to {@link get/3} with `Type' set to transaction.
%% @end
%%------------------------------------------------------------------------------
-spec get(table(), key()) -> {ok, [{key(), value()}]} | {error, term()}.
get(Table, Key) -> get(Table, Key, transaction).

%%------------------------------------------------------------------------------
%% @doc
%% Retrieves the entry for the given key from a table. Specifying `dirty' will
%% issue a faster dirty read operation (no isolation/atomicity).
%% @end
%%------------------------------------------------------------------------------
-spec get(table(), key(), dirty | transaction) ->
                 {ok, [{key(), value()}]} | {error, term()}.
get(Table, Key, dirty)       -> dirty_r(Table, Key);
get(Table, Key, transaction) -> do(fun() -> strip_l(r(Table, Key, read)) end).

%%------------------------------------------------------------------------------
%% @doc
%% Similar to {@link match_key/3} with `Type' set to transaction.
%% @end
%%------------------------------------------------------------------------------
-spec match_key(table(), key()) -> {ok, [{key(), value()}]} | {error, term()}.
match_key(Table, KeySpec) -> match_key(Table, KeySpec, transaction).

%%------------------------------------------------------------------------------
%% @doc
%% Retrieves the entries that match the given key spec from a table. Specifying
%% `dirty' will issue a faster dirty select operation (no isolation/atomicity).
%% @end
%%------------------------------------------------------------------------------
-spec match_key(table(), key(), dirty | transaction) ->
                       {ok, [{key(), value()}]} | {error, term()}.
match_key(Table, KeySpec, Type) -> match(Table, KeySpec, '_', Type).

%%------------------------------------------------------------------------------
%% @doc
%% Similar to {@link match/4} with `Type' set to transaction.
%% @end
%%------------------------------------------------------------------------------
-spec match(table(), key(), value()) ->
                   {ok, [{key(), value()}]} | {error, term()}.
match(Table, KeySpec, ValueSpec) ->
    match(Table, KeySpec, ValueSpec, transaction).

%%------------------------------------------------------------------------------
%% @doc
%% Retrieves the entries that match the given key and value spec from a table.
%% Specifying `dirty' will issue a faster dirty select operation (no
%% isolation/atomicity).
%% @end
%%------------------------------------------------------------------------------
-spec match(table(), key(), value(), dirty | transaction) ->
                   {ok, [{key(), value()}]} | {error, term()}.
match(Table, KeySpec, ValueSpec, dirty) ->
    dirty_m(Table, KeySpec, ValueSpec);
match(Table, KeySpec, ValueSpec, transaction) ->
    do(fun() -> strip_l(m(Table, KeySpec, ValueSpec, read)) end).

%%------------------------------------------------------------------------------
%% @doc
%% Updates all mappings of a table. This function can be used to modify or
%% delete random mappings.
%%
%% `Fun' will be invoked consecutively for all table entries and will be invoked
%% exactly once per contained key. To modify a mapping simply return
%% `{value, NewVal}', to preserve the current mapping just return a tuple with
%% the old value. All other return values will cause the current mapping to be
%% deleted. {@link update/2} returns a list with all previous mappings.
%% @end
%%------------------------------------------------------------------------------
-spec update(table(), update_fun()) ->
                    {ok, [{key(), value()}], [{key(), value()}]} |
                    {error, term()}.
update(Table, Fun) when is_function(Fun) ->
    do(fun() -> strip_t(u(Table, Fun)) end).

%%------------------------------------------------------------------------------
%% @doc
%% Updates the mapping associated with `Key'. This function can be used to
%% modify, add or delete a mapping to `Key'.
%%
%% `Fun' will be called regardless whether a mapping currently exists or not.
%% The argument passed to the function is either `{value, Value}' or `undefined'
%% if there is no mapping at the moment.
%%
%% To add or change a mapping return `{value, NewValue}'. Returning this with
%% the old value will simply preserve the mapping. Returning anything else will
%% remove the mapping from the table. {@link update/3} returns a list with the
%% previous mappings as well as a list of current mappings for `Key'. The form
%% is `{ok, {OldMappings, NewMappings}}'
%% @end
%%------------------------------------------------------------------------------
-spec update(table(), key(), update_fun()) ->
                    {ok, {[{key(), value()}], [{key(), value()}]}} |
                    {error, term()}.
update(Table, Key, Fun) when is_function(Fun) ->
    do(fun() -> strip_t(u(Table, Fun, Key)) end).

%%------------------------------------------------------------------------------
%% @doc
%% Print information about the `mnkv' state to stdout.
%% @end
%%------------------------------------------------------------------------------
-spec info() -> ok.
info() -> mnkv_mon:info().

%%%=============================================================================
%%% Application callbacks
%%%=============================================================================

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
start(_StartType, _StartArgs) -> supervisor:start_link(?MODULE, []).

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
stop(_State) -> ok.

%%%=============================================================================
%%% supervisor callbacks
%%%=============================================================================

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
init([]) -> {ok, {{one_for_one, 5, 1}, [spec(mnkv_mon, [])]}}.

%%%=============================================================================
%%% internal functions
%%%=============================================================================

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
spec(M, As) -> {M, {M, start_link, As}, permanent, 1000, worker, [M]}.

%%------------------------------------------------------------------------------
%% @private
%% Blocks the calling process until a certain table is available to this node.
%%------------------------------------------------------------------------------
await_table(Table) ->
    Timeout = application:get_env(?MODULE, wait_timeout, 10000),
    case mnesia:wait_for_tables([Table], Timeout) of
        ok ->
            ok;
        {timeout, [Table]} ->
            {error, timeout};
        Error ->
            Error
    end.

%%------------------------------------------------------------------------------
%% @private
%% Spawns `Fun' in a mnesia transaction. If aborted, a transaction may be
%% retried depending on the values set in the application environment. A
%% transaction abort might take place when replica nodes go down/up. In this
%% case retries could be successful because of the automatic replica management.
%% Retries are off by default.
%%------------------------------------------------------------------------------
do(Fun) ->
    do(Fun,
       application:get_env(?MODULE, retry_timeout, 500),
       application:get_env(?MODULE, retries, 0)).
do(Fun, Timeout, Retries) ->
    case mnesia:transaction(Fun) of
        {atomic, Result} ->
            {ok, Result};
        {aborted, _} when Retries > 0 ->
            ok = timer:sleep(Timeout),
            do(Fun, Timeout, Retries - 1);
        {aborted, Reason} ->
            {error, Reason}
    end.

%%------------------------------------------------------------------------------
%% @private
%% Get every entry associated with `Key' from `Tab' in a dirty manner, no
%% transaction required.
%%------------------------------------------------------------------------------
-spec dirty_r(table(), key()) -> [{key(), value()}] | {error, term()}.
dirty_r(Tab, Key) -> dirty(Tab, Key, dirty_read).

%%------------------------------------------------------------------------------
%% @private
%% Select every entry matching the given key and value spec from `Tab' in a
%% dirty manner, no transaction required.
%%------------------------------------------------------------------------------
-spec dirty_m(table(), key(), value()) -> [{key(), value()}] | {error, term()}.
dirty_m(Tab, KeySpec, ValueSpec) ->
    dirty(Tab, m_spec(KeySpec, ValueSpec), dirty_select).

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
-spec dirty(table(), key(), dirty_read | dirty_select) ->
                   [{key(), value()}] | {error, term()}.
dirty(Tab, Key, Function) ->
    try mnesia:Function(Tab, Key) of
        Records -> {ok, strip_l(Records)}
    catch
        exit:{aborted, Reason} -> {error, Reason}
    end.

%%------------------------------------------------------------------------------
%% @private
%% Read `Key' from `Tab', only allowed within transaction context.
%%------------------------------------------------------------------------------
-spec r(table(), key(), read | write) -> [#mnkv{}].
r(Tab, Key, Lock) -> mnesia:read(Tab, Key, Lock).

%%------------------------------------------------------------------------------
%% @private
%% Read every entry matching the given key and value specs from `Tab', only
%% allowed within transaction
%% context.
%%------------------------------------------------------------------------------
-spec m(table(), key(), value(), read | write) -> [#mnkv{}].
m(Tab, KeySpec, ValueSpec, Lock) ->
    mnesia:select(Tab, m_spec(KeySpec, ValueSpec), Lock).

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
-spec m_spec(key(), value()) -> [tuple()].
m_spec(KeySpec, ValueSpec) ->
    [{#mnkv{key = KeySpec, val = ValueSpec, _ = '_'}, [], ['$_']}].

%%------------------------------------------------------------------------------
%% @private
%% Establish mapping `Key' to `Val'in `Tab', only allowed within transaction
%% context.
%%------------------------------------------------------------------------------
-spec w(table(), key(), value(), version()) -> ok.
w(Tab, Key, Val, Ver) ->
    NewVer = mnkv_vclock:increment(node(), Ver),
    mnesia:write(Tab, #mnkv{key = Key, val = Val, ver = NewVer}, write).

%%------------------------------------------------------------------------------
%% @private
%% Delete mapping `Key' to `Val' from `Tab' (if any), only allowed within
%% transaction context.
%%------------------------------------------------------------------------------
-spec d(table(), key(), read | write) -> ok.
d(Tab, Key, Lock) -> mnesia:delete(Tab, Key, Lock).

%%------------------------------------------------------------------------------
%% @private
%% Read all mappings for `Key' or `Keys' in `Tab', delete them and return the
%% previous mappings. Only allowed within transaction context.
%%------------------------------------------------------------------------------
-spec r_and_d(table(), key() | [key()]) -> [#mnkv{}].
r_and_d(Tab, Keys) when is_list(Keys) ->
    lists:append([r_and_d(Tab, Key) || Key <- Keys]);
r_and_d(Tab, Key) ->
    Records = r(Tab, Key, write),
    ok = d(Tab, Key, write),
    Records.

%%------------------------------------------------------------------------------
%% @private
%% Similar to {@link r_and_w/3} but operates on an input list.
%%------------------------------------------------------------------------------
-spec r_and_w(table(), [{key(), value()}]) -> [#mnkv{}].
r_and_w(Tab, KeyValues) ->
    lists:append([r_and_w(Tab, Key, Val) || {Key, Val} <- KeyValues]).

%%------------------------------------------------------------------------------
%% @private
%% Read all mappings for `Key' in `Tab', establish a new mapping from `Key' to
%% `Val' and return the previous mappings. Only allowed within transaction
%% context.
%%------------------------------------------------------------------------------
-spec r_and_w(table(), key(), value()) -> [#mnkv{}].
r_and_w(Tab, Key, Val) ->
    case r(Tab, Key, write) of
        Records = [#mnkv{key = Key, val = Val} | _] ->
            ok; %% no change, no write
        Records = [#mnkv{key = Key, ver = Ver} | _] ->
            ok = w(Tab, Key, Val, Ver);
        Records = [] ->
            ok = w(Tab, Key, Val, mnkv_vclock:fresh())
    end,
    Records.

%%------------------------------------------------------------------------------
%% @private
%% Update the mappings for all `Key's in a table. Only allowed within
%% transaction context.
%%------------------------------------------------------------------------------
-spec u(table(), update_fun()) -> {[#mnkv{}], [#mnkv{}]}.
u(Tab, Fun) ->
    {A, B} = lists:unzip([u(Tab, Fun, Key) || Key <- mnesia:all_keys(Tab)]),
    {lists:append(A), lists:append(B)}.

%%------------------------------------------------------------------------------
%% @private
%% Update the mapping for `Key'. Only allowed within transaction context.
%%------------------------------------------------------------------------------
-spec u(table(), update_fun(), key()) -> {[#mnkv{}], [#mnkv{}]}.
u(Tab, Fun, Key) ->
    case r(Tab, Key, write) of
        Old = [#mnkv{key = Key, val = Val, ver = Ver} | _] ->
            case Fun(Key, {value, Val}) of
                {value, Val} ->
                    ok; %% no change, no write
                {value, NewVal} ->
                    ok = d(Tab, Key, write),
                    ok = w(Tab, Key, NewVal, Ver);
                _ ->
                    ok = d(Tab, Key, write)
            end;
        Old = [] ->
            case Fun(Key, undefined) of
                {value, Val} ->
                    ok = w(Tab, Key, Val, mnkv_vclock:fresh());
                _ ->
                    ok %% delete on non-existent entry
            end
    end,
    {Old, r(Tab, Key, read)}.

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
-spec strip_l([#mnkv{}]) -> [{key(), value()}].
strip_l(Records) -> [{K, V} || #mnkv{key = K, val = V} <- Records].

%%------------------------------------------------------------------------------
%% @private
%%------------------------------------------------------------------------------
-spec strip_t({[#mnkv{}], [#mnkv{}]}) ->
                     {[{key(), value()}], [{key(), value()}]}.
strip_t({A, B}) -> {strip_l(A), strip_l(B)}.