fix(postgres action): handle bad timestamp params more gracefully

Fixes https://emqx.atlassian.net/browse/EMQX-13306

See: emqx/epgsql#12

apps/emqx_bridge_pgsql/test/emqx_bridge_v2_pgsql_SUITE.erl

@@ -97,6 +97,7 @@ init_per_group(Group, Config) when
 ->
     [
         {bridge_type, group_to_type(Group)},
+        {action_type, group_to_type(Group)},
         {connector_type, group_to_type(Group)}
         | Config
     ];
@@ -145,7 +146,10 @@ init_per_testcase(TestCase, Config) ->
         {connector_name, Name},
         {connector_config, ConnectorConfig},
         {bridge_type, proplists:get_value(bridge_type, Config, ?BRIDGE_TYPE)},
+        {action_type, proplists:get_value(action_type, Config, ?BRIDGE_TYPE)},
         {bridge_name, Name},
+        {action_name, Name},
+        {action_config, BridgeConfig},
         {bridge_config, BridgeConfig}
         | NConfig
     ].
@@ -232,6 +236,16 @@ make_message() ->
         timestamp => 1668602148000
     }.
 
+create_connector_api(Config) ->
+    emqx_bridge_v2_testlib:simplify_result(
+        emqx_bridge_v2_testlib:create_connector_api(Config)
+    ).
+
+create_action_api(Config, Overrides) ->
+    emqx_bridge_v2_testlib:simplify_result(
+        emqx_bridge_v2_testlib:create_action_api(Config, Overrides)
+    ).
+
 %%------------------------------------------------------------------------------
 %% Testcases
 %%------------------------------------------------------------------------------
@@ -378,3 +392,37 @@ t_update_with_invalid_prepare(Config) ->
         WorkerPids
     ),
     ok.
+
+%% Checks that furnishing `epgsql' a value that cannot be encoded to a timestamp results
+%% in a pretty error instead of a crash.
+t_bad_datetime_param(Config) ->
+    {201, _} = create_connector_api(Config),
+    {201, _} = create_action_api(Config, #{
+        <<"parameters">> => #{
+            <<"sql">> => <<
+                "INSERT INTO mqtt_test(payload, arrived) "
+                "VALUES (${payload}, ${payload})"
+            >>
+        }
+    }),
+    RuleTopic = <<"bad/timestamp">>,
+    {ok, _} = emqx_bridge_v2_testlib:create_rule_and_action_http(
+        ?config(action_type, Config), RuleTopic, Config, #{}
+    ),
+    Payload = <<"2024-06-30 00:10:00">>,
+    ?assertMatch(
+        {_,
+            {ok, #{
+                context := #{
+                    reason := bad_param,
+                    type := timestamp,
+                    index := 1,
+                    value := Payload
+                }
+            }}},
+        ?wait_async_action(
+            emqx:publish(emqx_message:make(RuleTopic, Payload)),
+            #{?snk_kind := "postgres_bad_param_error"}
+        )
+    ),
+    ok.

apps/emqx_postgresql/rebar.config

@@ -2,7 +2,7 @@
 
 {erl_opts, [debug_info]}.
 {deps, [
-    {epgsql, {git, "https://github.com/emqx/epgsql", {tag, "4.7.1.3"}}},
+    {epgsql, {git, "https://github.com/emqx/epgsql", {tag, "4.7.1.4"}}},
     {emqx_connector, {path, "../../apps/emqx_connector"}},
     {emqx_resource, {path, "../../apps/emqx_resource"}}
 ]}.

apps/emqx_postgresql/src/emqx_postgresql.erl

@@ -794,6 +794,9 @@ handle_result({error, {unrecoverable_error, _Error}} = Res) ->
     Res;
 handle_result({error, disconnected}) ->
     {error, {recoverable_error, disconnected}};
+handle_result({error, #{reason := bad_param} = Context}) ->
+    ?tp("postgres_bad_param_error", #{context => Context}),
+    {error, {unrecoverable_error, Context}};
 handle_result({error, Error}) ->
     TranslatedError = translate_to_log_context(Error),
     {error, {unrecoverable_error, export_error(TranslatedError)}};

mix.exs

@@ -216,7 +216,7 @@ defmodule EMQXUmbrella.MixProject do
   def common_dep(:ra), do: {:ra, "2.15.0", override: true}
 
   # in conflict by emqx_connector and system_monitor
-  def common_dep(:epgsql), do: {:epgsql, github: "emqx/epgsql", tag: "4.7.1.3", override: true}
+  def common_dep(:epgsql), do: {:epgsql, github: "emqx/epgsql", tag: "4.7.1.4", override: true}
   def common_dep(:sasl_auth), do: {:sasl_auth, "2.3.3", override: true}
   def common_dep(:gen_rpc), do: {:gen_rpc, github: "emqx/gen_rpc", tag: "3.4.1", override: true}