Provides write-path sync support for Phoenix- or Plug-based apps.

Imagine you're building an application on sync. You've used the read-path sync utilities to sync data into the front-end. If the client then changes the data locally, these writes can be batched up and sent back to the server.

Phoenix.Sync.Writer provides a principled way of ingesting these local writes and applying them to Postgres. In a way that works-with and re-uses your existing authorization logic and your existing Ecto.Schemas and Ecto.Changeset validation functions.

This allows you to build instant, offline-capable applications that work with local optimistic state.

Requirements

Phoenix.Sync.Writer requires Ecto.

It's perfectly possible to use Phoenix.Sync without any database connection (syncing via a HTTP) but the write-path requires both a database and Ecto.

Controller example

For example, take a project management app that's using @TanStack/db to batch up local optimistic writes and POST them to the Phoenix.Controller below:

defmodule MutationController do
  use Phoenix.Controller, formats: [:json]

  alias Phoenix.Sync.Writer
  alias Phoenix.Sync.Writer.Format

  def mutate(conn, %{"transaction" => transaction} = _params) do
    user_id = conn.assigns.user_id

    {:ok, txid, _changes} =
      Writer.new()
      |> Writer.allow(
        Projects.Project,
        check: reject_invalid_params/1,
        load: &Projects.load_for_user(&1, user_id),
        validate: &Projects.Project.changeset/2
      )
      |> Writer.allow(
        Projects.Issue,
        # Use the sensible defaults:
        # validate: Projects.Issue.changeset/2
        # etc.
      )
      |> Writer.apply(
        transaction,
        Repo,
        format: Format.TanstackDB
      )

    json(conn, %{txid: txid})
  end
end

The controller constructs a Phoenix.Sync.Writer instance and pipes it through a series of allow/3 calls, registering functions against Ecto.Schemas (in this case Projects.Project and Projects.Issue) to validate and authorize each of these mutation operations before applying them as a single transaction.

This controller can become a single, unified entry point for ingesting writes into your application. You can extend the pipeline with allow/3 calls for every schema that you'd like to be able to ingest changes to.

The check, load, validate, etc. callbacks to the allow function are designed to allow you to re-use your authorization and validation logic from your existing Plug middleware and Ecto.Changeset functions.

Usage levels (high, mid, low)

You don't need to use Phoenix.Sync.Writer to ingest write operations using Phoenix. Phoenix already ships with primitives like Ecto.Multi and Ecto.Repo.transaction/2. However, Phoenix.Sync.Writer provides:

• a number of convenience functions that simplify ingesting mutation operations
• a high-level pipeline that dries up a lot of common boilerplate and allows you to re-use your existing Plug and Ecto.Changeset logic

High-level usage

The controller example above uses a higher level pipeline that dries up common boilerplate, whilst still allowing flexibility and extensibility. You create an ingest pipeline by instantiating a Phoenix.Sync.Writer instance and piping into allow/3 and apply/4 calls:

{:ok, txid, _changes} =
  Phoenix.Sync.Writer.new()
  |> Phoenix.Sync.Writer.allow(MyApp.Todo)
  |> Phoenix.Sync.Writer.allow(MyApp.OtherSchema)
  |> Phoenix.Sync.Writer.apply(transaction, Repo, format: MyApp.MutationFormat)

Or, instead of apply/4 you can use separate calls to ingest/3 and then transaction/2. This allows you to ingest multiple formats, for example:

{:ok, txid} =
  Phoenix.Sync.Writer.new()
  |> Phoenix.Sync.Writer.allow(MyApp.Todo)
  |> Phoenix.Sync.Writer.ingest(changes, format: MyApp.MutationFormat)
  |> Phoenix.Sync.Writer.ingest(other_changes, parser: &MyApp.MutationFormat.parse_other/1)
  |> Phoenix.Sync.Writer.ingest(more_changes, parser: {MyApp.MutationFormat, :parse_more, []})
  |> Phoenix.Sync.Writer.transaction(MyApp.Repo)

And at any point you can drop down / eject out to the underlying Ecto.Multi using to_multi/1 or to_multi/3:

multi =
  Phoenix.Sync.Writer.new()
  |> Phoenix.Sync.Writer.allow(MyApp.Todo)
  |> Phoenix.Sync.Writer.to_multi(changes, format: MyApp.MutationFormat)

# ... do anything you like with the multi ...

{:ok, changes} = Repo.transaction(multi)
{:ok, txid} = Phoenix.Sync.Writer.txid(changes)

Mid-level usage

