How I design a CRUD module in Elixir#

Most resources need the same operations: get, list, count, create, update, delete. The pattern I use:

a behaviour that every CRUD module implements;
a set of macros with the default implementation of each operation;
one standalone CRUD module per resource, mixing default and custom functions;
application errors only, never a nil or a raw changeset;
an event for every operation that changes the data.

The behaviour#

defmodule MyApp.Crud do
  alias MyApp.Errors.Error

  @callback get(id :: term(), opts :: keyword()) ::
              {:ok, struct()} | {:error, Error.t()}

  @callback list(params :: map(), opts :: keyword()) ::
              {:ok, [struct()]} | {:error, Error.t()}

  @callback list_query(query :: struct()) :: Ecto.Query.t()

  @callback count(params :: map()) ::
              {:ok, non_neg_integer()} | {:error, Error.t()}

  @callback count_query(query :: struct()) :: Ecto.Query.t()

  @callback create(attrs :: map()) ::
              {:ok, struct()} | {:error, Error.t()}

  @callback update(entity_or_id :: struct() | term(), attrs :: map()) ::
              {:ok, struct()} | {:error, Error.t()}

  @callback delete(entity_or_id :: struct() | term()) ::
              {:ok, struct()} | {:error, Error.t()}

  @callback topic(query :: map()) :: String.t()

  @callback topics(entity :: struct()) :: [String.t()]

  @callback publish_event(result, action :: atom(), opts :: keyword()) :: result
            when result: {:ok, struct()} | {:error, Error.t()} | struct()
end

Every CRUD module has the same API, whatever its implementation.
update/2 and delete/1 take the entity or its id.
count_query/1 turns a validated query into an Ecto.Query with the filters, used by count/1. list_query/1 adds what only the list needs (e.g. the order), used by list/2.
topic/1, topics/1 and publish_event/3 are for the events.

The default implementation: one macro per operation#

use MyApp.Crud stores the options in a @crud_config map. Then each macro defines one function of the behaviour, as a call to a plain do_* function with the config:

defmodule MyApp.Crud do
  import Ecto.Query,
    only: [from: 1, where: 3, order_by: 3, select: 3, offset: 2, limit: 2]

  alias MyApp.Crud.Events
  alias MyApp.Errors
  alias MyApp.Errors.Error
  alias MyApp.Repo

  # ... the callbacks above

  defmacro __using__(opts) do
    schema = Keyword.fetch!(opts, :schema)

    quote do
      @behaviour MyApp.Crud

      @crud_config %{
        module: __MODULE__,
        schema: unquote(schema),
        query: unquote(Keyword.fetch!(opts, :query)),
        topic_keys: unquote(Keyword.get(opts, :topic_keys, [:id])),
        name: unquote(opts[:name]) || unquote(schema) |> Module.split() |> List.last()
      }
    end
  end

  defmacro get do
    location = location(__CALLER__, :get, 2)

    quote do
      @impl MyApp.Crud
      def get(id, opts \\ []) do
        config = Map.put(@crud_config, :location, unquote(location))
        MyApp.Crud.do_get(id, opts, config)
      end
    end
  end

  defmacro list do
    location = location(__CALLER__, :list, 2)

    quote do
      @impl MyApp.Crud
      def list(params, opts \\ []) do
        config = Map.put(@crud_config, :location, unquote(location))
        MyApp.Crud.do_list(params, opts, config)
      end
    end
  end

  defmacro list_query do
    quote do
      @impl MyApp.Crud
      def list_query(query), do: MyApp.Crud.do_list_query(query, @crud_config)
    end
  end

  defmacro count do
    location = location(__CALLER__, :count, 1)

    quote do
      @impl MyApp.Crud
      def count(params) do
        config = Map.put(@crud_config, :location, unquote(location))
        MyApp.Crud.do_count(params, config)
      end
    end
  end

  defmacro count_query do
    quote do
      @impl MyApp.Crud
      def count_query(query), do: MyApp.Crud.filter(@crud_config.schema, query)
    end
  end

  defmacro create do
    location = location(__CALLER__, :create, 1)

    quote do
      @impl MyApp.Crud
      def create(attrs) do
        config = Map.put(@crud_config, :location, unquote(location))
        MyApp.Crud.do_create(attrs, config)
      end
    end
  end

  defmacro update do
    location = location(__CALLER__, :update, 2)

    quote do
      @impl MyApp.Crud
      def update(entity_or_id, attrs) do
        config = Map.put(@crud_config, :location, unquote(location))
        MyApp.Crud.do_update(entity_or_id, attrs, config)
      end
    end
  end

  defmacro delete do
    location = location(__CALLER__, :delete, 1)

    quote do
      @impl MyApp.Crud
      def delete(entity_or_id) do
        config = Map.put(@crud_config, :location, unquote(location))
        MyApp.Crud.do_delete(entity_or_id, config)
      end
    end
  end

  defmacro events do
    quote do
      @impl MyApp.Crud
      def topic(query), do: Events.topic(@crud_config.name, query, @crud_config.topic_keys)

      @impl MyApp.Crud
      def topics(entity), do: Events.topics(@crud_config.name, entity, @crud_config.topic_keys)

      @impl MyApp.Crud
      def publish_event(result, action, opts \\ []) do
        Events.publish(result, action, __MODULE__, opts)
      end
    end
  end

  # The location of the generated function, as Macro.Env.stacktrace/1 would return it
  defp location(caller, function, arity) do
    file = caller.file |> Path.relative_to_cwd() |> String.to_charlist()
    Macro.escape({caller.module, function, arity, [file: file, line: caller.line]})
  end

  @doc false
  def do_get(id, opts, config) do
    case Repo.get(config.schema, id) do
      nil ->
        {:error,
         Error.new(:not_found, %{
           message: "%{resource} %{id} not found",
           details: %{resource: config.name, id: id},
           location: config.location
         })}

      entity ->
        {:ok, Repo.preload(entity, Keyword.get(opts, :preload, []))}
    end
  end

  @doc false
  def do_list(params, opts, config) do
    with {:ok, query} <- config.query.validate(params) do
      query
      |> config.module.list_query()
      |> offset(^query.offset)
      |> limit(^query.limit)
      |> Repo.all()
      |> Repo.preload(Keyword.get(opts, :preload, []))
      |> then(&{:ok, &1})
    end
  end

  @doc false
  def do_list_query(query, config) do
    query
    |> config.module.count_query()
    |> order_by([e], desc: e.inserted_at, desc: e.id)
  end

  @doc false
  def do_count(params, config) do
    with {:ok, query} <- config.query.validate(params) do
      query
      |> config.module.count_query()
      |> select([e], count(e.id))
      |> Repo.one()
      |> then(&{:ok, &1})
    end
  end

  @doc false
  def do_create(attrs, config) do
    config.schema
    |> struct()
    |> config.schema.changeset(attrs)
    |> Repo.insert()
    |> Errors.do_map_changeset(config.location)
    |> config.module.publish_event(:created)
  end

  @doc false
  def do_update(%schema{} = entity, attrs, %{schema: schema} = config) do
    entity
    |> schema.changeset(attrs)
    |> Repo.update()
    |> Errors.do_map_changeset(config.location)
    |> config.module.publish_event(:updated, previous: entity)
  end

  def do_update(id, attrs, config) do
    with {:ok, entity} <- config.module.get(id) do
      do_update(entity, attrs, config)
    end
  end

  @doc false
  def do_delete(%schema{} = entity, %{schema: schema} = config) do
    entity
    |> Repo.delete()
    |> Errors.do_map_changeset(config.location)
    |> config.module.publish_event(:deleted)
  end

  def do_delete(id, config) do
    with {:ok, entity} <- config.module.get(id) do
      do_delete(entity, config)
    end
  end

  # Equality filter on every field of the query struct that is not nil
  def filter(schema, query) do
    query
    |> Map.from_struct()
    |> Map.drop([:offset, :limit])
    |> Enum.reduce(from(e in schema), fn
      {_key, nil}, q -> q
      {key, value}, q -> where(q, [e], field(e, ^key) == ^value)
    end)
  end
end

The macros generate only a call: the logic lives once in MyApp.Crud, not copied into every resource module. Less generated code, and plain functions are easier to read, debug and test.
The config holds all the compile-time data: module, schema, query, name, topic keys and location.
The location is added in the function body, not in the macro: the macros expand before the module body runs, so @crud_config is not set yet when they expand. Map.put/3 on a small map at each call costs nothing that matters.
Each macro captures its own location: the line of MyApp.Crud.get() in the resource module, with the generated function (get/2). The errors built in do_* point there, not inside MyApp.Crud.
config.module calls the module’s own functions (get/1, count_query/1, list_query/1, publish_event/3), default or custom.
count/1 and list/2 get their Ecto.Query from count_query/1 and list_query/1: a resource with special filters overrides only those, and keeps the default count/1 and list/2.
The filters live once, in count_query/1: the default list_query/1 starts from it, so the list and the count always match. The default filter is the equality one, enough for most resources.
The order is only in list_query/1: a count doesn’t need it, and Postgres refuses an ORDER BY on a column in an aggregate query without a GROUP BY.
Neither query has pagination, list/2 adds it: the same query can also feed a Repo.delete_all/1 or a Repo.stream/1.
The do_* functions are public because the generated functions call them from another module; @doc false hides them from the docs.
The conventions they rely on: the schema has a changeset/2, the query module has a validate/1 that returns {:ok, %Query{}} or {:error, %Error{}}.

