rustfs/crates/notify/src/runtime_facade.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::{Event, NotificationError, integration::NotificationMetrics, notifier::SharedNotifyTargetList};
use rustfs_targets::{
    BuiltinPluginRuntimeAdapter, PluginRuntimeAdapter, ReplayEvent, ReplayWorkerManager, RuntimeActivation, Target,
};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{RwLock, Semaphore};
use tracing::{debug, info};

const LOG_COMPONENT_NOTIFY: &str = "notify";
const LOG_SUBSYSTEM_RUNTIME: &str = "runtime";
const EVENT_NOTIFY_RUNTIME_LIFECYCLE: &str = "notify_runtime_lifecycle";
const EVENT_NOTIFY_RUNTIME_SHUTDOWN_FAILED: &str = "notify_runtime_shutdown_failed";

#[derive(Clone)]
pub struct NotifyRuntimeFacade {
    target_list: SharedNotifyTargetList,
    replay_workers: Arc<RwLock<ReplayWorkerManager>>,
    runtime_adapter: Arc<dyn PluginRuntimeAdapter<Event>>,
}

impl NotifyRuntimeFacade {
    pub fn new(
        target_list: SharedNotifyTargetList,
        replay_workers: Arc<RwLock<ReplayWorkerManager>>,
        concurrency_limiter: Arc<Semaphore>,
        metrics: Arc<NotificationMetrics>,
    ) -> Self {
        let replay_metrics = metrics;
        let runtime_adapter = BuiltinPluginRuntimeAdapter::new(
            Arc::new(move |event: ReplayEvent<Event>| {
                let metrics = replay_metrics.clone();
                Box::pin(async move {
                    match event {
                        ReplayEvent::Delivered { .. } => metrics.increment_processed(),
                        ReplayEvent::RetryableError { .. } => {}
                        ReplayEvent::Dropped { target, .. }
                        | ReplayEvent::PermanentFailure { target, .. }
                        | ReplayEvent::RetryExhausted { target, .. } => {
                            target.record_final_failure();
                            metrics.increment_failed();
                        }
                        ReplayEvent::UnreadableEntry { .. } => {}
                    }
                })
            }),
            Arc::new(|target_id, has_replay| {
                if has_replay {
                    info!(
                        event = EVENT_NOTIFY_RUNTIME_LIFECYCLE,
                        component = LOG_COMPONENT_NOTIFY,
                        subsystem = LOG_SUBSYSTEM_RUNTIME,
                        target_id = %target_id,
                        state = "replay_started",
                        "notify runtime lifecycle"
                    );
                } else {
                    debug!(
                        event = EVENT_NOTIFY_RUNTIME_LIFECYCLE,
                        component = LOG_COMPONENT_NOTIFY,
                        subsystem = LOG_SUBSYSTEM_RUNTIME,
                        target_id = %target_id,
                        state = "replay_skipped",
                        reason = "no_store_configured",
                        "notify runtime lifecycle"
                    );
                }
            }),
            Some(concurrency_limiter),
            Duration::from_secs(5),
            Duration::from_millis(500),
            "Stop event stream processing for target",
        );

        Self {
            target_list,
            replay_workers,
            runtime_adapter: Arc::new(runtime_adapter),
        }
    }

    pub async fn activate_targets_with_replay(
        &self,
        targets: Vec<Box<dyn Target<Event> + Send + Sync>>,
    ) -> RuntimeActivation<Event> {
        self.runtime_adapter.activate_with_replay(targets).await
    }

    pub async fn replace_targets(&self, activation: RuntimeActivation<Event>) -> Result<(), NotificationError> {
        // Lock order: replay_workers -> target_list (matches notify AGENTS.md).
        let mut replay_workers = self.replay_workers.write().await;
        let mut target_list = self.target_list.write().await;
        self.runtime_adapter
            .replace_runtime_targets(target_list.runtime_mut(), &mut replay_workers, activation)
            .await
            .map_err(NotificationError::Target)?;
        Ok(())
    }

    pub async fn stop_replay_workers(&self) {
        let mut replay_workers = self.replay_workers.write().await;
        self.runtime_adapter.stop_replay_workers(&mut replay_workers).await;
    }