The pattern above uses a lower-level transact/4 function. This abstracts the mechanical details of transaction management whilst still allowing you to handle and apply mutation operations yourself:

{:ok, txid} =
  Phoenix.Sync.Writer.transact(
    my_encoded_txn,
    MyApp.Repo,
    fn
      %{operation: :insert, relation: [_, "todos"], change: change} ->
        MyApp.Repo.insert(...)
      %{operation: :update, relation: [_, "todos"], data: data, change: change} ->
        MyApp.Repo.update(Ecto.Changeset.cast(...))
      %{operation: :delete, relation: [_, "todos"], data: data} ->
        # we don't allow deletes...
        {:error, "invalid delete"}
    end,
    format: Phoenix.Sync.Writer.Format.TanstackDB,
    timeout: 60_000
  )

However, with larger applications, this flexibility can become tiresome as you end up repeating boilerplate and defining your own pipeline to authorize, validate and apply changes with the right error handling and return values.

Low-level usage (DIY)

For the more advanced cases, if you're comfortable parsing, validating and persisting changes yourself then the simplest way to use Phoenix.Sync.Writer is to use txid!/1 within Ecto.Repo.transaction/2:

{:ok, txid} =
  MyApp.Repo.transaction(fn ->
    # ... save your changes to the database ...

    # Return the transaction id.
    Phoenix.Sync.Writer.txid!(MyApp.Repo)
  end)

This returns the database transaction ID that the changes were applied within. This allows you to return it to the client, which can then monitor the read-path sync stream to detect when the transaction syncs through. At which point the client can discard its local optimistic state.

A convenient way of doing this is to parse the request data into a list of Phoenix.Sync.Writer.Operations using a Phoenix.Sync.Writer.Format. You can then apply the changes yourself by matching on the operation data:

{:ok, %Transaction{operations: operations}} =
  Phoenix.Sync.Writer.parse_transaction(
    my_encoded_txn,
    format: Phoenix.Sync.Writer.Format.TanstackDB
  )

{:ok, txid} =
  MyApp.Repo.transaction(fn ->
    Enum.each(txn.operations, fn
      %{operation: :insert, relation: [_, "todos"], change: change} ->
        # insert a Todo
      %{operation: :update, relation: [_, "todos"], data: data, change: change} ->
        # update a Todo
      %{operation: :delete, relation: [_, "todos"], data: data} ->
        # for example, if you don't want to allow deletes...
        raise "invalid delete"
    end)

    Phoenix.Sync.Writer.txid!(MyApp.Repo)
  end, timeout: 60_000)

Transactions

The txid in the return value from apply/4 and txid/1 / txid!/1 allows the Postgres transaction ID to be returned to the client in the response data.

This allows clients to monitor the read-path sync stream and match on the arrival of the same transaction id. When the client receives this transaction id back through its sync stream, it knows that it can discard the local optimistic state for that transaction. (This is a more robust way of managing optimistic state than just matching on instance IDs, as it allows for local changes to be rebased on concurrent changes to the same date from other users).

Phoenix.Sync.Writer uses Ecto.Multi's transaction update mechanism under the hood, which means that either all the operations in a client transaction are accepted or none are. See to_multi/1 for how you can hook into the Ecto.Multi after applying your change data.

Client Libraries

Phoenix.Sync.Writer is not coupled to any particular client-side implementation. See Electric's write pattern guides and example code for implementation strategies and examples.

Instead, Phoenix.Sync.Writer provides an adapter pattern where you can register a format adapter or parser function to parse the expected payload format from a client side library into the struct that Phoenix.Sync.Writer expects.

The currently supported format adapters are:

• TanStack/db "A reactive client store for building super fast apps on sync"

  Integration:

  Phoenix.Sync.Writer.new()
  |> Phoenix.Sync.Writer.ingest(mutation_data, format: Phoenix.Sync.Writer.Format.TanstackDB)
  |> Phoenix.Sync.Writer.transaction(Repo)

Usage

Much as every controller action must be authenticated, authorized and validated to prevent users writing invalid data or data that they do not have permission to modify, mutations MUST be validated for both correctness (are the given values valid?) and permissions (is the current user allowed to apply the given mutation?).

This dual verification -- of data and permissions -- is performed by a pipeline of application-defined callbacks for every model that you allow writes to:

• check - a function that performs a "pre-flight" sanity check of the user-provided data in the mutation; this should just validate the data and not usually hit the database; checks are performed on all operations in a transaction before proceeding to the next steps in the pipeline; this allows for fast rejection of invalid data before performing more expensive operations
• load - a function that takes the original data and returns the existing model from the database, if it exists, for an update or delete operation
• validate - create and validate an Ecto.Changeset from the source data and mutation changes; this is intended to be compatible with using existing schema changeset functions; note that, as per any changeset function, the validate function can perform both authorization and validation
• pre_apply and post_apply - add arbitrary Ecto.Multi operations to the transaction based on the current operation

See apply/4 and the Callbacks for how the transaction is processed internally and how best to use these callback functions to express your app's authorization and validation requirements.

Calling new/0 creates an empty writer configuration with the given mutation parser. But this alone does not permit any mutations. In order to allow writes from clients you must call allow/3 with a schema module and some callback functions.

# create an empty writer configuration
writer = Phoenix.Sync.Writer.new()

# allow writes to the `Todos.Todo` table
# using `Todos.check_mutation/1` to validate mutation data before
# touching the database
writer = Phoenix.Sync.Writer.allow(writer, Todos.Todo, check: &Todos.check_mutation/1)

If the table name on the client differs from the Postgres table, then you can add a table option that specifies the client table name that this allow/3 call applies to:

# `client_todos` is the name of the `todos` table on the clients
writer =
  Phoenix.Sync.Writer.allow(
    writer,
    Todos.Todo,
    validate: &Todos.validate_mutation/2,
    table: "client_todos"
  )

Callbacks

Check

The check option should be a 1-arity function whose purpose is to test the mutation data against the authorization rules for the application and model before attempting any database access.

If the changes are valid then it should return :ok or {:error, reason} if they're invalid.

If any of the changes fail the auth test, then the entire transaction will be rejected.

This is the first line of defence against malicious writes as it provides a quick check of the data from the clients before any reads or writes to the database.

Note that the writer pipeline checks all the operations before proceeding to load, validate and apply each operation in turn.

def check(%Phoenix.Sync.Writer.Operation{} = operation) do
  # :ok or {:error, "..."}
end

Load

The load callback takes the data in update or delete mutations (i.e.: the original data before changes), and uses it to retrieve the original Ecto.Struct model from the database.

It can be a 1- or 2-arity function. The 1-arity version receives just the data parameters. The 2-arity version receives the Ecto.Repo that the transaction is being applied to and the data parameters.

# 1-arity version
def load(%{"column" => "value"} = data) do
  # Repo.get(...)
end

# 2-arity version
def load(repo, %{"column" => "value"} = data) do
  # repo.get(...)
end

If not provided defaults to using Ecto.Repo.get_by/3 using the primary key(s) defined on the model.

For insert operations this load function is not used. Instead, the original struct is created by calling the __struct__/0 function on the Ecto.Schema module.

Validate

The validate callback performs the usual role of a changeset function: to validate the changes against the model's data constraints using the functions in Ecto.Changeset.

It should return an Ecto.Changeset instance (or possibly the original schema struct in the case of deletes). If any of the transaction's changeset's are marked as invalid, then the entire transaction is aborted.

If not specified, the validate function is defaulted to the schema model's standard changeset/2 function if available.

The callback can be either a 2- or 3-arity function.

The 2-arity version will receive the Ecto.Schema struct returned from the load function and the mutation changes. The 3-arity version will receive the loaded struct, the changes and the operation.

# 2-arity version
def changeset(%Todo{} = data, %{} = changes) do
  data
  |> Ecto.Changeset.cast(changes, [:title, :completed])
  |> Ecto.Changeset.validate_required(changes, [:title])
end

# 3-arity version
def changeset(%Todo{} = data, %{} = changes, operation)
    when operation in [:insert, :update, :delete] do
  # ...
end

Primary keys

Whether the params for insert operations contains a value for the new primary key is application specific. It's certainly not required if you have declared your Ecto.Schema model with a primary key set to autogenerate: true.

It's worth noting that if you are accepting primary key values as part of your insert changes, then you should use UUID primary keys for your models to prevent conflicts.

pre_apply and post_apply

These callbacks, run before or after the actual insert, update or delete operation allow for the addition of side effects to the transaction.

They are passed an empty Ecto.Multi struct and which is then merged into the writer's transaction.

They also allow for more validation/authorization steps as any operation within the callback that returns an "invalid" operation will abort the entire transaction.

