%%--------------------------------------------------------------------
%% Copyright (c) 2020-2024 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_retainer_SUITE).

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

-include("emqx_retainer.hrl").

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

all() ->
    [
        {group, mnesia_without_indices},
        {group, mnesia_with_indices},
        {group, mnesia_reindex},
        {group, test_disable_then_start}
    ].

groups() ->
    [
        {mnesia_without_indices, [sequence], common_tests()},
        {mnesia_with_indices, [sequence], common_tests()},
        {mnesia_reindex, [sequence], [t_reindex]},
        {test_disable_then_start, [sequence], [test_disable_then_start]}
    ].

common_tests() ->
    emqx_common_test_helpers:all(?MODULE) -- [t_reindex].

-define(BASE_CONF, <<
    "retainer {\n"
    "    enable = true\n"
    "    msg_clear_interval = 0s\n"
    "    msg_expiry_interval = 0s\n"
    "    max_payload_size = 1MB\n"
    "    flow_control {\n"
    "        batch_read_number = 0\n"
    "        batch_deliver_number = 0\n"
    "     }\n"
    "   backend {\n"
    "        type = built_in_database\n"
    "        storage_type = ram\n"
    "        max_retained_messages = 0\n"
    "     }\n"
    "}"
>>).

%%--------------------------------------------------------------------
%% Setups
%%--------------------------------------------------------------------

init_per_suite(Config) ->
    Apps = emqx_cth_suite:start(
        [emqx, emqx_conf, app_spec()],
        #{work_dir => emqx_cth_suite:work_dir(Config)}
    ),
    [{suite_apps, Apps} | Config].

end_per_suite(Config) ->
    emqx_cth_suite:stop(?config(suite_apps, Config)).

init_per_group(mnesia_without_indices, Config) ->
    mnesia:clear_table(?TAB_INDEX_META),
    mnesia:clear_table(?TAB_INDEX),
    mnesia:clear_table(?TAB_MESSAGE),
    Config;
init_per_group(mnesia_reindex, Config) ->
    emqx_retainer_mnesia:populate_index_meta(),
    mnesia:clear_table(?TAB_INDEX),
    mnesia:clear_table(?TAB_MESSAGE),
    Config;
init_per_group(_, Config) ->
    emqx_retainer_mnesia:populate_index_meta(),
    mnesia:clear_table(?TAB_INDEX),
    mnesia:clear_table(?TAB_MESSAGE),
    Config.

end_per_group(_Group, Config) ->
    emqx_retainer_mnesia:populate_index_meta(),
    Config.

init_per_testcase(t_get_basic_usage_info, Config) ->
    mnesia:clear_table(?TAB_INDEX),
    mnesia:clear_table(?TAB_MESSAGE),
    emqx_retainer_mnesia:populate_index_meta(),
    Config;
init_per_testcase(_TestCase, Config) ->
    Config.

app_spec() ->
    {emqx_retainer, ?BASE_CONF}.

%%--------------------------------------------------------------------
%% Test Cases
%%--------------------------------------------------------------------
t_store_and_clean(_) ->
    {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
    {ok, _} = emqtt:connect(C1),

    emqtt:publish(
        C1,
        <<"retained">>,
        <<"this is a retained message">>,
        [{qos, 0}, {retain, true}]
    ),
    timer:sleep(100),

    {ok, _, List} = emqx_retainer:page_read(<<"retained">>, 1, 10),
    ?assertEqual(1, length(List)),
    ?assertMatch(
        {ok, [#message{payload = <<"this is a retained message">>}]},
        emqx_retainer:read_message(<<"retained">>)
    ),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(1, length(receive_messages(1))),
    ?assertMatch(
        {ok, [#message{payload = <<"this is a retained message">>}]},
        emqx_retainer:read_message(<<"retained">>)
    ),

    {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"retained">>),

    emqtt:publish(C1, <<"retained">>, <<"">>, [{qos, 0}, {retain, true}]),
    timer:sleep(100),
    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(0, length(receive_messages(1))),
    ?assertMatch(
        {ok, []},
        emqx_retainer:read_message(<<"retained">>)
    ),

    ok = emqx_retainer:clean(),
    {ok, _, List2} = emqx_retainer:page_read(<<"retained">>, 1, 10),
    ?assertEqual(0, length(List2)),
    ?assertMatch(
        {ok, []},
        emqx_retainer:read_message(<<"retained">>)
    ),

    ok = emqtt:disconnect(C1).

t_retain_handling(Config) ->
    {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
    {ok, _} = emqtt:connect(C1),

    ok = emqx_retainer:clean(),

    %% rh = 0, no wildcard, and with empty retained message
    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(0, length(receive_messages(1))),
    {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"retained">>),

    %% rh = 0, has wildcard, and with empty retained message
    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/#">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(0, length(receive_messages(1))),
    {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"retained/#">>),

    publish(
        C1,
        <<"retained">>,
        <<"this is a retained message">>,
        [{qos, 0}, {retain, true}],
        Config
    ),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(1, length(receive_messages(1))),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(1, length(receive_messages(1))),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/#">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(1, length(receive_messages(1))),

    {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"retained">>),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 1}]),
    ?assertEqual(1, length(receive_messages(1))),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 1}]),
    ?assertEqual(0, length(receive_messages(1))),

    {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"retained">>),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 2}]),
    ?assertEqual(0, length(receive_messages(1))),

    emqtt:publish(C1, <<"retained">>, <<"">>, [{qos, 0}, {retain, true}]),
    ok = emqtt:disconnect(C1).

