%%--------------------------------------------------------------------
%% Copyright (c) 2020-2022 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(emqx_telemetry_SUITE).

-compile(export_all).
-compile(nowarn_export_all).

-include_lib("common_test/include/ct.hrl").
-include_lib("eunit/include/eunit.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").

-import(proplists, [get_value/2]).

all() -> emqx_common_test_helpers:all(?MODULE).

init_per_suite(Config) ->
    ok = meck:new(emqx_authz, [non_strict, passthrough, no_history, no_link]),
    meck:expect(
        emqx_authz,
        acl_conf_file,
        fun() ->
            emqx_common_test_helpers:deps_path(emqx_authz, "etc/acl.conf")
        end
    ),
    emqx_common_test_helpers:start_apps(
        [emqx_conf, emqx_authn, emqx_authz, emqx_modules],
        fun set_special_configs/1
    ),
    Config.

end_per_suite(_Config) ->
    {ok, _} = emqx:update_config(
        [authorization],
        #{
            <<"no_match">> => <<"allow">>,
            <<"cache">> => #{<<"enable">> => <<"true">>},
            <<"sources">> => []
        }
    ),
    emqx_common_test_helpers:stop_apps([emqx_conf, emqx_authn, emqx_authz, emqx_modules]),
    meck:unload(emqx_authz),
    ok.

init_per_testcase(t_get_telemetry, Config) ->
    DataDir = ?config(data_dir, Config),
    mock_httpc(),
    ok = meck:new(emqx_telemetry, [non_strict, passthrough, no_history, no_link]),
    ok = meck:expect(
        emqx_telemetry,
        read_raw_build_info,
        fun() ->
            Path = filename:join([DataDir, "BUILD_INFO"]),
            {ok, Template} = file:read_file(Path),
            Vars0 = [
                {build_info_arch, "arch"},
                {build_info_wordsize, "64"},
                {build_info_os, "os"},
                {build_info_erlang, "erlang"},
                {build_info_elixir, "elixir"},
                {build_info_relform, "relform"}
            ],
            Vars = [
                {atom_to_list(K), iolist_to_binary(V)}
             || {K, V} <- Vars0
            ],
            Rendered = bbmustache:render(Template, Vars),
            {ok, Rendered}
        end
    ),
    Lwm2mDataDir = emqx_common_test_helpers:deps_path(
        emqx_gateway,
        "test/emqx_gateway_SUITE_data"
    ),
    ok = emqx_gateway_SUITE:setup_fake_usage_data(Lwm2mDataDir),
    {ok, _} = application:ensure_all_started(emqx_gateway),
    Config;
init_per_testcase(t_advanced_mqtt_features, Config) ->
    OldValues = emqx_modules:get_advanced_mqtt_features_in_use(),
    emqx_modules:set_advanced_mqtt_features_in_use(#{
        delayed => false,
        topic_rewrite => false,
        retained => false,
        auto_subscribe => false
    }),
    [{old_values, OldValues} | Config];
init_per_testcase(t_authn_authz_info, Config) ->
    mock_httpc(),
    {ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
    create_authn('mqtt:global', built_in_database),
    create_authn('tcp:default', redis),
    create_authn('ws:default', redis),
    create_authz(postgresql),
    Config;
init_per_testcase(t_enable, Config) ->
    ok = meck:new(emqx_telemetry, [non_strict, passthrough, no_history, no_link]),
    ok = meck:expect(emqx_telemetry, official_version, fun(_) -> true end),
    mock_httpc(),
    Config;
init_per_testcase(t_send_after_enable, Config) ->
    ok = meck:new(emqx_telemetry, [non_strict, passthrough, no_history, no_link]),
    ok = meck:expect(emqx_telemetry, official_version, fun(_) -> true end),
    mock_httpc(),
    Config;
init_per_testcase(t_rule_engine_and_data_bridge_info, Config) ->
    mock_httpc(),
    {ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
    {ok, _} = application:ensure_all_started(emqx_rule_engine),
    ok = application:start(emqx_bridge),
    ok = emqx_bridge_SUITE:setup_fake_telemetry_data(),
    ok = setup_fake_rule_engine_data(),
    Config;
init_per_testcase(t_exhook_info, Config) ->
    mock_httpc(),
    {ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
    ExhookConf =
        #{
            <<"exhook">> =>
                #{
                    <<"servers">> =>
                        [
                            #{
                                <<"name">> => "myhook",
                                <<"url">> => "http://127.0.0.1:9000"
                            }
                        ]
                }
        },
    {ok, _} = emqx_exhook_demo_svr:start(),
    ok = emqx_common_test_helpers:load_config(emqx_exhook_schema, ExhookConf),
    {ok, _} = application:ensure_all_started(emqx_exhook),
    Config;
