riak_core_ng

%% -------------------------------------------------------------------
%%
%% Copyright (c) 2013 Basho Technologies, Inc.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------
-module(riak_core_metadata).

-export([get/2,
         get/3,
         fold/3,
         fold/4,
         to_list/1,
         to_list/2,
         iterator/1,
         iterator/2,
         itr_next/1,
         itr_close/1,
         itr_done/1,
         itr_key_values/1,
         itr_key/1,
         itr_values/1,
         itr_value/1,
         itr_default/1,
         prefix_hash/1,
         put/3,
         put/4,
         delete/2,
         delete/3]).

-include("riak_core_metadata.hrl").

-export_type([iterator/0]).

%% Get Option Types
-type get_opt_default_val() :: {default, metadata_value()}.
-type get_opt_resolver()    :: {resolver, metadata_resolver()}.
-type get_opt_allow_put()   :: {allow_put, boolean()}.
-type get_opt()             :: get_opt_default_val() | get_opt_resolver() | get_opt_allow_put().
-type get_opts()            :: [get_opt()].

%% Iterator Types
-type it_opt_resolver()     :: {resolver, metadata_resolver() | lww}.
-type it_opt_default_fun()  :: fun((metadata_key()) -> metadata_value()).
-type it_opt_default()      :: {default, metadata_value() | it_opt_default_fun()}.
-type it_opt_keymatch()     :: {match, term()}.
-type it_opt()              :: it_opt_resolver() | it_opt_default() | it_opt_keymatch().
-type it_opts()             :: [it_opt()].
-type fold_opts()           :: it_opts().
-type iterator()            :: {riak_core_metadata_manager:metadata_iterator(), it_opts()}.

%% Put Option Types
-type put_opts()            :: [].

%% Delete Option types
-type delete_opts()         :: [].

-define(TOMBSTONE, '$deleted').

%% @doc same as get(FullPrefix, Key, [])
-spec get(metadata_prefix(), metadata_key()) -> metadata_value() | undefined.
get(FullPrefix, Key) ->
    get(FullPrefix, Key, []).

%% @doc Retrieves the local value stored at the given prefix and key.
%%
%% get/3 can take the following options:
%%  * default: value to return if no value is found, `undefined' if not given.
%%  * resolver:  A function that resolves conflicts if they are encountered. If not given
%%               last-write-wins is used to resolve the conflicts
%%  * allow_put: whether or not to write and broadcast a resolved value. defaults to `true'.
%%
%% NOTE: an update will be broadcast if conflicts are resolved and
%% `allow_put' is `true'. any further conflicts generated by
%% concurrenct writes during resolution are not resolved
-spec get(metadata_prefix(), metadata_key(), get_opts()) -> metadata_value().
get({Prefix, SubPrefix}=FullPrefix, Key, Opts)
  when (is_binary(Prefix) orelse is_atom(Prefix)) andalso
       (is_binary(SubPrefix) orelse is_atom(SubPrefix)) ->
    PKey = prefixed_key(FullPrefix, Key),
    Default = get_option(default, Opts, undefined),
    ResolveMethod = get_option(resolver, Opts, lww),
    AllowPut = get_option(allow_put, Opts, true),
    case riak_core_metadata_manager:get(PKey) of
        undefined -> Default;
        Existing ->
            maybe_tombstone(maybe_resolve(PKey, Existing, ResolveMethod, AllowPut), Default)
    end.

%% @doc same as fold(Fun, Acc0, FullPrefix, []).
-spec fold(fun(({metadata_key(),
                 [metadata_value() | metadata_tombstone()] |
                 metadata_value() | metadata_tombstone()}, any()) -> any()),
           any(),
           metadata_prefix()) -> any().
fold(Fun, Acc0, FullPrefix) ->
    fold(Fun, Acc0, FullPrefix, []).

%% @doc Fold over all keys and values stored under a given prefix/subprefix. Available
%% options are the same as those provided to iterator/2. To return early, throw
%% {break, Result} in your fold function.
-spec fold(fun(({metadata_key(),
                 [metadata_value() | metadata_tombstone()] |
                 metadata_value() | metadata_tombstone()}, any()) -> any()),
           any(),
           metadata_prefix(),
           fold_opts()) -> any().