t_wildcard_subscription(Config) ->
    {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
    {ok, _} = emqtt:connect(C1),
    emqtt:publish(
        C1,
        <<"retained/0">>,
        <<"this is a retained message 0">>,
        [{qos, 0}, {retain, true}]
    ),
    emqtt:publish(
        C1,
        <<"retained/1">>,
        <<"this is a retained message 1">>,
        [{qos, 0}, {retain, true}]
    ),
    emqtt:publish(
        C1,
        <<"retained/a/b/c">>,
        <<"this is a retained message 2">>,
        [{qos, 0}, {retain, true}]
    ),
    publish(
        C1,
        <<"/x/y/z">>,
        <<"this is a retained message 3">>,
        [{qos, 0}, {retain, true}],
        Config
    ),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/+">>, 0),
    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/+/b/#">>, 0),
    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"/+/y/#">>, 0),
    Msgs = receive_messages(4),
    ?assertEqual(4, length(Msgs), #{msgs => Msgs}),

    emqtt:publish(C1, <<"retained/0">>, <<"">>, [{qos, 0}, {retain, true}]),
    emqtt:publish(C1, <<"retained/1">>, <<"">>, [{qos, 0}, {retain, true}]),
    emqtt:publish(C1, <<"retained/a/b/c">>, <<"">>, [{qos, 0}, {retain, true}]),
    emqtt:publish(C1, <<"/x/y/z">>, <<"">>, [{qos, 0}, {retain, true}]),
    ok = emqtt:disconnect(C1).

t_message_expiry(Config) ->
    ConfMod = fun(Conf) ->
        Conf#{<<"delivery_rate">> := <<"infinity">>}
    end,
    Case = fun() ->
        {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
        {ok, _} = emqtt:connect(C1),

        emqtt:publish(
            C1,
            <<"retained/0">>,
            #{'Message-Expiry-Interval' => 0},
            <<"don't expire">>,
            [{qos, 0}, {retain, true}]
        ),
        emqtt:publish(
            C1,
            <<"retained/1">>,
            #{'Message-Expiry-Interval' => 2},
            <<"expire">>,
            [{qos, 0}, {retain, true}]
        ),
        emqtt:publish(
            C1,
            <<"retained/2">>,
            #{'Message-Expiry-Interval' => 5},
            <<"don't expire">>,
            [{qos, 0}, {retain, true}]
        ),
        emqtt:publish(
            C1,
            <<"retained/3">>,
            <<"don't expire">>,
            [{qos, 0}, {retain, true}]
        ),
        publish(
            C1,
            <<"$SYS/retained/4">>,
            <<"don't expire">>,
            [{qos, 0}, {retain, true}],
            Config
        ),

        {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/+">>, 0),
        {ok, #{}, [0]} = emqtt:subscribe(C1, <<"$SYS/retained/+">>, 0),
        ?assertEqual(5, length(receive_messages(5))),
        {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"retained/+">>),
        {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"$SYS/retained/+">>),

        timer:sleep(3000),
        {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/+">>, 0),
        {ok, #{}, [0]} = emqtt:subscribe(C1, <<"$SYS/retained/+">>, 0),
        ?assertEqual(4, length(receive_messages(5))),

        emqtt:publish(C1, <<"retained/0">>, <<"">>, [{qos, 0}, {retain, true}]),
        emqtt:publish(C1, <<"retained/1">>, <<"">>, [{qos, 0}, {retain, true}]),
        emqtt:publish(C1, <<"retained/2">>, <<"">>, [{qos, 0}, {retain, true}]),
        emqtt:publish(C1, <<"retained/3">>, <<"">>, [{qos, 0}, {retain, true}]),
        emqtt:publish(C1, <<"$SYS/retained/4">>, <<"">>, [{qos, 0}, {retain, true}]),

        ok = emqtt:disconnect(C1)
    end,
    with_conf(ConfMod, Case).

t_message_expiry_2(Config) ->
    ConfMod = fun(Conf) ->
        Conf#{<<"msg_expiry_interval">> := <<"2s">>}
    end,
    Case = fun() ->
        {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
        {ok, _} = emqtt:connect(C1),
        publish(C1, <<"retained">>, <<"expire">>, [{qos, 0}, {retain, true}], Config),

        {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 0}]),
        ?assertEqual(1, length(receive_messages(1))),
        timer:sleep(4000),
        {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 0}]),
        ?assertEqual(0, length(receive_messages(1))),
        {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"retained">>),

        emqtt:publish(C1, <<"retained">>, <<"">>, [{qos, 0}, {retain, true}]),

        ok = emqtt:disconnect(C1)
    end,
    with_conf(ConfMod, Case).

t_table_full(_) ->
    ConfMod = fun(Conf) ->
        Conf#{<<"backend">> => #{<<"max_retained_messages">> => <<"1">>}}
    end,
    Case = fun() ->
        {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
        {ok, _} = emqtt:connect(C1),
        emqtt:publish(C1, <<"retained/t/1">>, <<"a">>, [{qos, 0}, {retain, true}]),
        emqtt:publish(C1, <<"retained/t/2">>, <<"b">>, [{qos, 0}, {retain, true}]),

        {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/t/1">>, [{qos, 0}, {rh, 0}]),
        ?assertEqual(1, length(receive_messages(1))),
        {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/t/2">>, [{qos, 0}, {rh, 0}]),
        ?assertEqual(0, length(receive_messages(1))),

        ok = emqtt:disconnect(C1)
    end,
    with_conf(ConfMod, Case).

t_clean(Config) ->
    {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
    {ok, _} = emqtt:connect(C1),
    emqtt:publish(
        C1,
        <<"retained/0">>,
        <<"this is a retained message 0">>,
        [{qos, 0}, {retain, true}]
    ),
    emqtt:publish(
        C1,
        <<"retained/1">>,
        <<"this is a retained message 1">>,
        [{qos, 0}, {retain, true}]
    ),
    publish(
        C1,
        <<"retained/test/0">>,
        <<"this is a retained message 2">>,
        [{qos, 0}, {retain, true}],
        Config
    ),
    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/#">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(3, length(receive_messages(3))),

    ok = emqx_retainer:delete(<<"retained/test/0">>),
    ok = emqx_retainer:delete(<<"retained/+">>),
    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/#">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(0, length(receive_messages(3))),

    ok = emqtt:disconnect(C1).

t_stop_publish_clear_msg(_) ->
    emqx_retainer:update_config(#{<<"stop_publish_clear_msg">> => true}),
    {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
    {ok, _} = emqtt:connect(C1),
    emqtt:publish(
        C1,
        <<"retained/0">>,
        <<"this is a retained message 0">>,
        [{qos, 0}, {retain, true}]
    ),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/#">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(1, length(receive_messages(1))),

    emqtt:publish(C1, <<"retained/0">>, <<"">>, [{qos, 0}, {retain, true}]),
    ?assertEqual(0, length(receive_messages(1))),

    emqx_retainer:update_config(#{<<"stop_publish_clear_msg">> => false}),
    ok = emqtt:disconnect(C1).

t_flow_control(_) ->
    Rate = emqx_ratelimiter_SUITE:to_rate("1/1s"),
    LimiterCfg = make_limiter_cfg(Rate),
    JsonCfg = make_limiter_json(<<"1/1s">>),
    emqx_limiter_server:add_bucket(emqx_retainer, internal, LimiterCfg),
    emqx_retainer:update_config(#{
        <<"delivery_rate">> => <<"1/1s">>,
        <<"flow_control">> =>
            #{
                <<"batch_read_number">> => 1,
                <<"batch_deliver_number">> => 1,
                <<"batch_deliver_limiter">> => JsonCfg
            }
    }),
    {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
    {ok, _} = emqtt:connect(C1),
    emqtt:publish(
        C1,
        <<"retained/0">>,
        <<"this is a retained message 0">>,
        [{qos, 0}, {retain, true}]
    ),
    emqtt:publish(
        C1,
        <<"retained/1">>,
        <<"this is a retained message 1">>,
        [{qos, 0}, {retain, true}]
    ),
    emqtt:publish(
        C1,
        <<"retained/3">>,
        <<"this is a retained message 3">>,
        [{qos, 0}, {retain, true}]
    ),
    Begin = erlang:system_time(millisecond),
    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/#">>, [{qos, 0}, {rh, 0}]),
    ?assertEqual(3, length(receive_messages(3))),
    End = erlang:system_time(millisecond),
    Diff = End - Begin,

    ?assert(
        Diff > timer:seconds(2.5) andalso Diff < timer:seconds(3.9),
        lists:flatten(io_lib:format("Diff is :~p~n", [Diff]))
    ),

    ok = emqtt:disconnect(C1),

    emqx_limiter_server:del_bucket(emqx_retainer, internal),
    emqx_retainer:update_config(#{
        <<"flow_control">> =>
            #{
                <<"batch_read_number">> => 1,
                <<"batch_deliver_number">> => 1
            }
    }),
    ok.

t_clear_expired(_) ->
    ConfMod = fun(Conf) ->
        Conf#{
            <<"msg_clear_interval">> := <<"1s">>,
            <<"msg_expiry_interval">> := <<"3s">>
        }
    end,

    Case = fun() ->
        {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
        {ok, _} = emqtt:connect(C1),

        lists:foreach(
            fun(I) ->
                emqtt:publish(
                    C1,
                    <<"retained/", (I + 60):8/unsigned-integer>>,
                    #{'Message-Expiry-Interval' => 3},
                    <<"retained">>,
                    [{qos, 0}, {retain, true}]
                )
            end,
            lists:seq(1, 5)
        ),
        timer:sleep(1000),

        {ok, _, List} = emqx_retainer:page_read(<<"retained/+">>, 1, 10),
        ?assertEqual(5, erlang:length(List)),

        timer:sleep(4500),

        {ok, _, List2} = emqx_retainer:page_read(<<"retained/+">>, 1, 10),
        ?assertEqual(0, erlang:length(List2)),

        ok = emqtt:disconnect(C1)
    end,
    with_conf(ConfMod, Case).

t_max_payload_size(_) ->
    ConfMod = fun(Conf) -> Conf#{<<"max_payload_size">> := <<"1kb">>} end,
    Case = fun() ->
        emqx_retainer:clean(),
        timer:sleep(500),
        {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
        {ok, _} = emqtt:connect(C1),
        Payload = iolist_to_binary(lists:duplicate(1024, <<"0">>)),
        emqtt:publish(
            C1,
            <<"retained/1">>,
            #{},
            Payload,
            [{qos, 0}, {retain, true}]
        ),
        emqtt:publish(
            C1,
            <<"retained/2">>,
            #{},
            <<"1", Payload/binary>>,
            [{qos, 0}, {retain, true}]
        ),

        timer:sleep(500),
        {ok, _, List} = emqx_retainer:page_read(<<"retained/+">>, 1, 10),
        ?assertEqual(1, erlang:length(List)),

        ok = emqtt:disconnect(C1)
    end,
    with_conf(ConfMod, Case).

t_page_read(_) ->
    {ok, C1} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
    {ok, _} = emqtt:connect(C1),
    ok = emqx_retainer:clean(),
    timer:sleep(500),

    Fun = fun(I) ->
        emqtt:publish(
            C1,
            <<"retained/", (I + 60)>>,
            <<"this is a retained message">>,
            [{qos, 0}, {retain, true}]
        )
    end,
    lists:foreach(Fun, lists:seq(1, 9)),
    timer:sleep(200),

    {ok, _, List} = emqx_retainer:page_read(<<"retained/+">>, 1, 5),
    ?assertEqual(5, length(List)),

    {ok, _, List2} = emqx_retainer:page_read(<<"retained/+">>, 2, 5),
    ?assertEqual(4, length(List2)),

    ok = emqtt:disconnect(C1).

t_only_for_coverage(_) ->
    ?assertEqual(retainer, emqx_retainer_schema:namespace()),
    ignored = gen_server:call(emqx_retainer, unexpected),
    ok = gen_server:cast(emqx_retainer, unexpected),
    unexpected = erlang:send(erlang:whereis(emqx_retainer), unexpected),

    Dispatcher = emqx_retainer_dispatcher:worker(),
    ignored = gen_server:call(Dispatcher, unexpected),
    ok = gen_server:cast(Dispatcher, unexpected),
    unexpected = erlang:send(Dispatcher, unexpected),
    true = erlang:exit(Dispatcher, normal),
    ok.

t_reindex(_) ->
    {ok, C} = emqtt:start_link([{clean_start, true}, {proto_ver, v5}]),
    {ok, _} = emqtt:connect(C),

    ok = emqx_retainer:clean(),
    ok = emqx_retainer_mnesia:reindex([[1, 3]], false, fun(_Done) -> ok end),

    %% Prepare retained messages for "retained/N1/N2" topics
    ?check_trace(
        ?wait_async_action(
            lists:foreach(
                fun(N1) ->
                    lists:foreach(
                        fun(N2) ->
                            emqtt:publish(
                                C,
                                erlang:iolist_to_binary([
                                    <<"retained/">>,
                                    io_lib:format("~5..0w", [N1]),
                                    <<"/">>,
                                    io_lib:format("~5..0w", [N2])
                                ]),
                                <<"this is a retained message">>,
                                [{qos, 0}, {retain, true}]
                            )
                        end,
                        lists:seq(1, 10)
                    )
                end,
                lists:seq(1, 1000)
            ),
            #{?snk_kind := message_retained, topic := <<"retained/01000/00010">>},
            5000
        ),
        []
    ),

    ?check_trace(
        ?wait_async_action(
            begin
                %% Spawn reindexing in the background
                spawn_link(
                    fun() ->
                        timer:sleep(1000),
                        emqx_retainer_mnesia:reindex(
                            [[1, 4]],
                            false,
                            fun(Done) ->
                                ?tp(
                                    info,
                                    reindexing_progress,
                                    #{done => Done}
                                )
                            end
                        )
                    end
                ),

                %% Subscribe to "retained/N/+" for some time, while reindexing is in progress
                T = erlang:monotonic_time(millisecond),
                ok = test_retain_while_reindexing(C, T + 3000)
            end,
            #{?snk_kind := reindexing_progress, done := 10000},
            10000
        ),
        fun(Trace) ->
            ?assertMatch(
                [_ | _],
                lists:filter(
                    fun
                        (#{done := 10000}) -> true;
                        (_) -> false
                    end,
                    ?of_kind(reindexing_progress, Trace)
                )
            )
        end
    ).

t_get_basic_usage_info(_Config) ->
    ?assertEqual(#{retained_messages => 0}, emqx_retainer:get_basic_usage_info()),
    Context = emqx_retainer:context(),
    lists:foreach(
        fun(N) ->
            Num = integer_to_binary(N),
            Message = emqx_message:make(<<"retained/", Num/binary>>, <<"payload">>),
            ok = emqx_retainer:store_retained(Context, Message)
        end,
        lists:seq(1, 5)
    ),
    ?assertEqual(#{retained_messages => 5}, emqx_retainer:get_basic_usage_info()),
    ok.

%% test whether the app can start normally after disabling emqx_retainer
%% fix: https://github.com/emqx/emqx/pull/8911
test_disable_then_start(_Config) ->
    emqx_retainer:update_config(#{<<"enable">> => false}),
    ?assertNotEqual([], gproc_pool:active_workers(emqx_retainer_dispatcher)),
    ok = application:stop(emqx_retainer),
    timer:sleep(100),
    ?assertEqual([], gproc_pool:active_workers(emqx_retainer_dispatcher)),
    ok = application:ensure_started(emqx_retainer),
    timer:sleep(100),
    ?assertNotEqual([], gproc_pool:active_workers(emqx_retainer_dispatcher)),
    ok.

t_deliver_when_banned(_) ->
    Client1 = <<"c1">>,
    Client2 = <<"c2">>,

    {ok, C1} = emqtt:start_link([{clientid, Client1}, {clean_start, true}, {proto_ver, v5}]),
    {ok, _} = emqtt:connect(C1),

    lists:foreach(
        fun(I) ->
            Topic = erlang:list_to_binary(io_lib:format("retained/~p", [I])),
            Msg = emqx_message:make(Client2, 0, Topic, <<"this is a retained message">>),
            Msg2 = emqx_message:set_flag(retain, Msg),
            emqx:publish(Msg2)
        end,
        lists:seq(1, 3)
    ),

    Now = erlang:system_time(second),
    Who = emqx_banned:who(clientid, Client2),

    emqx_banned:create(#{
        who => Who,
        by => <<"test">>,
        reason => <<"test">>,
        at => Now,
        until => Now + 120
    }),

    timer:sleep(100),

    snabbkaffe:start_trace(),
    {ok, SubRef} =
        snabbkaffe:subscribe(
            ?match_event(#{?snk_kind := ignore_retained_message_deliver}),
            _NEvents = 3,
            _Timeout = 10000,
            0
        ),

    {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/+">>, [{qos, 0}, {rh, 0}]),

    {ok, Trace} = snabbkaffe:receive_events(SubRef),
    ?assertEqual(3, length(?of_kind(ignore_retained_message_deliver, Trace))),
    snabbkaffe:stop(),
    emqx_banned:delete(Who),
    {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"retained/+">>),
    ok = emqtt:disconnect(C1).

t_compatibility_for_deliver_rate(_) ->
    Parser = fun(Conf) ->
        {ok, RawConf} = hocon:binary(Conf, #{format => map}),
        hocon_tconf:check_plain(emqx_retainer_schema, RawConf, #{
            required => false, atom_key => false
        })
    end,
    Infinity = <<"retainer.deliver_rate = \"infinity\"">>,
    ?assertMatch(
        #{
            <<"retainer">> :=
                #{
                    <<"flow_control">> := #{
                        <<"batch_deliver_number">> := 0,
                        <<"batch_read_number">> := 0,
                        <<"batch_deliver_limiter">> := #{<<"rate">> := infinity}
                    }
                }
        },
        Parser(Infinity)
    ),

    R1 = <<"retainer.deliver_rate = \"1000/s\"">>,
    ?assertMatch(
        #{
            <<"retainer">> :=
                #{
                    <<"flow_control">> := #{
                        <<"batch_deliver_number">> := 1000,
                        <<"batch_read_number">> := 1000,
                        <<"batch_deliver_limiter">> := #{<<"client">> := #{<<"rate">> := 100.0}}
                    }
                }
        },
        Parser(R1)
    ),

    R2 = <<
        "retainer{deliver_rate = \"1000/s\"",
        "flow_control.batch_deliver_limiter.rate = \"500/s\"}"
    >>,
    ?assertMatch(
        #{
            <<"retainer">> :=
                #{
                    <<"flow_control">> := #{
                        <<"batch_deliver_number">> := 1000,
                        <<"batch_read_number">> := 1000,
                        <<"batch_deliver_limiter">> := #{<<"client">> := #{<<"rate">> := 100.0}}
                    }
                }
        },
        Parser(R2)
    ),

    DeliveryInf = <<"retainer.delivery_rate = \"infinity\"">>,
    ?assertMatch(
        #{
            <<"retainer">> :=
                #{
                    <<"flow_control">> := #{
                        <<"batch_deliver_number">> := 0,
                        <<"batch_read_number">> := 0,
                        <<"batch_deliver_limiter">> := #{<<"rate">> := infinity}
                    }
                }
        },
        Parser(DeliveryInf)
    ).

t_update_config(_) ->
    OldConf = emqx_config:get_raw([retainer]),
    NewConf = emqx_utils_maps:deep_put([<<"backend">>, <<"storage_type">>], OldConf, <<"disk">>),
    emqx_retainer:update_config(NewConf).

%%--------------------------------------------------------------------
%% Helper functions
%%--------------------------------------------------------------------

test_retain_while_reindexing(C, Deadline) ->
    case erlang:monotonic_time(millisecond) > Deadline of
        true ->
            ok;
        false ->
            N = rand:uniform(1000),
            Topic = iolist_to_binary([
                <<"retained/">>,
                io_lib:format("~5..0w", [N]),
                <<"/+">>
            ]),
            {ok, #{}, [0]} = emqtt:subscribe(C, Topic, [{qos, 0}, {rh, 0}]),
            Messages = receive_messages(10),
            ?assertEqual(10, length(Messages)),
            {ok, #{}, [0]} = emqtt:unsubscribe(C, Topic),
            test_retain_while_reindexing(C, Deadline)
    end.

receive_messages(Count) ->
    receive_messages(Count, []).
receive_messages(0, Msgs) ->
    Msgs;
receive_messages(Count, Msgs) ->
    receive
        {publish, Msg} ->
            ct:log("Msg: ~p ~n", [Msg]),
            receive_messages(Count - 1, [Msg | Msgs]);
        Other ->
            ct:log("Other Msg: ~p~n", [Other]),
            receive_messages(Count, Msgs)
    after 2000 ->
        Msgs
    end.

with_conf(ConfMod, Case) ->
    Conf = emqx:get_raw_config([retainer]),
    NewConf = ConfMod(Conf),
    emqx_retainer:update_config(NewConf),
    try
        Case(),
        {ok, _} = emqx_retainer:update_config(Conf)
    catch
        Type:Error:Strace ->
            emqx_retainer:update_config(Conf),
            erlang:raise(Type, Error, Strace)
    end.

make_limiter_cfg(Rate) ->
    Client = #{
        rate => Rate,
        initial => 0,
        burst => 0,
        low_watermark => 1,
        divisible => false,
        max_retry_time => timer:seconds(5),
        failure_strategy => force
    },
    #{client => Client, rate => Rate, initial => 0, burst => 0}.

make_limiter_json(Rate) ->
    Client = #{
        <<"rate">> => Rate,
        <<"initial">> => 0,
        <<"burst">> => <<"0">>,
        <<"low_watermark">> => 0,
        <<"divisible">> => <<"false">>,
        <<"max_retry_time">> => <<"5s">>,
        <<"failure_strategy">> => <<"force">>
    },
    #{
        <<"client">> => Client,
        <<"rate">> => <<"infinity">>,
        <<"initial">> => 0,
        <<"burst">> => <<"0">>
    }.

