brokkr-broker::utils::audit

Rust

Async Audit Logger for Brokkr.

This module provides an asynchronous audit logging service that buffers audit entries and writes them to the database in batches to avoid impacting request latency.

Usage

use brokkr_broker::utils::audit;

// Initialize during startup
audit::init_audit_logger(dal)?;

// Log an audit entry (non-blocking)
audit::log(NewAuditLog::new(
    ACTOR_TYPE_ADMIN,
    Some(admin_id),
    ACTION_AGENT_CREATED,
    RESOURCE_TYPE_AGENT,
    Some(agent_id),
)?);

Structs

struct AuditLoggerConfig

pub

Derives: Debug, Clone

Configuration for the audit logger.

Fields

Name	Type	Description
channel_size	usize	Channel buffer size.
batch_size	usize	Maximum batch size for writes.
flush_interval_ms	u64	Flush interval in milliseconds.

struct AuditLogger

pub

Derives: Clone

The async audit logger for buffering and batching audit entries.

Methods

new

pub

fn new(dal: DAL) -> Self

Creates a new audit logger and starts the background writer.

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer for database operations.

Returns:

An AuditLogger instance.

Source

    pub fn new(dal: DAL) -> Self {
        Self::with_config(dal, AuditLoggerConfig::default())
    }

with_​config

pub

fn with_config(dal: DAL, config: AuditLoggerConfig) -> Self

Creates a new audit logger with custom configuration.

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer for database operations.
config	AuditLoggerConfig	The logger configuration.

Returns:

An AuditLogger instance.

Source

    pub fn with_config(dal: DAL, config: AuditLoggerConfig) -> Self {
        let (sender, receiver) = mpsc::channel(config.channel_size);

        // Start the background writer task
        start_audit_writer(dal, receiver, config.batch_size, config.flush_interval_ms);

        info!(
            "Audit logger started (buffer: {}, batch: {}, flush: {}ms)",
            config.channel_size, config.batch_size, config.flush_interval_ms
        );

        Self { sender }
    }

log

pub

fn log(&self, entry: NewAuditLog)

Logs an audit entry asynchronously (non-blocking).

If the channel is full, the entry will be dropped and an error logged.

Parameters:

Name	Type	Description
entry	NewAuditLog	The audit log entry to record.

Source

    pub fn log(&self, entry: NewAuditLog) {
        let sender = self.sender.clone();
        let action = entry.action.clone();

        tokio::spawn(async move {
            match sender.send(entry).await {
                Ok(_) => {
                    debug!("Audit entry queued: {}", action);
                }
                Err(e) => {
                    error!(
                        "Failed to queue audit entry (action: {}): channel full or closed - {}",
                        action, e
                    );
                }
            }
        });
    }

log_​async

pub async

async fn log_async(
    &self,
    entry: NewAuditLog,
) -> Result<(), mpsc::error::SendError<NewAuditLog>>

Logs an audit entry, waiting for it to be accepted.

Parameters:

Name	Type	Description
entry	NewAuditLog	The audit log entry to record.

Returns:

Ok if the entry was accepted, Err if the channel is closed.

Source

    pub async fn log_async(
        &self,
        entry: NewAuditLog,
    ) -> Result<(), mpsc::error::SendError<NewAuditLog>> {
        let action = entry.action.clone();

        self.sender.send(entry).await.map_err(|e| {
            error!("Failed to queue audit entry (action: {}): {}", action, e);
            e
        })?;

        debug!("Audit entry queued (async): {}", action);
        Ok(())
    }

try_​log

pub

fn try_log(&self, entry: NewAuditLog) -> bool

Tries to log an audit entry without blocking.

Parameters:

Name	Type	Description
entry	NewAuditLog	The audit log entry to record.

Returns:

true if the entry was queued, false if the channel is full.

Source

    pub fn try_log(&self, entry: NewAuditLog) -> bool {
        match self.sender.try_send(entry) {
            Ok(_) => true,
            Err(mpsc::error::TrySendError::Full(_)) => {
                warn!("Audit log channel full, entry dropped");
                false
            }
            Err(mpsc::error::TrySendError::Closed(_)) => {
                error!("Audit log channel closed");
                false
            }
        }
    }

