class CreateOutboxEvents < ActiveRecord::Migration[7.1]
def change
create_table :outbox_events do |t|
t.string :event_type, null: false
t.string :aggregate_type, null: false
t.string :aggregate_id, null: false
t.string :dedupe_key, null: false
t.jsonb :payload, null: false, default: {}
t.datetime :published_at
t.integer :attempts, null: false, default: 0
t.timestamps
end
add_index :outbox_events, :dedupe_key, unique: true
add_index :outbox_events, :id,
where: "published_at IS NULL",
name: "index_outbox_events_unpublished"
end
end
class OutboxEvent < ApplicationRecord
scope :unpublished, -> { where(published_at: nil).order(:id) }
def self.record!(event_type:, aggregate:, payload:, dedupe_key: nil)
key = dedupe_key || default_dedupe_key(event_type, aggregate)
insert_all(
[{
event_type: event_type,
aggregate_type: aggregate.class.name,
aggregate_id: aggregate.id.to_s,
dedupe_key: key,
payload: payload,
created_at: Time.current,
updated_at: Time.current
}],
unique_by: :dedupe_key
)
end
def envelope
{
id: dedupe_key,
type: event_type,
aggregate: { type: aggregate_type, id: aggregate_id },
data: payload,
occurred_at: created_at.iso8601
}
end
def mark_published!
update_columns(published_at: Time.current, updated_at: Time.current)
end
def self.default_dedupe_key(event_type, aggregate)
Digest::SHA256.hexdigest("#{event_type}:#{aggregate.class.name}:#{aggregate.id}")
end
end
class OrderService
def self.place(customer:, line_items:)
order = nil
ActiveRecord::Base.transaction do
order = Order.create!(
customer: customer,
line_items_attributes: line_items,
status: "placed"
)
OutboxEvent.record!(
event_type: "order.placed",
aggregate: order,
payload: {
order_id: order.id,
customer_id: customer.id,
total_cents: order.total_cents,
currency: order.currency
}
)
end
OutboxRelayJob.perform_async
order
end
end
class OutboxRelayJob
include Sidekiq::Job
sidekiq_options queue: :outbox, retry: 5
BATCH_SIZE = 100
def perform
processed = 0
ActiveRecord::Base.transaction do
events = OutboxEvent
.unpublished
.lock("FOR UPDATE SKIP LOCKED")
.limit(BATCH_SIZE)
.to_a
events.each do |event|
publish(event)
event.mark_published!
processed += 1
end
end
OutboxRelayJob.perform_async if processed == BATCH_SIZE
end
private
def publish(event)
EventBroker.publish(
topic: event.aggregate_type.underscore,
key: event.aggregate_id,
message: event.envelope
)
rescue EventBroker::PublishError => e
event.increment!(:attempts)
raise e
end
end
The transactional outbox pattern solves a classic distributed-systems problem: a service needs to update its database and publish an event to a broker (Kafka, SNS, RabbitMQ), but there is no shared transaction across the two systems. If the code writes to the DB and then publishes, a crash in between drops the event; if it publishes first, a rolled-back transaction leaves a phantom event. The outbox sidesteps this by writing the event into a table inside the same transaction as the business change, then relaying it asynchronously.
The create_outbox_events migration sets up that durable queue. Each row carries an aggregate_type/aggregate_id for ordering context, a JSONB payload, and a dedupe_key with a unique index so retries never enqueue the same logical event twice. The partial index on published_at IS NULL keeps the relay's polling query fast even as the table grows, since only unpublished rows are indexed.
In OutboxEvent model, record! is the enrolment point. It is called from within the caller's transaction so the event and the aggregate commit atomically. The dedupe_key defaults to a deterministic hash of the event type and aggregate, and insert_all with unique_by makes insertion idempotent under concurrent writers. unpublished orders by id to preserve rough causal order, and mark_published! stamps published_at so the relay skips it next cycle.
The OrderService shows the pattern in use: inside a single transaction block it persists the Order and calls OutboxEvent.record!. Because both writes share the transaction, either both land or neither does — there is no window where an order exists without its event, or vice versa.
The OutboxRelayJob is the poller. It claims a batch with FOR UPDATE SKIP LOCKED, which lets multiple relay workers run in parallel without processing the same row twice. Each event is handed to the broker and marked published individually, so a publish failure only affects one row and leaves the rest recoverable. The job re-enqueues itself when a full batch is drained, draining backlog aggressively. The key trade-off is at-least-once delivery: consumers must be idempotent, which is why the dedupe_key is propagated into the payload envelope.
Related snips
class CommentsController < ApplicationController
before_action :set_post
def create
@comment = @post.comments.build(comment_params)System test: asserting Turbo Stream responses
class Post < ApplicationRecord
belongs_to :author, class_name: 'User'
has_many :comments, dependent: :destroy
scope :published, -> { where.not(published_at: nil).where('published_at <= ?', Time.current) }
scope :draft, -> { where(published_at: nil) }ActiveRecord scopes for reusable query logic
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
class PostsController < ApplicationController
def index
@posts = Post.includes(:author)
.order(created_at: :desc)
.page(params[:page])
.per(10)Turbo Frames: infinite scroll with lazy-loading frame
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
Share this code
Here's the card — post it anywhere.