fold(Fun, Acc0, FullPrefix, Opts) ->
    It = iterator(FullPrefix, Opts),
    try
        fold_it(Fun, Acc0, It)
    catch
        {break, Result} -> Result
    after
        ok = itr_close(It)
    end.

fold_it(Fun, Acc, It) ->
    case itr_done(It) of
        true ->
            Acc;
        false ->
            Next = Fun(itr_key_values(It), Acc),
            fold_it(Fun, Next, itr_next(It))
    end.

%% @doc same as to_list(FullPrefix, [])
-spec to_list(metadata_prefix()) -> [{metadata_key(),
                                      [metadata_value() | metadata_tombstone()] |
                                      metadata_value() | metadata_tombstone()}].
to_list(FullPrefix) ->
    to_list(FullPrefix, []).


%% @doc Return a list of all keys and values stored under a given prefix/subprefix. Available
%% options are the same as those provided to iterator/2.
-spec to_list(metadata_prefix(), fold_opts()) -> [{metadata_key(),
                                                   [metadata_value() | metadata_tombstone()] |
                                                   metadata_value() | metadata_tombstone()}].
to_list(FullPrefix, Opts) ->
    fold(fun({Key, ValOrVals}, Acc) ->
                 [{Key, ValOrVals} | Acc]
         end, [], FullPrefix, Opts).

%% @doc same as iterator(FullPrefix, []).
-spec iterator(metadata_prefix()) -> iterator().
iterator(FullPrefix) ->
    iterator(FullPrefix, []).

%% @doc Return an iterator pointing to the first key stored under a prefix
%%
%% iterator/2 can take the following options:
%%   * resolver: either the atom `lww' or a function that resolves conflicts if they
%%               are encounted (see get/3 for more details). Conflict resolution
%%               is performed when values are retrieved (see itr_value/1 and itr_key_values/1).
%%               If no resolver is provided no resolution is performed. The default is to
%%               not provide a resolver.
%%   * allow_put: whether or not to write and broadcast a resolved value. defaults to `true'.
%%   * default: Used when the value an iterator points to is a tombstone. default is
%%              either an arity-1 function or a value. If a function, the key the iterator
%%              points to is passed as the argument and the result is returned in place
%%              of the tombstone. If default is a value, the value is returned in place of
%%              the tombstone. This applies when using functions such as itr_values/1 and
%%              itr_key_values/1.
%%   * match: A tuple containing erlang terms and '_'s. Match can be used to iterate
%%            over a subset of keys -- assuming the keys stored are tuples
-spec iterator(metadata_prefix(), it_opts()) -> iterator().
iterator({Prefix, SubPrefix}=FullPrefix, Opts)
  when (is_binary(Prefix) orelse is_atom(Prefix)) andalso
       (is_binary(SubPrefix) orelse is_atom(SubPrefix)) ->
    KeyMatch = proplists:get_value(match, Opts),
    It = riak_core_metadata_manager:iterator(FullPrefix, KeyMatch),
    {It, Opts}.

%% @doc Advances the iterator
-spec itr_next(iterator()) -> iterator().
itr_next({It, Opts}) ->
    It1 = riak_core_metadata_manager:iterate(It),
    {It1, Opts}.

%% @doc Closes the iterator
-spec itr_close(iterator()) -> ok.
itr_close({It, _Ots}) ->
    riak_core_metadata_manager:iterator_close(It).

%% @doc Returns true if there is nothing more to iterate over
-spec itr_done(iterator()) -> boolean().
itr_done({It, _Opts}) ->
    riak_core_metadata_manager:iterator_done(It).

