%%--------------------------------------------------------------------
%% Copyright (c) 2022-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_bridge_greptimedb_SUITE).

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

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

%%------------------------------------------------------------------------------
%% CT boilerplate
%%------------------------------------------------------------------------------

all() ->
    [
        {group, with_batch},
        {group, without_batch}
    ].

groups() ->
    TCs = emqx_common_test_helpers:all(?MODULE),
    [
        {with_batch, [
            {group, sync_query},
            {group, async_query}
        ]},
        {without_batch, [
            {group, sync_query},
            {group, async_query}
        ]},
        {sync_query, [
            {group, grpcv1_tcp}
            %% uncomment tls when we are ready
            %% {group, grpcv1_tls}
        ]},
        {async_query, [
            {group, grpcv1_tcp}
            %% uncomment tls when we are ready
            %% {group, grpcv1_tls}
        ]},
        {grpcv1_tcp, TCs}
        %%{grpcv1_tls, TCs}
    ].

init_per_suite(Config) ->
    Config.

end_per_suite(_Config) ->
    delete_all_bridges(),
    emqx_mgmt_api_test_util:end_suite(),
    ok = emqx_connector_test_helpers:stop_apps([
        emqx_conf, emqx_bridge, emqx_resource, emqx_rule_engine
    ]),
    _ = application:stop(emqx_connector),
    ok.

init_per_group(GreptimedbType, Config0) when
    GreptimedbType =:= grpcv1_tcp;
    GreptimedbType =:= grpcv1_tls
->
    #{
        host := GreptimedbHost,
        port := GreptimedbPort,
        http_port := GreptimedbHttpPort,
        use_tls := UseTLS,
        proxy_name := ProxyName
    } =
        case GreptimedbType of
            grpcv1_tcp ->
                #{
                    host => os:getenv("GREPTIMEDB_GRPCV1_TCP_HOST", "toxiproxy"),
                    port => list_to_integer(os:getenv("GREPTIMEDB_GRPCV1_TCP_PORT", "4001")),
                    http_port => list_to_integer(os:getenv("GREPTIMEDB_HTTP_PORT", "4000")),
                    use_tls => false,
                    proxy_name => "greptimedb_grpc"
                };
            grpcv1_tls ->
                #{
                    host => os:getenv("GREPTIMEDB_GRPCV1_TLS_HOST", "toxiproxy"),
                    port => list_to_integer(os:getenv("GREPTIMEDB_GRPCV1_TLS_PORT", "4001")),
                    http_port => list_to_integer(os:getenv("GREPTIMEDB_HTTP_PORT", "4000")),
                    use_tls => true,
                    proxy_name => "greptimedb_tls"
                }
        end,
    case emqx_common_test_helpers:is_tcp_server_available(GreptimedbHost, GreptimedbHttpPort) of
        true ->
            ProxyHost = os:getenv("PROXY_HOST", "toxiproxy"),
            ProxyPort = list_to_integer(os:getenv("PROXY_PORT", "8474")),
            emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
            ok = start_apps(),
            {ok, _} = application:ensure_all_started(emqx_connector),
            {ok, _} = application:ensure_all_started(greptimedb),
            emqx_mgmt_api_test_util:init_suite(),
            Config = [{use_tls, UseTLS} | Config0],
            {Name, ConfigString, GreptimedbConfig} = greptimedb_config(
                grpcv1, GreptimedbHost, GreptimedbPort, Config
            ),
            EHttpcPoolNameBin = <<(atom_to_binary(?MODULE))/binary, "_http">>,
            EHttpcPoolName = binary_to_atom(EHttpcPoolNameBin),
            {EHttpcTransport, EHttpcTransportOpts} =
                case UseTLS of
                    true -> {tls, [{verify, verify_none}]};
                    false -> {tcp, []}
                end,
            EHttpcPoolOpts = [
                {host, GreptimedbHost},
                {port, GreptimedbHttpPort},
                {pool_size, 1},
                {transport, EHttpcTransport},
                {transport_opts, EHttpcTransportOpts}
            ],
            {ok, _} = ehttpc_sup:start_pool(EHttpcPoolName, EHttpcPoolOpts),
            [
                {proxy_host, ProxyHost},
                {proxy_port, ProxyPort},
                {proxy_name, ProxyName},
                {bridge_type, greptimedb},
                {bridge_name, Name},
                {bridge_config, GreptimedbConfig},
                {greptimedb_host, GreptimedbHost},
                {greptimedb_port, GreptimedbPort},
                {greptimedb_http_port, GreptimedbHttpPort},
                {greptimedb_type, grpcv1},
                {greptimedb_config, GreptimedbConfig},
                {greptimedb_config_string, ConfigString},
                {ehttpc_pool_name, EHttpcPoolName},
                {greptimedb_name, Name}
                | Config
            ];
        false ->
            {skip, no_greptimedb}
    end;
