fix sandbox

Author: magnetised

lib/phoenix/sync/sandbox/producer.ex

@@ -3,7 +3,8 @@ if Phoenix.Sync.sandbox_enabled?() do
     @moduledoc false
 
     alias Electric.Replication.Changes.{
-      Transaction,
+      Begin,
+      Commit,
       NewRecord,
       UpdatedRecord,
       DeletedRecord,
@@ -50,6 +51,12 @@ if Phoenix.Sync.sandbox_enabled?() do
 
     def init(stack_id) do
       state = %{txid: 10000, stack_id: stack_id}
+
+      Electric.LsnTracker.set_last_processed_lsn(
+        stack_id,
+        Electric.Postgres.Lsn.from_integer(0)
+      )
+
       {:ok, state}
     end
 
@@ -62,7 +69,7 @@ if Phoenix.Sync.sandbox_enabled?() do
       :ok =
         txid
         |> transaction(msgs)
-        |> ShapeLogCollector.store_transaction(ShapeLogCollector.name(stack_id))
+        |> ShapeLogCollector.handle_operations(ShapeLogCollector.name(stack_id))
 
       {:noreply, %{state | txid: next_txid}}
     end
@@ -73,21 +80,20 @@ if Phoenix.Sync.sandbox_enabled?() do
       :ok =
         state.txid
         |> transaction(changes)
-        |> ShapeLogCollector.store_transaction(ShapeLogCollector.name(state.stack_id))
+        |> ShapeLogCollector.handle_operations(ShapeLogCollector.name(state.stack_id))
 
       {:noreply, %{state | txid: state.txid + 100}}
     end
 
     defp transaction(txid, changes) do
-      %Transaction{
-        xid: txid,
-        lsn: Electric.Postgres.Lsn.from_integer(txid),
-        last_log_offset: Enum.at(changes, -1) |> Map.fetch!(:log_offset),
-        changes: changes,
-        num_changes: length(changes),
-        commit_timestamp: DateTime.utc_now(),
-        affected_relations: Enum.into(changes, MapSet.new(), & &1.relation)
-      }
+      [%Begin{xid: txid} | changes] ++
+        [
+          %Commit{
+            lsn: Electric.Postgres.Lsn.from_integer(txid),
+            transaction_size: 100,
+            commit_timestamp: DateTime.utc_now()
+          }
+        ]
     end
 
     defp msg_from_change({{:insert, schema_meta, values}, i}, lsn, txid) do

lib/phoenix/sync/sandbox/stack.ex

@@ -136,7 +136,6 @@ if Phoenix.Sync.sandbox_enabled?() do
         {Electric.ShapeCache.ShapeStatusOwner, stack_id: stack_id, storage: storage},
         {Electric.StatusMonitor, stack_id: stack_id},
         {Sandbox.Inspector, stack_id: stack_id, repo: repo},
-        {Sandbox.Producer, stack_id: stack_id},
         {DynamicSupervisor,
          name: Phoenix.Sync.Sandbox.Fetch.name(stack_id), strategy: :one_for_one},
         Supervisor.child_spec(
@@ -156,7 +155,8 @@ if Phoenix.Sync.sandbox_enabled?() do
             expiry_manager: {Phoenix.Sync.Sandbox.ExpiryManager, stack_id: stack_id}
           },
           restart: :temporary
-        )
+        ),
+        {Sandbox.Producer, stack_id: stack_id}
       ]
 
       Supervisor.init(children, strategy: :one_for_one)