itc

-module(event_server).

-behaviour(gen_server).
-export([start_link/1, init/1, stop/1, terminate/2, handle_call/3, handle_cast/2]).
-export([add/2, fork/2, list/1]).

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

start_link(Name) -> gen_server:start_link({local, Name}, ?MODULE, Name, []).

stop(PidOrName) -> gen_server:stop(PidOrName).

add(Self, Action) ->
    gen_server:call(Self, {add, Action}).

list(Self) ->
    gen_server:call(Self, list).

fork(Self, Other) ->
    {ok, _Pid} = start_link(Other),
    gen_server:call(Other, {join, Self}).

%%------------------------------------------------------------------------------

init(Node) -> {ok, #{node => Node, events => events:init()}}.

terminate(_Reason, _State) -> ok.

handle_call({add, Action}, _From, State = #{events := Events, node := Self}) ->
    NewEvents = events:append(Events, {action, Action}),
    sync_nodes(NewEvents, Self),
    {reply, ok, State#{events := NewEvents}};

handle_call({fork, Other}, _From, State = #{events := Events, node := Self}) ->
    [LeftEvents, RightEvents] = events:fork(Events),
    JoinedEvents = events:append(LeftEvents, {join, Other}),
    sync_nodes(JoinedEvents, Self),
    {reply, RightEvents, State#{events := JoinedEvents}};

handle_call({join, Other}, _From, State = #{node := Self, events := [_]}) ->
    Events = gen_server:call(Other, {fork, Self}),
    JoinedEvents = events:append(Events, {join, Other}),
    {reply, ok, State#{events := JoinedEvents}};

handle_call({join, _}, _From, State = #{events := Events}) ->
    Error = {
        error, "Join is allowed only after init", events:filter(Events, action)
    },
    {reply, Error, State};

handle_call(list, _From, State = #{events := Events}) ->
    {reply, events:filter(Events, action), State};

handle_call({sync_from, SyncFrom}, _From, State = #{events := Events}) ->
    {reply, events:list_unseen_events(Events, SyncFrom), State};

handle_call(M,  _From, State) -> {reply, {error, "unknown message", M}, State}.

handle_cast({sync, Tic, Other}, State = #{events := Events}) ->
    SyncFrom = events:get_last_seen_event_tic(Events, Tic),
    UnseenEvents = gen_server:call(Other, {sync_from, SyncFrom}),
    MergedEvents = events:merge(Events, UnseenEvents),
    {noreply, State#{events := MergedEvents}};

handle_cast(_, State) -> {noreply, State}.

%%------------------------------------------------------------------------------

sync_nodes(Events, Self) ->
    lists:foreach(
        fun (Node) ->
            gen_server:cast(Node, {sync, events:get_last_tic(Events), Self})
        end,
        get_other_nodes(Events, Self)
    ).

get_other_nodes(Events, Self) ->
    Nodes = lists:usort(events:filter(Events, join)),
    lists:filter(fun (Other) -> Other =/= Self end, Nodes).