rustfs/crates/notify/src/global.rs

// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::{
    BucketNotificationConfig, Event, EventArgs, LifecycleError, NotificationError, NotificationMetricSnapshot,
    NotificationSystem, NotificationTargetMetricSnapshot,
};
use rustfs_config::server_config::Config;
use rustfs_s3_types::EventName;
use rustfs_targets::arn::TargetID;
use std::sync::{Arc, OnceLock};
use tracing::error;

static NOTIFICATION_SYSTEM: OnceLock<Arc<NotificationSystem>> = OnceLock::new();
const LOG_COMPONENT_NOTIFY: &str = "notify";
const LOG_SUBSYSTEM_GLOBAL: &str = "global";
const EVENT_NOTIFY_GLOBAL_STATE: &str = "notify_global_state";

/// Initialize the global notification system with the given configuration.
/// This function should only be called once throughout the application life cycle.
pub async fn initialize(config: Config) -> Result<(), NotificationError> {
    // `new` is synchronous and responsible for creating instances
    let system = NotificationSystem::new(config);
    // `init` is asynchronous and responsible for performing I/O-intensive initialization
    system.init().await?;

    match NOTIFICATION_SYSTEM.set(Arc::new(system)) {
        Ok(_) => Ok(()),
        Err(losing_system) => {
            // Another initializer won the race. `init()` above already started this
            // system's targets and replay workers, so simply dropping it would leak
            // those background tasks. Shut the losing instance down cleanly before
            // reporting the conflict (backlog#984).
            losing_system.shutdown().await;
            Err(NotificationError::Lifecycle(LifecycleError::AlreadyInitialized))
        }
    }
}

/// Initialize the global notification system only for live in-process consumers.
///
/// This does not load configured notification targets or bucket rules. It exists so
/// ListenBucketNotification clients can receive live events even when external
/// notification targets are disabled.
pub fn initialize_live_events() -> Result<(), NotificationError> {
    let system = NotificationSystem::new(Config::new());

    match NOTIFICATION_SYSTEM.set(Arc::new(system)) {
        Ok(_) => Ok(()),
        Err(_) => Err(NotificationError::Lifecycle(LifecycleError::AlreadyInitialized)),
    }
}

/// Returns a handle to the global NotificationSystem instance.
/// Return None if the system has not been initialized.
pub fn notification_system() -> Option<Arc<NotificationSystem>> {
    NOTIFICATION_SYSTEM.get().cloned()
}

/// Returns aggregate notification delivery metrics for Prometheus collection.
pub fn notification_metrics_snapshot() -> NotificationMetricSnapshot {
    NOTIFICATION_SYSTEM
        .get()
        .map(|system| system.snapshot_metrics())
        .unwrap_or_default()
}

/// Returns per-target notification delivery metrics for Prometheus collection.
pub async fn notification_target_metrics() -> Vec<NotificationTargetMetricSnapshot> {
    if let Some(system) = notification_system() {
        system.snapshot_target_metrics().await
    } else {
        Vec::new()
    }
}

/// Check if the notification system has been initialized.
pub fn is_notification_system_initialized() -> bool {
    NOTIFICATION_SYSTEM.get().is_some()
}

/// A module providing the public API for event notification.
pub mod notifier_global {
    use super::*;
    use tracing::instrument;

    /// Notify an event asynchronously.
    /// This is the only entry point for all event notifications in the system.
    /// # Parameter
    /// - `args`: The event arguments containing details about the event to be notified.
    ///
    /// # Return value
    /// Returns `()`, indicating that the notification has been sent.
    ///
    /// # Using
    /// This function is used to notify events in the system, such as object creation, deletion, or updates.
    #[instrument(skip(args))]
    pub async fn notify(args: EventArgs) {
        // Dependency injection or service positioning mode obtain NotificationSystem instance
        let notification_sys = match notification_system() {
            // If the notification system itself cannot be retrieved, it will be returned directly
            Some(sys) => sys,
            None => {
                error!(
                    event = EVENT_NOTIFY_GLOBAL_STATE,
                    component = LOG_COMPONENT_NOTIFY,
                    subsystem = LOG_SUBSYSTEM_GLOBAL,
                    state = "uninitialized",
                    "notify global state"
                );
                return;
            }
        };

        // Avoid generating notifications for replica creation events
        if args.is_replication_request() {
            return;
        }

        // Create an event and send it
        let event = Arc::new(Event::new(args));
        notification_sys.send_event(event).await;
    }

