Set txn ids for buffered transactions

Author: robacourt

packages/sync-service/lib/electric/shapes/consumer/event_handler/subqueries/buffering.ex

@@ -265,13 +265,22 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Buffering do
     Enum.reduce_while(txns, {:ok, []}, fn txn, {:ok, acc} ->
       case Subqueries.convert_transaction(txn, state, views) do
         {:error, :truncate} -> {:halt, {:error, {:truncate, txn.xid}}}
-        {:ok, changes} -> {:cont, {:ok, acc ++ changes}}
+
+        {:ok, []} ->
+          {:cont, {:ok, acc}}
+
+        {:ok, changes} ->
+          {:cont,
+           {:ok,
+            [
+              %LogOp.AppendChanges{changes: changes, default_xid: txn.xid}
+              | acc
+            ]}}
       end
     end)
     |> case do
       {:error, _} = error -> error
-      {:ok, []} -> {:ok, []}
-      {:ok, changes} -> {:ok, [%LogOp.AppendChanges{changes: changes}]}
+      {:ok, ops} -> {:ok, Enum.reverse(ops)}
     end
   end
 

packages/sync-service/test/electric/shapes/consumer_test.exs

@@ -2083,7 +2083,7 @@ defmodule Electric.Shapes.ConsumerTest do
                },
                %{"headers" => %{"control" => "snapshot-end"}},
                %{
-                 "headers" => %{"operation" => "update"},
+                 "headers" => %{"operation" => "update", "txids" => [100]},
                  "key" => ~s'"public"."test_table"/"1"'
                }
              ] = get_log_items_from_storage(LogOffset.last_before_real_offsets(), shape_storage)