Magic EventBus

Crystal shard to capture Postgres database change events via LISTEN/NOTIFY mechanism and publish them to EventBus::EventHandlers for further processing. Shard comes with retry and watchdog functionality, and tries to reconnect automatically until it has exhausted all of the configured retry_attempts. To configure Retry Mechanics and WatchDog follow below configuration options.

Retry Configuration

• retry_attempts: Number of attempts to try before giving up. Use 0 or less for infinite attempts. Default 0
• retry_interval: Interval in seconds to wait for next attempt to re-connect. Default to 5 seconds.

Set on_error call back if you want to receive final give-up message, along with last exception received on re-connection attempt. If no on_error callback is configured, it will raise the last exception

WatchDog Configurations

Shard monitors the database connectivity in a separate connection and watch for disconnection/freeze like situations and follow the same semantics of retrying after configured interval. To configure watchdog heartbeat interval and connection time_out, configure EventBus with below configurations.

• watchdog_interval : Heart beat interval in seconds. Default to 5 seconds
• timeout: Interval in seconds to wait for network connectivity. Default to 5 seconds.

WatchDog will trigger at every watchdog_interval and wait for connection status for timeout seconds before timing out.

CDC Trigger Management

Requires PostgreSQL 14+ (CREATE OR REPLACE TRIGGER).

ensure_cdc_for / ensure_cdc_for_all_tables are idempotent: a catalog pre-check (which takes no lock on the table) skips all DDL when the trigger is already installed, so steady-state service boots acquire zero table locks. Installing or replacing triggers uses CREATE OR REPLACE TRIGGER, protected by a lock_timeout with jittered, bounded retries. Resetting a filtered policy also drops its UPDATE trigger, which can briefly require an exclusive table lock; the same timeout and retries apply.

• Bare table names resolve to the public schema. Pass "schema.table" for other schemas.
• disable_cdc_for(table) is a no-op by default: the trigger is shared infrastructure that other services rely on, and dropping it takes an ACCESS EXCLUSIVE lock that queues every read on the table behind it. Pass force: true to genuinely uninstall. It never raises.

DDL safety configuration (constructor option / environment variable / default):

• lock_timeout / EVENTBUS_LOCK_TIMEOUT: Postgres lock_timeout applied to CDC DDL. Default 2s
• ddl_attempts / EVENTBUS_DDL_ATTEMPTS: attempts before giving up when the table is lock-contended. Default 5
• ddl_backoff_ms / EVENTBUS_DDL_BACKOFF_MS: base backoff between attempts (exponential + jitter). Default 100

Event retention (environment variables, applied when the schema is installed):

• EVENTBUS_RETENTION: how long rows are kept in eventbus_cdc_events. Default 1 day
• EVENTBUS_CLEANUP_PROBABILITY: chance an insert triggers the retention cleanup (amortises the DELETE instead of running it on every change). Default 0.01

Ignore telemetry-only updates per table

Declare ignored columns when registering a table (an ORM can supply this from model metadata):

eventbus.ensure_cdc_for("displays", ignore_update_columns: ["last_seen", "current_item_id"])

An UPDATE changing only those columns produces no CDC event. The database still persists the update and enforces its constraints. An UPDATE changing any other column produces the normal event, including its full row and all changed fields. INSERT and DELETE continue to notify. On a filtered table, an UPDATE changing no values is also silent; unconfigured tables retain their existing behavior.

The filter runs in the SQL UPDATE trigger's WHEN condition, before EventBus computes the change payload, writes its event row or sends NOTIFY. No application-side filtering or per-update configuration lookup is required.

Configuration belongs to the table, so all subscribers share it. Omitting ignore_update_columns preserves the installed policy, including during ensure_cdc_for_all_tables. Repeating the same declaration is idempotent; conflicting declarations raise an error instead of silently replacing another service's policy. Ignored columns must exist in the table and cannot include the row identity column id. An explicit empty list on an unconfigured table leaves it unconfigured; it does not reserve an unfiltered policy against future declarations.

For an intentional policy change, supply the expected current policy. This prevents a deployment from overwriting a policy it did not expect:

eventbus.replace_cdc_update_policy(
  "displays",
  ignore_update_columns: ["last_seen"],
  expected_ignore_update_columns: ["last_seen", "current_item_id"]
)

# Restore ordinary UPDATE events, including no-op updates.
eventbus.replace_cdc_update_policy(
  "displays",
  ignore_update_columns: [] of String,
  expected_ignore_update_columns: ["last_seen"]
)

