brokkr-broker::utils::event_bus

Rust

Event emission utilities for webhook notifications.

This module provides a database-centric approach to event emission. Events are directly inserted into the webhook_deliveries table for matching subscriptions. No in-memory event bus is used.

Functions

fn emit_​event

pub

fn emit_event(dal: &DAL, event: &BrokkrEvent) -> usize

Emits an event by creating webhook deliveries for all matching subscriptions.

This function:

1. Finds all enabled subscriptions matching the event type
2. Discards those whose filters exclude this event’s payload
3. Creates a webhook_delivery record for each remaining subscription
4. Copies target_labels from subscription to delivery for routing

Parameters:

Name	Type	Description
dal	&DAL	The Data Access Layer instance.
event	&BrokkrEvent	The event to emit.

Returns:

The number of deliveries created.

Source

pub fn emit_event(dal: &DAL, event: &BrokkrEvent) -> usize {
    // Find all subscriptions matching this event type
    let subscriptions = match dal
        .webhook_subscriptions()
        .get_matching_subscriptions(&event.event_type)
    {
        Ok(subs) => subs,
        Err(e) => {
            error!(
                "Failed to get matching subscriptions for event {}: {:?}",
                event.event_type, e
            );
            return 0;
        }
    };

    if subscriptions.is_empty() {
        debug!(
            "No subscriptions match event {} (id: {})",
            event.event_type, event.id
        );
        return 0;
    }

    debug!(
        "Emitting event {} (id: {}) to {} subscription(s)",
        event.event_type,
        event.id,
        subscriptions.len()
    );

    let mut created_count = 0;

    // Create a delivery for each matching subscription
    for subscription in subscriptions {
        // Payload filtering (agent_id / stack_id). Note the deliberate
        // consequence: a subscription filtering on a field this event type never
        // carries receives nothing at all. `deployment.applied`/`.failed` carry
        // `agent_id` but only a `deployment_object_id`, so a `stack_id` filter
        // excludes them; resolving the owning stack would mean a DAL round trip
        // per subscription on the write path, and the fix belongs at the
        // emission site (put `stack_id` in the payload), not here.
        if !subscription_admits_event(&subscription, event) {
            debug!(
                "Subscription {} filtered out event {} (id: {})",
                subscription.id, event.event_type, event.id
            );
            continue;
        }

        // Copy target_labels from subscription to delivery
        let target_labels = subscription.target_labels.clone();

        match NewWebhookDelivery::new(subscription.id, event, target_labels) {
            Ok(new_delivery) => match dal.webhook_deliveries().create(&new_delivery) {
                Ok(delivery) => {
                    let target = if delivery.target_labels.is_some() {
                        "agent"
                    } else {
                        "broker"
                    };
                    debug!(
                        "Created delivery {} for subscription {} (event: {}, target: {})",
                        delivery.id, subscription.id, event.event_type, target
                    );
                    created_count += 1;
                }
                Err(e) => {
                    error!(
                        "Failed to create delivery for subscription {}: {:?}",
                        subscription.id, e
                    );
                }
            },
            Err(e) => {
                error!(
                    "Failed to create NewWebhookDelivery for subscription {}: {}",
                    subscription.id, e
                );
            }
        }
    }

    created_count
}