%% @doc Return the key and value(s) pointed at by the iterator. Before
%% calling this function, check the iterator is not complete w/ itr_done/1. If a resolver
%% was passed to iterator/0 when creating the given iterator, siblings will be resolved
%% using the given function or last-write-wins (if `lww' is passed as the resolver). If
%% no resolver was used then no conflict resolution will take place. If conflicts are
%% resolved, the resolved value is written to local metadata and a broadcast is submitted
%% to update other nodes in the cluster if `allow_put' is `true'. If `allow_put' is `false'
%% the values are resolved but are not written or broadcast. A single value is returned as the second
%% element of the tuple in the case values are resolved. If no resolution takes place then a list of
%% values will be returned as the second element (even if there is only a single sibling).
%%
%% NOTE: if resolution may be performed this function must be called at most once
%% before calling itr_next/1 on the iterator (at which point the function can be called
%% once more).
-spec itr_key_values(iterator()) -> {metadata_key(),
                                     [metadata_value() | metadata_tombstone()] |
                                     metadata_value() |
                                     metadata_tombstone()}.
itr_key_values({It, Opts}) ->
    Default = itr_default({It, Opts}),
    {Key, Obj} = riak_core_metadata_manager:iterator_value(It),
    AllowPut = get_option(allow_put, Opts, true),
    case get_option(resolver, Opts, undefined) of
        undefined ->
            {Key, maybe_tombstones(riak_core_metadata_object:values(Obj), Default)};
        Resolver ->
            Prefix = riak_core_metadata_manager:iterator_prefix(It),
            PKey = prefixed_key(Prefix, Key),
            Value = maybe_tombstone(maybe_resolve(PKey, Obj, Resolver, AllowPut), Default),
            {Key, Value}
    end.

%% @doc Return the key pointed at by the iterator. Before
%% calling this function, check the iterator is not complete w/ itr_done/1.
%% No conflict resolution will be performed as a result of calling this function.
-spec itr_key(iterator()) -> metadata_key().
itr_key({It, _Opts}) ->
    {Key, _} = riak_core_metadata_manager:iterator_value(It),
    Key.

%% @doc Return all sibling values pointed at by the iterator. Before
%% calling this function, check the iterator is not complete w/ itr_done/1.
%% No conflict resolution will be performed as a result of calling this function.
-spec itr_values(iterator()) -> [metadata_value() | metadata_tombstone()].
itr_values({It, Opts}) ->
    Default = itr_default({It, Opts}),
    {_, Obj} = riak_core_metadata_manager:iterator_value(It),
    maybe_tombstones(riak_core_metadata_object:values(Obj), Default).

%% @doc Return a single value pointed at by the iterator. If there are conflicts and
%% a resolver was specified in the options when creating this iterator, they will be
%% resolved. Otherwise, and error is returned. If conflicts are resolved, the resolved
%% value is written locally and a broadcast is performed to update other nodes
%% in the cluster if `allow_put' is `true' (the default value). If `allow_put' is `false',
%% values are resolved but not written or broadcast.
%%
%% NOTE: if resolution may be performed this function must be called at most once
%% before calling itr_next/1 on the iterator (at which point the function can be called
%% once more).
-spec itr_value(iterator()) -> metadata_value() | metadata_tombstone() | {error, conflict}.
itr_value({It, Opts}) ->
    Default = itr_default({It, Opts}),
    {Key, Obj} = riak_core_metadata_manager:iterator_value(It),
    AllowPut = get_option(allow_put, Opts, true),
    case get_option(resolver, Opts, undefined) of
        undefined ->
            case riak_core_metadata_object:value_count(Obj) of
                1 ->
                    maybe_tombstone(riak_core_metadata_object:value(Obj), Default);
                _ ->
                    {error, conflict}
            end;
        Resolver ->
            Prefix = riak_core_metadata_manager:iterator_prefix(It),
            PKey = prefixed_key(Prefix, Key),
            maybe_tombstone(maybe_resolve(PKey, Obj, Resolver, AllowPut), Default)
    end.

%% @doc Returns the value returned when an iterator points to a tombstone. If the default
%% used when creating the given iterator is a function it will be applied to the current
%% key the iterator points at. If no default was provided the tombstone value was returned.
%% This function should only be called after checking itr_done/1.
-spec itr_default(iterator()) -> metadata_tombstone() | metadata_value() | it_opt_default_fun().
itr_default({_, Opts}=It) ->
    case proplists:get_value(default, Opts, ?TOMBSTONE) of
        Fun when is_function(Fun) ->
            Fun(itr_key(It));
        Val -> Val
    end.

%% @doc Return the local hash associated with a full-prefix or prefix. The hash value is
%% updated periodically and does not always reflect the most recent value. This function
%% can be used to determine when keys stored under a full-prefix or prefix have changed.
%% If the tree has not yet been updated or there are no keys stored the given
%% (full-)prefix. `undefined' is returned.
-spec prefix_hash(metadata_prefix() | binary() | atom()) -> binary() | undefined.
prefix_hash(Prefix) when is_tuple(Prefix) or is_atom(Prefix) or is_binary(Prefix) ->
    riak_core_metadata_hashtree:prefix_hash(Prefix).

%% @doc same as put(FullPrefix, Key, Value, [])
-spec put(metadata_prefix(), metadata_key(), metadata_value() | metadata_modifier()) -> ok.
put(FullPrefix, Key, ValueOrFun) ->
    put(FullPrefix, Key, ValueOrFun, []).

%% @doc Stores or updates the value at the given prefix and key locally and then
%% triggers a broadcast to notify other nodes in the cluster. Currently, there
%% are no put options
%%
%% NOTE: because the third argument to this function can be a metadata_modifier(),
%% used to resolve conflicts on write, metadata values cannot be functions.
%% To store functions in metadata wrap them in another type like a tuple.
-spec put(metadata_prefix(),
          metadata_key(),
          metadata_value() | metadata_modifier(),
          put_opts()) -> ok.
put({Prefix, SubPrefix}=FullPrefix, Key, ValueOrFun, _Opts)
  when (is_binary(Prefix) orelse is_atom(Prefix)) andalso
       (is_binary(SubPrefix) orelse is_atom(SubPrefix)) ->
    PKey = prefixed_key(FullPrefix, Key),
    CurrentContext = current_context(PKey),
    Updated = riak_core_metadata_manager:put(PKey, CurrentContext, ValueOrFun),
    broadcast(PKey, Updated).

%% @doc same as delete(FullPrefix, Key, [])
-spec delete(metadata_prefix(), metadata_key()) -> ok.
delete(FullPrefix, Key) ->
    delete(FullPrefix, Key, []).

%% @doc Removes the value associated with the given prefix and key locally and then
%% triggers a broradcast to notify other nodes in the cluster. Currently there are
%% no delete options
%%
%% NOTE: currently deletion is logical and no GC is performed.
-spec delete(metadata_prefix(), metadata_key(), delete_opts()) -> ok.
delete(FullPrefix, Key, _Opts) ->
    put(FullPrefix, Key, ?TOMBSTONE, []).

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% @private
current_context(PKey) ->
    case riak_core_metadata_manager:get(PKey) of
        undefined -> riak_core_metadata_object:empty_context();
        CurrentMeta -> riak_core_metadata_object:context(CurrentMeta)
    end.

%% @private
maybe_resolve(PKey, Existing, Method, AllowPut) ->
    SibCount = riak_core_metadata_object:value_count(Existing),
    maybe_resolve(PKey, Existing, SibCount, Method, AllowPut).

%% @private
maybe_resolve(_PKey, Existing, 1, _Method, _AllowPut) ->
    riak_core_metadata_object:value(Existing);
maybe_resolve(PKey, Existing, _, Method, AllowPut) ->
    Reconciled = riak_core_metadata_object:resolve(Existing, Method),
    RContext = riak_core_metadata_object:context(Reconciled),
    RValue = riak_core_metadata_object:value(Reconciled),
    case AllowPut of
        false ->
            ok;
        true ->
            Stored = riak_core_metadata_manager:put(PKey, RContext, RValue),
            broadcast(PKey, Stored)
    end,
    RValue.

%% @private
maybe_tombstones(Values, Default) ->
    [maybe_tombstone(Value, Default) || Value <- Values].

%% @private
maybe_tombstone(?TOMBSTONE, Default) ->
    Default;
maybe_tombstone(Value, _Default) ->
    Value.

%% @private
broadcast(PKey, Obj) ->
    Broadcast = #metadata_broadcast{pkey = PKey,
                                    obj  = Obj},
    riak_core_broadcast:broadcast(Broadcast, riak_core_metadata_manager).

%% @private
-spec prefixed_key(metadata_prefix(), metadata_key()) -> metadata_pkey().
prefixed_key(FullPrefix, Key) ->
    {FullPrefix, Key}.

get_option(Key, Opts, Default) ->
    proplists:get_value(Key, Opts, Default).