defmodule Phoenix.Sync do @moduledoc """ Real-time sync for Postgres-backed Phoenix applications. See the [docs](../../README.md) for more information. """ alias Electric.Client.ShapeDefinition alias Phoenix.Sync.PredefinedShape @shape_keys [:namespace, :where, :columns] @shape_params @shape_keys |> Enum.map(&to_string/1) @json if(Code.ensure_loaded?(Jason), do: Jason, else: JSON) @type queryable() :: Ecto.Queryable.t() | Ecto.Schema.t() | Ecto.Changeset.t() @type shape_specification :: [ unquote(NimbleOptions.option_typespec(Phoenix.Sync.PredefinedShape.schema())) ] @type shape_definition :: String.t() | queryable() | shape_specification() @type param_override :: {:namespace, String.t()} | {:table, String.t()} | {:where, String.t()} | {:columns, String.t()} @type param_overrides :: [param_override()] @type match_shape_params() :: %{ table: String.t(), namespace: nil | String.t(), where: nil | String.t(), params: nil | %{String.t() => String.t()}, columns: nil | [String.t(), ...] } @doc """ Returns the required adapter configuration for your Phoenix Endpoint or `Plug.Router`. ## Phoenix Configure your endpoint with the configuration at runtime by passing the `phoenix_sync` configuration to your endpoint in the `Application.start/2` callback: def start(_type, _args) do children = [ # ... {MyAppWeb.Endpoint, phoenix_sync: Phoenix.Sync.plug_opts()} ] end ## Plug Add the configuration to the Plug opts in your server configuration: children = [ {Bandit, plug: {MyApp.Router, phoenix_sync: Phoenix.Sync.plug_opts()}} ] Your `Plug.Router` must be configured with [`copy_opts_to_assign`](https://hexdocs.pm/plug/Plug.Builder.html#module-options) and you should `use` the rele defmodule MyApp.Router do use Plug.Router, copy_opts_to_assign: :options use Phoenix.Sync.Controller use Phoenix.Sync.Router plug :match plug :dispatch sync "/shapes/todos", Todos.Todo get "/shapes/user-todos" do %{"user_id" => user_id} = conn.params sync_render(conn, from(t in Todos.Todo, where: t.owner_id == ^user_id) end end """ defdelegate plug_opts(), to: Phoenix.Sync.Application @doc false defdelegate plug_opts(config), to: Phoenix.Sync.Application defdelegate client!(), to: Phoenix.Sync.Client, as: :new! @doc false def json_library, do: @json _ = """ Use request query parameters to create a `Electric.Client.ShapeDefinition`. Useful when creating authorization endpoints that validate a user's access to a specific shape. ## Parameters ### Required - `table` - the Postgres [table name](https://electric-sql.com/docs/guides/shapes#table) Note: `table` is not required in the parameters if a `:table` override is set. ### Optional - `where` - the [Shape's where clause](https://electric-sql.com/docs/guides/shapes#where-clause) - `columns` - The columns to include in the shape. - `namespace` - The Postgres namespace (also called `SCHEMA`). See [`Electric.Client.ShapeDefinition.new/2`](`Electric.Client.ShapeDefinition.new/2`) for more details on the parameters. ### Examples # pass the Plug.Conn struct for a request iex> Phoenix.Sync.shape_from_params(%Plug.Conn{params: %{"table" => "items", "where" => "visible = true" }}) {:ok, %Electric.Client.ShapeDefinition{table: "items", where: "visible = true"}} # or a simple parameter map iex> Phoenix.Sync.shape_from_params(%{"table" => "items", "columns" => "id,name,value" }) {:ok, %Electric.Client.ShapeDefinition{table: "items", columns: ["id", "name", "value"]}} iex> Phoenix.Sync.shape_from_params(%{"columns" => "id,name,value" }) {:error, "Missing `table` parameter"} ## Overriding Parameter Values If you want to hard-code some elements of the shape, ignoring the values from the request, or to set defaults, then use the `overrides` to set specific values for elements of the shape. ### Examples iex> Phoenix.Sync.shape_from_params(%{"columns" => "id,name,value"}, table: "things") {:ok, %Electric.Client.ShapeDefinition{table: "things", columns: ["id", "name", "value"]}} iex> Phoenix.Sync.shape_from_params(%{"table" => "ignored"}, table: "things") {:ok, %Electric.Client.ShapeDefinition{table: "things"}} """ @doc false @spec shape_from_params(Plug.Conn.t() | Plug.Conn.params(), overrides :: param_overrides()) :: {:ok, Electric.Client.ShapeDefinition.t()} | {:error, String.t()} def shape_from_params(conn_or_map, overrides \\ []) def shape_from_params(%Plug.Conn{} = conn, overrides) do %{params: params} = Plug.Conn.fetch_query_params(conn) shape_from_params(params, overrides) end def shape_from_params(params, overrides) when is_map(params) do shape_params = params |> Map.take(@shape_params) |> Map.new(fn {"columns", ""} -> {:columns, nil} {"columns", v} when is_binary(v) -> {:columns, :binary.split(v, ",", [:global, :trim_all])} {k, v} -> {String.to_existing_atom(k), v} end) if table = Keyword.get(overrides, :table, Map.get(params, "table")) do ShapeDefinition.new( table, Enum.map(@shape_keys, fn k -> {k, Keyword.get(overrides, k, Map.get(shape_params, k))} end) ) else {:error, "Missing `table` parameter"} end end @doc """ Interrupts all long-polling requests matching the given shape definition. The broader the shape definition, the more requests will be interrupted. Returns the number of interrupted requests. ### Examples To interrupt all shapes on the `todos` table: Phoenix.Sync.interrupt("todos") Phoenix.Sync.interrupt(table: "todos") or the same using an `Ecto.Schema` module: Phoenix.Sync.interrupt(Todos.Todo) all shapes with the given parameterized where clause: Phoenix.Sync.interrupt(table: "todos", where: "user_id = $1") or a single shape for the given user: Phoenix.Sync.interrupt( from(t in Todos.Todo, where: t.user_id == ^user_id) ) # or Phoenix.Sync.interrupt( table: "todos", where: "user_id = $1", params: [user_id] ) # or Phoenix.Sync.interrupt( table: "todos", where: "user_id = '\#{user_id}'" ) If you want more control over the match, you can pass a function that will receive a normalized shape definition and should return `true` if the active shape matches. Phoenix.Sync.interrupt(fn %{table: _, where: _, params: _} = shape -> shape.table == "todos" && shape.where == "user_id = $1" && shape.params["0"] == user_id end) The normalized shape argument is a map with the following keys: - `table`, e.g. `"todos"` - `namespace`, e.g. `"public"` - `where`, e.g. `"where user_id = $1"` - `params`, a map of argument position to argument value, e.g. `%{"0" => "true", "1" => "..."}` - `columns`, e.g. `["id", "title"]` All except `table` may be `nil`. ### Interrupting Ecto Query-based Shapes Be careful when mixing `Ecto` query-based shapes with interrupt calls using hand-written where clauses. The shape Phoenix.Sync.Controller.sync_stream(conn, params, fn -> from(t in Todos.Todo, where: t.user_id == ^user_id) end) will **not** be interrupted by Phoenix.Sync.interrupt( table: "todos", where: "user_id = '\#{user_id}'" ) because the where clause matching is a simple *exact string* match and `Ecto` query generated where clauses will generally be different from the equivalent hand-written version. If you want to interrupt a query-based shape you should use the same query as the interrupt criteria. > #### Writing interrupts {: .tip} > > It's better to be too broad with your interrupt calls than too narrow. > Only clients whose shape definition changes after the `interrupt/1` call > will be affected. ## Supported options The more options you give the more specific the interrupt call will be. Only the table name is required. - `table` - Required. Interrupts all shapes matching the given table. E.g. `"todos"` - `namespace` - The table namespace. E.g. `"public"` - `where` - The shape's where clause. Can in be parameterized and will match all shapes with the same where filter irrespective of the parameters (unless provided). E.g. `"status = $1"`, `"completed = true"` - `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_specification()) :: {:ok, non_neg_integer()} def interrupt(shape, shape_opts \\ []) do Phoenix.Sync.ShapeRequestRegistry.interrupt_matching(shape, shape_opts) end @doc """ Returns a shape definition for the given params. ## Examples - An `Ecto.Schema` module: Phoenix.Sync.shape!(MyPlugApp.Todos.Todo) - An `Ecto` query: Phoenix.Sync.shape!(from(t in Todos.Todo, where: t.owner_id == ^user_id)) - A `changeset/1` function which defines the table and columns: Phoenix.Sync.shape!(&Todos.Todo.changeset/1) - A `changeset/1` function plus a where clause: Phoenix.Sync.shape!( &Todos.Todo.changeset/1, where: "completed = false" ) or a parameterized where clause: Phoenix.Sync.shape!( &Todos.Todo.changeset/1, where: "completed = $1", params: [false] ) - A keyword list defining the shape parameters: Phoenix.Sync.shape!( table: "todos", namespace: "my_app", where: "completed = $1", params: [false] ) ## Transforms Using the `transform` option it's possible to modify the sync messages before they are sent to the clients via the [`sync`](`Phoenix.Sync.Router.sync/3`) router macro or [`sync_render`](`Phoenix.Sync.Controller.sync_render/4`) within your controllers. Phoenix.Sync.shape!( table: "todos", transform: &MyApp.Todos.transform/1 ) The transform function is passed the change messages in raw form and can transform the messages to a limited extent as required by the application. The `transform` process is effectively a `Stream.flat_map/2` operation over the sync messages, so if you want to use pattern matching to perform some kind of additional filtering operation to remove messages, e.g. based on some authorization logic, then you can simply return an empty list: # don't send delete messages to the client def transform(%{"headers" => %{"operation" => "delete"}}), do: [] def transform(message), do: [message] Removing messages from the stream can be useful for cases where you want to perform additional runtime filtering for authorization reasons that you're not able to do at the database level. Be aware that this can impact consistency of the client state so is an advanced feature that should be used with care. The messages passed to the transform function are of the form: %{ "key" => key, "headers" => %{"operation" => operation, ...}, "value" => %{"column_name" => column_value, ...} } - `key` is a unique identifier for the row formed of the namespaced table name plus the values of the row's primary key(s). **DO NOT MODIFY THIS VALUE**. - `headers` is a map of metadata about the change. The `operation` key will have one of the values `"insert"`, `"update"` or `"delete"`. You should leave this as-is unless you have a very good reason to modify it. - `value` is the actual row data. Unless the shape is defined with `replica: :full` only `insert` operations will contain the full row data. `update` operations will only contain the columns that have changed and `delete` operations will only contain the primary key columns. You can modify the values of the `value` map, as required by you application, but you should only modify values in a way that's compatible with the column's datatype. E.g. don't concat a integer column with a string (unless the resulting string will parse as an integer...). It is also unwise to modify the primary key values of any row unless you can be sure not to cause conflicts. Any column values you add that aren't in the backing Postgres table will be passed through to the client as-is. When using the raw [`stream/2`](`Phoenix.Sync.Client.stream/2`) function to receive a sync stream directly, the `transform` option is unnecessary and hence ignored. You should use the functions available in `Enum` and `Stream` to perform any data transforms. ### Transform via Ecto.Schema If you have custom field types in your `Ecto.Schema` module you can set up a transform that passes the raw data from the replication stream through the `Ecto` load machinery to ensure that the sync stream values match the values you would see when using `Ecto` to load data directly from the database. To do this pass the `Ecto.Schema` module as the transform function: Phoenix.Sync.shape!( MyApp.Todos.Todo, transform: MyApp.Todos.Todo ) or in a route: sync "todos", MyApp.Todos.Todo, transform: MyApp.Todos.Todo For this to work you need to implement `Jason.Encoder` for your schema module (or `JSON.Encoder` if you're on Elixir >= 1.18 but if [`Jason`](https://hex.pm/packages/jason) is available then it will be used), e.g.: defmodule MyApp.Todos.Todo do use Ecto.Schema @derive {Jason.Encoder, except: [:__meta__]} schema "todos" do field :title, :string field :completed, :boolean, default: false end end > #### Effect of `transform` on server load {: .warning} > > Normally `Phoenix.Sync` simply passes the raw encoded JSON message stream > from the backend server straight to the clients, which puts very little load > on the application server. > > The `transform` mechanism requires intercepting, decoding, mutating and > re-encoding every message from the backend server before they are sent to the > client. This could be costly for busy shapes or lots of connected clients. ### Limitations See the documentation of `Phoenix.Sync.Router.sync/3` for additional constraints on transform functions when defined within a route. ## Options When defining a shape via a keyword list, it supports the following options: #{NimbleOptions.docs(PredefinedShape.schema())} """ @spec shape!(shape_definition(), shape_specification()) :: PredefinedShape.t() def shape!(shape, shape_opts \\ []) do PredefinedShape.new!(shape, shape_opts) end @doc false def sandbox_enabled? do Code.ensure_loaded?(Electric) && Code.ensure_loaded?(Ecto.Adapters.SQL.Sandbox) end end