-module(regis_pstore_ets).

-export([ new/0
        , delete/1
        , register/4
        , unregister/3
        , unregister_all/2
        , whereis_job/2
        , jobs/2
        , num_registered_pids/1
        , fold_names/3
        , fold_pids/3
        ]).

-record(pstore, {proc_store, proc_pids}).
-opaque pstore() :: #pstore{}.
-export_type([pstore/0]).

new() ->
    JobsId = ets:new(pscomp_proc_jobs, [public, named_table]),
    PidsId = ets:new(pscomp_proc_pids, [public, named_table]),
    {ok, _State=#pstore{proc_store=JobsId, proc_pids=PidsId}}.

delete(_State=#pstore{proc_store=JobsId, proc_pids=PidsId}) ->
    true = ets:delete(JobsId),
    true = ets:delete(PidsId),
    {ok, #pstore{proc_store=undefined, proc_pids=undefined}}.    

-spec register(pstore(), term(), term(), pid()) -> {ok, pstore()} | {error, exists_proc_store, none_proc_pids} | {error, none_proc_store, exists_proc_pids}.
register(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta={_App, _CallBack, _ProcId}, VNodeId, Pid) when is_pid(Pid) -> 
    register_( State
             , JobMeta
             , _VnodeItem={VNodeId, Pid}
             , { ets:lookup(JobsId, JobMeta), ets:lookup(PidsId, Pid) }
             ).

register_(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta, VnodeItem={_VNodeId, Pid}, {[], []}) ->
    link(Pid),
    true = ets:insert(JobsId, {JobMeta, [ VnodeItem ]}),
    true = ets:insert(PidsId, {Pid, [ JobMeta ]}),
    {ok, State};
register_(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta, VnodeItem={_VNodeId, Pid}, {[{_JobMeta, VnodeItems}], [{_Pid, JobMetas}]}) ->
    VnodeItem1 = [VnodeItem] ++ [ VI || VI <- VnodeItems, VI/=VnodeItem],
    JobMeta1   = [JobMeta]   ++ [ PI || PI <- JobMetas,   PI/=JobMeta],
    true = ets:insert(JobsId, {JobMeta, VnodeItem1}),
    true = ets:insert(PidsId, {Pid, JobMeta1}),
    {ok, State};
register_(_State, _JobMeta, _VnodeItem, {[{_JobMeta, _VnodeItems}], []}) ->
    {error, exists_proc_store, none_proc_pids};
register_(_State, _JobMeta, _VnodeItem, {[], [{_Pid, _JobMetas}]}) ->
    {error, none_proc_store, exists_proc_pids}.

-spec unregister(pstore(), term(), pid()) -> {ok, pstore()} | {error, exists_proc_store, none_proc_pids} | {error, none_proc_store, exists_proc_pids}.
unregister(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta={_App, _CallBack, _ProcId}, Pid) when is_pid(Pid) ->
    unregister_( State
               , JobMeta
               , Pid
               , { ets:lookup(JobsId, JobMeta), ets:lookup(PidsId, Pid) }
    ).

unregister_(State, _JobMeta, _Pid, {[], []}) ->
    {ok, State};
unregister_(State=#pstore{proc_store=JobsId, proc_pids=PidsId}, JobMeta={_App, _CallBack, _ProcId}, Pid, {[{_JobMeta, VnodeItems}], [{_Pid, JobMetas}]}) ->
    VnodeItemsFiltered = [{VNodeId, P} || {VNodeId, P} <- VnodeItems, P/=Pid],
    JobMetaFiltered = JobMetas -- [ JobMeta ],
    case {VnodeItemsFiltered, JobMetaFiltered} of
        {[], []} ->
            unlink(Pid),
            true = ets:delete(JobsId, JobMeta),
            true = ets:delete(PidsId, Pid),
            {ok, State};
        {[], JobMetas1} ->
            true = ets:delete(JobsId, JobMeta),
            true = ets:insert(PidsId, {Pid, JobMetas1}),
            {ok, State};
        {VnodeItems1, []} ->
            true = ets:insert(JobsId, {JobMeta, VnodeItems1}),
            true = ets:delete(PidsId, Pid),
            {ok, State};
        {VnodeItems2, JobMetas2} ->
            true = ets:insert(JobsId, {JobMeta, VnodeItems2}),
            true = ets:insert(PidsId, {Pid, JobMetas2}),
            {ok, State}
    end;
unregister_(_State, _JobMeta, _Pid, {[{_JobMeta, _VnodeItems}], []}) ->
    {error, exists_proc_store, none_proc_pids};
unregister_(_State, _JobMeta, _Pid, {[], [{_Pid, _JobMetas}]}) ->
    {error, none_proc_store, exists_proc_pids}.


-spec unregister_all(pstore(), pid()) -> {ok, pstore()}.
unregister_all(State=#pstore{proc_store=_JobsId, proc_pids=PidsId}, Pid) ->
    case ets:lookup(PidsId, Pid) of
        [] -> 
            {ok, State};
        [{_Pid, JobMetas}] ->
            [ unregister(State, JobMeta, Pid) || JobMeta <- JobMetas ],
            {ok, State}
    end.

-spec num_registered_pids(pstore()) -> integer().
num_registered_pids(_State=#pstore{proc_store=_JobsId, proc_pids=PidsId}) ->
    ets:info(PidsId, size).

-spec whereis_job(pstore() | atom(), term()) -> pid() | undefined.
whereis_job(_State=#pstore{proc_store=JobsId, proc_pids=_PidsId}, JobMeta) ->
    whereis_job(JobsId, JobMeta);
whereis_job(JobsId, JobMeta) ->
    case jobs(JobsId, JobMeta) of
        [{_, Pid}] -> Pid;
        _ -> undefined
    end.

-spec jobs(pstore() | atom(), term()) -> [{term(), pid()}].
jobs(_State=#pstore{proc_store=JobsId, proc_pids=_PidsId}, JobMeta) ->
    jobs(JobsId, JobMeta);
jobs(JobsId, JobMeta) ->
    case ets:lookup(JobsId, JobMeta) of
        [] -> [];
        [{_JobMeta, VnodeItems}] ->
            [{VNodeId, Pid} || {VNodeId, Pid} <- VnodeItems]
    end.

-spec fold_names(function(), term(), pstore()) -> term().
fold_names(UserFun, Acc0, _State=#pstore{proc_store=JobsId, proc_pids=_PidsId}) when is_function(UserFun, 3) ->
    Fun = fun({JobMeta, Values}, Acc) -> UserFun(JobMeta, Values, Acc) end,
    ets:foldl(Fun, Acc0, JobsId).

-spec fold_pids(function(), term(), pstore()) -> term().
fold_pids(UserFun, Acc0, _State=#pstore{proc_store=_JobsId, proc_pids=PidsId}) when is_function(UserFun, 3) ->
    Fun = fun({Pid, JobMetas}, Acc) -> UserFun(Pid, JobMetas, Acc) end,
    ets:foldl(Fun, Acc0, PidsId).

%% ===================================================================
%% EUnit tests
%% ===================================================================
-include_lib("eunit/include/eunit.hrl").
-define(setup(F), {setup, fun start/0, fun stop/1, F}).

%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% TESTS DESCRIPTIONS %%%
%%%%%%%%%%%%%%%%%%%%%%%%%%
start_stop_test_() ->
    {"1. Модуль должен реализовывать ETS-реестр процессов, который можно создать при запуске процесса и удалить при остановке",
    ?setup(fun is_registered/1)}.

register_test_() ->
    [
        {"1 Register. Должна иметься возможность регистрации Pid и Job из одной функции", ?setup(fun register_t/1)},
        {"2 UnRegister. Должна иметься возможность снятия с регистрации Pid и Job из одной функции обладая полями (JobMeta и Pid)", ?setup(fun unregister_t/1)},
        {"3 Search. Должна быть возможность определения Pid по имени Job, извлечения всех зарегистрированных джобов и количества зарегистрированных Pid", ?setup(fun search_t/1)}
    ].

%%%%%%%%%%%%%%%%%%%%%%%
%%% SETUP FUNCTIONS %%%
%%%%%%%%%%%%%%%%%%%%%%%
start() ->
        {ok, State} = new(),
        State.
         
stop(State) ->
        {ok, State1} = delete(State),
        State1.

%%%%%%%%%%%%%%%%%%%%
%%% ACTUAL TESTS %%%
%%%%%%%%%%%%%%%%%%%%
is_registered(State) ->
        #pstore{proc_store=JobsId, proc_pids=PidsId} = State,
        [ 
          ?_assertEqual(pscomp_proc_jobs, JobsId),
          ?_assertEqual(pscomp_proc_pids, PidsId)
        ].

register_t(State) ->
        Self = self(),
        JobMeta1 = {pscomp1, sample1, Self},
        VnodeItem1 = {111111111112234455666, Self},
        #pstore{proc_store=JobsId, proc_pids=PidsId}=State,
        [
            ?_assertMatch({ok, _State}, register(State, JobMeta1, VnodeItem1, Self) ),
            ?_assertEqual(1, num_items(JobsId, JobMeta1) ),
            ?_assertEqual(1, num_items(PidsId, Self) )
        ].

unregister_t(State) ->
        Self = self(),
        JobMeta1 = {pscomp1, sample1, Self},
        VnodeItem1 = {111111111112234455666, Self},
        VnodeItem2 = {212313123486898080821, Self},
        #pstore{proc_store=JobsId, proc_pids=PidsId}=State,
        [ register(State, JobMeta1, VnodeItem, Self) || VnodeItem <- [ VnodeItem1, VnodeItem2 ] ],
        [
            ?_assertEqual(2, num_items(JobsId, JobMeta1) ),
            ?_assertEqual(1, num_items(PidsId, Self) ),
            ?_assertMatch({ok, _State}, unregister(State, JobMeta1, Self) ),
            ?_assertEqual(0, num_items(JobsId, JobMeta1) ),
            ?_assertEqual(0, num_items(PidsId, Self) )
        ].

search_t(State) ->
        Self = self(),
        JobMeta1 = {pscomp1, sample1, Self},
        VnodeItem1 = {111111111112234455666, Self},
        VnodeItem2 = {212313123486898080821, Self},
        [
            ?_assertMatch({ok, _State}, register(State, JobMeta1, VnodeItem1, Self) ),
            ?_assertEqual(true, is_pid( whereis_job(State, JobMeta1) ) ),
            ?_assertMatch({ok, _State}, register(State, JobMeta1, VnodeItem2, Self) ),
            ?_assertEqual(1, num_registered_pids(State) ),
            ?_assertEqual(2, lists:sum( lists:map(fun(_X) -> 1 end, jobs(State, JobMeta1) ) ) )
        ].

%%%%%%%%%%%%%%%%%%%%%%%%
%%% HELPER FUNCTIONS %%%
%%%%%%%%%%%%%%%%%%%%%%%%
num_items(TabId, LookupItem) ->
    case ets:lookup(TabId, LookupItem) of
        [] ->
            0;
        [{_Pid, Items}] ->
            lists:sum( lists:map(fun(_X) -> 1 end, Items) )
    end.