Functions

fn init_​audit_​logger

pub

fn init_audit_logger(dal: DAL) -> Result<(), String>

Initializes the global audit logger.

This should be called once during broker startup.

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer for database operations.

Returns:

Ok(()) if initialization succeeded, Err if already initialized.

Source

pub fn init_audit_logger(dal: DAL) -> Result<(), String> {
    init_audit_logger_with_config(dal, AuditLoggerConfig::default())
}

fn init_​audit_​logger_​with_​config

pub

fn init_audit_logger_with_config(
    dal: DAL,
    config: AuditLoggerConfig,
) -> Result<(), String>

Initializes the global audit logger with custom configuration.

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer for database operations.
config	AuditLoggerConfig	The logger configuration.

Returns:

Ok(()) if initialization succeeded, Err if already initialized.

Source

pub fn init_audit_logger_with_config(dal: DAL, config: AuditLoggerConfig) -> Result<(), String> {
    let logger = AuditLogger::with_config(dal, config);
    AUDIT_LOGGER
        .set(Arc::new(logger))
        .map_err(|_| "Audit logger already initialized".to_string())
}

fn get_​audit_​logger

pub

fn get_audit_logger() -> Option<Arc<AuditLogger>>

Gets the global audit logger.

Returns:

The audit logger, or None if not initialized.

Source

pub fn get_audit_logger() -> Option<Arc<AuditLogger>> {
    AUDIT_LOGGER.get().cloned()
}

fn log

pub

fn log(entry: NewAuditLog)

Logs an audit entry to the global audit logger.

This is a convenience function for logging without needing to get the logger directly.

Parameters:

Name	Type	Description
entry	NewAuditLog	The audit log entry to record.

Source

pub fn log(entry: NewAuditLog) {
    match get_audit_logger() {
        Some(logger) => {
            logger.log(entry);
        }
        _ => {
            warn!(
                "Audit logger not initialized, entry dropped: {}",
                entry.action
            );
        }
    }
}

fn try_​log

pub

fn try_log(entry: NewAuditLog) -> bool

Tries to log an audit entry without blocking.

Parameters:

Name	Type	Description
entry	NewAuditLog	The audit log entry to record.

Returns:

true if logged, false if channel full or logger not initialized.

Source

pub fn try_log(entry: NewAuditLog) -> bool {
    match get_audit_logger() {
        Some(logger) => logger.try_log(entry),
        _ => {
            warn!(
                "Audit logger not initialized, entry dropped: {}",
                entry.action
            );
            false
        }
    }
}

fn log_​action

pub

fn log_action(
    actor_type: &str,
    actor_id: Option<uuid::Uuid>,
    action: &str,
    resource_type: &str,
    resource_id: Option<uuid::Uuid>,
    details: Option<serde_json::Value>,
    ip_address: Option<String>,
    user_agent: Option<String>,
)

Helper to create and log an audit entry in one call.

Parameters:

Name	Type	Description
actor_type	&str	Type of actor (admin, agent, generator, system).
actor_id	Option<uuid::Uuid>	ID of the actor.
action	&str	The action performed.
resource_type	&str	Type of resource affected.
resource_id	Option<uuid::Uuid>	ID of the affected resource.
details	Option<serde_json::Value>	Optional additional details.
ip_address	Option<String>	Optional client IP address.
user_agent	Option<String>	Optional client user agent.

Source

pub fn log_action(
    actor_type: &str,
    actor_id: Option<uuid::Uuid>,
    action: &str,
    resource_type: &str,
    resource_id: Option<uuid::Uuid>,
    details: Option<serde_json::Value>,
    ip_address: Option<String>,
    user_agent: Option<String>,
) {
    match NewAuditLog::new(actor_type, actor_id, action, resource_type, resource_id) {
        Ok(mut entry) => {
            if let Some(d) = details {
                entry = entry.with_details(d);
            }
            if let Some(ip) = ip_address {
                entry = entry.with_ip_address(ip);
            }
            if let Some(ua) = user_agent {
                entry = entry.with_user_agent(ua);
            }
            log(entry);
        }
        Err(e) => {
            error!("Failed to create audit entry: {}", e);
        }
    }
}