elixir-client: Build shape definition from changeset (#2801)

Allows for generating a ShapeDefinition from an Ecto.Changeset

Also bundles a few quality-of-life changes:

- ShapeDefinition.params/1 returns a `%{binary => binary}` map in query
string mode, rather than a mix of atom and binary keys
- Widen accepted types for where parameters from `:string` to [:string,
:integer, :float, :boolean]` to make life simpler for user. these types
are trivial to convert to strings so why not do that for people
- Give ShapeDefinitions a `replica` setting. This is how it should have
been to begin with -- the replica mode is a property of the shape, not
of the stream
- Expose the ShapeDefinition validation schema so we can split options
and validate partial shape parameters (Phoenix.Sync.PredefinedShape
accepts a big bag of options that we need to assign appropriately)

Author: magnetised

.changeset/lazy-rivers-camp.md

@@ -0,0 +1,5 @@
+---
+"@core/elixir-client": patch
+---
+
+Support generating shape definitions from Ecto.Changeset structs, add replica mode to client ShapeDefinitions, ensure client parameters are always of type %{binary() => binary()} and expose some options schema information for use in Phoenix.Sync

packages/elixir-client/lib/electric/client.ex

@@ -239,6 +239,15 @@ defmodule Electric.Client do
           endpoint: URI.t(),
           fetch: {module(), term()}
         }
+  if Code.ensure_loaded?(Ecto) do
+    @type ecto_shape() :: Ecto.Queryable.t() | Ecto.Changeset.t() | (map() -> Ecto.Changeset.t())
+
+    # queryable is a schema module, an ecto query, a function that returns a changeset
+    # or a changeset
+    defguardp is_ecto_shape(ecto_queryable)
+              when is_atom(ecto_queryable) or is_struct(ecto_queryable, Ecto.Query) or
+                     is_function(ecto_queryable, 1) or is_struct(ecto_queryable, Ecto.Changeset)
+  end
 
   @doc """
   Create a new client.
@@ -356,6 +365,13 @@ defmodule Electric.Client do
       end
     end
 
+    defp validate_queryable!(%Ecto.Query{} = query), do: query
+    defp validate_queryable!(%Ecto.Changeset{} = changeset), do: changeset
+
+    defp validate_queryable!(changeset_fun) when is_function(changeset_fun, 1) do
+      changeset_fun
+    end
+
     @doc """
     Create a [`ShapeDefinition`](`Electric.Client.ShapeDefinition`) from an `Ecto` query.
 
@@ -364,20 +380,26 @@ defmodule Electric.Client do
 
         iex> query = from(t in MyApp.Todo, where: t.completed == false)
         iex> Elixir.Client.shape!(query)
-        %Electric.Client.ShapeDefinition{table: "todos" where: "(\\"completed\\" = FALSE)"}
+        %Electric.Client.ShapeDefinition{table: "todos", where: "(\\"completed\\" = FALSE)"}
+
+    Also takes an `Ecto.Changeset` or 1-arity function returning an `Ecto.Changeset`:
+
+        iex> Elixir.Client.shape!(&MyApp.Todo.changeset/1)
+        %Electric.Client.ShapeDefinition{table: "todos", columns: ["title", "completed"]}
+
+    Passing a changeset will filter the table columns according to the applied
+    validations so if you want a shape to include a column, you must ensure
+    that it has some kind of validation, using e.g.
+    `Ecto.Changeset.validate_required/2`
 
     Values from the Electric change stream will be mapped to instances of the
     passed `Ecto.Schema` module.
     """
-    @spec shape!(Ecto.Queryable.t()) :: ShapeDefinition.t() | no_return()
-    def shape!(queryable) when is_atom(queryable) do
+    @spec shape!(ecto_shape()) :: ShapeDefinition.t() | no_return()
+    def shape!(queryable) when is_ecto_shape(queryable) do
       queryable
       |> validate_queryable!()
-      |> Electric.Client.EctoAdapter.shape_from_query!()
-    end
-
-    def shape!(%Ecto.Query{} = query) do
-      Electric.Client.EctoAdapter.shape_from_query!(query)
+      |> Electric.Client.EctoAdapter.shape!()
     end
   end
 
@@ -389,10 +411,8 @@ defmodule Electric.Client do
   @doc """
   A shortcut to [`ShapeDefinition.new!/2`](`Electric.Client.ShapeDefinition.new!/2`).
   """
-  def shape!(table_or_query, opts \\ [])
-
   @spec shape!(String.t(), ShapeDefinition.options()) :: ShapeDefinition.t() | no_return()
-  def shape!(table_name, opts) when is_binary(table_name) do
+  def shape!(table_name, opts \\ []) when is_binary(table_name) do
     ShapeDefinition.new!(table_name, opts)
   end
 
@@ -424,7 +444,7 @@ defmodule Electric.Client do
   - Using a full shape definition:
 
         {:ok, client} = Electric.Client.new(base_url: "http://localhost:3000")
-        {:ok, shape} = Electric.Client.shape("todos", where: "completed = false")
+        {:ok, shape} = Electric.Client.shape("todos", where: "completed = false", replica: :full)
         stream = Electric.Client.stream(client, shape)
 
   - Or with an `Ecto` query or `Ecto.Schema` struct:
@@ -448,7 +468,7 @@ defmodule Electric.Client do
   end
 
   if Code.ensure_loaded?(Ecto) do
-    @spec stream(t(), String.t() | Ecto.Queryable.t()) :: Enumerable.t(message())
+    @spec stream(t(), String.t() | ecto_shape()) :: Enumerable.t(message())
   else
     @spec stream(t(), String.t()) :: Enumerable.t(message())
   end
@@ -482,16 +502,11 @@ defmodule Electric.Client do
   @spec stream(t(), shape(), stream_options()) :: Enumerable.t(message())
 
   if Code.ensure_loaded?(Ecto) do
-    def stream(%Client{} = client, queryable, opts) when is_atom(queryable) do
+    def stream(%Client{} = client, queryable, opts) when is_ecto_shape(queryable) do
       shape_definition = shape!(queryable)
 
       stream(client, shape_definition, opts)
     end
-
-    def stream(%Client{} = client, %Ecto.Query{} = query, opts) do
-      shape_definition = shape!(query)
-      stream(client, shape_definition, opts)
-    end
   end
 
   def stream(%Client{} = client, table_name, opts) when is_binary(table_name) do

packages/elixir-client/lib/electric/client/authenticator/mock_authenticator.ex

@@ -22,7 +22,7 @@ defmodule Electric.Client.Authenticator.MockAuthenticator do
   end
 
   defp shape_hash(params, config) do
-    auth_hash([:table, :columns, :where], params, config)
+    auth_hash(["table", "columns", "where"], params, config)
   end
 
   defp put_request_params(request, hash) do

packages/elixir-client/lib/electric/client/ecto_adapter.ex

@@ -6,9 +6,18 @@ if Code.ensure_loaded?(Ecto) do
 
     @behaviour Electric.Client.ValueMapper
 
+    def shape!(schema, opts \\ [])
+
+    def shape!(schema, opts) when is_atom(schema), do: shape_from_query!(schema, opts)
+    def shape!(%Ecto.Query{} = query, opts), do: shape_from_query!(query, opts)
+    def shape!(%Ecto.Changeset{} = changeset, opts), do: shape_from_changeset!(changeset, opts)
+
+    def shape!(changeset_fun, opts) when is_function(changeset_fun, 1),
+      do: shape_from_changeset!(changeset_fun, opts)
+
     @doc false
     @spec shape_from_query!(Ecto.Queryable.t()) :: ShapeDefinition.t()
-    def shape_from_query!(queryable) do
+    def shape_from_query!(queryable, opts \\ []) do
       query = Ecto.Queryable.to_query(queryable)
 
       validate_query!(query)
@@ -20,14 +29,79 @@ if Code.ensure_loaded?(Ecto) do
       columns = query_columns(query)
       where = where(query)
 
-      ShapeDefinition.new!(table_name,
-        namespace: namespace,
-        where: where,
-        columns: columns,
-        parser: {__MODULE__, struct}
+      ShapeDefinition.new!(
+        table_name,
+        merge_shape_opts(
+          [namespace: namespace, where: where, columns: columns, parser: {__MODULE__, struct}],
+          opts
+        )
       )
     end
 
+    @shape_from_changeset_opts_schema ShapeDefinition.schema_definition()
+                                      |> Keyword.drop([:columns])
+                                      |> NimbleOptions.new!()
+
+    @type shape_from_changeset_opts() :: [
+            unquote(NimbleOptions.option_typespec(@shape_from_changeset_opts_schema))
+          ]
+
+    @spec shape_from_changeset!(
+            (map() -> Ecto.Changeset.t()) | Ecto.Changeset.t(),
+            shape_from_changeset_opts()
+          ) :: ShapeDefinition.t()
+    def shape_from_changeset!(changeset_or_fun, opts \\ [])
+
+    def shape_from_changeset!(%Ecto.Changeset{} = changeset, opts) do
+      generate_shape_from_changeset(changeset, opts)
+    end
+
+    def shape_from_changeset!(changeset_fun, opts) when is_function(changeset_fun, 1) do
+      case changeset_fun.(%{}) do
+        %Ecto.Changeset{} = changeset ->
+          generate_shape_from_changeset(changeset, opts)
+
+        invalid ->
+          raise ArgumentError,
+            message:
+              "Changeset function returned #{inspect(invalid)}, was expecting an Ecto.Changeset struct"
+      end
+    end
+
+    defp generate_shape_from_changeset(%Ecto.Changeset{data: %schema{}} = changeset, opts) do
+      if !function_exported?(schema, :__schema__, 1),
+        do:
+          raise(ArgumentError,
+            message:
+              "cannot generate a shape from a schema-less changeset. Use #{inspect(ShapeDefinition)}.new/2"
+          )
+
+      table_name = schema.__schema__(:source)
+      namespace = schema.__schema__(:prefix)
+      pks = schema.__schema__(:primary_key)
+
+      columns =
+        [pks, changeset.required, Keyword.keys(changeset.validations)]
+        |> Enum.concat()
+        |> Enum.uniq()
+        |> Enum.map(&schema.__schema__(:field_source, &1))
+        |> Enum.map(&to_string/1)
+
+      ShapeDefinition.new!(
+        table_name,
+        merge_shape_opts(
+          [columns: columns, parser: {__MODULE__, schema}, namespace: namespace],
+          opts
+        )
+      )
+    end
+
+    defp merge_shape_opts(base, overrides) do
+      Keyword.merge(base, overrides, fn _key, base, over ->
+        if is_nil(over), do: base, else: over
+      end)
+    end
+
     defp table_name(%{
            prefix: query_prefix,
            from: %{prefix: source_prefix, source: {table_name, struct}}

packages/elixir-client/lib/electric/client/embedded.ex

@@ -57,10 +57,10 @@ if Code.ensure_loaded?(Electric.Shapes.Api) do
       Map.merge(
         request.params,
         %{
-          offset: to_string(request.offset),
-          handle: request.shape_handle,
-          live: request.live,
-          replica: request.replica
+          "offset" => to_string(request.offset),
+          "handle" => request.shape_handle,
+          "live" => request.live,
+          "replica" => request.replica
         }
       )
     end

packages/elixir-client/lib/electric/client/fetch/request.ex

@@ -13,6 +13,10 @@ defmodule Electric.Client.Fetch.Request do
     :shape_handle,
     :live,
     :next_cursor,
+    # `replica` does not belong here, it's part of the shape definition that we
+    # receive via the client params. the other keys in this struct are part
+    # of consuming a stream rather than defining the shape.
+    # kept for backwards compatibility
     replica: :default,
     method: :get,
     offset: Client.Offset.before_all(),