refactor(app): add iam and notify interface boundaries (#1948)

rustfs/src/app/admin_usecase.rs

@@ -271,14 +271,18 @@ impl DefaultAdminUsecase {
     }
 
     pub fn execute_collect_dependency_readiness(&self) -> DependencyReadiness {
-        if let Some(context) = &self.context {
+        let iam_ready = self
-            let _ = context.object_store();
+            .context
-            let _ = context.iam();
+            .as_ref()
-        }
+            .map(|context| {
+                let _ = context.object_store();
+                context.iam().is_ready()
+            })
+            .unwrap_or(false);
 
         DependencyReadiness {
             storage_ready: new_object_layer_fn().is_some(),
-            iam_ready: rustfs_iam::get().is_ok(),
+            iam_ready,
         }
     }
 }

rustfs/src/app/bucket_usecase.rs

@@ -15,7 +15,7 @@
 //! Bucket application use-case contracts.
 #![allow(dead_code)]
 
-use crate::app::context::{AppContext, get_global_app_context};
+use crate::app::context::{AppContext, default_notify_interface, get_global_app_context};
 use crate::auth::get_condition_values;
 use crate::error::ApiError;
 use crate::server::RemoteAddr;
@@ -46,7 +46,6 @@ use rustfs_ecstore::client::object_api_utils::to_s3s_etag;
 use rustfs_ecstore::error::StorageError;
 use rustfs_ecstore::new_object_layer_fn;
 use rustfs_ecstore::store_api::{BucketOptions, DeleteBucketOptions, MakeBucketOptions, StorageAPI};