init_per_testcase(_Testcase, Config) ->
    mock_httpc(),
    Config.

end_per_testcase(t_get_telemetry, _Config) ->
    meck:unload([httpc, emqx_telemetry]),
    application:stop(emqx_gateway),
    ok;
end_per_testcase(t_advanced_mqtt_features, Config) ->
    OldValues = ?config(old_values, Config),
    emqx_modules:set_advanced_mqtt_features_in_use(OldValues);
end_per_testcase(t_authn_authz_info, _Config) ->
    meck:unload([httpc]),
    emqx_authz:update({delete, postgresql}, #{}),
    lists:foreach(
        fun(ChainName) ->
            catch emqx_authn_test_lib:delete_authenticators(
                [authentication],
                ChainName
            )
        end,
        ['mqtt:global', 'tcp:default', 'ws:default']
    ),
    ok;
end_per_testcase(t_enable, _Config) ->
    meck:unload([httpc, emqx_telemetry]);
end_per_testcase(t_send_after_enable, _Config) ->
    meck:unload([httpc, emqx_telemetry]);
end_per_testcase(t_rule_engine_and_data_bridge_info, _Config) ->
    meck:unload(httpc),
    lists:foreach(
        fun(App) ->
            ok = application:stop(App)
        end,
        [
            emqx_bridge,
            emqx_rule_engine
        ]
    ),
    ok;
end_per_testcase(t_exhook_info, _Config) ->
    meck:unload(httpc),
    emqx_exhook_demo_svr:stop(),
    application:stop(emqx_exhook),
    ok;
end_per_testcase(_Testcase, _Config) ->
    meck:unload([httpc]),
    ok.

%%------------------------------------------------------------------------------
%% Internal functions
%%------------------------------------------------------------------------------

t_node_uuid(_) ->
    UUID = emqx_telemetry:generate_uuid(),
    Parts = binary:split(UUID, <<"-">>, [global, trim]),
    ?assertEqual(5, length(Parts)),
    {ok, NodeUUID2} = emqx_telemetry:get_node_uuid(),
    emqx_telemetry:disable(),
    emqx_telemetry:enable(),
    emqx_modules_conf:set_telemetry_status(false),
    emqx_modules_conf:set_telemetry_status(true),
    {ok, NodeUUID3} = emqx_telemetry:get_node_uuid(),
    {ok, NodeUUID4} = emqx_telemetry_proto_v1:get_node_uuid(node()),
    ?assertEqual(NodeUUID2, NodeUUID3),
    ?assertEqual(NodeUUID3, NodeUUID4),
    ?assertMatch({badrpc, nodedown}, emqx_telemetry_proto_v1:get_node_uuid('fake@node')).

t_cluster_uuid(_Config) ->
    {ok, ClusterUUID0} = emqx_telemetry:get_cluster_uuid(),
    {ok, ClusterUUID1} = emqx_telemetry_proto_v1:get_cluster_uuid(node()),
    ?assertEqual(ClusterUUID0, ClusterUUID1),
    {ok, NodeUUID0} = emqx_telemetry:get_node_uuid(),

    Node = start_slave(n1),
    try
        ok = setup_slave(Node),
        {ok, ClusterUUID2} = emqx_telemetry_proto_v1:get_cluster_uuid(Node),
        ?assertEqual(ClusterUUID0, ClusterUUID2),
        {ok, NodeUUID1} = emqx_telemetry_proto_v1:get_node_uuid(Node),
        ?assertNotEqual(NodeUUID0, NodeUUID1),
        ok
    after
        ok = stop_slave(Node)
    end,
    ok.

t_official_version(_) ->
    true = emqx_telemetry:official_version("0.0.0"),
    true = emqx_telemetry:official_version("1.1.1"),
    true = emqx_telemetry:official_version("10.10.10"),
    false = emqx_telemetry:official_version("0.0.0.0"),
    false = emqx_telemetry:official_version("1.1.a"),
    true = emqx_telemetry:official_version("0.0-alpha.1"),
    true = emqx_telemetry:official_version("1.1-alpha.1"),
    true = emqx_telemetry:official_version("10.10-alpha.10"),
    false = emqx_telemetry:official_version("1.1-alpha.0"),
    true = emqx_telemetry:official_version("1.1-beta.1"),
    true = emqx_telemetry:official_version("1.1-rc.1"),
    false = emqx_telemetry:official_version("1.1-alpha.a").

t_get_telemetry(_Config) ->
    {ok, TelemetryData} = emqx_telemetry:get_telemetry(),
    OTPVersion = bin(erlang:system_info(otp_release)),
    ?assertEqual(OTPVersion, get_value(otp_version, TelemetryData)),
    {ok, NodeUUID} = emqx_telemetry:get_node_uuid(),
    {ok, ClusterUUID} = emqx_telemetry:get_cluster_uuid(),
    ?assertEqual(NodeUUID, get_value(uuid, TelemetryData)),
    ?assertEqual(ClusterUUID, get_value(cluster_uuid, TelemetryData)),
    ?assertNotEqual(NodeUUID, ClusterUUID),
    ?assertEqual(0, get_value(num_clients, TelemetryData)),
    BuildInfo = get_value(build_info, TelemetryData),
    ?assertMatch(
        #{
            <<"arch">> := <<_/binary>>,
            <<"elixir">> := <<_/binary>>,
            <<"erlang">> := <<_/binary>>,
            <<"os">> := <<_/binary>>,
            <<"relform">> := <<_/binary>>,
            <<"wordsize">> := Wordsize
        } when is_integer(Wordsize),
        BuildInfo
    ),
    VMSpecs = get_value(vm_specs, TelemetryData),
    ?assert(is_integer(get_value(num_cpus, VMSpecs))),
    ?assert(0 =< get_value(num_cpus, VMSpecs)),
    ?assert(is_integer(get_value(total_memory, VMSpecs))),
    ?assert(0 =< get_value(total_memory, VMSpecs)),
    MQTTRTInsights = get_value(mqtt_runtime_insights, TelemetryData),
    ?assert(is_number(maps:get(messages_sent_rate, MQTTRTInsights))),
    ?assert(is_number(maps:get(messages_received_rate, MQTTRTInsights))),
    ?assert(is_integer(maps:get(num_topics, MQTTRTInsights))),
    ?assert(is_map(get_value(authn_authz, TelemetryData))),
    GatewayInfo = get_value(gateway, TelemetryData),
    ?assert(is_map(GatewayInfo)),
    lists:foreach(
        fun({GatewayType, GatewayData}) ->
            ?assertMatch(
                #{
                    authn := GwAuthn,
                    num_clients := NClients,
                    listeners := Ls
                } when
                    is_binary(GwAuthn) andalso
                        is_integer(NClients) andalso
                        is_list(Ls),
                GatewayData,
                #{gateway_type => GatewayType}
            ),
            ListenersData = maps:get(listeners, GatewayData),
            lists:foreach(
                fun(L) -> assert_gateway_listener_shape(L, GatewayType) end,
                ListenersData
            )
        end,
        maps:to_list(GatewayInfo)
    ),
    ok.

