ExESDB.PubSubIntegration (ex_esdb v0.11.0)

Helper module for consistent pubsub integration across ExESDB modules.

This module provides standardized functions for broadcasting events to the ExESDBGater pubsub system, with built-in error handling and optional enablement/disablement of pubsub features.

This module now uses ExESDB.OperationalMessageHelpers to ensure consistent node tracking fields, millisecond-precision timestamps, and message validation across all operational message types.

Configuration

Add to your config to enable pubsub integration:

config :ex_esdb,
  pubsub_integration: true,
  health_broadcast_interval: 30_000,
  metrics_broadcast_interval: 60_000

Usage Examples

# System lifecycle events with standardized nodes and timestamps
ExESDB.PubSubIntegration.broadcast_system_lifecycle(:started, :ex_esdb, "1.0.0")

# Health updates with consistent node tracking
ExESDB.PubSubIntegration.broadcast_health_update(:store_worker, :healthy, %{store_id: "main"})

# Metrics with precise timestamps
ExESDB.PubSubIntegration.broadcast_metrics(:persistence, %{operations: 100, errors: 0})

# Alerts with ExESDBGater compatibility
ExESDB.PubSubIntegration.broadcast_alert(:node_failure, :critical, "Node down", %{node: :node1})

Summary

Functions

broadcast_alert(alert_type, severity, message, context \\ %{}, opts \\ [])

Broadcast critical alerts and notifications.

broadcast_audit_event(event_type, actor, resource, action, details \\ %{}, opts \\ [])

Broadcast audit trail events for compliance and tracking.

broadcast_batch(events, opts \\ [])

Broadcast multiple events in a batch for efficiency.

broadcast_cluster_health(nodes, failed_nodes, event_type, opts \\ [])

Broadcast cluster-wide health status.

broadcast_diagnostic(diagnostic_type, component, data, opts \\ [])

Broadcast diagnostic information for debugging and troubleshooting.

broadcast_health_update(component, status, details \\ %{}, opts \\ [])

Legacy function for backward compatibility - broadcasts to component_health topic.

broadcast_lifecycle_event(event_type, process_name, details \\ %{}, opts \\ [])

Broadcast process and node lifecycle events.

broadcast_metrics(component, metrics, opts \\ [])

Broadcast component performance metrics.

broadcast_store_health(store_id, component, status, details \\ %{}, opts \\ [])

Broadcast store-specific component health updates using store_id for topic generation.

broadcast_system_config(component, changes, opts \\ [])

Broadcast system configuration changes.

broadcast_system_lifecycle(event, system_name, version, opts \\ [])

Broadcast system lifecycle events (startup, shutdown, config changes).

disable!()

Disable pubsub integration at runtime.

enable!()

Enable pubsub integration at runtime.

enabled?()

Check if pubsub integration is enabled.

health_broadcast_interval()

Get the configured health broadcast interval.

metrics_broadcast_interval()

Get the configured metrics broadcast interval.

validate_messaging_compatibility()

Validate messaging compatibility between ExESDB and ExESDBGater.

Functions

broadcast_alert(alert_type, severity, message, context \\ %{}, opts \\ [])

Broadcast critical alerts and notifications.

broadcast_audit_event(event_type, actor, resource, action, details \\ %{}, opts \\ [])

Broadcast audit trail events for compliance and tracking.

broadcast_batch(events, opts \\ [])

Broadcast multiple events in a batch for efficiency.

broadcast_cluster_health(nodes, failed_nodes, event_type, opts \\ [])

Broadcast cluster-wide health status.

broadcast_diagnostic(diagnostic_type, component, data, opts \\ [])

Broadcast diagnostic information for debugging and troubleshooting.

broadcast_health_update(component, status, details \\ %{}, opts \\ [])

Legacy function for backward compatibility - broadcasts to component_health topic.

broadcast_lifecycle_event(event_type, process_name, details \\ %{}, opts \\ [])

Broadcast process and node lifecycle events.

broadcast_metrics(component, metrics, opts \\ [])

Broadcast component performance metrics.

broadcast_store_health(store_id, component, status, details \\ %{}, opts \\ [])

Broadcast store-specific component health updates using store_id for topic generation.

broadcast_system_config(component, changes, opts \\ [])

Broadcast system configuration changes.

broadcast_system_lifecycle(event, system_name, version, opts \\ [])

Broadcast system lifecycle events (startup, shutdown, config changes).

disable!()

Disable pubsub integration at runtime.

enable!()

Enable pubsub integration at runtime.

enabled?()

Check if pubsub integration is enabled.

health_broadcast_interval()

Get the configured health broadcast interval.

metrics_broadcast_interval()

Get the configured metrics broadcast interval.

validate_messaging_compatibility()

Validate messaging compatibility between ExESDB and ExESDBGater.

This function tests that operational messages can be created and validated using both systems' message patterns.