def pre_or_post_apply(%Ecto.Multi{} = multi, %Ecto.Changeset{} = change, %Phoenix.Sync.Writer.Context{} = context) do
  multi
  # add some side-effects
  # |> Ecto.Multi.run(Phoenix.Sync.Writer.operation_name(context, :image), fn _changes ->
  #   with :ok <- File.write(image.name, image.contents) do
  #    {:ok, nil}
  #   end
  # end)
  #
  # validate the current transaction and abort using an {:error, value} tuple
  # |> Ecto.Multi.run(Phoenix.Sync.Writer.operation_name(context, :my_validation), fn _changes ->
  #   {:error, "reject entire transaction"}
  # end)
end

Note the use of operation_name/2 when adding operations. Every name in the final Ecto.Multi struct must be unique, operation_name/2 generates names that are guaranteed to be unique to the current operation and callback.

Per-operation callbacks

If you want to re-use an existing function on a per-operation basis, then in your write configuration you can define both top-level and per operation callbacks:

Phoenix.Sync.Writer.allow(
  Todos.Todo,
  load: &Todos.fetch_for_user(&1, user_id),
  check: &Todos.check_mutation(&1, user_id),
  validate: &Todos.Todo.changeset/2,
  update: [
    # for inserts and deletes, use &Todos.Todo.changeset/2 but for updates
    # use this function
    validate: &Todos.Todo.update_changeset/2,
    pre_apply: &Todos.pre_apply_update_todo/3
  ],
  insert: [
    # optional validate, pre_apply and post_apply
    # overrides for insert operations
  ],
  delete: [
    # optional validate, pre_apply and post_apply
    # overrides for delete operations
  ],
)

End-to-end usage

The combination of the check, load, validate, pre_apply and post_apply functions can be composed to provide strong guarantees of validity.

The aim is to provide an interface as similar to that used in controller functions as possible.

Here we show an example controller module that allows updating of Todos via a standard HTTP PUT update handler and also via HTTP POSTs to the mutation handler which applies optimistic writes via this module.

We use the load function to validate the ownership of the original Todo by looking up the data using both the id and the user_id. This makes it impossible for user a to update todos belonging to user b.

defmodule MyController do
  use Phoenix.Controller, formats: [:html, :json]

  alias Phoenix.Sync.Writer

  # The standard HTTP PUT update handler
  def update(conn, %{"todo" => todo_params}) do
    user_id = conn.assigns.user_id

    with {:ok, todo} <- fetch_for_user(params, user_id),
         {:ok, params} <- validate_params(todo, todo_params, user_id),
         {:ok, updated_todo} <- Todos.update(todo, params) do
      redirect(conn, to: ~p"/todos/#{updated_todo.id}")
    end
  end

  # The HTTP POST mutations handler which receives JSON data
  def mutations(conn, %{"transaction" => transaction} = _params) do
    user_id = conn.assigns.user_id

    {:ok, txid, _changes} =
      Writer.new()
      |> Writer.allow(
        Todos.Todo,
        check: &validate_mutation(&1, user_id),
        load: &fetch_for_user(&1, user_id),
      )
      |> Writer.apply(transaction, Repo, format: Writer.Format.TanstackDB)

    json(conn, %{txid: txid})
  end

  # Included here for completeness but in a real app would be a
  # public function in the Todos context.
  # Because we're validating the ownership of the Todo here we add an
  # extra layer of auth checks, preventing one user from modifying
  # the Todos of another.
  defp fetch_for_user(%{"id" => id}, user_id) do
    from(t in Todos.Todo, where: t.id == ^id  and t.user_id == ^user_id)
    |> Repo.one()
  end

  defp validate_mutation(%Writer.Operation{} = op, user_id) do
    with :ok <- validate_params(op.data, user_id) do
      validate_params(op.changes, user_id)
    end
  end

  defp validate_params(%{"user_id" => user_id}, user_id), do: :ok
  defp validate_params(%{} = _params, _user_id), do: {:error, "invalid user_id"}
end

Because Phoenix.Sync.Write leverages Ecto.Multi to do the work of applying changes and managing errors, you're also free to extend the actions that are performed with every transaction using pre_apply and post_apply callbacks configured per-table or per-table per-action (insert, update, delete). See allow/3 for more information on the configuration options for each table.

The result of to_multi/1 or to_multi/3 is an Ecto.Multi instance so you can also just append operations using the normal Ecto.Multi functions:

{:ok, txid, _changes} =
  Writer.new()
  |> Writer.allow(Todo, ...)
  |> Writer.to_multi(transaction, parser: &my_transaction_parser/1)
  |> Ecto.Multi.insert(:my_action, %Event{})
  |> Writer.transaction(Repo)