EventBus

EventBus uses ETS (Erlang Term Storage) tables for temporary event storage and status tracking. When an event is configured, two tables are created:

1. Event Store (:eb_es_<<topic>>): A read-heavy table where event data is temporarily saved. Subscribers query this table to fetch the data they need to process.
2. Event Watcher (:eb_ew_<<topic>>): A table used by the Observation Manager to track the status of event subscribers (who has processed, skipped, or is still pending).

Important: ETS storage is not persistent. Data in both tables is automatically deleted by the Observation Manager once all subscribers have processed the event. To monitor unprocessed events, you can query the watcher table directly.

To get a list of unprocessed events for a topic (e.g., :hello_received):

# Returns a list of {id, {subscribers, completers, skippers}}
:ets.tab2list(:eb_ew_hello_received)

The EventBus.EventSource module provides a block-based builder pattern to simplify event creation and notification. It automatically handles timestamps (initialized_at, occurred_at) and can handle errors by routing them to an error_topic.

Using EventSource.build/2

Use this to create an %Event{} struct from a block of logic. The return value of the block becomes the event's data.

Using EventSource.notify/2

Use this to execute logic and immediately notify subscribers with the result. If the block returns an {:error, reason} tuple, the event is automatically published to the error_topic (if provided) instead of the primary topic.

use EventBus.EventSource

# 1. Build an event struct
params = %{id: "unique_id", topic: :user_created, source: "my_app"}
EventSource.build(params) do
  # Logic here
  %{email: "user@example.com"}
end

# 2. Notify subscribers directly
params = %{id: "unique_id", topic: :user_created, error_topic: :user_error}
EventSource.notify(params) do
  # If this returns {:error, reason}, it goes to :user_error
  {:error, %{reason: "failed"}}
end

Because EventBus uses ETS for temporary storage, you must implement your own persistence logic if you need to keep event data long-term.

The recommended pattern is to subscribe a dedicated module to all event types using the [."*" ] wildcard topic pattern, then save the fetched event data to a persistent database.

Follow this pattern:

1. Subscribe to [."*" ].
2. In the process/1 callback, fetch the event data using EventBus.fetch_event/1.
3. Save the data to your persistent store.
4. Call EventBus.mark_as_completed/1 to notify the Observation Manager.

# 1. Subscribe to all topics
EventBus.subscribe({MyDataStore, [.".*" ]})

# 2. Implement the subscriber module
defmodule MyDataStore do
  # The process/1 callback receives the event shadow
  def process({topic, id} = event_shadow) do
    GenServer.cast(__MODULE__, event_shadow)
    :ok
  end

  def handle_cast({topic, id}, state) do
    # 3. Fetch the actual event data from ETS
    event = EventBus.fetch_event({topic, id})
    
    # 4. Write logic to save event_data to a persistent store (e.g. Postgres)
    # ...

    # 5. Mark as completed so the Observation Manager can clean up ETS
    EventBus.mark_as_completed({__MODULE__, {topic, id}})
    {:noreply, state}
  end
end

EventBus.subscribe({MyDataStore, [.".*" ]})

defmodule MyDataStore do
  def process({topic, id} = event_shadow) do
    GenServer.cast(__MODULE__, event_shadow)
    :ok
  end

  def handle_cast({topic, id}, state) do
    event = EventBus.fetch_event({topic, id})
    # write your logic to save event_data to a persistent store

    EventBus.mark_as_completed({__MODULE__, {topic, id}})
    {:noreply, state}
  end
end