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, join/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).

join(Self, Other) ->
    gen_server:call(Self, {join, Other}).

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

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}),
    Others = get_other_nodes(Events, Self),
    lists:foreach(fun (Node) -> update_node(NewEvents, Node) end, Others),
    {reply, ok, State#{events := events:append(Events, {action, Action})}};

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

handle_call({sync, Tic}, _From, State = #{events := Events}) ->
    SyncFrom = events:get_last_seen_event_tic(Events, Tic),
    {reply, SyncFrom, State};

handle_call({fork, Other}, _From, State = #{events := Events, node := Self}) ->
    [LeftEvents, RightEvents] = events:fork(Events),
    JoinedEvents = events:append(LeftEvents, {join, Other}),
    Others = get_other_nodes(Events, Self),
    lists:foreach(fun (Node) -> update_node(JoinedEvents, Node) end, Others),
    {reply, RightEvents, State#{events := JoinedEvents}};

%% join is allowed only after init
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(_,  _From, State) -> {stop, shutdown, State}.

handle_cast({update, UnseenEvents}, State = #{events := Events}) ->
    {noreply, State#{events := events:merge(Events, UnseenEvents)}};

handle_cast(_, State) -> {stop, shutdown, State}.

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

update_node(Events, Node) ->
    LastTic = events:get_last_tic(Events),
    SyncFrom = gen_server:call(Node, {sync, LastTic}),
    UnseenEvents = events:list_unseen_events(Events, SyncFrom),
    gen_server:cast(Node, {update, UnseenEvents}).

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