publish(Client, Topic, Payload, Opts, TCConfig) ->
    PublishOpts = publish_opts(TCConfig),
    do_publish(Client, Topic, Payload, Opts, PublishOpts).

publish_opts(TCConfig) ->
    Timeout = proplists:get_value(publish_wait_timeout, TCConfig, undefined),
    Predicate =
        case proplists:get_value(publish_wait_predicate, TCConfig, undefined) of
            undefined -> undefined;
            {NEvents, Pred} -> {predicate, {NEvents, Pred, Timeout}};
            Pred -> {predicate, {1, Pred, Timeout}}
        end,
    Sleep =
        case proplists:get_value(sleep_after_publish, TCConfig, undefined) of
            undefined -> undefined;
            Time -> {sleep, Time}
        end,
    emqx_maybe:define(Predicate, Sleep).

do_publish(Client, Topic, Payload, Opts, undefined) ->
    emqtt:publish(Client, Topic, Payload, Opts);
do_publish(Client, Topic, Payload, Opts, {predicate, {NEvents, Predicate, Timeout}}) ->
    %% Do not delete this clause: it's used by other retainer implementation tests
    {ok, SRef0} = snabbkaffe:subscribe(Predicate, NEvents, Timeout),
    Res = emqtt:publish(Client, Topic, Payload, Opts),
    {ok, _} = snabbkaffe:receive_events(SRef0),
    Res;
do_publish(Client, Topic, Payload, Opts, {sleep, Time}) ->
    %% Do not delete this clause: it's used by other retainer implementation tests
    Res = emqtt:publish(Client, Topic, Payload, Opts),
    ct:sleep(Time),
    Res.