add support for changeset shapes

and clean up the shape definition path, removing phoenix.sync from the
chain of "things that understand shapes" so now if you change the
client's view of a shape it should propagate to phoenix.sync without
changes

Author: magnetised

lib/phoenix/sync/client.ex

@@ -124,26 +124,16 @@ defmodule Phoenix.Sync.Client do
   end
 
   @doc false
-  def stream(shape, stream_opts, sync_opts) do
-    client = new!(sync_opts)
-    {shape, shape_stream_opts} = resolve_shape(shape)
-    Electric.Client.stream(client, shape, Keyword.merge(shape_stream_opts, stream_opts))
+  # 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)
   end
 
-  defp resolve_shape(table) when is_binary(table) do
-    {table, []}
-  end
-
-  defp resolve_shape(definition) when is_list(definition) do
-    shape = PredefinedShape.new!(definition)
-    PredefinedShape.to_stream_params(shape)
-  end
-
-  defp resolve_shape(schema) when is_atom(schema) do
-    {schema, []}
-  end
-
-  defp resolve_shape(%Ecto.Query{} = query) do
-    {query, []}
+  defp resolve_shape(shape, stream_opts) do
+    shape
+    |> PredefinedShape.new!(stream_opts)
+    |> Phoenix.Sync.PredefinedShape.to_stream_params()
   end
 end

lib/phoenix/sync/controller.ex

@@ -41,9 +41,37 @@ defmodule Phoenix.Sync.Controller do
         end
       end
 