init_per_group(sync_query, Config) ->
    [{query_mode, sync} | Config];
init_per_group(async_query, Config) ->
    [{query_mode, async} | Config];
init_per_group(with_batch, Config) ->
    [{batch_size, 100} | Config];
init_per_group(without_batch, Config) ->
    [{batch_size, 1} | Config];
init_per_group(_Group, Config) ->
    Config.

end_per_group(Group, Config) when
    Group =:= grpcv1_tcp;
    Group =:= grpcv1_tls
->
    ProxyHost = ?config(proxy_host, Config),
    ProxyPort = ?config(proxy_port, Config),
    EHttpcPoolName = ?config(ehttpc_pool_name, Config),
    emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
    ehttpc_sup:stop_pool(EHttpcPoolName),
    delete_bridge(Config),
    _ = application:stop(greptimedb),
    ok;
end_per_group(_Group, _Config) ->
    ok.

init_per_testcase(_Testcase, Config) ->
    delete_all_rules(),
    delete_all_bridges(),
    Config.

end_per_testcase(_Testcase, Config) ->
    ProxyHost = ?config(proxy_host, Config),
    ProxyPort = ?config(proxy_port, Config),
    ok = snabbkaffe:stop(),
    emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
    delete_all_rules(),
    delete_all_bridges(),
    ok.

%%------------------------------------------------------------------------------
%% Helper fns
%%------------------------------------------------------------------------------

start_apps() ->
    %% some configs in emqx_conf app are mandatory
    %% we want to make sure they are loaded before
    %% ekka start in emqx_common_test_helpers:start_apps/1
    emqx_common_test_helpers:render_and_load_app_config(emqx_conf),
    ok = emqx_common_test_helpers:start_apps([emqx_conf]),
    ok = emqx_connector_test_helpers:start_apps([emqx_resource, emqx_bridge, emqx_rule_engine]).

example_write_syntax() ->
    %% N.B.: this single space character is relevant
    <<"${topic},clientid=${clientid}", " ", "payload=${payload},",
        "${clientid}_int_value=${payload.int_key}i,",
        "uint_value=${payload.uint_key}u,"
        "float_value=${payload.float_key},", "undef_value=${payload.undef},",
        "${undef_key}=\"hard-coded-value\",", "bool=${payload.bool}">>.

greptimedb_config(grpcv1 = Type, GreptimedbHost, GreptimedbPort, Config) ->
    BatchSize = proplists:get_value(batch_size, Config, 100),
    QueryMode = proplists:get_value(query_mode, Config, sync),
    UseTLS = proplists:get_value(use_tls, Config, false),
    Name = atom_to_binary(?MODULE),
    WriteSyntax = example_write_syntax(),
    ConfigString =
        io_lib:format(
            "bridges.greptimedb.~s {\n"
            "  enable = true\n"
            "  server = \"~p:~b\"\n"
            "  dbname = public\n"
            "  username = greptime_user\n"
            "  password = greptime_pwd\n"
            "  precision = ns\n"
            "  write_syntax = \"~s\"\n"
            "  resource_opts = {\n"
            "    request_ttl = 1s\n"
            "    query_mode = ~s\n"
            "    batch_size = ~b\n"
            "  }\n"
            "  ssl {\n"
            "    enable = ~p\n"
            "    verify = verify_none\n"
            "  }\n"
            "}\n",
            [
                Name,
                GreptimedbHost,
                GreptimedbPort,
                WriteSyntax,
                QueryMode,
                BatchSize,
                UseTLS
            ]
        ),
    {Name, ConfigString, parse_and_check(ConfigString, Type, Name)}.

