EventStore

To implement the competing consumers pattern and increase throughput, you can configure a subscription to support multiple concurrent subscribers.

Concurrency Options

• concurrency_limit: The maximum number of concurrent subscribers allowed. Defaults to 1. If exceeded, returns {:error, :too_many_subscribers}.
• buffer_size: Limits the number of in-flight events sent to the subscriber before an acknowledgement is required. Defaults to 1.
• partition_by: A function used to distribute events to subscribers. It receives an EventStore.RecordedEvent and returns a partition key.

Ordering Guarantee with Partitioning

When using multiple subscribers, global ordering is lost. To maintain ordering for specific groups (e.g., ensuring all events for a single stream are processed in order), use partition_by to return a key like stream_uuid. This ensures all events for a specific stream go to the same subscriber, while different streams are processed concurrently by different subscribers.

alias EventStore.RecordedEvent
alias MyApp.EventStore

# Partition by stream_uuid to guarantee per-stream ordering with 10 concurrent subscribers
by_stream = fn %RecordedEvent{stream_uuid: stream_uuid} -> stream_uuid end

{:ok, _subscription} =
  EventStore.subscribe_to_stream(stream_uuid, "example", self(),
    concurrency_limit: 10,
    partition_by: by_stream
  )

EventStore provides two distinct subscription models for consuming events:

1. Transient subscriptions: These broadcast new events to subscribers immediately after they are appended to storage. They do not require acknowledgement and terminate when the subscriber process stops. They are suitable for real-time notifications where missing an event during downtime is acceptable.
2. Persistent subscriptions: These guarantee at-least-once delivery of every persisted event. They provide back-pressure and allow the subscriber to start, pause, and resume from any position (including the stream's origin). They are suitable for critical processing where every event must be handled.

Under the hood, EventStore uses PostgreSQL's LISTEN and NOTIFY mechanism. An after-update trigger on the streams table executes a NOTIFY for each batch of inserted events. A single process connects to the database to listen for these notifications and broadcasts them to all active subscriptions. This architecture allows EventStore to run on multiple nodes without requiring a distributed Erlang cluster.

EventStore supports two types of deletion:

Soft Delete

Marks a stream as deleted without removing its events.

• Events still appear in the global $all stream and linked streams.
• The stream cannot be read or appended to directly.
• Subscriptions to the specific stream will stop receiving events, but subscriptions to linked streams (like $all) will still receive them.
• Default behavior: delete_stream/3 uses :soft if no type is specified.

Hard Delete

Permanently removes the stream and all its events. This is irreversible and removes data from the $all stream and linked streams.

• Disabled by default: You must explicitly enable hard deletes in your module definition or configuration.

Enabling Hard Deletes

# In module definition
defmodule MyApp.EventStore do
  use EventStore, otp_app: :my_app, enable_hard_deletes: true
end

# OR in config
config :my_app, MyApp.EventStore, enable_hard_deletes: true

# Soft delete a stream
:ok = MyApp.EventStore.delete_stream("stream1", :any_version, :soft)

# Hard delete a stream (requires enable_hard_deletes: true)
:ok = MyApp.EventStore.delete_stream("stream1", :any_version, :hard)

When using PostgreSQL with EventStore, the system maintains three distinct types of database connections. If you configure a pool_size of $N$, you should expect a total of $N + 2$ connections to the database.

1. Pooled Connection: Used for standard database operations like reading and appending events. This is configured via config/config.exs and uses an :exp (exponential) back-off strategy.
2. Notification Connection: A dedicated connection used to listen for event notifications via Postgres' LISTEN / NOTIFY mechanism.
3. Subscription Connection: A dedicated connection used for managing advisory locks. This ensures that only one instance of a subscription runs across multiple nodes.

Note that the Notification and Subscription connections use a :stop back-off strategy. This means the connection process terminates when the database connection is broken, allowing the EventStore.MonitoredServer to monitor the exit and trigger appropriate after_exit or after_restart callbacks to reacquire locks or notify related processes.

If you prefer using Poison for JSON serialization, you must implement the EventStore.Serializer behaviour. The implementation requires two functions: serialize/1 to convert a term to a JSON binary, and deserialize/2 to convert a binary back into a specific type. When deserializing, you can use the :type key in the config argument to specify the target struct.

defmodule JsonSerializer do
  @moduledoc """
  A serializer that uses the JSON format.
  """

  @behaviour EventStore.Serializer

  @doc """
  Serialize given term to JSON binary data.
  """
  def serialize(term) do
    Poison.encode!(term)
  end

  @doc """
  Deserialize given JSON binary data to the expected type.
  """
  def deserialize(binary, config) do
    type = case Keyword.get(config, :type, nil) do
      nil -> []
      type -> type |> String.to_existing_atom |> struct
    end
    Poison.decode!(binary, as: type)
  end
end

To use EventStore in your Elixir application, follow these steps:

1. Add dependency: Add {:eventstore, "~> 1.4"} to your mix.exs dependencies and run mix deps.get.
2. Define your module: Create an event store module using the EventStore macro.
3. Configure database: Add connection details (username, password, database, hostname, or a url) to your environment configuration.
4. Register the module: Add your module to the event_stores list in your application's main configuration so mix tasks can find it.
5. Initialize database: Run mix event_store.create and mix event_store.init to set up the tables.
6. Supervise the module: Add your event store module to your application's supervision tree.

# 1. mix.exs
def deps do
  [{:eventstore, "~> 1.4"}]
end

# 2. Define module
defmodule MyApp.EventStore do
  use EventStore, otp_app: :my_app

  def init(config) do
    {:ok, config}
  end
end

# 3. config/dev.exs
config :my_app, MyApp.EventStore,
  serializer: EventStore.JsonSerializer,
  username: "postgres",
  password: "postgres",
  database: "eventstore",
  hostname: "localhost"

# 4. config/config.exs
config :my_app, event_stores: [MyApp.EventStore]

# 5. Terminal
$ mix do event_store.create, event_store.init

# 6. lib/my_app/application.ex
defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [MyApp.EventStore]
    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end