+  ## Shape definitions
+
+  Shape definitions can be any of the following:
+
+  - An `Ecto.Schema` module:
+
+        sync_render(conn, MyPlugApp.Todos.Todo)
+
+  - An `Ecto` query:
+
+        sync_render(conn, params, from(t in Todos.Todo, where: t.owner_id == ^user_id))
+
+  - A `changeset/1` function which defines the table and columns:
+
+        sync_render(conn, params, &Todos.Todo.changeset/1)
+
+  - A `changeset/1` function plus a where clause:
+
+        sync_render(conn, params, &Todos.Todo.changeset/1, where: "completed = false")
+
+    or a parameterized where clause:
+
+        sync_render(conn, params, &Todos.Todo.changeset/1, where: "completed = $1", params: [false])
+
+  - A keyword list defining the shape parameters:
+
+        sync_render(conn, params, table: "todos", namespace: "my_app", where: "completed = $1", params: [false])
   """
 
   alias Phoenix.Sync.Plug.CORS
+  alias Phoenix.Sync.PredefinedShape
 
   defmacro __using__(opts \\ []) do
     # validate that we're being used in the context of a Plug.Router impl
@@ -78,25 +106,31 @@ defmodule Phoenix.Sync.Controller do
   @doc """
   Return the sync events for the given shape as a `Plug.Conn` response.
   """
-  @spec sync_render(Plug.Conn.t(), Plug.Conn.params(), Electric.Shapes.Api.shape_opts()) ::
-          Plug.Conn.t()
-  def sync_render(%{private: %{phoenix_endpoint: endpoint}} = conn, params, shape) do
+  @spec sync_render(
+          Plug.Conn.t(),
+          Plug.Conn.params(),
+          PredefinedShape.shape(),
+          PredefinedShape.options()
+        ) :: Plug.Conn.t()
+  def sync_render(conn, params, shape, shape_opts \\ [])
+
+  def sync_render(%{private: %{phoenix_endpoint: endpoint}} = conn, params, shape, shape_opts) do
     api =
       endpoint.config(:phoenix_sync) ||
         raise RuntimeError,
           message:
             "Please configure your Endpoint with [phoenix_sync: Phoenix.Sync.plug_opts()] in your `c:Application.start/2`"
 
-    sync_render_api(conn, api, params, shape)
+    sync_render_api(conn, api, params, shape, shape_opts)
   end
 
   # the Plug.{Router, Builder} version
-  def sync_render(%{private: %{phoenix_sync_api: api}} = conn, params, shape) do
-    sync_render_api(conn, api, params, shape)
+  def sync_render(%{private: %{phoenix_sync_api: api}} = conn, params, shape, shape_opts) do
+    sync_render_api(conn, api, params, shape, shape_opts)
   end
 
-  defp sync_render_api(conn, api, params, shape) do
-    predefined_shape = Phoenix.Sync.PredefinedShape.new!(shape)
+  defp sync_render_api(conn, api, params, shape, shape_opts) do
+    predefined_shape = PredefinedShape.new!(shape, shape_opts)
 
     {:ok, shape_api} = Phoenix.Sync.Adapter.PlugApi.predefined_shape(api, predefined_shape)
 

lib/phoenix/sync/predefined_shape.ex

@@ -3,163 +3,149 @@ defmodule Phoenix.Sync.PredefinedShape do
 
   # A self-contained way to hold shape definition information, alongside stream
   # configuration, compatible with both the embedded and HTTP API versions.
+  # Defers to the client code to validate shape options, so we can keep up with
+  # changes to the api without duplicating changes here
 
   alias Electric.Client.ShapeDefinition
 
-  @keys [
-    :relation,
-    :where,
-    :columns,
-    :replica,
-    :storage
-  ]
+  shape_schema_gen = fn required? ->
+    Keyword.take(
+      [table: [type: :string, required: required?]] ++ ShapeDefinition.schema_definition(),
+      ShapeDefinition.public_keys()
+    )
+  end
 
-  @schema NimbleOptions.new!(
-            table: [type: :string],
-            query: [type: {:or, [:atom, {:struct, Ecto.Query}]}, doc: false],
-            namespace: [type: :string, default: "public"],
-            where: [type: :string],
-            columns: [type: {:list, :string}],
-            replica: [type: {:in, [:default, :full]}],
-            storage: [type: {:or, [:map, nil]}]
-          )
+  @shape_definition_schema shape_schema_gen.(false)
+  @keyword_shape_schema shape_schema_gen.(true)
 
-  defstruct [:query | @keys]
+  @api_schema_opts [
+    storage: [type: {:or, [:map, nil]}]
+  ]
+
+  @shape_schema NimbleOptions.new!(@shape_definition_schema)
+  @api_schema NimbleOptions.new!(@api_schema_opts)
+  @stream_schema Electric.Client.Stream.options_schema()
+  @public_schema NimbleOptions.new!(@shape_definition_schema ++ @api_schema_opts)
+
+  @api_schema_keys Keyword.keys(@api_schema_opts)
+  @stream_schema_keys Keyword.keys(@stream_schema.schema)
+  @shape_definition_keys ShapeDefinition.public_keys()
+
+  # we hold the query separate from the shape definition in order to allow
+  # for transformation of a query to a shape definition at runtime rather
+  # than compile time.
+  defstruct [
+    :shape_config,
+    :api_config,
+    :stream_config,
+    :query
+  ]
 
   @type t :: %__MODULE__{}
+  @type options() :: [unquote(NimbleOptions.option_typespec(@public_schema))]
 
-  def schema, do: @schema
-  def keys, do: @keys
+  if Code.ensure_loaded?(Ecto) do
+    @type shape() :: options() | Electric.Client.ecto_shape()
+  else
+    @type shape() :: options()
+  end
+
+  def schema, do: @public_schema
 
+  @spec new!(shape(), options()) :: t()
   def new!(opts, config \\ [])
 
-  def new!(shape, opts) when is_list(opts) and is_list(shape) do
-    config = NimbleOptions.validate!(Keyword.merge(shape, opts), @schema)
-    new(Keyword.put(config, :relation, build_relation!(config)))
+  def new!(shape, opts) when is_list(shape) and is_list(opts) do
+    shape
+    |> Keyword.merge(opts)
+    |> split_and_validate_opts!(mode: :keyword)
+    |> new()
   end
 
-  def new!(schema, opts) when is_atom(schema) do
-    new(Keyword.put(opts, :query, schema))
+  def new!(table, opts) when is_binary(table) and is_list(opts) do
+    new!([table: table], opts)
   end
 
-  def new!(%Ecto.Query{} = query, opts) do
-    new(Keyword.put(opts, :query, query))
+  if Code.ensure_loaded?(Ecto) do
+    def new!(ecto_shape, opts)
+        when is_atom(ecto_shape) or is_struct(ecto_shape, Ecto.Query) or
+               is_function(ecto_shape, 1) or
+               is_struct(ecto_shape, Ecto.Changeset) do
+      opts
+      |> split_and_validate_opts!(mode: :ecto)
+      |> Keyword.merge(query: ecto_shape)
+      |> new()
+    end
   end
 
-  defp new(opts) do
-    struct(__MODULE__, opts)
-  end
+  defp new(opts), do: struct(__MODULE__, opts)
 
-  defp build_relation!(opts) do
-    build_relation(opts) ||
-      raise ArgumentError,
-        message: "missing relation or table in #{inspect(opts)}"
-  end
+  defp split_and_validate_opts!(opts, mode) do
+    {shape_opts, other_opts} = Keyword.split(opts, @shape_definition_keys)
+    {api_opts, other_opts} = Keyword.split(other_opts, @api_schema_keys)
 
-  defp build_relation(opts) do
-    case Keyword.get(opts, :relation) do
-      {_namespace, _table} = relation ->
-        relation
+    stream_opts =
+      case Keyword.split(other_opts, @stream_schema_keys) do
+        {stream_opts, []} ->
+          stream_opts
 
-      nil ->
-        case Keyword.get(opts, :table) do
-          table when is_binary(table) ->
-            namespace = Keyword.get(opts, :namespace, "public")
-            {namespace, table}
+        {_stream_opts, invalid_opts} ->
+          raise ArgumentError,
+            message: "received invalid options to a shape definition: #{inspect(invalid_opts)}"
+      end
 
-          _ ->
-            nil
-        end
+    shape_config = validate_shape_config(shape_opts, mode)
+    api_config = NimbleOptions.validate!(api_opts, @api_schema)
 
-      _ ->
-        nil
-    end
+    # remove replica value from the stream because it will override the shape
+    # setting and since we've removed the `:replica` value earlier
+    # it'll always be set to default
+    stream_config =
+      NimbleOptions.validate!(stream_opts, @stream_schema)
+      |> Enum.reject(&is_nil(elem(&1, 1)))
+      |> Enum.reject(&(elem(&1, 0) == :replica))
+
+    [shape_config: shape_config, api_config: api_config, stream_config: stream_config]
+  end
+
+  # If we're defining a shape with a keyword list then we need at least the
+  # `table`. Coming from some ecto value, the table is already present
+  defp validate_shape_config(shape_opts, mode: :keyword) do
+    NimbleOptions.validate!(shape_opts, @keyword_shape_schema)
+  end
+
+  defp validate_shape_config(shape_opts, _mode) do
+    NimbleOptions.validate!(shape_opts, @shape_schema)
   end
 
   def client(%Electric.Client{} = client, %__MODULE__{} = predefined_shape) do
     Electric.Client.merge_params(client, to_client_params(predefined_shape))
   end
 
-  defp to_client_params(%__MODULE__{} = predefined_shape) do
-    {{namespace, table}, shape} =
-      predefined_shape
-      |> resolve_query()
-      |> to_list()
-      |> Keyword.pop!(:relation)
-
-    # Remove storage as it's not currently supported as a query param
-    shape
-    |> Keyword.put(:table, ShapeDefinition.url_table_name(namespace, table))
-    |> Keyword.delete(:storage)
-    |> columns_to_query_param()
+  def to_client_params(%__MODULE__{} = predefined_shape) do
+    predefined_shape
+    |> to_shape_definition()
+    |> ShapeDefinition.params()
   end
 
   def to_api_params(%__MODULE__{} = predefined_shape) do
     predefined_shape
-    |> resolve_query()
-    |> to_list()
+    |> to_shape_definition()
+    |> ShapeDefinition.params(format: :keyword)
+    |> Keyword.merge(predefined_shape.api_config)
   end
 
   def to_stream_params(%__MODULE__{} = predefined_shape) do
-    {{namespace, table}, shape} =
-      predefined_shape
-      |> resolve_query()
-      |> to_list()
-      |> Keyword.pop!(:relation)
-
-    {shape_opts, stream_opts} = Keyword.split(shape, ShapeDefinition.public_keys())
-
-    {:ok, shape_definition} =
-      ShapeDefinition.new(table, Keyword.merge(shape_opts, namespace: namespace))
-
-    {shape_definition, stream_opts}
+    {to_shape_definition(predefined_shape), predefined_shape.stream_config}
   end
 
-  defp resolve_query(%__MODULE__{query: nil} = predefined_shape) do
-    predefined_shape
+  defp to_shape_definition(%__MODULE__{query: nil, shape_config: shape_config}) do
+    ShapeDefinition.new!(shape_config)
   end
 
   # we resolve the query at runtime to avoid compile-time dependencies in
   # router modules
-  defp resolve_query(%__MODULE__{} = predefined_shape) do
-    from_queryable!(predefined_shape)
-  end
-
-  defp from_queryable!(%{query: queryable} = predefined_shape) do
-    queryable
-    |> Electric.Client.EctoAdapter.shape_from_query!()
-    |> from_shape_definition(predefined_shape)
-  end
-
-  defp from_shape_definition(%ShapeDefinition{} = shape_definition, predefined_shape) do
-    %{
-      namespace: namespace,
-      table: table,
-      where: where,
-      columns: columns
-    } = shape_definition
-
-    %{predefined_shape | relation: {namespace || "public", table}, columns: columns}
-    |> put_if(:where, where)
-  end
-
-  defp put_if(shape, _key, nil), do: shape
-  defp put_if(shape, key, value), do: Map.put(shape, key, value)
-
-  defp to_list(%__MODULE__{} = shape) do
-    Enum.flat_map(@keys, fn key ->
-      value = Map.fetch!(shape, key)
-
-      if !is_nil(value),
-        do: [{key, value}],
-        else: []
-    end)
-  end
-
-  defp columns_to_query_param(shape) do
-    case Keyword.get(shape, :columns) do
-      columns when is_list(columns) -> Keyword.put(shape, :columns, Enum.join(columns, ","))
-      _ -> shape
-    end
+  defp to_shape_definition(%__MODULE__{query: queryable, shape_config: shape_config}) do
+    Electric.Client.EctoAdapter.shape!(queryable, shape_config)
   end
 end