parse_and_check(ConfigString, Type, Name) ->
    {ok, RawConf} = hocon:binary(ConfigString, #{format => map}),
    TypeBin = greptimedb_type_bin(Type),
    hocon_tconf:check_plain(emqx_bridge_schema, RawConf, #{required => false, atom_key => false}),
    #{<<"bridges">> := #{TypeBin := #{Name := Config}}} = RawConf,
    Config.

greptimedb_type_bin(grpcv1) ->
    <<"greptimedb">>.

create_bridge(Config) ->
    create_bridge(Config, _Overrides = #{}).

create_bridge(Config, Overrides) ->
    Type = greptimedb_type_bin(?config(greptimedb_type, Config)),
    Name = ?config(greptimedb_name, Config),
    GreptimedbConfig0 = ?config(greptimedb_config, Config),
    GreptimedbConfig = emqx_utils_maps:deep_merge(GreptimedbConfig0, Overrides),
    emqx_bridge:create(Type, Name, GreptimedbConfig).

delete_bridge(Config) ->
    Type = greptimedb_type_bin(?config(greptimedb_type, Config)),
    Name = ?config(greptimedb_name, Config),
    emqx_bridge:remove(Type, Name).

delete_all_bridges() ->
    lists:foreach(
        fun(#{name := Name, type := Type}) ->
            emqx_bridge:remove(Type, Name)
        end,
        emqx_bridge:list()
    ).

delete_all_rules() ->
    lists:foreach(
        fun(#{id := RuleId}) ->
            ok = emqx_rule_engine:delete_rule(RuleId)
        end,
        emqx_rule_engine:get_rules()
    ).

create_rule_and_action_http(Config) ->
    create_rule_and_action_http(Config, _Overrides = #{}).

create_rule_and_action_http(Config, Overrides) ->
    GreptimedbName = ?config(greptimedb_name, Config),
    Type = greptimedb_type_bin(?config(greptimedb_type, Config)),
    BridgeId = emqx_bridge_resource:bridge_id(Type, GreptimedbName),
    Params0 = #{
        enable => true,
        sql => <<"SELECT * FROM \"t/topic\"">>,
        actions => [BridgeId]
    },
    Params = emqx_utils_maps:deep_merge(Params0, Overrides),
    Path = emqx_mgmt_api_test_util:api_path(["rules"]),
    AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
    case emqx_mgmt_api_test_util:request_api(post, Path, "", AuthHeader, Params) of
        {ok, Res} -> {ok, emqx_utils_json:decode(Res, [return_maps])};
        Error -> Error
    end.

send_message(Config, Payload) ->
    Name = ?config(greptimedb_name, Config),
    Type = greptimedb_type_bin(?config(greptimedb_type, Config)),
    BridgeId = emqx_bridge_resource:bridge_id(Type, Name),
    Resp = emqx_bridge:send_message(BridgeId, Payload),
    Resp.

query_by_clientid(Topic, ClientId, Config) ->
    GreptimedbHost = ?config(greptimedb_host, Config),
    GreptimedbPort = ?config(greptimedb_http_port, Config),
    EHttpcPoolName = ?config(ehttpc_pool_name, Config),
    UseTLS = ?config(use_tls, Config),
    Path = <<"/v1/sql?db=public">>,
    Scheme =
        case UseTLS of
            true -> <<"https://">>;
            false -> <<"http://">>
        end,
    URI = iolist_to_binary([
        Scheme,
        list_to_binary(GreptimedbHost),
        ":",
        integer_to_binary(GreptimedbPort),
        Path
    ]),
    Headers = [
        {"Authorization", "Basic Z3JlcHRpbWVfdXNlcjpncmVwdGltZV9wd2Q="},
        {"Content-Type", "application/x-www-form-urlencoded"}
    ],
    Body = <<"sql=select * from \"", Topic/binary, "\" where clientid='", ClientId/binary, "'">>,
    {ok, 200, _Headers, RawBody0} =
        ehttpc:request(
            EHttpcPoolName,
            post,
            {URI, Headers, Body},
            _Timeout = 10_000,
            _Retry = 0
        ),

    case emqx_utils_json:decode(RawBody0, [return_maps]) of
        #{
            <<"code">> := 0,
            <<"output">> := [
                #{
                    <<"records">> := #{
                        <<"rows">> := Rows,
                        <<"schema">> := Schema
                    }
                }
            ]
        } ->
            make_row(Schema, Rows);
        #{
            <<"code">> := Code,
            <<"error">> := Error
        } ->
            GreptimedbName = ?config(greptimedb_name, Config),
            Type = greptimedb_type_bin(?config(greptimedb_type, Config)),
            BridgeId = emqx_bridge_resource:bridge_id(Type, GreptimedbName),

            ?SLOG(error, #{
                msg => "failed_to_query",
                code => Code,
                connector => BridgeId,
                reason => Error
            }),
            %% TODO(dennis): check the error by code
            case binary:match(Error, <<"Table not found">>) of
                nomatch ->
                    {error, Error};
                _ ->
                    %% Table not found
                    #{}
            end
    end.

make_row(null, _Rows) ->
    #{};
make_row(_Schema, []) ->
    #{};
