fix: provide microsecond granularity by using binary pgoutput mode

Author: filipecabaco

lib/realtime/adapters/postgres/decoder.ex

@@ -218,19 +218,13 @@ defmodule Realtime.Adapters.Postgres.Decoder do
   defp decode_message_impl(<<"I", relation_id::integer-32, "N", number_of_columns::integer-16, tuple_data::binary>>) do
     {<<>>, decoded_tuple_data} = decode_tuple_data(tuple_data, number_of_columns)
 
-    %Insert{
-      relation_id: relation_id,
-      tuple_data: decoded_tuple_data
-    }
+    %Insert{relation_id: relation_id, tuple_data: decoded_tuple_data}
   end
 
   defp decode_message_impl(<<"U", relation_id::integer-32, "N", number_of_columns::integer-16, tuple_data::binary>>) do
     {<<>>, decoded_tuple_data} = decode_tuple_data(tuple_data, number_of_columns)
 
-    %Update{
-      relation_id: relation_id,
-      tuple_data: decoded_tuple_data
-    }
+    %Update{relation_id: relation_id, tuple_data: decoded_tuple_data}
   end
 
   defp decode_message_impl(
@@ -242,10 +236,7 @@ defmodule Realtime.Adapters.Postgres.Decoder do
 
     {<<>>, decoded_tuple_data} = decode_tuple_data(new_tuple_binary, new_number_of_columns)
 
-    base_update_msg = %Update{
-      relation_id: relation_id,
-      tuple_data: decoded_tuple_data
-    }
+    base_update_msg = %Update{relation_id: relation_id, tuple_data: decoded_tuple_data}
 
     case key_or_old do
       "K" -> Map.put(base_update_msg, :changed_key_tuple_data, old_decoded_tuple_data)
@@ -259,9 +250,7 @@ defmodule Realtime.Adapters.Postgres.Decoder do
        when key_or_old == "K" or key_or_old == "O" do
     {<<>>, decoded_tuple_data} = decode_tuple_data(tuple_data, number_of_columns)
 
-    base_delete_msg = %Delete{
-      relation_id: relation_id
-    }
+    base_delete_msg = %Delete{relation_id: relation_id}
 
     case key_or_old do
       "K" -> Map.put(base_delete_msg, :changed_key_tuple_data, decoded_tuple_data)
@@ -313,19 +302,39 @@ defmodule Realtime.Adapters.Postgres.Decoder do
   defp decode_tuple_data(<<"u", rest::binary>>, columns_remaining, accumulator),
     do: decode_tuple_data(rest, columns_remaining - 1, [:unchanged_toast | accumulator])
 
+  @start_date "2000-01-01T00:00:00Z"
   defp decode_tuple_data(
-         <<"t", column_length::integer-32, rest::binary>>,
+         <<"b", column_length::integer-32, rest::binary>>,
          columns_remaining,
          accumulator
-       ),
-       do:
-         decode_tuple_data(
-           :erlang.binary_part(rest, {byte_size(rest), -(byte_size(rest) - column_length)}),
-           columns_remaining - 1,
-           [
-             :erlang.binary_part(rest, {0, column_length}) | accumulator
-           ]
-         )
+       ) do
+    data = :erlang.binary_part(rest, {0, column_length})
+    remainder = :erlang.binary_part(rest, {byte_size(rest), -(byte_size(rest) - column_length)})
+
+    data =
+      case data do
+        <<1>> ->
+          true
+
+        <<0>> ->
+          false
+
+        <<1, binary::binary-size(column_length - 1)>> ->
+          binary
+
+        <<microseconds::signed-big-64>> ->
+          NaiveDateTime.from_iso8601!(@start_date)
+          |> NaiveDateTime.add(microseconds, :microsecond)
+
+        <<uuid_binary::binary-16>> ->
+          UUID.binary_to_string!(uuid_binary)
+
+        data when is_binary(data) ->
+          data
+      end
+
+    decode_tuple_data(remainder, columns_remaining - 1, [data | accumulator])
+  end
 
   defp decode_columns(binary, accumulator \\ [])
   defp decode_columns(<<>>, accumulator), do: Enum.reverse(accumulator)
@@ -345,7 +354,6 @@ defmodule Realtime.Adapters.Postgres.Decoder do
         name: name,
         flags: decoded_flags,
         type: OidDatabase.name_for_type_id(data_type_id),
-        # type: data_type_id,
         type_modifier: type_modifier
       }
       | accumulator

lib/realtime/api/message.ex

@@ -17,7 +17,7 @@ defmodule Realtime.Api.Message do
     field(:event, :string)
     field(:private, :boolean)
 
-    timestamps()
+    timestamps(type: :naive_datetime_usec)
   end
 
   def changeset(message, attrs) do

lib/realtime/monitoring/prom_ex/plugins/tenant.ex

@@ -245,7 +245,7 @@ defmodule Realtime.PromEx.Plugins.Tenant do
           unit: :second,
           description: "Latency of database inserted_at until reaches server to be broadcasted",
           tags: [:tenant],
-          reporter_options: [buckets: [1, 2, 5]]
+          reporter_options: [buckets: [10, 250, 5000]]
         )
       ]
     )

lib/realtime/tenants/replication_connection.ex

@@ -57,7 +57,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
             publication_name: nil,
             replication_slot_name: nil,
             output_plugin: "pgoutput",
-            proto_version: 1,
+            proto_version: 3,
             relations: %{},
             buffer: [],
             monitored_pid: nil,
@@ -236,7 +236,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
     )
 
     query =
-      "START_REPLICATION SLOT #{replication_slot_name} LOGICAL 0/0 (proto_version '#{proto_version}', publication_names '#{publication_name}')"
+      "START_REPLICATION SLOT #{replication_slot_name} LOGICAL 0/0 (proto_version '#{proto_version}', publication_names '#{publication_name}', binary 'true')"
 
     {:stream, query, [], %{state | step: :streaming}}
   end
@@ -318,8 +318,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
            payload: Map.put_new(payload, "id", id)
          },
          :ok <- BatchBroadcast.broadcast(nil, tenant, %{messages: [broadcast_message]}, true) do
-      inserted_at = NaiveDateTime.from_iso8601!(inserted_at)
-      latency_inserted_at = NaiveDateTime.utc_now() |> NaiveDateTime.diff(inserted_at)
+      latency_inserted_at = NaiveDateTime.utc_now(:microsecond) |> NaiveDateTime.diff(inserted_at)
 
       Telemetry.execute(
         [:realtime, :tenants, :broadcast_from_database],
@@ -377,7 +376,6 @@ defmodule Realtime.Tenants.ReplicationConnection do
     |> Map.new(fn
       {nil, %{name: name}} -> {name, nil}
       {value, %{name: name, type: "jsonb"}} -> {name, Jason.decode!(value)}
-      {value, %{name: name, type: "bool"}} -> {name, value == "t"}
       {value, %{name: name}} -> {name, value}
     end)
   end