-use rustfs_notify::notifier_global;
 use rustfs_policy::policy::{
     action::{Action, S3Action},
     {BucketPolicy, BucketPolicyArgs, Effect, Validator},
@@ -1189,7 +1188,12 @@ impl DefaultBucketUsecase {
             .map_err(ApiError::from)?;
 
         let region = resolve_notification_region(rustfs_ecstore::global::get_global_region(), request_region);
-        let clear_rules = notifier_global::clear_bucket_notification_rules(&bucket);
+        let notify = self
+            .context
+            .as_ref()
+            .map(|context| context.notify())
+            .unwrap_or_else(default_notify_interface);
+        let clear_rules = notify.clear_bucket_notification_rules(&bucket);
         let parse_rules = async {
             let mut event_rules = Vec::new();
 
@@ -1223,7 +1227,8 @@ impl DefaultBucketUsecase {
             event_rules_result.map_err(|e| s3_error!(InvalidArgument, "Invalid ARN in notification configuration: {e}"))?;
         warn!("notify event rules: {:?}", &event_rules);
 
-        notifier_global::add_event_specific_rules(&bucket, &region, &event_rules)
+        notify
+            .add_event_specific_rules(&bucket, &region, &event_rules)
             .await
             .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))?;
 

rustfs/src/app/context.rs

@@ -17,14 +17,18 @@
 //! for storage, IAM, and KMS handles.
 #![allow(dead_code)]
 
+use async_trait::async_trait;
 use rustfs_ecstore::store::ECStore;
 use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
 use rustfs_kms::KmsServiceManager;
+use rustfs_notify::{EventArgs, NotificationError, notifier_global};
+use rustfs_targets::{EventName, arn::TargetID};
 use std::sync::{Arc, OnceLock};
 
 /// IAM interface for application-layer use-cases.
 pub trait IamInterface: Send + Sync {
     fn handle(&self) -> Arc<IamSys<ObjectStore>>;
+    fn is_ready(&self) -> bool;
 }
 
 /// KMS interface for application-layer use-cases.
@@ -32,6 +36,21 @@ pub trait KmsInterface: Send + Sync {
     fn handle(&self) -> Arc<KmsServiceManager>;
 }
 
+/// Notify interface for application-layer use-cases.
+#[async_trait]
+pub trait NotifyInterface: Send + Sync {
+    async fn notify(&self, args: EventArgs);
+
+    async fn add_event_specific_rules(
+        &self,
+        bucket_name: &str,
+        region: &str,
+        event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
+    ) -> Result<(), NotificationError>;
+
+    async fn clear_bucket_notification_rules(&self, bucket_name: &str) -> Result<(), NotificationError>;
+}
+
 /// Default IAM interface adapter.
 pub struct IamHandle {
     iam: Arc<IamSys<ObjectStore>>,
@@ -47,6 +66,10 @@ impl IamInterface for IamHandle {
     fn handle(&self) -> Arc<IamSys<ObjectStore>> {
         self.iam.clone()
     }
+
+    fn is_ready(&self) -> bool {
+        rustfs_iam::get().is_ok()
+    }
 }
 
 /// Default KMS interface adapter.
@@ -66,17 +89,47 @@ impl KmsInterface for KmsHandle {
     }
 }
 
+/// Default notify interface adapter.
+#[derive(Default)]
+pub struct NotifyHandle;
+
+#[async_trait]
+impl NotifyInterface for NotifyHandle {
+    async fn notify(&self, args: EventArgs) {
+        notifier_global::notify(args).await;
+    }
+
+    async fn add_event_specific_rules(
+        &self,
+        bucket_name: &str,
+        region: &str,
+        event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
+    ) -> Result<(), NotificationError> {
+        notifier_global::add_event_specific_rules(bucket_name, region, event_rules).await
+    }
+
+    async fn clear_bucket_notification_rules(&self, bucket_name: &str) -> Result<(), NotificationError> {
+        notifier_global::clear_bucket_notification_rules(bucket_name).await
+    }
+}
+
 /// Application-layer context with explicit dependencies.
 #[derive(Clone)]
 pub struct AppContext {
     object_store: Arc<ECStore>,
     iam: Arc<dyn IamInterface>,
     kms: Arc<dyn KmsInterface>,
+    notify: Arc<dyn NotifyInterface>,
 }
 
 impl AppContext {
     pub fn new(object_store: Arc<ECStore>, iam: Arc<dyn IamInterface>, kms: Arc<dyn KmsInterface>) -> Self {
-        Self { object_store, iam, kms }
+        Self {
+            object_store,
+            iam,
+            kms,
+            notify: default_notify_interface(),
+        }
     }
 
     pub fn with_default_interfaces(
@@ -98,6 +151,14 @@ impl AppContext {
     pub fn kms(&self) -> Arc<dyn KmsInterface> {
         self.kms.clone()
     }
+
+    pub fn notify(&self) -> Arc<dyn NotifyInterface> {
+        self.notify.clone()
+    }
+}
+
+pub fn default_notify_interface() -> Arc<dyn NotifyInterface> {
+    Arc::new(NotifyHandle)
 }
 
 static GLOBAL_APP_CONTEXT: OnceLock<Arc<AppContext>> = OnceLock::new();

rustfs/src/app/object_usecase.rs

@@ -15,7 +15,7 @@
 //! Object application use-case contracts.
 #![allow(dead_code)]
 
-use crate::app::context::{AppContext, get_global_app_context};
+use crate::app::context::{AppContext, default_notify_interface, get_global_app_context};
 use crate::config::workload_profiles::get_global_buffer_config;
 use crate::error::ApiError;
 use crate::storage::access::{ReqInfo, authorize_request, has_bypass_governance_header};
@@ -72,7 +72,7 @@ use rustfs_filemeta::{
     REPLICATE_INCOMING_DELETE, ReplicationStatusType, ReplicationType, RestoreStatusOps, VersionPurgeStatusType,
     parse_restore_obj_status,
 };
-use rustfs_notify::{EventArgsBuilder, notifier_global};
+use rustfs_notify::EventArgsBuilder;
 use rustfs_policy::policy::action::{Action, S3Action};
 use rustfs_rio::{CompressReader, DecryptReader, EncryptReader, EtagReader, HardLimitReader, HashReader, Reader, WarpReader};
 use rustfs_s3select_api::{
@@ -2647,6 +2647,11 @@ impl DefaultObjectUsecase {
         }
 
         let req_headers = req.headers.clone();
+        let notify = self
+            .context
+            .as_ref()
+            .map(|context| context.notify())
+            .unwrap_or_else(default_notify_interface);
         tokio::spawn(async move {
             for res in delete_results {
                 if let Some(dobj) = res.delete_object {
@@ -2671,7 +2676,7 @@ impl DefaultObjectUsecase {
                     .user_agent(get_request_user_agent(&req_headers))
                     .build();
 
-                    notifier_global::notify(event_args).await;
+                    notify.notify(event_args).await;
                 }
             }
         });