make_row(#{<<"column_schemas">> := ColumnsSchemas}, [Row]) ->
    Columns = lists:map(fun(#{<<"name">> := Name}) -> Name end, ColumnsSchemas),
    maps:from_list(lists:zip(Columns, Row)).

assert_persisted_data(ClientId, Expected, PersistedData) ->
    ClientIdIntKey = <<ClientId/binary, "_int_value">>,
    maps:foreach(
        fun
            (int_value, ExpectedValue) ->
                ?assertMatch(
                    ExpectedValue,
                    maps:get(ClientIdIntKey, PersistedData)
                );
            (Key, ExpectedValue) ->
                ?assertMatch(
                    ExpectedValue,
                    maps:get(atom_to_binary(Key), PersistedData),
                    #{expected => ExpectedValue}
                )
        end,
        Expected
    ),
    ok.

resource_id(Config) ->
    Type = greptimedb_type_bin(?config(greptimedb_type, Config)),
    Name = ?config(greptimedb_name, Config),
    emqx_bridge_resource:resource_id(Type, Name).

%%------------------------------------------------------------------------------
%% Testcases
%%------------------------------------------------------------------------------

t_start_ok(Config) ->
    QueryMode = ?config(query_mode, Config),
    ?assertMatch(
        {ok, _},
        create_bridge(Config)
    ),
    ClientId = emqx_guid:to_hexstr(emqx_guid:gen()),
    Payload = #{
        int_key => -123,
        bool => true,
        float_key => 24.5,
        uint_key => 123
    },
    SentData = #{
        <<"clientid">> => ClientId,
        <<"topic">> => atom_to_binary(?FUNCTION_NAME),
        <<"payload">> => Payload,
        <<"timestamp">> => erlang:system_time(millisecond)
    },
    ?check_trace(
        begin
            case QueryMode of
                async ->
                    ?assertMatch(ok, send_message(Config, SentData)),
                    ct:sleep(500);
                sync ->
                    ?assertMatch({ok, _}, send_message(Config, SentData))
            end,
            PersistedData = query_by_clientid(atom_to_binary(?FUNCTION_NAME), ClientId, Config),
            Expected = #{
                bool => true,
                int_value => -123,
                uint_value => 123,
                float_value => 24.5,
                payload => emqx_utils_json:encode(Payload)
            },
            assert_persisted_data(ClientId, Expected, PersistedData),
            ok
        end,
        fun(Trace0) ->
            Trace = ?of_kind(greptimedb_connector_send_query, Trace0),
            ?assertMatch([#{points := [_]}], Trace),
            [#{points := [Point0]}] = Trace,
            {Measurement, [Point]} = Point0,
            ct:pal("sent point: ~p", [Point]),
            ?assertMatch(#{dbname := _, table := _, timeunit := _}, Measurement),
            ?assertMatch(
                #{
                    fields := #{},
                    tags := #{},
                    timestamp := TS
                } when is_integer(TS),
                Point
            ),
            #{fields := Fields} = Point,
            ?assert(lists:all(fun is_binary/1, maps:keys(Fields))),
            ?assertNot(maps:is_key(<<"undefined">>, Fields)),
            ?assertNot(maps:is_key(<<"undef_value">>, Fields)),
            ok
        end
    ),
    ok.

t_start_stop(Config) ->
    %% we can't use this test case directly because `greptimedb_worker' apparently leaks
    %% atoms...
    %% ok = emqx_bridge_testlib:t_start_stop(Config, greptimedb_client_stopped),
    BridgeType = ?config(bridge_type, Config),
    BridgeName = ?config(bridge_name, Config),
    BridgeConfig = ?config(bridge_config, Config),
    StopTracePoint = greptimedb_client_stopped,
    ?check_trace(
        begin
            ProbeRes0 = emqx_bridge_testlib:probe_bridge_api(
                BridgeType,
                BridgeName,
                BridgeConfig
            ),
            ?assertMatch({ok, {{_, 204, _}, _Headers, _Body}}, ProbeRes0),
            ?assertMatch({ok, _}, emqx_bridge:create(BridgeType, BridgeName, BridgeConfig)),
            ResourceId = emqx_bridge_resource:resource_id(BridgeType, BridgeName),

            %% Since the connection process is async, we give it some time to
            %% stabilize and avoid flakiness.
            ?retry(
                _Sleep = 1_000,
                _Attempts = 20,
                ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceId))
            ),

            %% `start` bridge to trigger `already_started`
            ?assertMatch(
                {ok, {{_, 204, _}, _Headers, []}},
                emqx_bridge_testlib:op_bridge_api("start", BridgeType, BridgeName)
            ),

            ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceId)),

            ?assertMatch(
                {{ok, _}, {ok, _}},
                ?wait_async_action(
                    emqx_bridge_testlib:op_bridge_api("stop", BridgeType, BridgeName),
                    #{?snk_kind := StopTracePoint},
                    5_000
                )
            ),

            ?assertEqual(
                {error, resource_is_stopped}, emqx_resource_manager:health_check(ResourceId)
            ),

            ?assertMatch(
                {ok, {{_, 204, _}, _Headers, []}},
                emqx_bridge_testlib:op_bridge_api("stop", BridgeType, BridgeName)
            ),

            ?assertEqual(
                {error, resource_is_stopped}, emqx_resource_manager:health_check(ResourceId)
            ),

            ?assertMatch(
                {ok, {{_, 204, _}, _Headers, []}},
                emqx_bridge_testlib:op_bridge_api("start", BridgeType, BridgeName)
            ),

            ?retry(
                _Sleep = 1_000,
                _Attempts = 20,
                ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceId))
            ),

            %% Disable the bridge, which will also stop it.
            ?assertMatch(
                {{ok, _}, {ok, _}},
                ?wait_async_action(
                    emqx_bridge:disable_enable(disable, BridgeType, BridgeName),
                    #{?snk_kind := StopTracePoint},
                    5_000
                )
            ),

            ok
        end,
        fun(Trace) ->
            ResourceId = emqx_bridge_resource:resource_id(BridgeType, BridgeName),
            %% one for probe, two for real
            ?assertMatch(
                [_, #{instance_id := ResourceId}, #{instance_id := ResourceId}],
                ?of_kind(StopTracePoint, Trace)
            ),
            ok
        end
    ),
    ok.

