Function process_event

nodedb/src/control/event_trigger.rs:59–116

Process a single change event against all matching EventDefinitions.

(shared: &SharedState, event: &ChangeEvent)

/// Process a single change event against all matching EventDefinitions.
async fn process_event(shared: &SharedState, event: &ChangeEvent) {
    let catalog = match shared.credentials.catalog() {
        Some(c) => c,
        None => return,
    };

    let coll = match catalog.get_collection(
        DatabaseId::DEFAULT,
        event.tenant_id.as_u64(),
        &event.collection,
    ) {
        Ok(Some(c)) => c,
        _ => return,
    };

    if coll.event_defs.is_empty() {
        return;
    }

    let op_str = event.operation.as_str();

    for event_def in &coll.event_defs {
        let when_upper = event_def.when_condition.to_uppercase();
        let matches = match when_upper.as_str() {
            "INSERT" => event.operation == ChangeOperation::Insert,
            "UPDATE" => event.operation == ChangeOperation::Update,
            "DELETE" => event.operation == ChangeOperation::Delete,
            "ANY" | "*" | "TRUE" => true,
            compound => compound.contains(op_str),
        };

        if !matches {
            continue;
        }

        debug!(
            event = event_def.name,
            collection = event.collection,
            document_id = event.document_id,
            operation = op_str,
            action = event_def.then_action,
            "event trigger fired"
        );

        // Execute the THEN action as SQL.
        execute_then_action(shared, event, &event_def.then_action, &event_def.name).await;

        shared.audit_record(
            crate::control::security::audit::AuditEvent::AdminAction,
            Some(event.tenant_id),
            "event_trigger",
            &format!(
                "event '{}' on '{}': doc={}, op={}, action={}",
                event_def.name, event.collection, event.document_id, op_str, event_def.then_action
            ),
        );
    }
}