The query module#

list/2 takes the params as they come (e.g. from a controller) and validates them with an embedded schema, one per resource:

defmodule MyApp.Organizations.Crud.Query do
  use Ecto.Schema
  import Ecto.Changeset

  require MyApp.Errors

  alias MyApp.Errors

  @primary_key false
  embedded_schema do
    field :offset, :integer, default: 0
    field :limit, :integer, default: 25
    field :user_id, :binary_id
  end

  def validate(params) do
    %__MODULE__{}
    |> cast(params, [:offset, :limit, :user_id])
    |> validate_required([:user_id])
    |> apply_action(:validate)
    |> Errors.map_changeset()
  end
end

Only the fields in the schema can be filtered: unknown params are dropped by cast/3.
validate_required makes a filter mandatory: here you can never list the organizations of every user.
validate/1 returns an application error itself, so list/2 doesn’t need to map it.

Always return an application error#

The CRUD module never returns a nil, a raw changeset or a library error: always {:ok, result} or {:error, %Error{}} (see Errors: model them once, catch them once).

Repo.get/2 returns nil → get/2 maps it to a :not_found error.
A changeset error (invalid data, a failed constraint) → Errors.map_changeset/1 maps it to an :unprocessable_entity error, with the field errors in details.

The mapping is not specific to the CRUD (any code that writes a changeset needs it), so it lives in the Errors module, next to the new macro:

defmodule MyApp.Errors do
  alias MyApp.Errors.Error

  # ... the `new` macro

  # A macro, like `new`: the error location points to the caller, not here
  defmacro map_changeset(result) do
    [location] = Macro.Env.stacktrace(__CALLER__)

    quote do
      MyApp.Errors.do_map_changeset(unquote(result), unquote(Macro.escape(location)))
    end
  end

  # Public: called by the macro, and by code that already has a location (MyApp.Crud)
  def do_map_changeset(result, location) do
    case result do
      {:ok, value} ->
        {:ok, value}

      {:error, %Ecto.Changeset{} = changeset} ->
        {:error,
         Error.new(:unprocessable_entity, %{
           message: "Invalid data",
           details: %{errors: changeset_errors(changeset)},
           location: location
         })}
    end
  end

  def changeset_errors(%Ecto.Changeset{} = changeset) do
    Ecto.Changeset.traverse_errors(changeset, fn {message, opts} ->
      Regex.replace(~r"%{(\w+)}", message, fn _, key ->
        opts |> Keyword.get(String.to_existing_atom(key), key) |> to_string()
      end)
    end)
  end
end

Other code (custom CRUD functions, the query modules) calls the map_changeset macro; MyApp.Crud calls do_map_changeset/2 with the location from its config.

The location points to the resource module, at the line of the macro call, and to the generated function:

{:error,
 %MyApp.Errors.Error{
   reason: :unprocessable_entity,
   message: "Invalid data",
   details: %{errors: %{name: ["can't be blank"]}},
   location: {MyApp.Organizations.Crud, :create, 1, [file: ~c"lib/my_app/organizations/crud.ex", line: 9]}
 }}

Ref: Ecto.Changeset.traverse_errors/2

One standalone module per resource#

Each resource has its own CRUD module, e.g. MyApp.Organizations.Crud. It picks the default functions it needs and writes the others by hand.

All defaults:

defmodule MyApp.Organizations.Crud do
  use MyApp.Crud,
    schema: MyApp.Organizations.Organization,
    query: MyApp.Organizations.Crud.Query,
    topic_keys: [:id, :user_id]

  MyApp.Crud.get()
  MyApp.Crud.list()
  MyApp.Crud.list_query()
  MyApp.Crud.count()
  MyApp.Crud.count_query()
  MyApp.Crud.create()
  MyApp.Crud.update()
  MyApp.Crud.delete()
  MyApp.Crud.events()
end

A mix: files have a search on the name and a soft delete:

defmodule MyApp.Files.Crud do
  use MyApp.Crud,
    schema: MyApp.Files.File,
    query: MyApp.Files.Crud.Query,
    topic_keys: [:id, :organization_id, :parent_id]

  import Ecto.Query, only: [where: 3, order_by: 3]

  require MyApp.Errors

  alias MyApp.Crud
  alias MyApp.Errors
  alias MyApp.Files.File
  alias MyApp.Files.Crud.Query
  alias MyApp.Repo

  Crud.get()
  Crud.list()
  Crud.count()
  Crud.create()
  Crud.update()
  Crud.events()

  # Custom: the standard filter, plus a search on the name, without the deleted files
  @impl Crud
  def count_query(%Query{} = query) do
    File
    |> Crud.filter(%{query | search: nil})
    |> search(query.search)
    |> where([f], is_nil(f.deleted_at))
  end

  # Custom: the same filters, ordered by name
  @impl Crud
  def list_query(%Query{} = query) do
    query
    |> count_query()
    |> order_by([f], asc: f.name)
  end

  defp search(query, nil), do: query
  defp search(query, text), do: where(query, [f], ilike(f.name, ^"%#{text}%"))

  # Custom: soft delete
  @impl Crud
  def delete(%File{} = file) do
    file
    |> Ecto.Changeset.change(deleted_at: DateTime.utc_now(:second))
    |> Repo.update()
    |> Errors.map_changeset()
    |> publish_event(:deleted)
  end

  def delete(id) do
    with {:ok, file} <- get(id) do
      delete(file)
    end
  end
end

The custom functions reuse the building blocks (Crud.filter/2, Errors.map_changeset/1, publish_event/3), so they behave like the defaults: same errors, same events.
The behaviour checks the result: a missing operation is a compile warning.
A resource can add its own functions beyond the standard ones, e.g. upsert/2 or list_recent/1.

Permissions: outside the CRUD#

The CRUD is not responsible for the permissions: the caller (e.g. the context module) checks them before calling it.

The CRUD is plain data access: jobs, seeds, tests and other contexts use it without a user.
The permission rules live in one layer, instead of being spread across every operation.

Most of the time, checking the query the caller passes is enough. To list the files of an organization, the query is %{organization_id: org_id}: if the user has access to org_id, they can run the query. The caller can:

check the query: verify that the user has access to the organization in the params;
force the query: put in the params the organization it already checked, e.g. the one in the scope.

defmodule MyApp.Files do
  alias MyApp.Accounts.Scope
  alias MyApp.Files.Crud
  alias MyApp.Organizations

  # Check the query: the user must have access to the organization it asks for.
  # Organizations.authorize/2 returns :ok or {:error, %Error{}}
  def list_files(params, %Scope{} = scope) do
    with :ok <- Organizations.authorize(params["organization_id"], scope) do
      Crud.list(params)
    end
  end

  # Force the query: the organization comes from the scope, not from the params
  def list_organization_files(params, %Scope{organization: organization}) do
    Crud.list(Map.put(params, "organization_id", organization.id))
  end

  # One entity: check the organization it belongs to
  def get_file(id, %Scope{} = scope) do
    with {:ok, file} <- Crud.get(id),
         :ok <- Organizations.authorize(file.organization_id, scope) do
      {:ok, file}
    end
  end
end

The params come from the controller, with string keys: Map.put/3 uses a string key too, because cast/3 raises on a map with mixed keys.
validate_required in the query module is the safety net (as for user_id above): a list without organization_id is an error, never all the files.

The downside: nothing stops a controller from calling the CRUD directly and skipping the checks. The rule is a convention: the web layer calls only the context, never a CRUD module.

Events#

This part is a raw idea, I’m not sure it’s the right direction.

Events from the CRUD are useful in many places, for example:

update a LiveView live, when another user changes the data;
push the change to a client over a WebSocket (e.g. a Phoenix Channel);
send it to an external system, e.g. track it in PostHog.

Only for things where losing an event is acceptable: the events are async, with no delivery guarantee. If the system goes down between the publish and the moment the listener processes the event, the event is lost. For work that must happen, I don’t rely on them.

The event is generic, not only for the CRUD: a name and a payload.

defmodule MyApp.Events.Event do
  @enforce_keys [:name, :payload]
  defstruct [:name, :payload]
end

Every CRUD operation that changes the data publishes a standard one: entity_created, entity_updated or entity_deleted, with the entity in the payload.

Events.publish/4 takes the result of the operation, so it goes at the end of a pipe:

{:error, _} → nothing is published, the error goes on;
{:ok, entity} or entity → the event and its topics are built from the entity, and the result is returned as it is.

The topic keys live in the CRUD module, so publish/4 needs it. Each CRUD module wraps it in publish_event/3, so its code only says the action: |> publish_event(:deleted).

One topic per query#

The event is published on many topics: each topic is a query (or a sub-query) the entity belongs to. A listener subscribes to the topic of its query and receives only the events of the entities that match it, more or less.

The topic keys are the fields you can query on. A file with topic_keys: [:id, :organization_id, :parent_id] is published on every combination of each key with its value or * (any):

File/id:f1/organization_id:o1/parent_id:p1   # this file
File/id:*/organization_id:o1/parent_id:p1    # the files in folder p1
File/id:*/organization_id:o1/parent_id:*     # the files in organization o1
File/id:*/organization_id:*/parent_id:*      # all the files
...                                          # 8 topics in total

defmodule MyApp.Crud.Events do
  alias MyApp.Events.Event

  def publish(result, action, crud, opts \\ [])

  def publish({:error, _} = error, _action, _crud, _opts), do: error

  def publish({:ok, entity}, action, crud, opts) do
    publish(entity, action, crud, opts)
    {:ok, entity}
  end

  def publish(%_{} = entity, action, crud, opts) when action in [:created, :updated, :deleted] do
    event = %Event{name: "entity_#{action}", payload: %{entity: entity}}

    entity
    |> event_topics(crud, opts)
    |> Enum.each(&Phoenix.PubSub.broadcast(MyApp.PubSub, &1, event))

    entity
  end

  # With `previous:`, the old topics too: the entity may have left a query
  defp event_topics(entity, crud, opts) do
    case Keyword.get(opts, :previous) do
      nil -> crud.topics(entity)
      previous -> Enum.uniq(crud.topics(previous) ++ crud.topics(entity))
    end
  end

  def subscribe(topic), do: Phoenix.PubSub.subscribe(MyApp.PubSub, topic)

  # The topic of one query: a missing key matches any value
  def topic(name, query, keys) do
    keys
    |> Enum.sort()
    |> Enum.map(&segment(&1, Map.get(query, &1, :any)))
    |> then(&Enum.join([name | &1], "/"))
  end

  # All the topics of an entity: every combination of its value or "*" for each key
  def topics(name, entity, keys) do
    keys
    |> Enum.sort()
    |> Enum.reduce([[]], fn key, paths ->
      value = Map.fetch!(entity, key)

      for path <- paths,
          segment <- [segment(key, value), segment(key, :any)],
          do: path ++ [segment]
    end)
    |> Enum.map(&Enum.join([name | &1], "/"))
  end

  defp segment(key, :any), do: "#{key}:*"
  defp segment(key, nil), do: "#{key}:nil"
  defp segment(key, value), do: "#{key}:#{value}"
end

A LiveView that shows a folder subscribes to the topic of its query:

def mount(%{"organization_id" => organization_id, "parent_id" => parent_id}, _session, socket) do
  if connected?(socket) do
    # "File/id:*/organization_id:o1/parent_id:p1"
    Events.subscribe(Files.Crud.topic(%{organization_id: organization_id, parent_id: parent_id}))
  end

  # ...
end

def handle_info(%Event{name: "entity_created", payload: %{entity: file}}, socket) do
  {:noreply, stream_insert(socket, :files, file)}
end

Open questions#

Number of topics: n keys → 2^n topics per event. Fine for 2–4 keys, not for 10.
Only equality: a search or a date range can’t be a topic. The listener subscribes to a broader topic and filters the events itself.
Entities that leave a query: an update that moves a file to another folder is published on the old topics too (previous:), so the old folder hears about it. The listener must check whether the entity still matches its query, and remove it when not.
Transactions: the event is published right after the write. If the CRUD runs inside a bigger Repo.transaction that rolls back, the event announces a change that never happened.

Ref: Phoenix.PubSub

Wrapping up#

One behaviour for every CRUD module: the same API everywhere.
One macro per operation: pick the defaults, write the rest by hand with the same building blocks.
Always an application error: nil becomes :not_found, a changeset becomes :unprocessable_entity.
Permissions outside the CRUD: the caller checks or forces the query, the CRUD stays plain data access.
Events on query topics: a listener subscribes to its query and receives only the events it cares about (still an experiment).

Since every CRUD module has the same API, the behaviour is also a base to build on: a standard controller and routes to expose the APIs, with new macros on top of the CRUD ones, so that every API stays aligned and written the same way. That’s for a future post.