t_start_already_started(Config) ->
    Type = greptimedb_type_bin(?config(greptimedb_type, Config)),
    Name = ?config(greptimedb_name, Config),
    GreptimedbConfigString = ?config(greptimedb_config_string, Config),
    ?assertMatch({ok, _}, create_bridge(Config)),
    ResourceId = resource_id(Config),
    TypeAtom = binary_to_atom(Type),
    NameAtom = binary_to_atom(Name),
    {ok, #{bridges := #{TypeAtom := #{NameAtom := GreptimedbConfigMap}}}} = emqx_hocon:check(
        emqx_bridge_schema, GreptimedbConfigString
    ),
    ?check_trace(
        emqx_bridge_greptimedb_connector:on_start(ResourceId, GreptimedbConfigMap),
        fun(Result, Trace) ->
            ?assertMatch({ok, _}, Result),
            ?assertMatch([_], ?of_kind(greptimedb_connector_start_already_started, Trace)),
            ok
        end
    ),
    ok.

t_start_ok_timestamp_write_syntax(Config) ->
    GreptimedbType = ?config(greptimedb_type, Config),
    GreptimedbName = ?config(greptimedb_name, Config),
    GreptimedbConfigString0 = ?config(greptimedb_config_string, Config),
    GreptimedbTypeCfg =
        case GreptimedbType of
            grpcv1 -> "greptimedb"
        end,
    WriteSyntax =
        %% N.B.: this single space characters are relevant
        <<"${topic},clientid=${clientid}", " ", "payload=${payload},",
            "${clientid}_int_value=${payload.int_key}i,",
            "uint_value=${payload.uint_key}u,"
            "bool=${payload.bool}", " ", "${timestamp}">>,
    %% append this to override the config
    GreptimedbConfigString1 =
        io_lib:format(
            "bridges.~s.~s {\n"
            "  write_syntax = \"~s\"\n"
            "}\n",
            [GreptimedbTypeCfg, GreptimedbName, WriteSyntax]
        ),
    GreptimedbConfig1 = parse_and_check(
        GreptimedbConfigString0 ++ GreptimedbConfigString1,
        GreptimedbType,
        GreptimedbName
    ),
    Config1 = [{greptimedb_config, GreptimedbConfig1} | Config],
    ?assertMatch(
        {ok, _},
        create_bridge(Config1)
    ),
    ok.

t_start_ok_no_subject_tags_write_syntax(Config) ->
    GreptimedbType = ?config(greptimedb_type, Config),
    GreptimedbName = ?config(greptimedb_name, Config),
    GreptimedbConfigString0 = ?config(greptimedb_config_string, Config),
    GreptimedbTypeCfg =
        case GreptimedbType of
            grpcv1 -> "greptimedb"
        end,
    WriteSyntax =
        %% N.B.: this single space characters are relevant
        <<"${topic}", " ", "payload=${payload},", "${clientid}_int_value=${payload.int_key}i,",
            "uint_value=${payload.uint_key}u,"
            "bool=${payload.bool}", " ", "${timestamp}">>,
    %% append this to override the config
    GreptimedbConfigString1 =
        io_lib:format(
            "bridges.~s.~s {\n"
            "  write_syntax = \"~s\"\n"
            "}\n",
            [GreptimedbTypeCfg, GreptimedbName, WriteSyntax]
        ),
    GreptimedbConfig1 = parse_and_check(
        GreptimedbConfigString0 ++ GreptimedbConfigString1,
        GreptimedbType,
        GreptimedbName
    ),
    Config1 = [{greptimedb_config, GreptimedbConfig1} | Config],
    ?assertMatch(
        {ok, _},
        create_bridge(Config1)
    ),
    ok.

