[wip]

Author: magnetised

lib/phoenix/sync/application.ex

@@ -9,6 +9,8 @@ defmodule Phoenix.Sync.Application do
 
   @impl true
   def start(_type, _args) do
+    base_children = [Phoenix.Sync.Shape.Supervisor]
+
     children =
       case children() do
         {:ok, children} ->
@@ -19,7 +21,10 @@ defmodule Phoenix.Sync.Application do
           []
       end
 
-    Supervisor.start_link(children, strategy: :one_for_one, name: Phoenix.Sync.Supervisor)
+    Supervisor.start_link(base_children ++ children,
+      strategy: :one_for_one,
+      name: Phoenix.Sync.Supervisor
+    )
   end
 
   @doc false

lib/phoenix/sync/client.ex

@@ -106,6 +106,13 @@ defmodule Phoenix.Sync.Client do
         columns: ["id", "title"]
       )
 
+      # consumer a pre-defined shape from a remote server, e.g.
+      # one defined using `Phoenix.Sync.Controller.sync_render/3` or
+      # `Phoenix.Sync.Router.sync/2`.
+      stream = Phoenix.Sync.Client.stream(
+        "https://myapp.com/sync/todos?user_id=1234"
+      )
+
       # once you have a stream, consume it as usual
       Enum.each(stream, &IO.inspect/1)
 
@@ -116,18 +123,30 @@ defmodule Phoenix.Sync.Client do
   Elixir/Ecto types, rather than raw column data in the form `%{"column_name"
   => "column_value"}`.
   """
+  # stream(Todo, replica: full, client: client)
+  # stream("todos", where: "...", replica: full, client: client)
+  # stream(table: "todos", where: "...", replica: full, client: client)
   @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(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(table, stream_opts) when is_binary(table) and is_list(stream_opts) do
+    stream(Keyword.put(stream_opts, :table, table), [])
   end
 
-  @doc false
-  def stream(shape, stream_opts, sync_opts) do
-    client = new!(sync_opts)
+  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)
-    Electric.Client.stream(client, shape, Keyword.merge(shape_stream_opts, stream_opts))
+    stream_opts = Keyword.merge(shape_stream_opts, stream_opts)
+    Electric.Client.stream(client, shape, stream_opts)
   end
 
   defp resolve_shape(table) when is_binary(table) do

lib/phoenix/sync/predefined_shape.ex

@@ -31,6 +31,13 @@ defmodule Phoenix.Sync.PredefinedShape do
   def schema, do: @schema
   def keys, do: @keys
 
+  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 new!(opts, config \\ [])
 
   def new!(shape, opts) when is_list(opts) and is_list(shape) do

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,17 @@ if Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) do
       client
     end
 
+    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, 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,