%%% -------------------------------------------------------
%%% @author huangyongxing@yeah.net
%%% @doc
%%% 跨服中心节点管理
%%% @end
%%% @since 2017-03-03
%%% -------------------------------------------------------
-module(clusters_center).
-author("huangyongxing@yeah.net").
-behaviour(gen_server).
%% API
-export([
        start_link/0
        ,apply_call/2
        ,apply_call/4
        ,apply_cast/2
        ,apply_cast/4
        ,apply_to_all_node/3
        ,apply_to_all_node/4
        ,add_node/1
        ,get_nodes/0
        ,get_nodes_info/0
        ,get_open_time/1
        ,get_nodes_server_id/0
        ,get_nodes_open_time/0
        ,get_target_nodes_open_time/1
        ,check_node_connect/1
        ,get_node_by_role_id/1
        ,find_node/1
        ,find_now_server_id/1
        ,find_server_num/1
        ,compare_server_id/2
        ,proc_apply/1                       %% 通过clusters_center进程，执行MFA接口，带有全部Nodes信息
    ]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
%% for gm
-export([fix_nodes/0]).

-include("common.hrl").
-include("clusters.hrl").

-define(SERVER, ?MODULE).

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

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

%% Key -> Node :: node() | ServerId :: integer()
apply_call(Key, MFA) ->
    if
        is_integer(Key) ->
            case find_node(Key) of
                undefined -> error;
                Node -> clusters_api:rpc_call_do_mfa(Node, MFA)
            end;
        is_atom(Key) ->
            clusters_api:rpc_call_do_mfa(Key, MFA);
        true ->
            error
    end.

%% Key -> Node :: node() | ServerId :: integer()
apply_cast(Key, MFA) ->
    if
        is_integer(Key) ->
            case find_node(Key) of
                undefined -> error;
                Node -> clusters_api:rpc_cast_do_mfa(Node, MFA)
            end;
        is_atom(Key) ->
            clusters_api:rpc_cast_do_mfa(Key, MFA);
        true ->
            error
    end.

apply_call(Key, Module, Func, Args) ->
    apply_call(Key, {Module, Func, Args}).

apply_cast(Key, Module, Func, Args) ->
    apply_cast(Key, {Module, Func, Args}).

%% 通知所有节点执行
apply_to_all_node(Module, Func, Args) ->
    gen_server:cast(?SERVER, {'apply_to_all_node', Module, Func, Args}).

%% 通知所有节点执行
%% Timeout:毫秒 —— 这里限制Timeout小于600秒，正常逻辑不应出现这么大的延迟数值
apply_to_all_node(Module, Func, Args, Timeout) ->
    gen_server:cast(?SERVER, {'apply_to_all_node', Module, Func, Args, Timeout}).

%% 添加跨服节点
add_node(GameNode) when is_record(GameNode, game_node) ->
    gen_server:cast(?SERVER, {add_node, GameNode});
add_node(_) ->
    error.

%% 获得游戏服开服时间
%% Key :: node() | {Platform, ServerId}
get_open_time(Key) ->
    gen_server:call(?SERVER, {get_open_time, Key}).

%% 获取所有节点
get_nodes() ->
    gen_server:call(?SERVER, get_nodes).

%% 获取所有节点详细信息
get_nodes_info() ->
    gen_server:call(?SERVER, get_nodes_info).

%% 获取所有节点服务器ID
get_nodes_server_id() ->
    gen_server:call(?SERVER, get_nodes_server_id).

%% 获取所有节点开服时间
get_nodes_open_time() ->
    gen_server:call(?SERVER, get_nodes_open_time).

%% 获取指定节点开服时间
get_target_nodes_open_time(ServIds) ->
    gen_server:call(?SERVER, {get_target_nodes_open_time, ServIds}).

%% 判断游戏服是否链接跨服中心:{_Platform, _ServerId}
check_node_connect(Key) ->
    gen_server:call(?SERVER, {check_node_connect, Key}).

%% 根据ServerId查找Node节点名
find_node(ServerId) ->
    case ets:lookup(?SERVER_INFO, ServerId) of
        [#server_info{node = Node}] -> Node;
        _ -> undefined
    end.

%% 根据ServerId查找游戏服编号
find_server_num(ServerId) ->
    case ets:lookup(?SERVER_INFO, ServerId) of
        [#server_info{server_num = ServerNum}|_] -> ServerNum;
        _ -> 0
    end.

%% 根据ServerId查找合服后游戏服最新的ServerId
find_now_server_id(ServerId) ->
    case ets:lookup(?SERVER_INFO, ServerId) of
        [#server_info{now_server_id = NewServerId}] ->
            NewServerId;
        _ -> ServerId
    end.

%% 对比两个ServerId是否指向同一个服
compare_server_id(ServerId, ServerId) -> true;
compare_server_id(ServerIdA, ServerIdB) ->
    case find_now_server_id(ServerIdA) of
        ServerIdB -> true;
        ServerIdNow ->
            ServerIdNow =:= find_now_server_id(ServerIdB)
    end.

%% 通过玩家ID获得玩家节点
get_node_by_role_id(RoleID)->
    ServId = mod_player_create:get_serid_by_id(RoleID),
    find_node(ServId).

%% 通过clusters_center进程，执行具体的MFA接口，带有全部Nodes信息
proc_apply(MFA) ->
    case util:valid_mfa(MFA) of
        true  -> gen_server:cast(?SERVER, {proc_apply, MFA});
        false -> ignore
    end.

%% 有时候运维部署有问题，节点数据被覆盖掉了，需要等运维把不要的节点关闭后，调用一下fix_nodes
fix_nodes() ->
    erlang:send(?SERVER, fix_nodes).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init([]) ->
    process_flag(trap_exit, true),
    ets:new(?SERVER_INFO, [named_table, protected, set, {keypos, #server_info.server_id}, {read_concurrency, true}]),
    db_clean_connected(),
    State = #clusters_center{
        node = node(),
        nodes = []
    },
    {ok, State}.

handle_cast({proc_apply, MFA}, State = #clusters_center{nodes = Nodes}) ->
    proc_lib:spawn_link(fun() ->
                NewMFA =
                case MFA of
                    {M, F, A}   -> {M, F, [Nodes | A]};
                    {F, A}      -> {F, [Nodes | A]};
                    F           -> {F, [Nodes]}
                end,
                util:do_mfa(NewMFA)
        end),
    {noreply, State};

%% 添加跨服节点
handle_cast({add_node, GameNode}, State = #clusters_center{nodes = Nodes, node = CenterNode})->
    case GameNode of
        #game_node{node = Node, server_id = ServerId, pid = Pid} ->
            CheckRet =
            case lists:keyfind(Node, #game_node.node, Nodes) of
                #game_node{server_id = ServerId} -> true;
                #game_node{} -> false;
                false -> true
            end
            andalso
            case lists:keyfind(ServerId, #game_node.server_id, Nodes) of
                #game_node{node = Node} -> true;
                #game_node{} -> false;
                false -> true
            end,
            case CheckRet of
                true  ->
                    catch clusters_api:add_node_db(GameNode),
                    set_server_info(GameNode),
                    Monitor = erlang:monitor(process, Pid),
                    NewNodes = lists:keystore(Node, #game_node.node, Nodes, GameNode#game_node{monitor = Monitor}),
                    ?INFO("Built connection to cluster node [~w]", [Node]),
                    rpc:cast(Node, clusters_node, connected, [CenterNode, self()]),
                    proc_lib:spawn_link(fun() -> clusters_api:center_connected(GameNode) end),
                    %% 检查游戏服是否已在跨服注册
                    center_svr_reg:check_reg(GameNode),
                    NewState = State#clusters_center{nodes = NewNodes},
                    {noreply, NewState};
                false ->
                    %% 数据已经存在，但是节点与ServerId不一致，运维配重复数据了
                    rpc:cast(Node, clusters_node, notice_invalid, [CenterNode, GameNode]),
                    ?ERR("Invalid game config, node[~w], server_id[~w]", [Node, ServerId]),
                    {noreply, State}
            end;
        _ ->
            {noreply, State}
    end;

%% 通知所有节点执行
handle_cast({'apply_to_all_node', Module, Func, Args} , State = #clusters_center{nodes = Nodes}) ->
    [rpc:cast(Node, Module, Func, Args) || #game_node{node = Node} <- Nodes],
    {noreply, State};

%% 通知所有节点
%% timeout延迟发送
handle_cast({'apply_to_all_node', Module, Func, Args, Timeout} , State = #clusters_center{nodes = Nodes}) ->
    proc_lib:spawn_link(fun() ->
                NewTimeout = min(Timeout, 600000),
                [
                    begin
                            rpc:cast(Node, Module, Func, Args),
                            timer:sleep(NewTimeout)
                    end || #game_node{node = Node} <- Nodes
                ]
        end),
    {noreply, State};

handle_cast({'apply_cast', ServerId, MFA}, State = #clusters_center{nodes = Nodes}) when is_integer(ServerId) ->
    case lists:keyfind(ServerId, #game_node.server_id, Nodes) of
        #game_node{node = Node}  ->
            %% ?ERR("ServerId:~w is found, cast mfa ~w", [ServerId, MFA]),
            clusters_api:rpc_cast_do_mfa(Node, MFA);
        false ->
            %% ?ERR("ServerId:~w not found, cast mfa ~w", [ServerId, MFA]),
            false
    end,
    {noreply, State};

handle_cast({'apply_cast', Node, MFA} , State = #clusters_center{nodes = Nodes}) when is_atom(Node) ->
    case lists:keymember(Node, #game_node.node, Nodes) of
        true  ->
            clusters_api:rpc_cast_do_mfa(Node, MFA);
        false ->
            % ?ERR("node:~ts not found, cast mfa ~w", [Node, MFA]),
            false
    end,
    {noreply, State};

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

handle_call({'apply_call', ServerId, MFA}, From, State = #clusters_center{nodes = Nodes}) when is_integer(ServerId) ->
    case lists:keyfind(ServerId, #game_node.server_id, Nodes) of
        #game_node{node = Node} ->
            case util:valid_mfa(MFA) of
                true  -> clusters_api:async_apply_call(Node, From, MFA);
                false -> gen_server:reply(From, error)
            end;
        false ->
            % ?ERR("ServerId:~w not found, cast mfa ~w", [ServerId, MFA]),
            false
    end,
    {noreply, State};

handle_call({'apply_call', Node, MFA}, From, State = #clusters_center{nodes = Nodes}) when is_atom(Node) ->
    case lists:keymember(Node, #game_node.node, Nodes) of
        true  ->
            case util:valid_mfa(MFA) of
                true  -> clusters_api:async_apply_call(Node, From, MFA);
                false -> gen_server:reply(From, error)
            end;
        false ->
            % ?ERR("node:~ts not found, cast mfa ~w", [Node, MFA]),
            false
    end,
    {noreply, State};

handle_call(get_nodes, _From, State = #clusters_center{nodes = Nodes}) ->
    NodeNames = [Node || #game_node{node = Node} <- Nodes],
    {reply, NodeNames, State};

handle_call(get_nodes_info, _From, State = #clusters_center{nodes = Nodes}) ->
    %% merge_server_ids列表可能很长，不传递过去了。
    NodeList = [GameNode#game_node{merge_server_ids = []} || GameNode <- Nodes],
    {reply, NodeList, State};

handle_call(get_nodes_server_id, _From, State = #clusters_center{nodes = Nodes}) ->
    ServerIds = [ServerId || #game_node{server_id = ServerId} <- Nodes],
    {reply, ServerIds, State};

handle_call({get_open_time, Key}, _From, State = #clusters_center{nodes = Nodes}) ->
    Ret =
    if
        is_integer(Key) ->
            lists:keyfind(Key, #game_node.server_id, Nodes);
        is_atom(Key) ->
            lists:keyfind(Key, #game_node.node, Nodes);
        true ->
            false
    end,
    case Ret of
        #game_node{open_time = OpenTime} ->
            {reply, OpenTime, State};
        _ ->
            {reply, 0, State}
    end;

handle_call(get_nodes_open_time, _From, State = #clusters_center{nodes = Nodes}) ->
    OpenTimes = [{ServerId, Time} || #game_node{server_id = ServerId, open_time = Time} <- Nodes],
    {reply, OpenTimes, State};

handle_call({get_target_nodes_open_time, ServIds}, _From, State = #clusters_center{nodes = Nodes}) ->
    OpenTimes = lists:filtermap(fun(ServId) -> 
        case lists:keyfind(ServId, #game_node.server_id, Nodes) of
            #game_node{open_time = OpenTime} ->
                {true, {ServId, OpenTime}};
            _ ->
                false
        end
     end, ServIds),
    {reply, OpenTimes, State};

handle_call({check_node_connect, ServerKey}, _From, State = #clusters_center{nodes = Nodes}) ->
    IsConnected =
    if
        is_integer(ServerKey) ->
            lists:keymember(ServerKey, #game_node.server_id, Nodes);
        is_atom(ServerKey) ->
            lists:keymember(ServerKey, #game_node.node, Nodes);
        true ->
            false
    end,
    {reply, IsConnected, State};

handle_call(_R, _From, State) ->
    {noreply, State}.

handle_info({'DOWN', Monitor, _Type, _Obj, _Reason}, State = #clusters_center{nodes = Nodes}) ->
    case lists:keyfind(Monitor, #game_node.monitor, Nodes) of
        #game_node{node = _Node, server_id = _ServerId} = GameNode ->
            catch clusters_api:del_node_db(GameNode),
            NewState = State#clusters_center{nodes = lists:keydelete(Monitor, #game_node.monitor, Nodes)},
            clusters_api:center_disconnected(GameNode),
            proc_lib:spawn_link(fun() -> clusters_api:center_disconnected(GameNode) end),
            % ServerList = ets:match_object(?SERVER_INFO, #server_info{now_server_id = _ServerId, _ = '_'}),
            % ets:insert(?SERVER_INFO, [DownServer#server_info{alive = 0} ||DownServer <- ServerList]),
            % 改用主服信息重新生成已合并服信息，避免ets:match_object/2调用
            set_server_info(GameNode, 0),
            ?INFO("Lost connection to cluster node [~w]", [_Node]),
            {noreply, NewState};
        false ->
            {noreply, State}
    end;

handle_info({'EXIT', Pid, Why}, State) ->
    case Why of
        normal   -> ignore;
        shutdown -> ignore;
        _        -> ?ERR("link pid[~w] exit:~p", [Pid, Why])
    end,
    {noreply, State};

handle_info(fix_nodes, State = #clusters_center{nodes = Nodes}) ->
    %% 重新根据节点信息写入所有游戏服信息（并未清空旧数据）
    lists:foreach(
        fun (GameNode = #game_node{}) ->
                set_server_info(GameNode);
            (_) ->
                ignore
        end, Nodes
    ),
    {noreply, State};

handle_info(_R, State) ->
    {noreply, State}.

terminate(Reason, State) ->
    ?INFO("clusters center terminate, reason:~w", [Reason]),
    {ok, State}.

code_change(_OldVsn, State, _Extra)->
    {ok, State}.

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

%% 清空DB中的连接数据（后续添加节点时写入数据，主要供后台查看）
db_clean_connected() ->
    % Nodes = db:get_all(?SQL_GET_CLUSTER_NODES),
    db:execute(?SQL_TRUNCATE_CLUSTERS),
    ok.

set_server_info(GameNode) ->
    set_server_info(GameNode, 1).

%% Alive表示节点状态（1为连接，0为断开）
set_server_info(#game_node{
        node = Node
        ,server_id = ServerId
        ,server_num = ServerNum
        ,platform = Platform
        ,merge_server_ids = MergeServerIds
    }, Alive) ->
    ServerIds =
    case lists:member(ServerId, MergeServerIds) of
        true  -> MergeServerIds;
        false -> [ServerId | MergeServerIds]
    end,
    ServerInfoList = [
        #server_info{
            node = Node
            ,server_id = Id
            ,now_server_id = ServerId
            ,server_num = ServerNum
            ,platform = Platform
            ,alive = Alive
        } || Id <- ServerIds
    ],
    ets:insert(?SERVER_INFO, ServerInfoList).

