The Outbox Pattern

defmodule Shop.MixProject do
  use Mix.Project

  def project do
    [app: :shop, version: "0.1.0", elixir: "~> 1.20", deps: deps()]
  end

  def application do
    [mod: {Shop.Application, []}]
  end

  defp deps do
    [{:ecto_sqlite3, "~> 0.26"}]
  end
end

import Config

config :shop, ecto_repos: [Shop.Repo]
config :shop, Shop.Repo, database: "shop.db", log: false

# Map columns are stored as JSON text. Elixir's built-in
# `JSON` module does the encoding, so no extra dependency.
config :ecto_sqlite3, json_library: JSON

defmodule Shop.Repo do
  use Ecto.Repo, otp_app: :shop, adapter: Ecto.Adapters.SQLite3
end

defmodule Shop.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [Shop.Repo, Shop.Relay]
    Supervisor.start_link(children, strategy: :one_for_one, name: Shop.Supervisor)
  end
end

defmodule Shop.Repo.Migrations.CreateOrders do
  use Ecto.Migration

  def change do
    create table(:orders) do
      add :item, :string, null: false
    end

    # Events waiting to be published. `published_at`
    # stays empty until the relay has sent the event.
    create table(:outbox) do
      add :topic, :string, null: false
      add :payload, :map, null: false
      add :published_at, :utc_datetime
    end
  end
end

defmodule Shop.Order do
  use Ecto.Schema

  schema "orders" do
    field :item, :string
  end
end

defmodule Shop.OutboxEvent do
  use Ecto.Schema

  schema "outbox" do
    field :topic, :string
    field :payload, :map
    field :published_at, :utc_datetime
  end
end

defmodule Shop.Broker do
  @moduledoc """
  A stand-in for a message broker (Kafka, RabbitMQ, SNS):
  another system, which keeps what it was sent even
  when the shop's node goes down. Here it's an
  append-only file.
  """

  @log "broker.log"

  def publish(id, topic, payload) do
    File.write!(@log, "#{id} #{topic} #{payload["item"]}\n", [:append])
  end

  def received do
    case File.read(@log) do
      {:ok, text} -> String.split(text, "\n", trim: true)
      {:error, :enoent} -> []
    end
  end
end

defmodule Shop do
  alias Shop.{Broker, Order, OutboxEvent, Repo}
  alias Ecto.Multi

  def orders, do: Order |> Repo.all() |> Enum.map(&{&1.id, &1.item})

  # The dual write: save the order, then tell the broker.
  # `crash_after_commit: true` stops the node between the
  # two, like a deploy or an out-of-memory kill would.
  def place_order_dual_write(item, opts \\ []) do
    {:ok, order} = Repo.insert(%Order{item: item})
    if opts[:crash_after_commit], do: crash("order #{order.id} committed, node stops")
    Broker.publish("dual-#{order.id}", "order_placed", %{"item" => item})
    {:ok, order.id}
  end

  # The outbox: the order and the event are saved in the
  # same transaction. Nothing is sent here. The relay
  # sends it later, from the table.
  def place_order(item, opts \\ []) do
    {:ok, %{order: order}} =
      Multi.new()
      |> Multi.insert(:order, %Order{item: item})
      |> Multi.insert(:event, %OutboxEvent{topic: "order_placed", payload: %{"item" => item}})
      |> Repo.transaction()

    if opts[:crash_after_commit], do: crash("order #{order.id} committed, node stops")
    {:ok, order.id}
  end

  # Stands in for a deploy or an out-of-memory kill: the
  # node stops on the spot, with exit status 1.
  def crash(message) do
    IO.puts(message)
    System.halt(1)
  end
end

defmodule Shop.Relay do
  @moduledoc """
  Publishes unsent outbox events, oldest first, and
  marks each one as published. It polls once a second;
  `flush/1` runs a pass right now.
  """
  use GenServer
  import Ecto.Query
  alias Shop.{Broker, OutboxEvent, Repo}

  @interval 1_000

  def start_link(_), do: GenServer.start_link(__MODULE__, nil, name: __MODULE__)
  def flush(opts \\ []), do: GenServer.call(__MODULE__, {:flush, opts})

  @impl true
  def init(nil) do
    Process.send_after(self(), :poll, @interval)
    {:ok, nil}
  end

  @impl true
  def handle_call({:flush, opts}, _from, state) do
    {:reply, publish_pending(opts), state}
  end

  @impl true
  def handle_info(:poll, state) do
    publish_pending([])
    Process.send_after(self(), :poll, @interval)
    {:noreply, state}
  end

  defp publish_pending(opts) do
    pending = Repo.all(from e in OutboxEvent, where: is_nil(e.published_at), order_by: e.id)

    for event <- pending do
      Broker.publish("outbox-#{event.id}", event.topic, event.payload)
      # Crash between sending and marking as sent.
      if opts[:crash_before_mark], do: Shop.crash("outbox-#{event.id} published, node stops")
      now = DateTime.utc_now(:second)
      Repo.update!(Ecto.Changeset.change(event, published_at: now))
    end

    length(pending)
  end
end