%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2025, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 20. 4月 2026 00:00
%%%-------------------------------------------------------------------
-module(efka_iot_client).
-author("anlicheng").
-include("efka_tables.hrl").

-behaviour(gen_statem).

%% API
-export([start_link/0]).
-export([metric_data/2, ping/13, task_event_stream/3, close_task_event_stream/2]).
-export([is_activated/0, dropped_message_count/0]).

%% gen_statem callbacks
-export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]).

-define(SERVER, ?MODULE).

%% 标记当前agent的状态，只有在 activated 状态下才可以正常的发送数据
-define(STATE_DISCONNECTED, disconnected).
%% 等待校验中
-define(STATE_AUTH, auth).
%% 激活状态下
-define(STATE_ACTIVATED, activated).
-define(SSL_PING_INTERVAL, 30000).
-define(OUTBOX_SEGMENT_RECORD_LIMIT, 2000).
-define(OUTBOX_MAX_SEGMENTS, 5).
-define(OUTBOX_MAX_RECORD_BYTES, 16 * 1024 * 1024).

-record(state, {
    socket :: undefined | ssl:sslsocket(),
    outbox :: efka_iot_outbox:outbox(),
    %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
    auth_ref = undefined :: undefined | binary(),
    ping_timer_ref = undefined :: undefined | reference(),
    dropped_message_count = 0 :: non_neg_integer()
}).

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

%% 发送数据
-spec metric_data(RouteKey :: binary(), Metric :: binary()) -> ok.
metric_data(RouteKey, Metric) when is_binary(RouteKey), is_binary(Metric) ->
    gen_statem:cast(?SERVER, {metric_data, RouteKey, Metric}).