t_advanced_mqtt_features(_) ->
    {ok, TelemetryData} = emqx_telemetry:get_telemetry(),
    AdvFeats = get_value(advanced_mqtt_features, TelemetryData),
    ?assertEqual(
        #{
            retained => 0,
            topic_rewrite => 0,
            auto_subscribe => 0,
            delayed => 0
        },
        AdvFeats
    ),
    lists:foreach(
        fun(TelemetryKey) ->
            EnabledFeats = emqx_modules:get_advanced_mqtt_features_in_use(),
            emqx_modules:set_advanced_mqtt_features_in_use(EnabledFeats#{TelemetryKey => true}),
            {ok, Data} = emqx_telemetry:get_telemetry(),
            #{TelemetryKey := Value} = get_value(advanced_mqtt_features, Data),
            ?assertEqual(1, Value, #{key => TelemetryKey})
        end,
        [
            retained,
            topic_rewrite,
            auto_subscribe,
            delayed
        ]
    ),
    ok.

t_authn_authz_info(_) ->
    {ok, TelemetryData} = emqx_telemetry:get_telemetry(),
    AuthnAuthzInfo = get_value(authn_authz, TelemetryData),
    ?assertEqual(
        #{
            authn =>
                [
                    <<"password_based:built_in_database">>,
                    <<"password_based:redis">>
                ],
            authn_listener => #{<<"password_based:redis">> => 2},
            authz => [postgresql]
        },
        AuthnAuthzInfo
    ).

t_enable(_) ->
    ok = emqx_telemetry:enable(),
    ok = emqx_telemetry:disable().

t_send_after_enable(_) ->
    ok = emqx_telemetry:disable(),
    ok = snabbkaffe:start_trace(),
    try
        ok = emqx_telemetry:enable(),
        ?assertMatch({ok, _}, ?block_until(#{?snk_kind := telemetry_data_reported}, 2000, 100)),
        receive
            {request, post, _URL, _Headers, Body} ->
                {ok, Decoded} = emqx_json:safe_decode(Body, [return_maps]),
                ?assertMatch(
                    #{
                        <<"uuid">> := _,
                        <<"messages_received">> := _,
                        <<"messages_sent">> := _,
                        <<"build_info">> := #{},
                        <<"vm_specs">> :=
                            #{
                                <<"num_cpus">> := _,
                                <<"total_memory">> := _
                            },
                        <<"mqtt_runtime_insights">> :=
                            #{
                                <<"messages_received_rate">> := _,
                                <<"messages_sent_rate">> := _,
                                <<"num_topics">> := _
                            },
                        <<"advanced_mqtt_features">> :=
                            #{
                                <<"retained">> := _,
                                <<"topic_rewrite">> := _,
                                <<"auto_subscribe">> := _,
                                <<"delayed">> := _
                            }
                    },
                    Decoded
                )
        after 2100 ->
            exit(telemetry_not_reported)
        end
    after
        ok = snabbkaffe:stop()
    end.

