walex (agoodway/walex) | MoltPulse

Back to Directory

walex

Postgres change events (CDC) in Elixir [WAL]

agoodway/walex00

MoltPulse

Based on repository activity, growth velocity and community engagement.

31

Growth4/30

Activity10/25

Popularity10/25

Trust7/20

362

Stars

High

Sentiment

Votes

362

README.md

WalEx

Simple and reliable Postgres Change Data Capture (CDC) in Elixir.

Walex mascot

WalEx allows you to listen to change events on your Postgres tables then perform callback-like actions with the data via the Event DSL. For example:

Stream database changes to an external service
Send a user a welcome email after they create a new account
Augment an existing Postgres-backed application with business logic
Send events to third party services (analytics, CRM, webhooks, etc))
Update index / invalidate cache whenever a record is changed

You can learn more about CDC and what you can do with it here: Why capture changes?

Credit

This library borrows liberally from realtime from Supabase, which in turn draws heavily on cainophile.

Alternatives

If you don't need to subscribe to WAL events directly in Elixir or are not an Elixir user, take a look at Sequin, which allows you to stream Postgres changes to a variety of destinations, including Kafka, S3, and more.

Installation

If available in Hex, the package can be installed by adding walex to your list of dependencies in mix.exs:

def deps do
  [
    {:walex, "~> 4.8.0"}
  ]
end

ALTER SYSTEM SET wal_level = 'logical';

command: [ "postgres", "-c", "wal_level=logical" ]

CREATE PUBLICATION events FOR ALL TABLES;

CREATE PUBLICATION events FOR TABLE user, todo;

CREATE PUBLICATION user_event FOR TABLE user WHERE (active IS TRUE);

ALTER TABLE user REPLICA IDENTITY FULL;
ALTER TABLE todo REPLICA IDENTITY FULL;

rds.logical_replication = 1
max_replication_slots = 5
max_slot_wal_keep_size = 2048

# config.exs

config :my_app, WalEx,
  hostname: "localhost",
  username: "postgres",
  password: "postgres",
  port: "5432",
  database: "postgres",
  publication: "events",
  subscriptions: ["user", "todo"],
  # WalEx assumes your module names match this pattern: MyApp.Events.User, MyApp.Events.ToDo, etc
  # but you can also specify custom modules like so:
  # modules: [MyApp.CustomModule.User, MyApp.OtherCustomModule.ToDo],
  name: MyApp

# config.exs

config :my_app, WalEx,
  url: "postgres://username:password@hostname:port/database"
  publication: "events",
  subscriptions: ["user", "todo"],
  name: MyApp

config :my_app, WalEx,
  url: "postgres://username:password@hostname:port/database",
  publication: "events",
  subscriptions: [:all_tables],
  name: MyApp,

WalEx.Configs.add_config(MyApp, :subscriptions, ["new_subscriptions_1", "new_subscriptions_2"])
WalEx.Configs.remove_config(MyApp, :subscriptions, "subscriptions")
WalEx.Configs.replace_config(MyApp, :password, "new_password")

defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      {WalEx.Supervisor, Application.get_env(:my_app, WalEx)}
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

[
  %Walex.Event{
    name: :user,
    type: :update,
    source: %WalEx.Event.Source{
      name: "WalEx",
      version: "3.8.0",
      db: "todos",
      schema: "public",
      table: "user",
      columns: %{
        id: "integer",
        name: "varchar",
        created_at: "timestamptz"
      }
    },
    new_record: %{
      id: 1234,
      name: "Chase Pursley",
      created_at: #DateTime<2023-08-18 14:09:05.988369-04:00 -04 Etc/UTC-4>
    },
    # we don't show old_record for update to reduce payload size
    # however, you can see any old values that changed under "changes"
    old_record: nil,
    changes: %{
      name: %{
        new_value: "Chase Pursley",
        old_value: "Chase"
      }
    },
    timestamp: ~U[2023-12-18 15:50:08.329504Z]
  }
]

defmodule MyApp.Events.User do
  use WalEx.Event, name: MyApp

  # any subscribed event
  on_event(:all, fn events ->
    IO.inspect(events: events)
  end)

  # any user event
  on_event(:user, fn users ->
    IO.inspect(on_event: users)
    # do something with users data
  end)

  # any user insert event
  on_insert(:user, fn users ->
    IO.inspect(on_insert: users)
  end)

  on_update(:user, fn users ->
    IO.inspect(on_update: users)
  end)

  on_delete(:user, fn users ->
    IO.inspect(on_delete: users)
  end)
end

defmodule MyApp.Events.User do
  use WalEx.Event, name: MyApp

  @filters %{
    unwatched_records: %{event_subscribe: false},
    unwatched_fields: ~w(event_id updated_at todos_count)a
  }

  on_insert(:user, @filters, fn users ->
    IO.inspect(on_insert: users)
    # resulting users data is filtered
  end)
end

defmodule MyApp.Events.User do
  use WalEx.Event, name: MyApp

  @filters %{unwatched_records: %{event_subscribe: false}}
  @functions ~w(send_welcome_email add_to_crm clear_cache)a

  on_insert(:user, @filters, @functions, fn users ->
    IO.inspect(on_insert: users)
    # resulting users data is first filtered then functions are applied
  end)

  def send_welcome_email(user) do
    # logic for sending welcome email to new user
  end

  def add_to_crm(user) do
  # logic for adding user to crm system
  end

  def clear_cache(user) do
  # logic for clearing user cache
  end
end

# config.exs

config :my_app, WalEx,
  # ...
  durable_slot: true,
  slot_name: my_app_replication_slot

config :my_app, WalEx,
  # ...
  message_middleware: fn message, app_name -> ... end

MIX_ENV=test mix walex.setup
MIX_ENV=test mix test