t_const_timestamp(Config) ->
    QueryMode = ?config(query_mode, Config),
    Const = erlang:system_time(nanosecond),
    ConstBin = integer_to_binary(Const),
    ?assertMatch(
        {ok, _},
        create_bridge(
            Config,
            #{
                <<"write_syntax">> =>
                    <<"mqtt,clientid=${clientid} foo=${payload.foo}i,bar=5i ", ConstBin/binary>>
            }
        )
    ),
    ClientId = emqx_guid:to_hexstr(emqx_guid:gen()),
    Payload = #{<<"foo">> => 123},
    SentData = #{
        <<"clientid">> => ClientId,
        <<"topic">> => atom_to_binary(?FUNCTION_NAME),
        <<"payload">> => Payload,
        <<"timestamp">> => erlang:system_time(millisecond)
    },
    case QueryMode of
        async ->
            ?assertMatch(ok, send_message(Config, SentData)),
            ct:sleep(500);
        sync ->
            ?assertMatch({ok, _}, send_message(Config, SentData))
    end,
    PersistedData = query_by_clientid(<<"mqtt">>, ClientId, Config),
    Expected = #{foo => 123},
    assert_persisted_data(ClientId, Expected, PersistedData),
    TimeReturned = maps:get(<<"greptime_timestamp">>, PersistedData),
    ?assertEqual(Const, TimeReturned).

t_boolean_variants(Config) ->
    QueryMode = ?config(query_mode, Config),
    ?assertMatch(
        {ok, _},
        create_bridge(Config)
    ),
    BoolVariants = #{
        true => true,
        false => false,
        <<"t">> => true,
        <<"f">> => false,
        <<"T">> => true,
        <<"F">> => false,
        <<"TRUE">> => true,
        <<"FALSE">> => false,
        <<"True">> => true,
        <<"False">> => false
    },
    maps:foreach(
        fun(BoolVariant, Translation) ->
            ClientId = emqx_guid:to_hexstr(emqx_guid:gen()),
            Payload = #{
                int_key => -123,
                bool => BoolVariant,
                uint_key => 123
            },
            SentData = #{
                <<"clientid">> => ClientId,
                <<"topic">> => atom_to_binary(?FUNCTION_NAME),
                <<"timestamp">> => erlang:system_time(millisecond),
                <<"payload">> => Payload
            },
            case QueryMode of
                sync ->
                    ?assertMatch({ok, _}, send_message(Config, SentData));
                async ->
                    ?assertMatch(ok, send_message(Config, SentData))
            end,
            case QueryMode of
                async -> ct:sleep(500);
                sync -> ok
            end,
            PersistedData = query_by_clientid(atom_to_binary(?FUNCTION_NAME), ClientId, Config),
            Expected = #{
                bool => Translation,
                int_value => -123,
                uint_value => 123,
                payload => emqx_utils_json:encode(Payload)
            },
            assert_persisted_data(ClientId, Expected, PersistedData),
            ok
        end,
        BoolVariants
    ),
    ok.

t_bad_timestamp(Config) ->
    GreptimedbType = ?config(greptimedb_type, Config),
    GreptimedbName = ?config(greptimedb_name, Config),
    QueryMode = ?config(query_mode, Config),
    BatchSize = ?config(batch_size, Config),
    GreptimedbConfigString0 = ?config(greptimedb_config_string, Config),
    GreptimedbTypeCfg =
        case GreptimedbType of
            grpcv1 -> "greptimedb"
        end,
    WriteSyntax =
        %% N.B.: this single space characters are relevant
        <<"${topic}", " ", "payload=${payload},", "${clientid}_int_value=${payload.int_key}i,",
            "uint_value=${payload.uint_key}u,"
            "bool=${payload.bool}", " ", "bad_timestamp">>,
    %% append this to override the config
    GreptimedbConfigString1 =
        io_lib:format(
            "bridges.~s.~s {\n"
            "  write_syntax = \"~s\"\n"
            "}\n",
            [GreptimedbTypeCfg, GreptimedbName, WriteSyntax]
        ),
    GreptimedbConfig1 = parse_and_check(
        GreptimedbConfigString0 ++ GreptimedbConfigString1,
        GreptimedbType,
        GreptimedbName
    ),
    Config1 = [{greptimedb_config, GreptimedbConfig1} | Config],
    ?assertMatch(
        {ok, _},
        create_bridge(Config1)
    ),
    ClientId = emqx_guid:to_hexstr(emqx_guid:gen()),
    Payload = #{
        int_key => -123,
        bool => false,
        uint_key => 123
    },
    SentData = #{
        <<"clientid">> => ClientId,
        <<"topic">> => atom_to_binary(?FUNCTION_NAME),
        <<"timestamp">> => erlang:system_time(millisecond),
        <<"payload">> => Payload
    },
    ?check_trace(
        ?wait_async_action(
            send_message(Config1, SentData),
            #{?snk_kind := greptimedb_connector_send_query_error},
            10_000
        ),
        fun(Result, Trace) ->
            ?assertMatch({_, {ok, _}}, Result),
            {Return, {ok, _}} = Result,
            IsBatch = BatchSize > 1,
            case {QueryMode, IsBatch} of
                {async, true} ->
                    ?assertEqual(ok, Return),
                    ?assertMatch(
                        [#{error := points_trans_failed}],
                        ?of_kind(greptimedb_connector_send_query_error, Trace)
                    );
                {async, false} ->
                    ?assertEqual(ok, Return),
                    ?assertMatch(
                        [
                            #{
                                error := [
                                    {error, {bad_timestamp, <<"bad_timestamp">>}}
                                ]
                            }
                        ],
                        ?of_kind(greptimedb_connector_send_query_error, Trace)
                    );
                {sync, false} ->
                    ?assertEqual(
                        {error, [
                            {error, {bad_timestamp, <<"bad_timestamp">>}}
                        ]},
                        Return
                    );
                {sync, true} ->
                    ?assertEqual({error, {unrecoverable_error, points_trans_failed}}, Return)
            end,
            ok
        end
    ),
    ok.

