Materialize shape into an ETS table (#77)

Fixes #58

Author: magnetised

apps/phoenix_sync_example/mix.lock

@@ -10,7 +10,7 @@
   "ecto": {:hex, :ecto, "3.13.2", "7d0c0863f3fc8d71d17fc3ad3b9424beae13f02712ad84191a826c7169484f01", [:mix], [{:decimal, "~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "669d9291370513ff56e7b7e7081b7af3283d02e046cf3d403053c557894a0b3e"},
   "ecto_sql": {:hex, :ecto_sql, "3.13.2", "a07d2461d84107b3d037097c822ffdd36ed69d1cf7c0f70e12a3d1decf04e2e1", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "539274ab0ecf1a0078a6a72ef3465629e4d6018a3028095dc90f60a19c371717"},
   "electric": {:hex, :electric, "1.0.24", "f17ee7971390cf710a731a349456f6da43750fbc6582d62793c8702c636ab203", [:mix], [{:backoff, "~> 1.1", [hex: :backoff, repo: "hexpm", optional: false]}, {:bandit, "~> 1.6", [hex: :bandit, repo: "hexpm", optional: false]}, {:dotenvy, "~> 1.1", [hex: :dotenvy, repo: "hexpm", optional: false]}, {:ecto, "~> 3.12", [hex: :ecto, repo: "hexpm", optional: false]}, {:electric_cubdb, "~> 2.0", [hex: :electric_cubdb, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:nimble_options, "~> 1.1", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:opentelemetry, "~> 1.5", [hex: :opentelemetry, repo: "hexpm", optional: true]}, {:opentelemetry_exporter, "~> 1.8", [hex: :opentelemetry_exporter, repo: "hexpm", optional: true]}, {:opentelemetry_semantic_conventions, "~> 1.27", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}, {:opentelemetry_telemetry, "~> 1.1", [hex: :opentelemetry_telemetry, repo: "hexpm", optional: false]}, {:otel_metric_exporter, "~> 0.3.9", [hex: :otel_metric_exporter, repo: "hexpm", optional: true]}, {:pg_query_ex, "0.7.0", [hex: :pg_query_ex, repo: "hexpm", optional: false]}, {:plug, "~> 1.17", [hex: :plug, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.20", [hex: :postgrex, repo: "hexpm", optional: false]}, {:remote_ip, "~> 1.2", [hex: :remote_ip, repo: "hexpm", optional: false]}, {:req, "~> 0.5", [hex: :req, repo: "hexpm", optional: false]}, {:retry, "~> 0.19", [hex: :retry, repo: "hexpm", optional: false]}, {:sentry, "~> 10.9", [hex: :sentry, repo: "hexpm", optional: true]}, {:stream_split, "~> 0.1", [hex: :stream_split, repo: "hexpm", optional: false]}, {:telemetry_metrics_prometheus_core, "~> 1.1", [hex: :telemetry_metrics_prometheus_core, repo: "hexpm", optional: true]}, {:telemetry_metrics_statsd, "~> 0.7", [hex: :telemetry_metrics_statsd, repo: "hexpm", optional: true]}, {:telemetry_poller, "~> 1.2", [hex: :telemetry_poller, repo: "hexpm", optional: false]}, {:tls_certificate_check, "~> 1.27", [hex: :tls_certificate_check, repo: "hexpm", optional: false]}, {:tz, "~> 0.28", [hex: :tz, repo: "hexpm", optional: false]}], "hexpm", "91e3a8b957c1e02d07da3a0e1e902420f32e1d7d5da25814475175517698fb61"},
-  "electric_client": {:hex, :electric_client, "0.6.4", "a582b5df5aa6c94296e4d11c98431114f21136766c7f336f14cb1dabd44800d5", [:mix], [{:ecto_sql, "~> 3.12", [hex: :ecto_sql, repo: "hexpm", optional: true]}, {:electric, "~> 1.0.6", [hex: :electric, repo: "hexpm", optional: true]}, {:gen_stage, "~> 1.2", [hex: :gen_stage, repo: "hexpm", optional: true]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:nimble_options, "~> 1.1", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:req, "~> 0.5", [hex: :req, repo: "hexpm", optional: false]}], "hexpm", "911768e0740cbe738e815c0ab9e378682ea4f800aa87442424f841d0b2c06fe9"},
+  "electric_client": {:hex, :electric_client, "0.6.5-beta-3", "0344418d2ee25b5e43159a59092ca69942c41420804994330d8eb20f801d6ef0", [:mix], [{:ecto_sql, "~> 3.12", [hex: :ecto_sql, repo: "hexpm", optional: true]}, {:electric, "~> 1.0.6", [hex: :electric, repo: "hexpm", optional: true]}, {:gen_stage, "~> 1.2", [hex: :gen_stage, repo: "hexpm", optional: true]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:nimble_options, "~> 1.1", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:req, "~> 0.5", [hex: :req, repo: "hexpm", optional: false]}], "hexpm", "ed49bfc57c914c6dd049dd7bfb1a079acdf600a5334ca3652cac31117617a213"},
   "electric_cubdb": {:hex, :electric_cubdb, "2.0.2", "36f86e3c52dc26f4e077a49fbef813b1a38d3897421cece851f149190b34c16c", [:mix], [], "hexpm", "0c0e24b31fb76ad1b33c5de2ab35c41a4ff9da153f5c1f9b15e2de78575acaf2"},
   "elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"},
   "file_system": {:hex, :file_system, "1.1.0", "08d232062284546c6c34426997dd7ef6ec9f8bbd090eb91780283c9016840e8f", [:mix], [], "hexpm", "bfcf81244f416871f2a2e15c1b515287faa5db9c6bcf290222206d120b3d43f6"},

lib/phoenix/sync.ex

@@ -12,19 +12,14 @@ defmodule Phoenix.Sync do
   @shape_keys [:namespace, :where, :columns]
   @shape_params @shape_keys |> Enum.map(&to_string/1)
 
-  @type shape_options :: [
-          unquote(NimbleOptions.option_typespec(PredefinedShape.schema()))
+  @type queryable() :: Ecto.Queryable.t() | Ecto.Schema.t() | Ecto.Changeset.t()
+  @type shape_specification :: [
+          unquote(NimbleOptions.option_typespec(Phoenix.Sync.PredefinedShape.schema()))
         ]
-
-  if Code.ensure_loaded?(Ecto) do
-    @type shape_definition ::
-            String.t()
-            | Ecto.Queryable.t()
-            | shape_options()
-  else
-    @type shape_definition() :: shape_options()
-  end
-
+  @type shape_definition ::
+          String.t()
+          | queryable()
+          | shape_specification()
   @type param_override ::
           {:namespace, String.t()}
           | {:table, String.t()}
@@ -287,7 +282,7 @@ defmodule Phoenix.Sync do
   - `columns` - The columns included in the shape. E.g. `["id", "title", "completed"]`
   - `params` - The values associated with a parameterized where clause. E.g. `[true, 1, "alive"]`, `%{1 => true}`
   """
-  @spec interrupt(shape_definition() | (match_shape_params() -> boolean()), shape_options()) ::
+  @spec interrupt(shape_definition() | (match_shape_params() -> boolean()), shape_specification()) ::
           {:ok, non_neg_integer()}
   def interrupt(shape, shape_opts \\ []) do
     Phoenix.Sync.ShapeRequestRegistry.interrupt_matching(shape, shape_opts)
@@ -339,7 +334,7 @@ defmodule Phoenix.Sync do
 
   #{NimbleOptions.docs(PredefinedShape.schema())}
   """
-  @spec shape!(shape_definition(), shape_options()) :: PredefinedShape.t()
+  @spec shape!(shape_definition(), shape_specification()) :: PredefinedShape.t()
   def shape!(shape, shape_opts \\ []) do
     PredefinedShape.new!(shape, shape_opts)
   end

lib/phoenix/sync/application.ex

@@ -9,7 +9,12 @@ defmodule Phoenix.Sync.Application do
 
   @impl true
   def start(_type, _args) do
-    base_children = [Phoenix.Sync.ShapeRequestRegistry]
+    base_children = [Phoenix.Sync.Shape.Supervisor, Phoenix.Sync.ShapeRequestRegistry]
+
+    # Electric is noisy by default which isn't helpful when developing or in
+    # production so up its log level to info (unless otherwise configured using
+    # the `ELECTRIC_LOG_LEVEL` env var.
+    Logger.put_application_level(:electric, electric_log_level())
 
     children =
       case children() do
@@ -102,4 +107,16 @@ defmodule Phoenix.Sync.Application do
 
     config
   end
+
+  defp electric_log_level do
+    valid_levels = Enum.map(Logger.levels(), &to_string/1)
+    configured_level = System.get_env("ELECTRIC_LOG_LEVEL", "info")
+
+    if configured_level in valid_levels do
+      String.to_existing_atom(configured_level)
+    else
+      Logger.warning("Invalid ELECTRIC_LOG_LEVEL: #{configured_level}. Defaulting to :info.")
+      :info
+    end
+  end
 end

lib/phoenix/sync/client.ex

@@ -119,16 +119,23 @@ defmodule Phoenix.Sync.Client do
   @spec stream(Phoenix.Sync.shape_definition(), Electric.Client.stream_options()) :: Enum.t()
   def stream(shape, stream_opts \\ [])
 
-  def stream(shape, stream_opts) do
-    stream(shape, stream_opts, nil)
+  def stream(table, stream_opts) when is_binary(table) and is_list(stream_opts) do
+    stream(Keyword.put(stream_opts, :table, table), [])
   end
 
-  @doc false
-  # used for testing. `config` replace the application configuration
-  def stream(shape, stream_opts, config) do
-    client = new!(config)
-    {shape, stream_opts} = resolve_shape(shape, stream_opts)
-    Electric.Client.stream(client, shape, stream_opts)
+  def stream(shape, []) when is_list(shape) do
+    {client, shape} = Keyword.pop_lazy(shape, :client, &new!/0)
+
+    {shape, shape_stream_opts} = resolve_shape(shape, [])
+
+    Electric.Client.stream(client, shape, shape_stream_opts)
+  end
+
+  def stream(shape, stream_opts) when not is_list(shape) and is_list(stream_opts) do
+    {client, stream_opts} = Keyword.pop_lazy(stream_opts, :client, &new!/0)
+    {shape, shape_stream_opts} = resolve_shape(shape, stream_opts)
+
+    Electric.Client.stream(client, shape, shape_stream_opts)
   end
 
   defp resolve_shape(shape, stream_opts) do

lib/phoenix/sync/predefined_shape.ex

@@ -44,7 +44,7 @@ defmodule Phoenix.Sync.PredefinedShape do
   @type options() :: [option()]
 
   if Code.ensure_loaded?(Ecto) do
-    @type shape() :: options() | Electric.Client.ecto_shape()
+    @type shape() :: options() | Phoenix.Sync.queryable()
   else
     @type shape() :: options()
   end
@@ -55,6 +55,14 @@ defmodule Phoenix.Sync.PredefinedShape do
   @doc false
   def schema, do: @keyword_shape_schema
 
+  def is_queryable?(schema) when is_atom(schema) do
+    Code.ensure_loaded?(schema) && function_exported?(schema, :__schema__, 1) &&
+      !is_nil(schema.__schema__(:source))
+  end
+
+  def is_queryable?(q) when is_struct(q, Ecto.Query) or is_struct(q, Ecto.Changeset), do: true
+  def is_queryable?(_), do: false
+
   @doc false
   @spec new!(shape(), options()) :: t()
   def new!(opts, config \\ [])

lib/phoenix/sync/sandbox.ex

@@ -187,6 +187,11 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
 
     alias __MODULE__
 
+    defmodule Error do
+      @moduledoc false
+      defexception [:message]
+    end
+
     @type start_opts() :: [{:shared, boolean()}]
 
     @registry __MODULE__.Registry
@@ -282,7 +287,7 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
 
         stack_id ->
           case GenServer.whereis(__MODULE__.Stack.name(stack_id)) do
-            nil -> raise RuntimeError, message: "no stack found for #{inspect(stack_id)}"
+            nil -> raise Error, message: "no stack found for #{inspect(stack_id)}"
             _pid -> :ok
           end
       end
@@ -330,7 +335,8 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
 
     @doc false
     def stack_id! do
-      stack_id() || raise "No stack_id found. Did you call Phoenix.Sync.Sandbox.start!/1?"
+      stack_id() ||
+        raise Error, message: "No stack_id found. Did you call Phoenix.Sync.Sandbox.start!/1?"
     end
 
     @doc false
@@ -344,14 +350,15 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
       # has even started
       case GenServer.whereis(Sandbox.StackRegistry) do
         nil ->
-          raise """
-          Phoenix.Sync.Sandbox is not running. Have you set the mode to `:sandbox` in `config/test.exs`?
+          raise Error,
+            message: """
+            Phoenix.Sync.Sandbox is not running. Have you set the mode to `:sandbox` in `config/test.exs`?
 
-          # config/test.exs
-          config :phoenix_sync,
-            env: config_env(),
-            mode: :sandbox
-          """
+            # config/test.exs
+            config :phoenix_sync,
+              env: config_env(),
+              mode: :sandbox
+            """
 
         registry_pid when is_pid(registry_pid) ->
           pids
@@ -380,16 +387,19 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
         |> lookup_stack_id()
         |> case do
           nil ->
-            raise RuntimeError,
-                  "No stack_id found for process #{inspect(parent)}. Did you call Phoenix.Sync.Sandbox.start!/1?"
+            raise Error,
+              message:
+                "No stack_id found for process #{inspect(parent)}. Did you call Phoenix.Sync.Sandbox.start!/1?"
 
           stack_id ->
             case GenServer.whereis(name_or_pid) do
               pid when is_pid(pid) ->
                 Sandbox.StackRegistry.register(pid, stack_id)
 
               other ->
-                raise "`allow/4` expects a PID or a locally registered process name but lookup returned: #{inspect(other)}"
+                raise Error,
+                  message:
+                    "`allow/4` expects a PID or a locally registered process name but lookup returned: #{inspect(other)}"
             end
         end
       end
@@ -424,7 +434,7 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
       if stack_id = stack_id() do
         Sandbox.StackRegistry.get_client(stack_id)
       else
-        {:error, "No stack_id found. Did you call Phoenix.Sync.Sandbox.start!/1?"}
+        raise Error, message: "No stack_id found. Did you call Phoenix.Sync.Sandbox.start!/1?"
       end
     end
 
@@ -439,6 +449,18 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
       client
     end
 
+    @doc false
+    def must_refetch!(shape) do
+      stack_id = stack_id!()
+
+      {%{namespace: namespace, table: table}, _} =
+        shape
+        |> Phoenix.Sync.PredefinedShape.new!()
+        |> Phoenix.Sync.PredefinedShape.to_stream_params()
+
+      Sandbox.Producer.truncate(stack_id, {namespace || "public", table})
+    end
+
     @impl true
     @doc false
     def init(_) do

lib/phoenix/sync/sandbox/producer.ex

@@ -6,7 +6,8 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
       Transaction,
       NewRecord,
       UpdatedRecord,
-      DeletedRecord
+      DeletedRecord,
+      TruncatedRelation
     }
 
     alias Electric.Replication.LogOffset
@@ -33,6 +34,10 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
       GenServer.cast(name(stack_id), {:emit_changes, changes})
     end
 
+    def truncate(stack_id, relation) do
+      GenServer.cast(name(stack_id), {:truncate, relation})
+    end
+
     def name(stack_id) do
       Phoenix.Sync.Sandbox.name({__MODULE__, stack_id})
     end
@@ -60,6 +65,17 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
       {:noreply, %{state | txid: next_txid}}
     end
 
+    def handle_cast({:truncate, relation}, state) do
+      changes = [%TruncatedRelation{relation: relation}]
+
+      :ok =
+        state.txid
+        |> transaction(changes)
+        |> ShapeLogCollector.store_transaction(ShapeLogCollector.name(state.stack_id))
+
+      {:noreply, %{state | txid: state.txid + 100}}
+    end
+
     defp transaction(txid, changes) do
       %Transaction{
         xid: txid,