Policy metadata is stored in comments on both managed triggers and is preserved by ordinary registration. Do not replace these comments manually; they let EventBus repair a missing or altered trigger without losing the policy. Forced uninstall removes both triggers and their metadata.

Install the policy before starting heartbeat writers if the first write must be silent. Upgrade every service that installs EventBus triggers before enabling filtering: older installers can restore the legacy combined trigger while leaving the new UPDATE trigger, causing both unwanted and duplicate events. Suppression applies to every writer of the ignored fields, and consumers relying on those telemetry events must read the persisted values directly or use another notification path.

EventBus::EventHandler Lifecycle methods

Below lifecycle methods are invoked for all registered handlers

• on_start - invoked when EventBus is going to start. Override this method
• on_connect - invoked when EventBus PG listener get connected to Postgres.
• on_event - invoked when an event is received. Refer to EventBus::Event struct for structure
• on_close - invoked when EventBus is going to shutdown

EventBus::Event structure

• timestamp : Time - PG Timestamp when event occurred
• schema : String - Schema name of PG
• table : String - PG Table name where event occurred
• action : EventBus::Action - contains one of INSERT | UPDATE | DELETE based on event
• id : JSON::Any - PG Table column id value
• data : String - JSON object in String representing table row data
• changes : String? - JSON object in String, contains array of hash with column name, old and new value. This field is set for UPDATE events only.

Installation

1. Add the dependency to your shard.yml:

   dependencies:
     eventbus:
       github: spider-gazelle/eventbus
   

2. Run shards install

Usage

require "eventbus"

# Instantiate EventBus object with Postgres URI
eventbus =  EventBus.new(PG_DATABASE_URL, retry_attempts: 5, retry_interval: 5)

# Register Custom EventHandlers which will receive events

eventbus.add_handler MyLogger.new, MyRedisPub.new

# Register Error handler which will get invoked on fatal error
eventbus.on_error ->(ex : EventBus::ErrHandlerType) {
  puts " Received Fatal error from EventBus\n"
  puts ex
  puts "\n terminating gracefully"
  eventbus.close rescue nil
}

# Enable CDC mechanism on all or particular table

 eventbus.ensure_cdc_for_all_tables
 # OR
 eventbus.ensure_cdc_for("MyTable")

 # Start Event Bus

 eventbus.start # for async (non-blocking mode)
 # OR
 eventbus.run # for sync (blocking mode)

 # Once done

 eventbus.close

Examples

Located under example folder.

Application (application.cr)

Complete demo application which make use of handlers under example folder and publishes PG events to Redis Cluster. To run demo application ensure you set below environment variables for it to work.

# Database config:
PG_DATABASE_URL=postgresql://user:password@hostname/database
REDIS_URL=redis://user:password@redis:port/database

Client Subscriber (subscriber.cr)

Demo client application which connects to REDIS_URL and subscribe to CHANNEL for events.

To run demo client subscriber ensure you set below environment variables for it to work.

# Database config:
REDIS_URL=redis://user:password@redis:port/database
CHANNEL="name of your application published channel"

EventBus::EventHandler implementations.

Below sample implementations are provided which are used by demo application.

EventLogger (example/handlers/log.cr)

Sample implementation which simply logs events as they are received.

RedisPublisher (example/handlers/redis.cr)

Sample implementation which publish events to Redis Cluster for publishing to subscribers.

Events are captured and published to Redis channels which are built using scheme schema.table.cdc_events, where schema and table referred to your database schema and table name.

e.g. If you want to subscribe to change events for table mytable located in public schema, you should subscribe to channel public.mytable.cdc_events

Event payload

All change events are published to channels in JSON format

{
  "timestamp": "Timestamp with timezone",
  "schema": "PG Schema",
  "table": "PG Table name",
  "action": "one of insert|update|delete",
  "id": "Table row ID",
  "data": "JSON object representing table row data",
  "changes": "JSON Array object representing updated columns with old and new value. this is only set for update events"
}

Testing

Given you have the following dependencies...

• docker
• docker-compose

It is simple to develop the service with docker.

Install dependencies and build the lint tool before running checks:

$ shards install
$ mkdir -p bin
$ crystal build -o bin/ameba lib/ameba/bin/ameba.cr
$ ./bin/ameba
$ crystal tool format --check

With Docker

• Run specs, tearing down the docker-compose environment upon completion.

$ ./test

Without Docker

• To run tests

$ crystal spec

NOTE: The upstream dependencies specified in docker-compose.yml are required...

Compiling

$ shards build