# `Outboxer`
[🔗](https://github.com/epinault/outboxer/blob/v0.1.0/lib/outboxer.ex#L1)

A transactional outbox for Ecto.

You write a message to the outbox table **in the same database transaction**
as your business data. Outboxer then relays it to Kafka, SQS, Oban or your
own sink, in order, retrying until it gets through. The message exists if and
only if the transaction committed, which is the guarantee that "write to the
database, then publish to the broker" cannot give you.

## Getting started

1. Create the table (see `Outboxer.Migration`):

       def up, do: Outboxer.Migration.up()
       def down, do: Outboxer.Migration.down()

2. Define a message (see `Outboxer.Message`):

       defmodule MyApp.Events.OrderPlaced do
         use Outboxer.Message, outbox: :orders
         defstruct [:order_id, :total]

         @impl true
         def group_key(%__MODULE__{order_id: id}), do: "order:#{id}"
       end

3. Add Outboxer to your supervision tree, after the repo. Outboxer has no
   application of its own; this child is the whole runtime:

       children = [
         MyApp.Repo,
         {Outboxer,
          repo: MyApp.Repo,
          outboxes: [
            orders: [relay: {Outboxer.Relay.Kafka, client: :my_brod_client, topic: "orders"}]
          ]}
       ]

4. Publish inside your transaction:

       MyApp.Repo.transaction(fn ->
         order = MyApp.Repo.insert!(changeset)
         {:ok, _envelope} = Outboxer.publish(%OrderPlaced{order_id: order.id, total: order.total})
         order
       end)

## Options

* `:outboxes` (`t:keyword/0`) - The outboxes, as `name: options`. See "Outbox options" below. The default value is `[]`.

* `:name` (`t:atom/0`) - Name of this instance. Only needed when you run more than one. The default value is `Outboxer`.

* `:repo` (`t:atom/0`) - Required. The `Ecto.Repo` that holds the outbox table. It must be the repo your business writes use.

* `:table` (`t:String.t/0`) - Name of the table used by every outbox that does not set its own `:table`. Must match what was given to `Outboxer.Migration`. The default value is `"outboxer_messages"`.

* `:prefix` - Postgres schema of the outbox tables. The default value is `nil`.

* `:storage` (`t:atom/0`) - An `Outboxer.Storage` module. Detected from the repo's adapter when omitted.

* `:notify` (`t:boolean/0`) - Postgres only. Wake workers through `LISTEN/NOTIFY` when a publishing transaction commits, instead of waiting for the next poll. Uses one extra database connection. The default value is `false`.

* `:relay` (`t:boolean/0`) - Whether this node delivers messages. `false` makes the instance publish-only: nothing is started, no relay is initialised (so the outboxes need no `:relay` and the node needs no broker dependency), and `publish/2` and the admin functions still work. Use it on web nodes when a separate node, started with the same outboxes and `relay: true`, does the delivering. The default value is `true`.

* `:testing` - For your test environment. `:manual` starts no workers: messages are stored but only delivered when you call `Outboxer.Testing.drain/2`. The default value is `:disabled`.

* `:require_transaction` (`t:boolean/0`) - Raise when `Outboxer.publish/2` is called outside a repo transaction. The default value is `false`.

* `:prune_interval` (`t:pos_integer/0`) - Milliseconds between pruning passes, for outboxes using `retention: {:keep, seconds}`. The default value is `60000`.

## Outbox options

Each entry under `:outboxes` accepts:

* `:relay` - The relay that delivers this outbox's messages: a module implementing `Outboxer.Relay`, or `{module, opts}`. Required, except on an instance started with `relay: false`, where it is ignored.

* `:table` (`t:String.t/0`) - Table holding this outbox's messages, for example `"order_outbox"`. Defaults to the instance's `:table`. Create it with `Outboxer.Migration.up(table: "order_outbox")`.

* `:ordering` - `:group` keeps order between messages that share a group key (messages without one are unordered). `:total` keeps one global order for the whole outbox (forces a single partition; one failing message blocks everything behind it). `:none` ignores group keys entirely. The default value is `:group`.

* `:partitions` (`t:pos_integer/0`) - How many partitions, and so how many concurrent delivery workers, the outbox has. A group always lands in the same partition. Do not change this while the outbox holds undelivered messages. The default value is `1`.

* `:batch_size` (`t:pos_integer/0`) - Maximum number of messages fetched and handed to the relay at once. The default value is `100`.

* `:poll_interval` (`t:pos_integer/0`) - Milliseconds an idle worker waits before looking for new messages. The default value is `1000`.

* `:max_attempts` - `:infinity` retries a failing message forever, which blocks its group and never breaks order. A number moves the message to the dead state after that many failed attempts and lets the group continue. The default value is `:infinity`.

* `:backoff` - Delay before a failed message is retried. Either `[min: ms, max: ms]` for exponential backoff with jitter, or a function taking the number of failed attempts and returning milliseconds. The default value is `[min: 1000, max: 60000]`.

* `:retention` - `:delete` removes a message as soon as it is delivered (smallest table, recommended). `{:keep, seconds}` marks it delivered and prunes it after that many seconds. The default value is `:delete`.

* `:serialize_writes` (`t:boolean/0`) - Postgres only. Take a transaction-scoped advisory lock on the group key when publishing, so that concurrent transactions writing to the same group commit in id order. Only needed when nothing else (such as a row lock on the aggregate) already serialises those writers. The default value is `false`.

## What gets started

Only what the configuration needs:

    Outboxer (Supervisor)
    ├── Registry                       worker lookup
    ├── Outboxer.Notifier              only with notify: true
    ├── one supervisor per outbox
    │   ├── the relay's own children   e.g. a brod client, if the relay starts one
    │   └── one worker per partition
    └── Outboxer.Pruner                only if an outbox uses retention: {:keep, _}

With `relay: false` or `testing: :manual` none of these are started and
`publish/2` still works.

## Publishing and relaying on different nodes

Start the same outboxes everywhere and switch delivery off where it is not
wanted, for example on web pods:

    {Outboxer,
     repo: MyApp.Repo,
     relay: System.get_env("OUTBOXER_RELAY", "true") == "true",
     outboxes: [...]}

A `relay: false` instance only publishes. It ignores the `:relay` of its
outboxes, so it needs neither broker settings nor the broker's dependency.
See the operations guide for what the two sides must agree on.

## One table per outbox

Every outbox can live in its own table:

    outboxes: [
      orders: [table: "order_outbox", relay: ...],
      invoices: [table: "invoice_outbox", relay: ...]
    ]

Create each one with `Outboxer.Migration.up(table: "order_outbox")`. Outboxes
without `:table` share the instance's table. Message ids are unique within a
table, not across tables.

## Several instances

All functions take a `:name` option, defaulting to `Outboxer`. Run a second
instance (for another repo, say) by giving it a name:

    {Outboxer, name: MyApp.BillingOutbox, repo: MyApp.BillingRepo, outboxes: [...]}

    Outboxer.publish(message, name: MyApp.BillingOutbox)

## Telemetry

| Event | Measurements | Metadata |
| --- | --- | --- |
| `[:outboxer, :publish]` | `count` | `name`, `outboxes` |
| `[:outboxer, :relay, :batch, :start]` | `system_time` | `name`, `outbox`, `partition`, `relay`, `count` |
| `[:outboxer, :relay, :batch, :stop]` | `duration` | the above plus `fetched`, `delivered`, `failed`, `dead` |
| `[:outboxer, :relay, :batch, :exception]` | `duration` | `name`, `outbox`, `partition`, `relay`, `kind`, `reason`, `stacktrace` |
| `[:outboxer, :message, :retry]` | `attempt` | `name`, `outbox`, `id`, `type`, `group_key`, `error` |
| `[:outboxer, :message, :dead]` | `attempt` | `name`, `outbox`, `id`, `type`, `group_key`, `error` |
| `[:outboxer, :prune]` | `count` | `name`, `outbox` |

# `message`

```elixir
@type message() :: struct()
```

# `name`

```elixir
@type name() :: atom()
```

# `counts`

```elixir
@spec counts(atom(), keyword()) :: %{
  pending: non_neg_integer(),
  delivered: non_neg_integer(),
  dead: non_neg_integer()
}
```

Number of messages of an outbox by state: `%{pending: n, delivered: n, dead: n}`.

`pending` is the backlog. `delivered` is only ever non-zero with
`retention: {:keep, seconds}`.

## Options

  * `:name` - the Outboxer instance, default `Outboxer`.

# `discard`

```elixir
@spec discard(atom(), [pos_integer()], keyword()) :: non_neg_integer()
```

Permanently deletes messages of an outbox by id, whatever their state.
Returns how many were deleted.

This is how you skip a poison message that blocks a group under
`max_attempts: :infinity`: find it with `list_pending/2` and discard it.

    Outboxer.discard(:orders, [4711])

## Options

  * `:name` - the Outboxer instance, default `Outboxer`.

# `list_dead`

```elixir
@spec list_dead(atom(), keyword()) :: [Outboxer.Envelope.t()]
```

Lists dead messages of an outbox, oldest first, as envelopes.

A message is dead when it failed `max_attempts` times. `envelope.last_error`
holds the reason for the final failure.

## Options

  * `:name` - the Outboxer instance, default `Outboxer`.
  * `:limit` - maximum number of messages, default `100`.

# `list_pending`

```elixir
@spec list_pending(atom(), keyword()) :: [Outboxer.Envelope.t()]
```

Lists pending (not yet delivered) messages of an outbox, oldest first.

Messages with `attempts > 0` have failed and are waiting for a retry; with
ordering on, they are what is blocking their group.

Takes the same options as `list_dead/2`.

# `publish`

```elixir
@spec publish(message() | [message()], keyword()) ::
  {:ok, Outboxer.Envelope.t() | [Outboxer.Envelope.t()]}
```

Writes one message, or a list of messages, to the outbox.

Call it inside `Repo.transaction/2` together with your business writes. The
insert uses the repo's current connection, so it commits or rolls back with
everything else in that transaction.

Returns `{:ok, envelope}` for a single message and `{:ok, envelopes}` for a
list. Raises `ArgumentError` for something that is not a message or names an
unknown outbox; database errors raise as they do for `Repo.insert_all/3`.

## Options

  * `:name` - the Outboxer instance, default `Outboxer`.
  * `:outbox` - overrides the message's `c:Outboxer.Message.outbox/0`.
  * `:group_key` - overrides the message's `c:Outboxer.Message.group_key/1`.
  * `:headers` - merged over the message's `c:Outboxer.Message.headers/1`.

## Outside a transaction

Publishing without a surrounding transaction is allowed and stores the
message atomically on its own. That is fine for a message that is not tied
to another write. Set `require_transaction: true` on the instance if you
would rather have that raise.

# `publish_multi`

```elixir
@spec publish_multi(
  Ecto.Multi.t(),
  Ecto.Multi.name(),
  message() | [message()] | (map() -> term()),
  keyword()
) :: Ecto.Multi.t()
```

Adds a publish step to an `Ecto.Multi`.

`message` is a message, a list of messages, or a function that receives the
changes so far and returns either.

    Ecto.Multi.new()
    |> Ecto.Multi.insert(:order, changeset)
    |> Outboxer.publish_multi(:event, fn %{order: order} ->
      %OrderPlaced{order_id: order.id, total: order.total}
    end)
    |> MyApp.Repo.transaction()

Takes the same options as `publish/2`.

# `retry_dead`

```elixir
@spec retry_dead(atom(), keyword()) :: non_neg_integer()
```

Puts dead messages back in the outbox for delivery, with their attempt
counter reset. Returns how many were revived.

They keep their original id, so they rejoin their group in their original
position. Messages of the same group published later may already have been
delivered; that gap is what `max_attempts` trades for not blocking.

## Options

  * `:name` - the Outboxer instance, default `Outboxer`.
  * `:ids` - only these message ids. Default: every dead message of the outbox.

# `start_link`

```elixir
@spec start_link(keyword()) :: Supervisor.on_start()
```

Starts an Outboxer instance. See the module documentation for the options.

Raises `ArgumentError` or `NimbleOptions.ValidationError` when the options
are invalid, so that a misconfigured outbox fails at boot rather than at the
first message.

---

*Consult [api-reference.md](api-reference.md) for complete listing*