Update to support electric 1.4

- Fix invalid protocol typing
- Support more esoteric version tuples

Author: magnetised

.gitignore

@@ -24,3 +24,5 @@ electric_phoenix-*.tar
 
 # Temporary files, for example, from tests.
 /tmp/
+.claude
+

lib/mix/tasks/phoenix_sync.install.ex

@@ -300,7 +300,7 @@ if Code.ensure_loaded?(Igniter) do
     defp required_electric_version do
       Phoenix.Sync.MixProject.project()
       |> Keyword.fetch!(:deps)
-      |> Enum.find(&match?({:electric, _, _}, &1))
+      |> Enum.find(&(elem(&1, 0) == :electric))
       |> elem(1)
     end
 

lib/phoenix/sync/electric.ex

@@ -640,8 +640,9 @@ defmodule Phoenix.Sync.Electric do
   end
 end
 
-if Code.ensure_loaded?(Electric.Shapes.Api) &&
-     Code.ensure_loaded?(Phoenix.Sync.Electric.ApiAdapter) do
+if Code.ensure_loaded?(Electric.Shapes.Api) do
+  Code.ensure_loaded(Phoenix.Sync.Electric.ApiAdapter)
+
   defimpl Phoenix.Sync.Adapter.PlugApi, for: Electric.Shapes.Api do
     alias Electric.Shapes
 
@@ -701,7 +702,7 @@ if Code.ensure_loaded?(Electric.Shapes.Api) &&
       end
     end
 
-    def send_response(%ApiAdapter{}, conn, {request, response}) do
+    def send_response(_api, conn, {request, response}) do
       conn
       |> content_type()
       |> Plug.Conn.assign(:request, request)

lib/phoenix/sync/plug/cors.ex

@@ -13,10 +13,12 @@ defmodule Phoenix.Sync.Plug.CORS do
   @electric_headers [
     "electric-cursor",
     "electric-handle",
+    "electric-has-data",
     "electric-offset",
     "electric-schema",
     "electric-up-to-date",
-    "electric-internal-known-error"
+    "electric-internal-known-error",
+    "retry-after"
   ]
 
   @expose_headers ["transfer-encoding" | @electric_headers]

lib/phoenix/sync/sandbox.ex

@@ -253,13 +253,14 @@ if Phoenix.Sync.sandbox_enabled?() do
 
           :ok = maybe_set_shared_mode(owner, stack_id, opts)
 
-          # give the inspector access to the sandboxed connection
-          Ecto.Adapters.SQL.Sandbox.allow(repo, owner, Sandbox.Inspector.name(stack_id))
-
           # mark the stack as ready
           Electric.StatusMonitor.mark_pg_lock_acquired(stack_id, owner)
           Electric.StatusMonitor.mark_replication_client_ready(stack_id, owner)
-          Electric.StatusMonitor.mark_connection_pool_ready(stack_id, owner)
+          Electric.StatusMonitor.mark_connection_pool_ready(stack_id, :admin, owner)
+          Electric.StatusMonitor.mark_connection_pool_ready(stack_id, :snapshot, owner)
+          Electric.StatusMonitor.mark_integrety_checks_passed(stack_id, owner)
+          Electric.StatusMonitor.mark_shape_log_collector_ready(stack_id, owner)
+          Electric.StatusMonitor.mark_supervisor_processes_ready(stack_id, owner)
 
           api_config = Sandbox.Stack.config(stack_id, repo)
           api = Electric.Application.api(api_config)
@@ -292,7 +293,8 @@ if Phoenix.Sync.sandbox_enabled?() do
     defp generate_stack_id(opts) do
       tags = Keyword.get(opts, :tags, %{})
       # with parameterised tests the same file:line can be running simultaneously
-      uid = System.unique_integer() |> to_string()
+      uid = System.unique_integer([:positive, :monotonic]) |> to_string()
+      now = System.monotonic_time(:microsecond)
 
       suffix =
         case Map.fetch(tags, :line) do
@@ -306,7 +308,7 @@ if Phoenix.Sync.sandbox_enabled?() do
           :error -> ""
         end
 
-      "#{inspect(__MODULE__.Stack)}#{uid}#{prefix}#{suffix}"
+      "#{inspect(__MODULE__.Stack)}#{now}-#{uid}#{prefix}#{suffix}"
     end
 
     defp maybe_set_shared_mode(owner, stack_id, opts) do

lib/phoenix/sync/sandbox/cleanup.ex

@@ -0,0 +1,24 @@
+defmodule Phoenix.Sync.Sandbox.Cleanup do
+  # We use PureFileStorage in the sandbox to remove compatibility issues
+  # this ensures that the storage dir is removed when the stack is torn
+  # down
+  @moduledoc false
+
+  use GenServer
+
+  def start_link({stack_id, storage_dir}) do
+    GenServer.start_link(__MODULE__, {stack_id, storage_dir})
+  end
+
+  @impl GenServer
+  def init({stack_id, storage_dir}) do
+    Process.flag(:trap_exit, true)
+    {:ok, Path.join(storage_dir, stack_id)}
+  end
+
+  @impl GenServer
+  def terminate(_reason, storage_dir) do
+    File.rm_rf!(storage_dir)
+    :ok
+  end
+end

lib/phoenix/sync/sandbox/expiry_manager.ex

