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 @api_schema_opts [ storage: [type: {:or, [:map, nil]}] ] @sync_schema_opts [ transform: [ type: {:or, [:mfa, {:fun, 1}, :atom]}, doc: """ A transform function to apply to each row. This can be either a MFA tuple (`{MyModule, :my_fun, [arg1, ...]}`), a `&MyModule.fun/1` capture or an `Ecto.Schema` module (depending on use). See the documentation of `Phoenix.Sync.shape!/2` for more details. """, type_doc: ~s/`(map() -> [map()]) | mfa() | module()`/ ] ] shape_schema_gen = fn required? -> Keyword.take( [table: [type: :string, required: required?]] ++ ShapeDefinition.schema_definition(), ShapeDefinition.public_keys() ) ++ @sync_schema_opts end @shape_definition_schema shape_schema_gen.(false) @keyword_shape_schema shape_schema_gen.(true) @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) @sync_schema NimbleOptions.new!(@sync_schema_opts) @api_schema_keys Keyword.keys(@api_schema_opts) @stream_schema_keys Keyword.keys(@stream_schema.schema) @shape_definition_keys ShapeDefinition.public_keys() @sync_schema_keys Keyword.keys(@sync_schema_opts) # 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, :sync_config ] @type t :: %__MODULE__{} @type option() :: unquote(NimbleOptions.option_typespec(@public_schema)) @type options() :: [option()] if Code.ensure_loaded?(Ecto) do @type shape() :: options() | Phoenix.Sync.queryable() else @type shape() :: options() end @doc false def shape_schema, do: @shape_schema @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 \\ []) 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!(table, opts) when is_binary(table) and is_list(opts) do new!([table: table], opts) end 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) defp split_and_validate_opts!(opts, mode) do {shape_config, api_config, stream_config, sync_config} = opts |> split_opts!() |> validate_opts!(mode) [ shape_config: shape_config, api_config: api_config, stream_config: stream_config, sync_config: sync_config ] end defp split_opts!(opts) do {shape_opts, other_opts} = Keyword.split(opts, @shape_definition_keys) {api_opts, other_opts} = Keyword.split(other_opts, @api_schema_keys) {sync_opts, other_opts} = Keyword.split(other_opts, @sync_schema_keys) stream_opts = case Keyword.split(other_opts, @stream_schema_keys) do {stream_opts, []} -> stream_opts {_stream_opts, invalid_opts} -> raise ArgumentError, message: "received invalid options to a shape definition: #{inspect(invalid_opts)}" end {shape_opts, api_opts, stream_opts, sync_opts} end defp validate_opts!({shape_opts, api_opts, stream_opts, sync_opts}, mode) do shape_config = validate_shape_config(shape_opts, mode) api_config = NimbleOptions.validate!(api_opts, @api_schema) sync_config = NimbleOptions.validate!(sync_opts, @sync_schema) # 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, api_config, stream_config, sync_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 @doc false def new_macro!(queryable, shape_opts, caller, macro_opts \\ []) def new_macro!(queryable, shape_opts, caller, macro_opts) when is_tuple(queryable) and is_list(shape_opts) do function_caller = case Keyword.fetch(macro_opts, :function) do {:ok, function} -> %{caller | function: function} :error -> caller end case Macro.expand_literals(queryable, function_caller) do schema when is_atom(schema) -> new!(schema, macro_sanitise_opts(shape_opts, caller, macro_opts)) query when is_tuple(query) -> raise ArgumentError, message: "Router shape configuration only accepts a Ecto.Schema module as a query." <> " For Ecto.Query support please use `Phoenix.Sync.Controller.sync_render/3`" end end def new_macro!(shape, opts, caller, macro_opts) when is_list(shape) and is_list(opts) do shape = macro_sanitise_opts(shape, caller, macro_opts) opts = macro_sanitise_opts(opts, caller, macro_opts) new!(shape, opts) end defp macro_sanitise_opts(opts, caller, macro_opts) do opts |> Keyword.replace_lazy(:storage, fn storage_ast -> {storage, _binding} = Code.eval_quoted(storage_ast, [], caller) storage end) |> Keyword.replace_lazy(:transform, fn {:&, _, _} = transform_ast -> context = Keyword.fetch!(macro_opts, :context) {transform, _binding} = transform_ast |> macro_validate_capture!(context) |> Code.eval_quoted([], caller) transform {:{}, _, _} = transform_ast -> {transform, _binding} = Code.eval_quoted(transform_ast, [], caller) transform {:__aliases__, _, _} = transform_ast -> {transform, _binding} = Code.eval_quoted(transform_ast, [], caller) transform end) end # phoenix router does not support captures in the router defp macro_validate_capture!(_ast, :phoenix) do raise ArgumentError, message: "Invalid transform function specification in sync shape definition." <> " When using Phoenix Router please use an MFA tuple (`transform: {Mod, :fun, [arg1, ...]}`)" end # only Mod.fun/1 style captures are supported -- you can't quote a &mod.fun(&1) style capture defp macro_validate_capture!(ast, :plug) do case ast do {:&, _, [{:/, _, _}]} = capture_ast -> capture_ast _ -> raise ArgumentError, message: "Invalid transform function specification in sync shape definition." <> " Expected either a capture (&Mod.fun/1) or MFA tuple ({Mod, :fun, [arg1, ...]})" end end def client(%Electric.Client{} = client, %__MODULE__{} = predefined_shape) do Electric.Client.merge_params(client, to_client_params(predefined_shape)) end @doc false def to_client_params(%__MODULE__{} = predefined_shape) do predefined_shape |> to_shape_definition() |> ShapeDefinition.params() end @doc false def to_api_params(%__MODULE__{} = predefined_shape) do predefined_shape |> to_shape_definition() |> ShapeDefinition.params(format: :keyword) |> Keyword.merge(predefined_shape.api_config) end @doc false def to_shape_params(%__MODULE__{} = predefined_shape) do predefined_shape |> to_shape_definition() |> ShapeDefinition.params(format: :keyword) end @doc false def to_shape(%__MODULE__{} = predefined_shape) do to_shape_definition(predefined_shape) end @doc false def to_stream_params(%__MODULE__{} = predefined_shape) do {to_shape_definition(predefined_shape), predefined_shape.stream_config} end defp to_shape_definition(%__MODULE__{query: nil, shape_config: shape_config}) do ShapeDefinition.new!(shape_config) end if Code.ensure_loaded?(Ecto) do # we resolve the query at runtime to avoid compile-time dependencies in # router modules defp to_shape_definition(%__MODULE__{query: queryable, shape_config: shape_config}) do try do Electric.Client.EctoAdapter.shape!(queryable, shape_config) rescue e in Protocol.UndefinedError -> raise ArgumentError, message: "Invalid query `#{inspect(queryable)}`: #{e.description}" end end end @doc false def transform_fun(nil), do: nil def transform_fun(%__MODULE__{sync_config: sync_config}) do transform_fun(sync_config[:transform]) end def transform_fun({m, f, a}) when is_atom(m) and is_atom(f) and is_list(a) do fn row -> List.wrap(apply(m, f, [row | a])) end end def transform_fun(fun) when is_function(fun, 1) do fn msg -> List.wrap(fun.(msg)) end end def transform_fun(module) when is_atom(module) do ecto_transform_fun(module) end if Code.ensure_loaded?(Ecto) do defp ecto_transform_fun(module) when is_atom(module) do ecto_transform = Electric.Client.EctoAdapter.for_schema(%{}, module) &apply_ecto_transform(&1, ecto_transform) end defp apply_ecto_transform(%{"value" => value} = msg, transform_fun) do [Map.put(msg, "value", transform_fun.(value))] end else defp ecto_transform_fun(module) when is_atom(module) do raise ArgumentError, message: "Ecto not available: cannot generate transform function from #{inspect(module)}" end end end