    /// Add notification rules for the specified bucket and load configuration
    /// # Parameter
    /// - `bucket_name`: The name of the target bucket.
    /// - `region`: The area where bucket is located.
    /// - `event_names`: A list of event names that trigger notifications.
    /// - `prefix`: The prefix of the object key that triggers notifications.
    /// - `suffix`: The suffix of the object key that triggers notifications.
    /// - `target_ids`: A list of target IDs that will receive notifications.
    ///
    /// # Return value
    /// Returns `Result<(), NotificationError>`, Ok on success, and an error on failure
    ///
    /// # Using
    /// This function allows you to dynamically add notification rules for a specific bucket.
    pub async fn add_bucket_notification_rule(
        bucket_name: &str,
        region: &str,
        event_names: &[EventName],
        prefix: &str,
        suffix: &str,
        target_ids: &[TargetID],
    ) -> Result<(), NotificationError> {
        // Construct pattern using proper pattern function
        let pattern = crate::rules::pattern::new_pattern(
            if prefix.is_empty() { None } else { Some(prefix) },
            if suffix.is_empty() { None } else { Some(suffix) },
        );

        // Create BucketNotificationConfig
        let mut bucket_config = BucketNotificationConfig::new(region);
        for target_id in target_ids {
            bucket_config.add_rule(event_names, pattern.clone(), target_id.clone());
        }

        // Get global NotificationSystem
        let notification_sys = match notification_system() {
            Some(sys) => sys,
            None => return Err(NotificationError::Lifecycle(LifecycleError::NotInitialized)),
        };

        // Loading configuration
        notification_sys
            .load_bucket_notification_config(bucket_name, &bucket_config)
            .await
    }

    /// Dynamically add notification rules according to different event types.
    ///
    /// # Parameter
    /// - `bucket_name`: The name of the target bucket.
    /// - `region`: The area where bucket is located.
    /// - `event_rules`: Each rule contains a list of event types, prefixes, suffixes, and target IDs.
    ///
    /// # Return value
    /// Returns `Result<(), NotificationError>`, Ok on success, and an error on failure.
    ///
    /// # Using
    /// Supports notification rules for adding multiple event types, prefixes, suffixes, and targets to the same bucket in batches.
    pub async fn add_event_specific_rules(
        bucket_name: &str,
        region: &str,
        event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
    ) -> Result<(), NotificationError> {
        let mut bucket_config = BucketNotificationConfig::new(region);

        for (event_names, prefix, suffix, target_ids) in event_rules {
            // Use `new_pattern` to construct a matching pattern
            let pattern = crate::rules::pattern::new_pattern(Some(prefix.as_str()), Some(suffix.as_str()));

            for target_id in target_ids {
                bucket_config.add_rule(event_names, pattern.clone(), target_id.clone());
            }
        }

        // Get global NotificationSystem instance
        let notification_sys = notification_system().ok_or(NotificationError::Lifecycle(LifecycleError::NotInitialized))?;

        // Loading configuration
        notification_sys
            .load_bucket_notification_config(bucket_name, &bucket_config)
            .await
    }

    /// Clear all notification rules for the specified bucket.
    /// # Parameter
    /// - `bucket_name`: The name of the target bucket.
    /// # Return value
    /// Returns `Result<(), NotificationError>`, Ok on success, and an error on failure.
    /// # Using
    /// This function allows you to clear all notification rules for a specific bucket.
    /// This is useful when you want to reset the notification configuration for a bucket.
    ///
    pub async fn clear_bucket_notification_rules(bucket_name: &str) -> Result<(), NotificationError> {
        // Get global NotificationSystem instance
        let notification_sys = notification_system().ok_or(NotificationError::Lifecycle(LifecycleError::NotInitialized))?;

        // Clear configuration
        notification_sys.remove_bucket_notification_config(bucket_name).await;
        Ok(())
    }
}