@@ -0,0 +1,11 @@
+defmodule Phoenix.Sync.Sandbox.ExpiryManager do
+  use GenServer
+
+  def start_link(args) do
+    GenServer.start_link(__MODULE__, args, name: Electric.ShapeCache.ExpiryManager.name(args))
+  end
+
+  def init(_) do
+    {:ok, []}
+  end
+end

lib/phoenix/sync/sandbox/initialize_stack.ex

@@ -0,0 +1,25 @@
+defmodule Phoenix.Sync.Sandbox.InitializeStack do
+  @moduledoc false
+
+  # synchronously initializes the stack
+
+  use GenServer
+
+  def start_link(args) do
+    GenServer.start_link(__MODULE__, args)
+  end
+
+  @impl GenServer
+  def init(args) do
+    {:ok, stack_id} = Keyword.fetch(args, :stack_id)
+
+    :ok = Electric.ShapeCache.ShapeStatusOwner.initialize(stack_id)
+
+    Electric.LsnTracker.set_last_processed_lsn(
+      stack_id,
+      Electric.Postgres.Lsn.from_integer(0)
+    )
+
+    :ignore
+  end
+end

lib/phoenix/sync/sandbox/inspector.ex

@@ -34,6 +34,13 @@ if Phoenix.Sync.sandbox_enabled?() do
     @impl Electric.Postgres.Inspector
     def list_relations_with_stale_cache(_), do: {:ok, []}
 
+    @impl Electric.Postgres.Inspector
+    def load_supported_features(stack_id) do
+      with {:ok, pid} <- validate_stack_alive(stack_id) do
+        GenServer.call(pid, :load_supported_features)
+      end
+    end
+
     def start_link(args) do
       GenServer.start_link(__MODULE__, args, name: name(args[:stack_id]))
     end
@@ -56,8 +63,20 @@ if Phoenix.Sync.sandbox_enabled?() do
     def init(args) do
       {:ok, stack_id} = Keyword.fetch(args, :stack_id)
       {:ok, repo} = Keyword.fetch(args, :repo)
-
-      {:ok, %{repo: repo, stack_id: stack_id, relations: %{}, columns: %{}, oids: %{}}}
+      {:ok, owner} = Keyword.fetch(args, :owner)
+
+      # give the inspector access to the sandboxed connection
+      Ecto.Adapters.SQL.Sandbox.allow(repo, owner, self())
+
+      {:ok,
+       %{
+         repo: repo,
+         stack_id: stack_id,
+         relations: %{},
+         columns: %{},
+         oids: %{},
+         supported_features: %{}
+       }}
     end
 
     @impl GenServer
@@ -88,6 +107,15 @@ if Phoenix.Sync.sandbox_enabled?() do
       {:reply, result, state}
     end
 
+    def handle_call(:load_supported_features, _from, state) do
+      {result, state} =
+        fetch_lazy(state, :supported_features, nil, fn ->
+          Electric.Postgres.Inspector.DirectInspector.load_supported_features(pool(state))
+        end)
+
+      {:reply, result, state}
+    end
+
     defp pool(state) do
       %{pid: pool} = Ecto.Adapter.lookup_meta(state.repo.get_dynamic_repo())
       pool

lib/phoenix/sync/sandbox/producer.ex

@@ -3,7 +3,8 @@ if Phoenix.Sync.sandbox_enabled?() do
     @moduledoc false
 
     alias Electric.Replication.Changes.{
-      Transaction,
+      TransactionFragment,
+      Commit,
       NewRecord,
       UpdatedRecord,
       DeletedRecord,
@@ -50,6 +51,7 @@ if Phoenix.Sync.sandbox_enabled?() do
 
     def init(stack_id) do
       state = %{txid: 10000, stack_id: stack_id}
+
       {:ok, state}
     end
 
@@ -62,7 +64,7 @@ if Phoenix.Sync.sandbox_enabled?() do
       :ok =
         txid
         |> transaction(msgs)
-        |> ShapeLogCollector.store_transaction(ShapeLogCollector.name(stack_id))
+        |> ShapeLogCollector.handle_event(stack_id)
 
       {:noreply, %{state | txid: next_txid}}
     end
@@ -73,20 +75,22 @@ if Phoenix.Sync.sandbox_enabled?() do
       :ok =
         state.txid
         |> transaction(changes)
-        |> ShapeLogCollector.store_transaction(ShapeLogCollector.name(state.stack_id))
+        |> ShapeLogCollector.handle_event(state.stack_id)
 
       {:noreply, %{state | txid: state.txid + 100}}
     end
 
     defp transaction(txid, changes) do
-      %Transaction{
+      %{log_offset: last_log_offset} = Enum.at(changes, -1)
+
+      %TransactionFragment{
         xid: txid,
         lsn: Electric.Postgres.Lsn.from_integer(txid),
-        last_log_offset: Enum.at(changes, -1) |> Map.fetch!(:log_offset),
+        last_log_offset: last_log_offset,
+        has_begin?: true,
+        commit: %Commit{},
         changes: changes,
-        num_changes: length(changes),
-        commit_timestamp: DateTime.utc_now(),
-        affected_relations: Enum.into(changes, MapSet.new(), & &1.relation)
+        affected_relations: MapSet.new(changes, & &1.relation)
       }
     end