feat(sync-service): Re-enable pluggable snapshot function

packages/sync-service/lib/electric/shapes/consumer/snapshotter.ex

@@ -114,7 +114,7 @@ defmodule Electric.Shapes.Consumer.Snapshotter do
         consumer,
         shape_handle,
         shape,
-        ctx
+        %{stack_id: stack_id} = ctx
       ) do
     supervisor = Electric.ProcessRegistry.name(ctx.stack_id, Electric.StackTaskSupervisor)
     self_pid = self()
@@ -137,9 +137,12 @@ defmodule Electric.Shapes.Consumer.Snapshotter do
 
     task =
       Task.Supervisor.async_nolink(supervisor, fn ->
+        snapshot_fun =
+          Electric.StackConfig.lookup(stack_id, :create_snapshot_fn, &stream_snapshot_from_db/5)
+
         try do
           result =
-            stream_snapshot_from_db(
+            snapshot_fun.(
               self_pid,
               consumer,
               shape_handle,
@@ -211,7 +214,6 @@ defmodule Electric.Shapes.Consumer.Snapshotter do
         %{storage: storage, stack_id: stack_id, db_pool: db_pool}
       ) do
     shape_attrs = telemetry_shape_attrs(shape_handle, shape)
-    chunk_bytes_threshold = Electric.StackConfig.lookup(stack_id, :chunk_bytes_threshold)
 
     Postgrex.transaction(
       db_pool,
@@ -270,35 +272,7 @@ defmodule Electric.Shapes.Consumer.Snapshotter do
               })
 
             stream =
-              Querying.stream_initial_data(conn, stack_id, shape, chunk_bytes_threshold)
-              |> Stream.transform(
-                fn -> false end,
-                fn item, acc ->
-                  if not acc do
-                    send(task_parent, :data_received)
-                    GenServer.cast(consumer, {:snapshot_started, shape_handle})
-                  end
-
-                  {[item], true}
-                end,
-                fn acc ->
-                  if not acc do
-                    # The stream has been read to the end but we haven't seen a single item in
-                    # it. Notify `consumer` anyway since an empty file will have been created by
-                    # the storage implementation for the API layer to read the snapshot data
-                    # from.
-                    send(task_parent, :data_received)
-                    GenServer.cast(consumer, {:snapshot_started, shape_handle})
-                  end
-
-                  {[], acc}
-                end,
-                fn acc ->
-                  # noop after fun just to be able to specify the last fun which is only
-                  # available in `Stream.transoform/5`.
-                  acc
-                end
-              )
+              stream_initial_data(task_parent, consumer, conn, stack_id, shape, shape_handle)
               |> Stream.concat([finishing_control_message])
 
             # could pass the shape and then make_new_snapshot! can pass it to row_to_snapshot_item
@@ -314,6 +288,42 @@ defmodule Electric.Shapes.Consumer.Snapshotter do
       raise SnapshotError.connection_not_available()
   end
 
+  @doc false
+  # Used by Phoenix.Sync
+  def stream_initial_data(task_parent, consumer, conn, stack_id, shape, shape_handle) do
+    chunk_bytes_threshold = Electric.StackConfig.lookup(stack_id, :chunk_bytes_threshold)
+
+    Querying.stream_initial_data(conn, stack_id, shape, chunk_bytes_threshold)
+    |> Stream.transform(
+      fn -> false end,
+      fn item, acc ->
+        if not acc do
+          send(task_parent, :data_received)
+          GenServer.cast(consumer, {:snapshot_started, shape_handle})
+        end
+
+        {[item], true}
+      end,
+      fn acc ->
+        if not acc do
+          # The stream has been read to the end but we haven't seen a single item in
+          # it. Notify `consumer` anyway since an empty file will have been created by
+          # the storage implementation for the API layer to read the snapshot data
+          # from.
+          send(task_parent, :data_received)
+          GenServer.cast(consumer, {:snapshot_started, shape_handle})
+        end
+
+        {[], acc}
+      end,
+      fn acc ->
+        # noop after fun just to be able to specify the last fun which is only
+        # available in `Stream.transoform/5`.
+        acc
+      end
+    )
+  end
+
   defp query_span!(conn, span_name, span_attrs, query, params, stack_id) do
     OpenTelemetry.with_span(
       span_name,

packages/sync-service/lib/electric/shapes/dynamic_consumer_supervisor.ex

@@ -14,7 +14,7 @@ defmodule Electric.Shapes.DynamicConsumerSupervisor do
   """
   def child_spec(opts) do
     stack_id = Keyword.fetch!(opts, :stack_id)
-    name = Keyword.get(opts, :name, name(stack_id))
+    {name, opts} = Keyword.pop(opts, :name, name(stack_id))
 
     # We're overriding Electric.Shapes.DynamicConsumerSupervisor's child_spec() function here
     # to make the usage of PartitionSupervisor transparent to the callers. As a consequence, we

packages/sync-service/lib/electric/stack_config.ex

@@ -17,6 +17,16 @@ defmodule Electric.StackConfig do
         message: "stack config value #{inspect(key)} is missing for stack #{stack_id}"
   end
 
+  @doc false
+  # Should provide all required values not defined dynamically at stack init
+  def default_seed_config do
+    [
+      snapshot_timeout_to_first_data: :timer.seconds(30),
+      shape_hibernate_after: Electric.Config.default(:shape_hibernate_after),
+      chunk_bytes_threshold: Electric.ShapeCache.LogChunker.default_chunk_size_threshold()
+    ]
+  end
+
   ###
 
   def name(stack_ref) do
@@ -37,9 +47,11 @@ defmodule Electric.StackConfig do
   def init(opts) do
     stack_id = Keyword.fetch!(opts, :stack_id)
 
+    seed_config = Keyword.merge(default_seed_config(), Keyword.get(opts, :seed_config, []))
+
     tab = table(stack_id)
     :ets.new(tab, [:public, :named_table, :set, read_concurrency: true])
-    :ets.insert(tab, Keyword.fetch!(opts, :seed_config))
+    :ets.insert(tab, seed_config)
 
     {:ok, nil}
   end

packages/sync-service/lib/electric/stack_supervisor.ex

@@ -363,7 +363,6 @@ defmodule Electric.StackSupervisor do
              chunk_bytes_threshold: config.chunk_bytes_threshold,
              snapshot_timeout_to_first_data: config.tweaks[:snapshot_timeout_to_first_data],
              inspector: inspector,
-             shape_changes_registry: shape_changes_registry_name,
              shape_hibernate_after: shape_hibernate_after
            ]},
           {Electric.AsyncDeleter,