riak_ensemble

%% -------------------------------------------------------------------
%%
%% 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_ensemble_state).

-include_lib("riak_ensemble_types.hrl").

-export([new/1, is_state/1]).
-export([add_member/3,
         del_member/3]).
-export([set_ensemble/3,
         update_ensemble/5,
         set_pending/4]).
-export([enable/1, enabled/1]).
-export([merge/2]).
-export([id/1, members/1, ensembles/1, pending/1]).

-type ensembles() :: orddict(ensemble_id(), ensemble_info()).
-type pending()   :: orddict(ensemble_id(), {vsn(), views()}).

-record(cluster_state, {id          :: any(),
                        enabled     :: boolean(),
                        members     :: {vsn(), ordsets(node())},
                        ensembles   :: ensembles(),
                        pending     :: pending()
                       }).

-type state() :: #cluster_state{}.
-export_type([state/0]).

-define(STATE, #cluster_state).

%%%===================================================================

-spec new(term()) -> state().
new(Id) ->
    ?STATE{id = Id,
           enabled = false,
           members = {vsn0(), ordsets:new()},
           ensembles = orddict:new(),
           pending = orddict:new()}.

-spec is_state(term()) -> boolean().
is_state(?STATE{}) ->
    true;
is_state(_) ->
    false.

-spec enable(state()) -> {ok, state()} | error.
enable(State=?STATE{enabled=false}) ->
    State2 = State?STATE{enabled=true},
    {ok, State2};
enable(?STATE{enabled=true}) ->
    error.

-spec enabled(state()) -> boolean().
enabled(?STATE{enabled=Enabled}) ->
    Enabled.

-spec id(state()) -> term().
id(?STATE{id=Id}) ->
    Id.

-spec members(state()) -> ordsets(node()).
members(?STATE{members={_Vsn, Nodes}}) ->
    Nodes.

-spec ensembles(state()) -> ensembles().
ensembles(?STATE{ensembles=Ensembles}) ->
    Ensembles.

-spec pending(state()) -> pending().
pending(?STATE{pending=Pending}) ->
    %% [{Id, Views} || {Id, {_Vsn, Views}} || <- Pending].
    Pending.

-spec add_member(vsn(), node(), state()) -> {ok, state()} | error.
add_member(Vsn, Node, State=?STATE{members={CurVsn, Nodes}}) ->
    case newer(CurVsn, Vsn) of
        true ->
            Nodes2 = ordsets:add_element(Node, Nodes),
            State2 = State?STATE{members={Vsn, Nodes2}},
            {ok, State2};
        false ->
            error
    end.

-spec del_member(vsn(), node(), state()) -> {ok, state()} | error.
del_member(Vsn, Node, State=?STATE{members={CurVsn, Nodes}}) ->
    case newer(CurVsn, Vsn) of
        true ->
            Nodes2 = ordsets:del_element(Node, Nodes),
            State2 = State?STATE{members={Vsn, Nodes2}},
            {ok, State2};
        false ->
            error
    end.

-spec set_ensemble(ensemble_id(), ensemble_info(), state()) -> {ok, state()} |
                                                               error.
set_ensemble(Ensemble, Info, State=?STATE{ensembles=Ensembles}) ->
    Vsn = Info#ensemble_info.vsn,
    CurVsn = case orddict:find(Ensemble, Ensembles) of
                 {ok, CurInfo} ->
                     CurInfo#ensemble_info.vsn;
                 error ->
                     undefined
             end,
    case newer(CurVsn, Vsn) of
        true ->
            Ensembles2 = orddict:store(Ensemble, Info, Ensembles),
            State2 = State?STATE{ensembles=Ensembles2},
            {ok, State2};
        false ->
            error
    end.

-spec update_ensemble(vsn(), ensemble_id(), peer_id(), views(), state())
                     -> {ok, state()} | error.
update_ensemble(Vsn, Ensemble, Leader, Views, State=?STATE{ensembles=Ensembles}) ->
    case orddict:find(Ensemble, Ensembles) of
        {ok, CurInfo} ->
            CurVsn = CurInfo#ensemble_info.vsn,
            case newer(CurVsn, Vsn) of
                true ->
                    NewInfo = CurInfo#ensemble_info{vsn=Vsn, leader=Leader, views=Views},
                    Ensembles2 = orddict:store(Ensemble, NewInfo, Ensembles),
                    State2 = State?STATE{ensembles=Ensembles2},
                    {ok, State2};
                false ->
                    error
            end;
        error ->
            error
    end.

-spec set_pending(vsn(), ensemble_id(), views(), state()) -> {ok, state()} |
                                                             error.
set_pending(Vsn, Ensemble, Views, State=?STATE{pending=Pending}) ->
    CurVsn = case orddict:find(Ensemble, Pending) of
                 {ok, {CV, _}} ->
                     CV;
                 error ->
                     undefined
             end,
    case newer(CurVsn, Vsn) of
        true ->
            Pending2 = orddict:store(Ensemble, {Vsn, Views}, Pending),
            State2 = State?STATE{pending=Pending2},
            {ok, State2};
        false ->
            error
    end.

-spec merge(state(), state()) -> state().
merge(A, B) when A?STATE.enabled and (A?STATE.id =/= B?STATE.id) ->
    _ = lager:warning("Ignoring cluster state with different id"),
    A;
merge(A=?STATE{members=MembersA, ensembles=EnsemblesA, pending=PendingA},
      _=?STATE{members=MembersB, ensembles=EnsemblesB, pending=PendingB}) ->
    A?STATE{members=merge_members(MembersA, MembersB),
            ensembles=merge_ensembles(EnsemblesA, EnsemblesB),
            pending=merge_pending(PendingA, PendingB)}.

%%%===================================================================

merge_members(A={VsnA, _}, B={VsnB, _}) ->
    case newer(VsnA, VsnB) of
        true ->
            B;
        false ->
            A
    end.

merge_ensembles(EnsemblesA, EnsemblesB) ->
    orddict:merge(fun merge_ensemble/3, EnsemblesA, EnsemblesB).

merge_ensemble(_, InfoA, InfoB) ->
    case newer(InfoA#ensemble_info.vsn, InfoB#ensemble_info.vsn) of
        true ->
            InfoB;
        false ->
            InfoA
    end.

merge_pending(PendingA, PendingB) ->
    orddict:merge(fun merge_pending_views/3, PendingA, PendingB).

merge_pending_views(_, A={VsnA, _}, B={VsnB, _}) ->
    case newer(VsnA, VsnB) of
        true ->
            B;
        false ->
            A
    end.

newer(VsnA, VsnB) ->
    ensure_vsn(VsnB) > ensure_vsn(VsnA).

ensure_vsn(undefined) ->
    vsn0();
ensure_vsn(Vsn={_,_}) ->
    Vsn.

vsn0() ->
    {-1,0}.