    pub async fn shutdown(&self) {
        info!(
            event = EVENT_NOTIFY_RUNTIME_LIFECYCLE,
            component = LOG_COMPONENT_NOTIFY,
            subsystem = LOG_SUBSYSTEM_RUNTIME,
            state = "stopping",
            "notify runtime lifecycle"
        );

        let active_targets = self.replay_workers.read().await.len();
        info!(
            event = EVENT_NOTIFY_RUNTIME_LIFECYCLE,
            component = LOG_COMPONENT_NOTIFY,
            subsystem = LOG_SUBSYSTEM_RUNTIME,
            state = "replay_stopping",
            active_targets,
            "notify runtime lifecycle"
        );

        {
            // Lock order: replay_workers -> target_list (matches notify AGENTS.md).
            let mut replay_workers = self.replay_workers.write().await;
            let mut target_list = self.target_list.write().await;
            if let Err(err) = self
                .runtime_adapter
                .shutdown(target_list.runtime_mut(), &mut replay_workers)
                .await
            {
                tracing::error!(
                    event = EVENT_NOTIFY_RUNTIME_SHUTDOWN_FAILED,
                    component = LOG_COMPONENT_NOTIFY,
                    subsystem = LOG_SUBSYSTEM_RUNTIME,
                    error = %err,
                    "Failed to shutdown notify runtime cleanly"
                );
            }
        }
        tokio::time::sleep(Duration::from_millis(500)).await;

        info!(
            event = EVENT_NOTIFY_RUNTIME_LIFECYCLE,
            component = LOG_COMPONENT_NOTIFY,
            subsystem = LOG_SUBSYSTEM_RUNTIME,
            state = "stopped",
            "notify runtime lifecycle"
        );
    }
}

#[cfg(test)]
mod tests {
    use super::NotifyRuntimeFacade;
    use crate::{
        Event, integration::NotificationMetrics, notifier::EventNotifier, rule_engine::NotifyRuleEngine,
        runtime_view::NotifyRuntimeView,
    };
    use async_trait::async_trait;
    use rustfs_targets::arn::TargetID;
    use rustfs_targets::store::{Key, Store};
    use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta};
    use rustfs_targets::{ReplayWorkerManager, SharedTarget, StoreError, Target, TargetError};
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tokio::sync::{RwLock, Semaphore};

    #[derive(Clone)]
    struct TestTarget {
        close_calls: Arc<AtomicUsize>,
        id: TargetID,
    }

    impl TestTarget {
        fn new(id: &str, name: &str) -> Self {
            Self {
                close_calls: Arc::new(AtomicUsize::new(0)),
                id: TargetID::new(id.to_string(), name.to_string()),
            }
        }
    }

    #[async_trait]
    impl<E> Target<E> for TestTarget
    where
        E: rustfs_targets::PluginEvent,
    {
        fn id(&self) -> TargetID {
            self.id.clone()
        }

        async fn is_active(&self) -> Result<bool, TargetError> {
            Ok(true)
        }

        async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
            Ok(())
        }

        async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
            Ok(())
        }

        async fn close(&self) -> Result<(), TargetError> {
            self.close_calls.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }

        fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
            None
        }

        fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
            Box::new(self.clone())
        }

        async fn init(&self) -> Result<(), TargetError> {
            Ok(())
        }

        fn is_enabled(&self) -> bool {
            true
        }
    }

    fn build_facade() -> (NotifyRuntimeFacade, Arc<EventNotifier>, Arc<RwLock<ReplayWorkerManager>>) {
        let metrics = Arc::new(NotificationMetrics::new());
        let notifier = Arc::new(EventNotifier::new(metrics.clone(), NotifyRuleEngine::new()));
        let target_list = notifier.target_list();
        let replay_workers = Arc::new(RwLock::new(ReplayWorkerManager::new()));
        let facade = NotifyRuntimeFacade::new(target_list, replay_workers.clone(), Arc::new(Semaphore::new(4)), metrics);
        (facade, notifier, replay_workers)
    }

    #[tokio::test]
    async fn runtime_facade_stops_empty_replay_workers() {
        let (facade, _, _) = build_facade();
        facade.stop_replay_workers().await;
    }

    #[tokio::test]
    async fn runtime_facade_activates_empty_target_list() {
        let (facade, _, _) = build_facade();
        let activation = facade.activate_targets_with_replay(Vec::new()).await;

        assert!(activation.targets.is_empty());
        assert_eq!(activation.replay_workers.len(), 0);
    }

    #[tokio::test]
    async fn runtime_facade_replace_targets_commits_runtime_state() {
        let (facade, notifier, replay_workers) = build_facade();
        let target = TestTarget::new("primary", "webhook");
        let activation = rustfs_targets::RuntimeActivation {
            replay_workers: ReplayWorkerManager::new(),
            targets: vec![Arc::new(target) as SharedTarget<Event>],
        };

        facade
            .replace_targets(activation)
            .await
            .expect("replace_targets should succeed");

        let runtime_view = NotifyRuntimeView::new(notifier.target_list(), replay_workers.clone());
        let active_targets = runtime_view.get_active_targets().await;
        assert_eq!(active_targets, vec![TargetID::new("primary".to_string(), "webhook".to_string())]);
        assert_eq!(replay_workers.read().await.len(), 0);
    }
}