t_mqtt_runtime_insights(_) ->
    State0 = emqx_telemetry:empty_state(),
    {MQTTRTInsights1, State1} = emqx_telemetry:mqtt_runtime_insights(State0),
    ?assertEqual(
        #{
            messages_sent_rate => 0.0,
            messages_received_rate => 0.0,
            num_topics => 0
        },
        MQTTRTInsights1
    ),
    %% add some fake stats
    emqx_metrics:set('messages.sent', 10_000_000_000),
    emqx_metrics:set('messages.received', 20_000_000_000),
    emqx_stats:setstat('topics.count', 30_000),
    {MQTTRTInsights2, _State2} = emqx_telemetry:mqtt_runtime_insights(State1),
    assert_approximate(MQTTRTInsights2, messages_sent_rate, "16.53"),
    assert_approximate(MQTTRTInsights2, messages_received_rate, "33.07"),
    ?assertEqual(30_000, maps:get(num_topics, MQTTRTInsights2)),
    ok.

t_rule_engine_and_data_bridge_info(_Config) ->
    {ok, TelemetryData} = emqx_telemetry:get_telemetry(),
    RuleInfo = get_value(rule_engine, TelemetryData),
    BridgeInfo = get_value(bridge, TelemetryData),
    ?assertEqual(
        #{num_rules => 2},
        RuleInfo
    ),
    ?assertEqual(
        #{
            data_bridge =>
                #{
                    http => #{num => 1, num_linked_by_rules => 3},
                    mqtt => #{num => 1, num_linked_by_rules => 1}
                },
            num_data_bridges => 2
        },
        BridgeInfo
    ),
    ok.

t_exhook_info(_Config) ->
    {ok, TelemetryData} = emqx_telemetry:get_telemetry(),
    ExhookInfo = get_value(exhook, TelemetryData),
    ?assertEqual(1, maps:get(num_servers, ExhookInfo)),
    [Server] = maps:get(servers, ExhookInfo),
    ?assertEqual(grpc, maps:get(driver, Server)),
    Hooks = maps:get(hooks, Server),
    ?assertEqual(
        [
            'client.authenticate',
            'client.authorize',
            'client.connack',
            'client.connect',
            'client.connected',
            'client.disconnected',
            'client.subscribe',
            'client.unsubscribe',
            'message.acked',
            'message.delivered',
            'message.dropped',
            'message.publish',
            'session.created',
            'session.discarded',
            'session.resumed',
            'session.subscribed',
            'session.takenover',
            'session.terminated',
            'session.unsubscribed'
        ],
        lists:sort(Hooks)
    ),
    ok.

%%------------------------------------------------------------------------------
%% Internal functions
%%------------------------------------------------------------------------------

assert_approximate(Map, Key, Expected) ->
    Value = maps:get(Key, Map),
    ?assertEqual(Expected, float_to_list(Value, [{decimals, 2}])).

bin(L) when is_list(L) ->
    list_to_binary(L);
bin(B) when is_binary(B) ->
    B.

mock_httpc() ->
    TestPID = self(),
    ok = meck:new(httpc, [non_strict, passthrough, no_history, no_link]),
    ok = meck:expect(httpc, request, fun(
        Method, {URL, Headers, _ContentType, Body}, _HTTPOpts, _Opts
    ) ->
        TestPID ! {request, Method, URL, Headers, Body}
    end).

create_authn(ChainName, built_in_database) ->
    emqx_authentication:initialize_authentication(
        ChainName,
        [
            #{
                mechanism => password_based,
                backend => built_in_database,
                enable => true,
                user_id_type => username,
                password_hash_algorithm => #{
                    name => plain,
                    salt_position => suffix
                }
            }
        ]
    );
