En57

DCB-compatible event store library in Ruby with support for PostgreSQL.

Usage

Set up the database schema

En57 owns its PostgreSQL schema and tracks the installed schema version in the database. Add the rake tasks to your application's Rakefile:

require "en57/tasks"

Then install or update the schema with DATABASE_URL:

DATABASE_URL=postgres://localhost:5432/en57 bundle exec rake en57:migrate

To inspect the current schema status without applying changes:

DATABASE_URL=postgres://localhost:5432/en57 bundle exec rake en57:status

Run en57:migrate before using the event store for the first time.

Connect with raw pg

Use EventStore.for_pg when En57 should own a pg connection.

event_store = En57::EventStore.for_pg("postgres://localhost:5432/en57")

Connect with Sequel

Use EventStore.for_sequel when your app already owns a Sequel database.

database = Sequel.connect("postgres://localhost:5432/en57")

event_store = En57::EventStore.for_sequel(database)

Connect with ActiveRecord

Use EventStore.for_active_record when your app uses ActiveRecord.

ActiveRecord::Base.establish_connection("postgres://localhost:5432/en57")

event_store = En57::EventStore.for_active_record

Append events unconditionally

event_store.append(
  [
    En57::Event.new(
      type: "OrderPlaced",
      data: { amount: 100 },
      tags: ["order_id:123", "customer:42"],
    ),
  ],
)

Read all events

events = event_store.read.each.to_a

Read events with positions

event, position = event_store.read.each_with_position.first

Read events filtered by tags

events = event_store.read.with_tag("order_id:123", "customer:42").each.to_a

Read events after a position

events = event_store.read.after(42).each.to_a

Read events filtered by merged scopes

orders = event_store.read.of_type("OrderPlaced").with_tag("order_id:123")
price_changes = event_store.read.of_type("PriceChanged")

events = (orders | price_changes).each.to_a

Conditional write (optimistic concurrency style)

Example: consume credits only once per account.

account_scope = event_store.read.with_tag("account:x")

begin
  event_store.append(
    [
      En57::Event.new(
        type: "CreditsUsed",
        data: { amount: 100 },
        tags: ["account:x"],
      ),
    ],
    fail_if: account_scope.of_type("CreditsUsed"),
  )
rescue En57::AppendConditionViolated
  # lost the race; another writer already consumed credits
end

To ignore events at or before a known position, scope the fail_if condition with after.

last_read_event_position = 42

event_store.append(
  [En57::Event.new(type: "CreditsUsed", tags: ["account:x"])],
  fail_if: event_store.read.of_type("CreditsUsed").after(last_read_event_position),
)

Conditional write for email uniqueness

Example: ensure no event exists with this email tag before writing.

email_tag = "email:alice@example.com"

begin
  event_store.append(
    [
      En57::Event.new(
        type: "UserRegistered",
        data: { name: "Alice" },
        tags: [email_tag],
      ),
    ],
    fail_if: event_store.read.with_tag(email_tag),
  )
rescue En57::AppendConditionViolated
  # email already used
end