t_get_status(Config) ->
    ProxyPort = ?config(proxy_port, Config),
    ProxyHost = ?config(proxy_host, Config),
    ProxyName = ?config(proxy_name, Config),
    {ok, _} = create_bridge(Config),
    ResourceId = resource_id(Config),
    ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceId)),
    emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
        ?assertEqual({ok, disconnected}, emqx_resource_manager:health_check(ResourceId))
    end),
    ok.

t_create_disconnected(Config) ->
    ProxyPort = ?config(proxy_port, Config),
    ProxyHost = ?config(proxy_host, Config),
    ProxyName = ?config(proxy_name, Config),
    ?check_trace(
        emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
            ?assertMatch({ok, _}, create_bridge(Config))
        end),
        fun(Trace) ->
            ?assertMatch(
                [#{error := greptimedb_client_not_alive, reason := _SomeReason}],
                ?of_kind(greptimedb_connector_start_failed, Trace)
            ),
            ok
        end
    ),
    ok.

t_start_error(Config) ->
    %% simulate client start error
    ?check_trace(
        emqx_common_test_helpers:with_mock(
            greptimedb,
            start_client,
            fun(_Config) -> {error, some_error} end,
            fun() ->
                ?wait_async_action(
                    ?assertMatch({ok, _}, create_bridge(Config)),
                    #{?snk_kind := greptimedb_connector_start_failed},
                    10_000
                )
            end
        ),
        fun(Trace) ->
            ?assertMatch(
                [#{error := some_error}],
                ?of_kind(greptimedb_connector_start_failed, Trace)
            ),
            ok
        end
    ),
    ok.

t_start_exception(Config) ->
    %% simulate client start exception
    ?check_trace(
        emqx_common_test_helpers:with_mock(
            greptimedb,
            start_client,
            fun(_Config) -> error(boom) end,
            fun() ->
                ?wait_async_action(
                    ?assertMatch({ok, _}, create_bridge(Config)),
                    #{?snk_kind := greptimedb_connector_start_exception},
                    10_000
                )
            end
        ),
        fun(Trace) ->
            ?assertMatch(
                [#{error := {error, boom}}],
                ?of_kind(greptimedb_connector_start_exception, Trace)
            ),
            ok
        end
    ),
    ok.

t_write_failure(Config) ->
    ProxyName = ?config(proxy_name, Config),
    ProxyPort = ?config(proxy_port, Config),
    ProxyHost = ?config(proxy_host, Config),
    QueryMode = ?config(query_mode, Config),
    {ok, _} = create_bridge(Config),
    ClientId = emqx_guid:to_hexstr(emqx_guid:gen()),
    Payload = #{
        int_key => -123,
        bool => true,
        float_key => 24.5,
        uint_key => 123
    },
    SentData = #{
        <<"clientid">> => ClientId,
        <<"topic">> => atom_to_binary(?FUNCTION_NAME),
        <<"timestamp">> => erlang:system_time(millisecond),
        <<"payload">> => Payload
    },
    ?check_trace(
        emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
            case QueryMode of
                sync ->
                    ?wait_async_action(
                        ?assertMatch(
                            {error, {resource_error, #{reason := timeout}}},
                            send_message(Config, SentData)
                        ),
                        #{?snk_kind := handle_async_reply, action := nack},
                        1_000
                    );
                async ->
                    ?wait_async_action(
                        ?assertEqual(ok, send_message(Config, SentData)),
                        #{?snk_kind := handle_async_reply},
                        1_000
                    )
            end
        end),
        fun(Trace0) ->
            case QueryMode of
                sync ->
                    Trace = ?of_kind(handle_async_reply, Trace0),
                    ?assertMatch([_ | _], Trace),
                    [#{result := Result} | _] = Trace,
                    ?assert(
                        not emqx_bridge_greptimedb_connector:is_unrecoverable_error(Result),
                        #{got => Result}
                    );
                async ->
                    Trace = ?of_kind(handle_async_reply, Trace0),
                    ?assertMatch([_ | _], Trace),
                    [#{result := Result} | _] = Trace,
                    ?assert(
                        not emqx_bridge_greptimedb_connector:is_unrecoverable_error(Result),
                        #{got => Result}
                    )
            end,
            ok
        end
    ),
    ok.

t_missing_field(Config) ->
    BatchSize = ?config(batch_size, Config),
    IsBatch = BatchSize > 1,
    {ok, _} =
        create_bridge(
            Config,
            #{
                <<"resource_opts">> => #{<<"worker_pool_size">> => 1},
                <<"write_syntax">> => <<"${clientid} foo=${foo}i">>
            }
        ),
    %% note: we don't select foo here, but we interpolate it in the
    %% fields, so it'll become undefined.
    {ok, _} = create_rule_and_action_http(Config, #{sql => <<"select * from \"t/topic\"">>}),
    ClientId0 = emqx_guid:to_hexstr(emqx_guid:gen()),
    ClientId1 = emqx_guid:to_hexstr(emqx_guid:gen()),
    %% Message with the field that we "forgot" to select in the rule
    Msg0 = emqx_message:make(ClientId0, <<"t/topic">>, emqx_utils_json:encode(#{foo => 123})),
    %% Message without any fields
    Msg1 = emqx_message:make(ClientId1, <<"t/topic">>, emqx_utils_json:encode(#{})),
    ?check_trace(
        begin
            emqx:publish(Msg0),
            emqx:publish(Msg1),
            NEvents = 1,
            {ok, _} =
                snabbkaffe:block_until(
                    ?match_n_events(NEvents, #{
                        ?snk_kind := greptimedb_connector_send_query_error
                    }),
                    _Timeout1 = 16_000
                ),
            ok
        end,
        fun(Trace) ->
            PersistedData0 = query_by_clientid(ClientId0, ClientId0, Config),
            PersistedData1 = query_by_clientid(ClientId1, ClientId1, Config),
            case IsBatch of
                true ->
                    ?assertMatch(
                        [#{error := points_trans_failed} | _],
                        ?of_kind(greptimedb_connector_send_query_error, Trace)
                    );
                false ->
                    ?assertMatch(
                        [#{error := [{error, no_fields}]} | _],
                        ?of_kind(greptimedb_connector_send_query_error, Trace)
                    )
            end,
            %% nothing should have been persisted
            ?assertEqual(#{}, PersistedData0),
            ?assertEqual(#{}, PersistedData1),
            ok
        end
    ),
    ok.

t_authentication_error_on_send_message(Config0) ->
    QueryMode = proplists:get_value(query_mode, Config0, sync),
    GreptimedbType = ?config(greptimedb_type, Config0),
    GreptimeConfig0 = proplists:get_value(greptimedb_config, Config0),
    GreptimeConfig =
        case GreptimedbType of
            grpcv1 -> GreptimeConfig0#{<<"password">> => <<"wrong_password">>}
        end,
    Config = lists:keyreplace(greptimedb_config, 1, Config0, {greptimedb_config, GreptimeConfig}),

    % Fake initialization to simulate credential update after bridge was created.
    emqx_common_test_helpers:with_mock(
        greptimedb,
        check_auth,
        fun(_) ->
            ok
        end,
        fun() ->
            {ok, _} = create_bridge(Config),
            ResourceId = resource_id(Config0),
            ?retry(
                _Sleep = 1_000,
                _Attempts = 10,
                ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceId))
            )
        end
    ),

    % Now back to wrong credentials
    ClientId = emqx_guid:to_hexstr(emqx_guid:gen()),
    Payload = #{
        int_key => -123,
        bool => true,
        float_key => 24.5,
        uint_key => 123
    },
    SentData = #{
        <<"clientid">> => ClientId,
        <<"topic">> => atom_to_binary(?FUNCTION_NAME),
        <<"timestamp">> => erlang:system_time(millisecond),
        <<"payload">> => Payload
    },
    case QueryMode of
        sync ->
            ?assertMatch(
                {error, {unrecoverable_error, <<"authorization failure">>}},
                send_message(Config, SentData)
            );
        async ->
            ?check_trace(
                begin
                    ?wait_async_action(
                        ?assertEqual(ok, send_message(Config, SentData)),
                        #{?snk_kind := handle_async_reply},
                        1_000
                    )
                end,
                fun(Trace) ->
                    ?assertMatch(
                        [#{error := <<"authorization failure">>} | _],
                        ?of_kind(greptimedb_connector_do_query_failure, Trace)
                    ),
                    ok
                end
            )
    end,
    ok.