Forja

Event Bus with dual-path processing for Elixir – PubSub latency with Oban delivery guarantees.

Leia em Portugues

Why Forja?

Most event systems force a trade-off: fast but unreliable (PubSub) or reliable but slow (persistent queues). Forja gives you both.

Every event travels two paths simultaneously:

• Fast path – GenStage via PubSub delivers events in milliseconds
• Guaranteed path – Oban persists the event and processes it as a background job

Three-layer deduplication (PostgreSQL advisory locks + processed_at column + Oban unique jobs) ensures exactly-once processing regardless of which path wins.

App Code
  |
  emit/3
  |
  +-- INSERT event + Oban job (single transaction)
  |
  +-- PubSub broadcast ---------> GenStage pipeline (fast)
  |                                    |
  +-- Oban polls -------> ProcessEventWorker (guaranteed)
                                       |
                          advisory lock + processed_at check
                                       |
                               Handler.handle_event/2

Installation

Add forja to your dependencies in mix.exs:

def deps do
  [
    {:forja, "~> 0.1.0"}
  ]
end

With Igniter (recommended)

If you have Igniter installed, run:

mix igniter.install forja

This automatically generates the migration, adds Forja to your supervision tree, and configures Oban queues.

Manual setup

1. Generate the migration:

mix forja.install
mix ecto.migrate

1. Configure Oban queues in config/config.exs:

config :my_app, Oban,
  repo: MyApp.Repo,
  queues: [
    default: 10,
    forja_events: 5,
    forja_reconciliation: 1
  ],
  plugins: [
    {Oban.Plugins.Cron, crontab: [
      {"0 * * * *", Forja.Workers.ReconciliationWorker,
       args: %{forja_name: "my_app"}}
    ]}
  ]

1. Add Forja to your supervision tree:

children = [
  MyApp.Repo,
  {Phoenix.PubSub, name: MyApp.PubSub},
  {Oban, Application.fetch_env!(:my_app, Oban)},
  {Forja,
   name: :my_app,
   repo: MyApp.Repo,
   pubsub: MyApp.PubSub,
   handlers: [
     MyApp.Events.OrderNotifier,
     MyApp.Events.AnalyticsTracker
   ]}
]

Usage

Emitting events

# Simple emission
Forja.emit(:my_app, "order:created",
  payload: %{"order_id" => order.id, "total" => order.total},
  source: "orders"
)

# Idempotent emission (prevents duplicate processing)
Forja.emit(:my_app, "payment:received",
  payload: %{"payment_id" => payment.id},
  idempotency_key: "payment-#{payment.id}"
)

Transactional emission

Compose event emission with your domain operations in a single database transaction:

def create_order(attrs) do
  Ecto.Multi.new()
  |> Ecto.Multi.insert(:order, Order.changeset(%Order{}, attrs))
  |> Forja.emit_multi(:my_app, "order:created",
    payload_fn: fn %{order: order} ->
      %{"order_id" => order.id, "total" => order.total}
    end,
    source: "orders"
  )
  |> Repo.transaction()
  |> case do
    {:ok, %{order: order}} -> {:ok, order}
    {:error, :order, changeset, _} -> {:error, changeset}
  end
end

Writing handlers

defmodule MyApp.Events.OrderNotifier do
  @behaviour Forja.Handler

  @impl Forja.Handler
  def event_types, do: ["order:created", "order:shipped"]

  @impl Forja.Handler
  def handle_event(%Forja.Event{type: "order:created"} = event, _meta) do
    order = MyApp.Orders.get_order!(event.payload["order_id"])
    MyApp.Mailer.send_confirmation(order)
    :ok
  end

  def handle_event(%Forja.Event{type: "order:shipped"} = event, _meta) do
    order = MyApp.Orders.get_order!(event.payload["order_id"])
    MyApp.Mailer.send_shipping_notification(order)
    :ok
  end
end

Use :all to handle every event type:

defmodule MyApp.Events.AuditLogger do
  @behaviour Forja.Handler

  @impl Forja.Handler
  def event_types, do: :all

  @impl Forja.Handler
  def handle_event(event, _meta) do
    MyApp.AuditLog.record(event.type, event.payload)
    :ok
  end
end

Dead letter handling

When an event exhausts all processing attempts, Forja can notify you:

defmodule MyApp.Events.DeadLetterHandler do
  @behaviour Forja.DeadLetter

  @impl Forja.DeadLetter
  def handle_dead_letter(event, reason) do
    MyApp.Alerts.notify_ops("Dead letter event", %{
      event_id: event.id,
      type: event.type,
      reason: reason
    })
    :ok
  end
end

# Configure in supervision tree:
{Forja,
 name: :my_app,
 repo: MyApp.Repo,
 pubsub: MyApp.PubSub,
 handlers: [...],
 dead_letter: MyApp.Events.DeadLetterHandler}

Testing

Forja provides test helpers for verifying event emission:

defmodule MyApp.OrderTest do
  use MyApp.DataCase
  import Forja.Testing

  test "emitting order event" do
    {:ok, _order} = MyApp.Orders.create_order(%{total: 5000})

    assert_event_emitted(:my_app, "order:created", %{"total" => 5000})
  end

  test "no duplicate events" do
    MyApp.Orders.create_order(%{total: 5000})

    assert_event_deduplicated(:my_app, "order-create-ref-123")
  end

  test "process pending events synchronously" do
    Forja.emit(:my_app, "order:created", payload: %{"id" => 1})

    process_all_pending(:my_app)

    # All handlers have now executed
  end
end

Telemetry

Forja emits telemetry events for observability:

Configuration

Reconciliation defaults

reconciliation: [
  enabled: true,
  interval_minutes: 60,
  threshold_minutes: 15,
  max_retries: 3
]

Architecture

Forja runs as a supervisor with three children:

• ObanListener – Watches for discarded Oban jobs to trigger dead letter handling
• EventProducer – GenStage producer that receives PubSub broadcasts and buffers event IDs
• EventConsumer – ConsumerSupervisor that spawns a transient Task per event

The Processor is the shared functional core called by both paths. It acquires an advisory lock, loads the event, dispatches to handlers, and marks the event as processed.

Two Oban workers run outside the supervision tree:

• ProcessEventWorker – Queue :forja_events, the guaranteed delivery path
• ReconciliationWorker – Queue :forja_reconciliation, periodic sweep for stale events

License

MIT License. See LICENSE for details.