create_authn(ChainName, redis) ->
    emqx_authentication:initialize_authentication(
        ChainName,
        [
            #{
                mechanism => password_based,
                backend => redis,
                enable => true,
                user_id_type => username,
                cmd => "HMGET mqtt_user:${username} password_hash salt is_superuser",
                password_hash_algorithm => #{
                    name => plain,
                    salt_position => suffix
                }
            }
        ]
    ).

create_authz(postgresql) ->
    emqx_authz:update(
        append,
        #{
            <<"type">> => <<"postgresql">>,
            <<"enable">> => true,
            <<"server">> => <<"127.0.0.1:27017">>,
            <<"pool_size">> => 1,
            <<"database">> => <<"mqtt">>,
            <<"username">> => <<"xx">>,
            <<"password">> => <<"ee">>,
            <<"auto_reconnect">> => true,
            <<"ssl">> => #{<<"enable">> => false},
            <<"query">> => <<"abcb">>
        }
    ).

assert_gateway_listener_shape(ListenerData, GatewayType) ->
    ?assertMatch(
        #{type := LType, authn := LAuthn} when
            is_atom(LType) andalso is_binary(LAuthn),
        ListenerData,
        #{gateway_type => GatewayType}
    ).

setup_fake_rule_engine_data() ->
    {ok, _} =
        emqx_rule_engine:create_rule(
            #{
                id => <<"rule:t_get_basic_usage_info:1">>,
                sql => <<"select 1 from topic">>,
                outputs =>
                    [
                        #{function => <<"erlang:hibernate">>, args => #{}},
                        #{function => console},
                        <<"http:my_http_bridge">>,
                        <<"http:my_http_bridge">>
                    ]
            }
        ),
    {ok, _} =
        emqx_rule_engine:create_rule(
            #{
                id => <<"rule:t_get_basic_usage_info:2">>,
                sql => <<"select 1 from topic">>,
                outputs =>
                    [
                        <<"mqtt:my_mqtt_bridge">>,
                        <<"http:my_http_bridge">>
                    ]
            }
        ),
    ok.

set_special_configs(emqx_authz) ->
    {ok, _} = emqx:update_config([authorization, cache, enable], false),
    {ok, _} = emqx:update_config([authorization, no_match], deny),
    {ok, _} = emqx:update_config([authorization, sources], []),
    ok;
set_special_configs(_App) ->
    ok.

start_slave(Name) ->
    % We want VMs to only occupy a single core
    CommonBeamOpts = "+S 1:1 ",
    {ok, Node} = slave:start_link(host(), Name, CommonBeamOpts ++ ebin_path()),
    Node.

%% for some unknown reason, gen_rpc running locally or in CI might
%% start with different `port_discovery' modes, which means that'll
%% either be listening at the port in the config (`tcp_server_port',
%% 5369) if `manual', else it'll listen on 5370 if started as
%% `stateless'.
find_gen_rpc_port() ->
    [EPort] = [
        EPort
     || {links, Ls} <- process_info(whereis(gen_rpc_server_tcp)),
        EPort <- Ls,
        is_port(EPort)
    ],
    {ok, {_, Port}} = inet:sockname(EPort),
    Port.

setup_slave(Node) ->
    TestNode = node(),
    Port = find_gen_rpc_port(),
    [ok = rpc:call(Node, application, load, [App]) || App <- [gen_rpc, emqx]],
    ok = rpc:call(
        Node,
        application,
        set_env,
        [gen_rpc, tcp_server_port, 9002]
    ),
    ok = rpc:call(
        Node,
        application,
        set_env,
        [gen_rpc, client_config_per_node, {internal, #{TestNode => Port}}]
    ),
    ok = rpc:call(
        Node,
        application,
        set_env,
        [gen_rpc, port_discovery, manual]
    ),
    Handler =
        fun
            (emqx) ->
                application:set_env(
                    emqx,
                    boot_modules,
                    []
                ),
                ekka:join(TestNode),
                ok;
            (_) ->
                ok
        end,
    ok = rpc:call(
        Node,
        emqx_common_test_helpers,
        start_apps,
        [
            [emqx_conf, emqx_modules],
            Handler
        ]
    ),
    ok.

stop_slave(Node) ->
    slave:stop(Node).

host() ->
    [_, Host] = string:tokens(atom_to_list(node()), "@"),
    Host.

ebin_path() ->
    string:join(["-pa" | lists:filter(fun is_lib/1, code:get_path())], " ").

is_lib(Path) ->
    string:prefix(Path, code:lib_dir()) =:= nomatch.