-spec task_event_stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> ok.
task_event_stream(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
    gen_statem:cast(?SERVER, {task_event_stream, TaskId, Type, Stream}).

-spec close_task_event_stream(TaskId :: integer(), Reason :: binary()) -> ok.
close_task_event_stream(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
    gen_statem:cast(?SERVER, {close_task_event_stream, TaskId, Reason}).

-spec is_activated() -> boolean().
is_activated() ->
    gen_statem:call(?SERVER, is_activated).

-spec dropped_message_count() -> non_neg_integer().
dropped_message_count() ->
    gen_statem:call(?SERVER, dropped_message_count).

-spec ping(term(), term(), term(), term(), term(), term(), term(), term(), term(), term(), term(), term(), term()) -> ok.
ping(AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces) ->
    gen_statem:cast(?SERVER, {ping, AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces}).

-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
start_link() ->
    gen_statem:start_link({local, ?SERVER}, ?MODULE, [], []).

%%%===================================================================
%%% gen_statem callbacks
%%%===================================================================

-spec init(list()) -> {ok, atom(), #state{}}.
init([]) ->
    case efka_iot_outbox:open(outbox_options()) of
        {ok, Outbox} ->
            erlang:start_timer(0, self(), create_transport),
            {ok, ?STATE_DISCONNECTED, #state{socket = undefined, outbox = Outbox}};
        {error, Reason} ->
            {stop, Reason}
    end.

-spec callback_mode() -> handle_event_function.
callback_mode() ->
    handle_event_function.

-spec outbox_options() -> map().
outbox_options() ->
    {ok, DetsDir} = application:get_env(efka, dets_dir),
    #{
        dir => filename:join(DetsDir, "iot_outbox"),
        segment_record_limit => ?OUTBOX_SEGMENT_RECORD_LIMIT,
        max_segments => ?OUTBOX_MAX_SEGMENTS,
        max_record_bytes => ?OUTBOX_MAX_RECORD_BYTES
    }.

%% 异步发送数据，连接存在时直接发送；否则写入持久化 outbox。
-spec handle_event(term(), term(), atom(), #state{}) -> term().
handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->
    Packet = term_to_binary({<<"message">>, {<<"data">>, #{<<"route_key">> => RouteKey, <<"metric">> => Metric}}}),
    case StateName of
        ?STATE_ACTIVATED ->
            ok = ssl:send(Socket, Packet),
            {keep_state, State};
        _ ->
            case efka_iot_outbox:append(Packet, State#state.outbox) of
                {ok, Outbox} ->
                    {keep_state, State#state{outbox = Outbox}};
                {dropped, capacity_reached, Outbox} ->
                    logger:warning("[efka_iot_client] outbox capacity reached, drop offline metric"),
                    {keep_state, State#state{
                        outbox = Outbox,
                        dropped_message_count = State#state.dropped_message_count + 1
                    }};
                {error, Reason} ->
                    logger:warning("[efka_iot_client] append outbox failed, reason: ~p", [Reason]),
                    {keep_state, State#state{dropped_message_count = State#state.dropped_message_count + 1}}
            end
    end;

%% Task的stream流，只做实时的
handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
    logger:debug("[efka_iot_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => Type, <<"stream">> => Stream}}}),
    ok = ssl:send(Socket, Packet),
    {keep_state, State};

handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => <<"close">>, <<"stream">> => Reason}}}),
    ok = ssl:send(Socket, Packet),
    {keep_state, State};

%% 其他情况下直接忽略
handle_event(cast, _, _, State = #state{}) ->
    {keep_state, State};

handle_event({call, From}, is_activated, ?STATE_ACTIVATED, State = #state{}) ->
    {keep_state, State, [{reply, From, true}]};
handle_event({call, From}, is_activated, _StateName, State = #state{}) ->
    {keep_state, State, [{reply, From, false}]};
handle_event({call, From}, dropped_message_count, _StateName, State = #state{dropped_message_count = DroppedCount}) ->
    {keep_state, State, [{reply, From, DroppedCount}]};

%% 异步建立到服务器的连接
handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) ->
    case connect_socket() of
        {ok, Socket} ->
            Ref = request_ref(),
            AuthPacket = auth_packet(Ref),
            ok = ssl:send(Socket, AuthPacket),
            logger:debug("[efka_iot_client] send auth request, ref: ~p", [Ref]),
            {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
        {error, _Reason} ->
            schedule_reconnect(),
            {keep_state, State#state{socket = undefined}}
    end;

handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket}) ->
    logger:debug("[efka_iot_client] auth request timeout"),
    disconnect(Socket),
    schedule_reconnect(),
    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};

handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) ->
    Packet = term_to_binary({<<"message">>, <<"ping">>}),
    case ssl:send(Socket, Packet) of
        ok ->
            {keep_state, schedule_ssl_ping(State)};
        {error, Reason} ->
            logger:warning("[efka_iot_client] send ssl ping failed, reason: ~p", [Reason]),
            disconnect(Socket),
            schedule_reconnect(),
            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
    end;
handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) ->
    {keep_state, State};

%% 将缓存中的数据推送到服务器端
handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
    case efka_iot_outbox:next(State#state.outbox) of
        {ok, Seq, Packet} ->
            ok = ssl:send(Socket, Packet),
            case efka_iot_outbox:ack(Seq, State#state.outbox) of
                {ok, Outbox} ->
                    {keep_state, State#state{outbox = Outbox}, [{next_event, info, flush_cache}]};
                {error, Reason} ->
                    logger:warning("[efka_iot_client] ack outbox failed, seq: ~p, reason: ~p", [Seq, Reason]),
                    {keep_state, State}
            end;
        eof ->
            {keep_state, State};
        {error, Reason} ->
            logger:warning("[efka_iot_client] read outbox failed, reason: ~p", [Reason]),
            {keep_state, State}
    end;
handle_event(info, flush_cache, _, State) ->
    {keep_state, State};

%% 处理收到的ssl消息
handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) ->
    try binary_to_term(PacketBin, [safe]) of
        Packet ->
            {keep_state, State, [{next_event, internal, Packet}]}
    catch
        error:Error ->
            logger:warning("[efka_iot_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
            disconnect(Socket),
            cancel_ssl_ping(State),
            schedule_reconnect(),
            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
    end;
handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
    logger:debug("[efka_iot_client] ssl error: ~p", [Reason]),
    disconnect(Socket),
    cancel_ssl_ping(State),
    schedule_reconnect(),
    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
handle_event(info, {ssl_closed, Socket}, _, State = #state{}) ->
    logger:debug("[efka_iot_client] ssl closed"),
    disconnect(Socket),
    cancel_ssl_ping(State),
    schedule_reconnect(),
    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};

%%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配

%% 容器管理命令由 iot 发起，使用 command/command_response 语义。
handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
    handle_container_command(Ref, Request, Socket),
    {keep_state, State};
handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) ->
    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
    {keep_state, State};

%% 处理response
handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
    logger:debug("[efka_iot_client] auth success"),
    State1 = schedule_ssl_ping(State#state{auth_ref = undefined}),
    {next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]};
handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
    logger:debug("[efka_iot_client] auth failed, reason: ~p", [Reason]),
    disconnect(Socket),
    schedule_reconnect(),
    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
handle_event(internal, {<<"response">>, _Ref, Reply}, StateName, State) ->
    logger:warning("[efka_iot_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
    {keep_state, State};
handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State) ->
    logger:warning("[efka_iot_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]),
    {keep_state, State};

%% 处理Pub/Sub机制
handle_event(internal, {<<"message">>, <<"pong">>}, ?STATE_ACTIVATED, State) ->
    {keep_state, State};
handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) ->
    logger:debug("[efka_iot_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
    efka_subscription:publish(Topic, Qos, Content),
    {keep_state, State};
handle_event(internal, Packet, _StateName, State) ->
    logger:warning("[efka_iot_client] ignore unknown packet: ~p", [Packet]),
    {keep_state, State};

handle_event(info, Info, _, State = #state{}) ->
    logger:notice("[efka_iot_client] get unknown info: ~p", [Info]),
    {keep_state, State}.

-spec handle_container_command(binary(), term(), ssl:sslsocket()) -> ok.
handle_container_command(Ref, #{<<"action">> := <<"list">>}, Socket) ->
    Reply = docker_commands:get_containers(),
    send_container_response(Socket, Ref, Reply),
    ok;
handle_container_command(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) ->
    Reply = docker_deploy_manager:deploy(TaskId, Params),
    send_container_response(Socket, Ref, Reply),
    ok;
handle_container_command(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) ->
    Reply = docker_commands:start_container(container_target(Target)),
    send_container_response(Socket, Ref, Reply),
    ok;
handle_container_command(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) ->
    Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
    send_container_response(Socket, Ref, Reply),
    ok;
handle_container_command(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) ->
    Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
    send_container_response(Socket, Ref, Reply),
    ok;
handle_container_command(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) ->
    Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
    send_container_response(Socket, Ref, Reply),
    ok;
handle_container_command(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) ->
    Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
    send_container_response(Socket, Ref, Reply),
    ok;
handle_container_command(Ref, Request, Socket) ->
    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
    ok.

-spec terminate(term(), atom(), #state{}) -> ok.
terminate(Reason, _StateName, State = #state{socket = Socket, outbox = Outbox}) ->
    cancel_ssl_ping(State),
    disconnect(Socket),
    efka_iot_outbox:close(Outbox),
    logger:notice("[efka_iot_client] terminate with reason: ~p", [Reason]),
    ok.

-spec code_change(term(), atom(), #state{}, term()) -> {ok, atom(), #state{}}.
code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
    {ok, StateName, State}.

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

-spec auth_packet(binary()) -> binary().
auth_packet(Ref) when is_binary(Ref) ->
    {ok, AuthInfo} = application:get_env(efka, auth),
    UUID = proplists:get_value(uuid, AuthInfo),
    Token = proplists:get_value(token, AuthInfo),

    Timestamp = efka_util:timestamp(),
    term_to_binary({<<"request">>, Ref, {<<"auth_request">>, #{
        <<"uuid">> => list_to_binary(UUID),
        <<"token">> => list_to_binary(Token),
        <<"timestamp">> => Timestamp
    }}}).

-spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
connect_socket() ->
    {ok, Props} = application:get_env(efka, iot_server),
    Host = proplists:get_value(host, Props),
    Port = proplists:get_value(tls_port, Props),
    SslOptions = [
        binary,
        {active, true},
        {packet, 4},
        {verify, verify_none}
    ],
    ssl:connect(Host, Port, SslOptions, 5000).

-spec disconnect(undefined | ssl:sslsocket()) -> ok.
disconnect(undefined) ->
    ok;
disconnect(Socket) ->
    catch ssl:close(Socket),
    ok.

-spec schedule_reconnect() -> reference().
schedule_reconnect() ->
    erlang:start_timer(5000, self(), create_transport).

-spec schedule_ssl_ping(#state{}) -> #state{}.
schedule_ssl_ping(State = #state{ping_timer_ref = TimerRef}) ->
    cancel_timer(TimerRef),
    State#state{ping_timer_ref = erlang:start_timer(?SSL_PING_INTERVAL, self(), ssl_ping)}.

-spec cancel_ssl_ping(#state{}) -> ok.
cancel_ssl_ping(#state{ping_timer_ref = TimerRef}) ->
    cancel_timer(TimerRef).

-spec cancel_timer(undefined | reference()) -> ok.
cancel_timer(undefined) ->
    ok;
cancel_timer(TimerRef) ->
    _ = erlang:cancel_timer(TimerRef),
    ok.

-spec request_ref() -> binary().
request_ref() ->
    crypto:strong_rand_bytes(16).

-spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok.
send_container_response(Socket, Ref, Reply) ->
    Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}),
    ok = ssl:send(Socket, Packet).

-spec safe_reply(term()) -> term().
safe_reply(ok) ->
    <<"ok">>;
safe_reply({ok, Result}) ->
    {<<"ok">>, safe_term(Result)};
safe_reply({error, Reason}) ->
    {<<"error">>, safe_term(Reason)}.

-spec safe_term(term()) -> term().
safe_term(true) ->
    true;
safe_term(false) ->
    false;
safe_term(undefined) ->
    undefined;
safe_term(Value) when is_atom(Value) ->
    atom_to_binary(Value, utf8);
safe_term(Value) when is_map(Value) ->
    maps:from_list([{safe_term(K), safe_term(V)} || {K, V} <- maps:to_list(Value)]);
safe_term(Value) when is_list(Value) ->
    [safe_term(Item) || Item <- Value];
safe_term(Value) when is_tuple(Value) ->
    list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]);
safe_term(Value) ->
    Value.

-spec container_target(map()) -> binary().
container_target(Target) when is_map(Target) ->
    NameBin = to_binary(maps:get(<<"name">>, Target, <<>>)),
    IdBin = to_binary(maps:get(<<"id">>, Target, <<>>)),
    case NameBin of
        <<>> ->
            true = IdBin =/= <<>>,
            IdBin;
        _ ->
            NameBin
    end.

-spec to_binary(binary() | list()) -> binary().
to_binary(Value) when is_binary(Value) ->
    Value;
to_binary(Value) when is_list(Value) ->
    unicode:characters_to_binary(Value).

-spec to_bool(true | false | 0 | 1) -> boolean().
to_bool(true) ->
    true;
to_bool(1) ->
    true;
to_bool(false) ->
    false;
to_bool(0) ->
    false.