chore(electric): Explicitly set those configuration parameters that a…

…ffect formatting of values of certain types (#414)

Closes VAX-974.

components/electric/lib/electric/replication/postgres/client.ex

@@ -165,6 +165,34 @@ defmodule Electric.Replication.Postgres.Client do
     |> :epgsql.start_replication(:erlang.binary_to_list(slot), handler, [], ~c"0/0", opts)
   end
 
+  @doc """
+  Explicitly set those configuration parameters that affect formatting of values of certain types.
+
+  By setting those parameters for the current session we're safe-guarding against non-standard configuration being used
+  in the Postgres database cluster or even the specific database Electric is configured to replicate from.
+
+  The parameters we're interested in are:
+
+    * `bytea_output` - determines how Postgres encodes bytea values. It can use either Hex- or Escape-based encoding.
+    * `DateStyle` - determines how Postgres interprets date values.
+    * `TimeZone` - affects the time zone offset Postgres uses for timestamptz and timetz values.
+    * `extra_float_digits` - determines whether floating-point values are rounded or are encoded precisely.
+  """
+  def set_display_settings_for_replication(conn) do
+    results =
+      :epgsql.squery(
+        conn,
+        """
+        SET bytea_output = 'hex';
+        SET DateStyle = 'ISO, DMY';
+        SET TimeZone = 'UTC';
+        SET extra_float_digits = 1;
+        """
+      )
+
+    :ok = Enum.each(results, &({:ok, [], []} = &1))
+  end
+
   @doc """
   Confirm successful processing of a WAL segment.
 

components/electric/lib/electric/replication/postgres/logical_replication_producer.ex

@@ -88,6 +88,7 @@ defmodule Electric.Replication.Postgres.LogicalReplicationProducer do
 
     with {:ok, conn} <- Client.connect(conn_opts),
          {:ok, _} <- Client.create_slot(conn, slot),
+         :ok <- Client.set_display_settings_for_replication(conn),
          :ok <- Client.start_replication(conn, publication, slot, self()) do
       Logger.metadata(pg_producer: origin)
       Logger.info("Starting replication from #{origin}")

components/electric/test/electric/replication/postgres/logical_replication_producer_test.exs

@@ -16,7 +16,8 @@ defmodule Electric.Replication.Postgres.LogicalReplicationProducerTest do
      [
        connect: fn _ -> {:ok, :conn} end,
        start_replication: fn :conn, _, _, _ -> :ok end,
-       create_slot: fn :conn, name -> {:ok, name} end
+       create_slot: fn :conn, name -> {:ok, name} end,
+       set_display_settings_for_replication: fn _ -> :ok end
      ]},
     {Connectors, [:passthrough],
      [