From 07d76d612de929d1a6e6e546f35c6904b0df4cc3 Mon Sep 17 00:00:00 2001 From: Catty Steve <4795515+Catty2014@user.noreply.gitee.com> Date: Wed, 22 Apr 2026 21:21:26 +0800 Subject: [PATCH] refactor!: move notification module to src/notify.rs BREAKING CHANGE: notification system is now separate module --- workshop-baker/src/finalize/notification.rs | 5 - .../src/finalize/notification/core.rs | 247 ------------------ .../src/finalize/notification/error.rs | 36 --- workshop-baker/src/notify.rs | 0 4 files changed, 288 deletions(-) delete mode 100644 workshop-baker/src/finalize/notification.rs delete mode 100644 workshop-baker/src/finalize/notification/core.rs delete mode 100644 workshop-baker/src/finalize/notification/error.rs create mode 100644 workshop-baker/src/notify.rs diff --git a/workshop-baker/src/finalize/notification.rs b/workshop-baker/src/finalize/notification.rs deleted file mode 100644 index e4f78a1..0000000 --- a/workshop-baker/src/finalize/notification.rs +++ /dev/null @@ -1,5 +0,0 @@ -pub mod core; -pub mod error; - -pub use core::{NotificationCore, NotificationRequest}; -pub use error::NotificationError; diff --git a/workshop-baker/src/finalize/notification/core.rs b/workshop-baker/src/finalize/notification/core.rs deleted file mode 100644 index 388eb41..0000000 --- a/workshop-baker/src/finalize/notification/core.rs +++ /dev/null @@ -1,247 +0,0 @@ -// Code in this module PARTIALLY OR FULLY utilized AI Coding Agent. -use crate::engine::{EventSender, ExecutionContext}; -use crate::finalize::config::{NotificationConfig, NotificationMethod}; -use crate::finalize::error::PluginError; -use crate::finalize::notification::error::NotificationError; -use crate::finalize::plugin::{call_named_plugin, PluginMap}; -use crate::types::build_status::{BuildStatus, StageResult}; -use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::sync::Arc; -use tokio::sync::{mpsc, oneshot}; - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct NotificationRequest { - pub pipeline_name: String, - pub build_id: String, - pub build_status: BuildStatus, - pub final_result: StageResult, - pub config: NotificationConfig, -} - -/// A single notification job in the queue -struct NotificationJob { - request: NotificationRequest, - ctx: ExecutionContext, - event_tx: EventSender, - /// Number of remaining retry attempts - lives: u32, - /// Earliest time this job can be processed (for backoff) - not_before: std::time::Instant, - /// Response channel - done: oneshot::Sender>, -} - -/// Core notification dispatcher with MQ logic -pub struct NotificationCore { - plugins: Arc, - tx: mpsc::Sender, - _worker: tokio::task::JoinHandle<()>, -} - -impl NotificationCore { - pub fn new(plugins: PluginMap) -> Self { - let (tx, mut rx) = mpsc::channel::(256); - let plugins_arc = Arc::new(plugins); - - let worker = tokio::spawn(async move { - while let Some(job) = rx.recv().await { - // Check not_before backoff - let now = std::time::Instant::now(); - if now < job.not_before { - tokio::time::sleep(job.not_before - now).await; - } - - let plugins = Arc::clone(&plugins_arc); - let result = Self::execute_job(job.request, plugins, &job.ctx, job.event_tx).await; - - match result { - Ok(()) => { - // Success - nothing to do - } - Err(e) => { - log::warn!("Notification job failed: {}", e); - // Job failed but channel already closed, can't retry - } - } - } - }); - - Self { plugins: Arc::new(PluginMap::new()), tx, _worker: worker } - } - - /// Dispatch a notification request. - /// Enqueues the request and blocks until it is fully processed. - pub async fn dispatch( - &self, - request: NotificationRequest, - ctx: &ExecutionContext, - event_tx: EventSender, - ) -> Result<(), NotificationError> { - let (done_tx, done_rx) = oneshot::channel(); - let job = NotificationJob { - request, - ctx: ctx.clone(), - event_tx, - lives: 0, // TODO: derive from config retry + 1 - not_before: std::time::Instant::now(), - done: done_tx, - }; - - self.tx - .send(job) - .await - .map_err(|_| NotificationError::QueueClosed)?; - done_rx - .await - .map_err(|_| NotificationError::WorkerDropped)? - } - - /// Execute a notification job: iterate priorities, fallback on failure - async fn execute_job( - request: NotificationRequest, - plugins: Arc, - ctx: &ExecutionContext, - event_tx: EventSender, - ) -> Result<(), NotificationError> { - let groups = Self::sort_by_priority(&request.config.groups); - - for (priority, methods) in groups { - // Standalone: filter out on_finish_of (not supported) - let active_methods: Vec<_> = methods - .iter() - .filter(|(_, m)| m.on_finish_of.is_none()) - .collect(); - - if active_methods.is_empty() { - continue; - } - - // Determine which template to use based on final_result - let results = futures_util::future::join_all( - active_methods - .iter() - .map(|(name, method)| { - Self::dispatch_single( - name, - method, - &request, - &plugins, - ctx, - event_tx.clone(), - ) - }), - ) - .await; - - let any_success = results.iter().any(|r| r.is_ok()); - if any_success { - return Ok(()); - } - - // Check if we should trigger fallback to next priority - let any_critical_failed = active_methods - .iter() - .zip(results.iter()) - .any(|((_, m), r)| !m.fallible && r.is_err()); - - if any_critical_failed { - log::warn!( - "Priority {} all failed, triggering fallback to next priority", - priority - ); - continue; - } else { - // All fallible, no fallback - return Err(NotificationError::AllFailed(priority)); - } - } - - // All priorities exhausted - Err(NotificationError::AllPrioritiesFailed) - } - - /// Dispatch a single notification method - async fn dispatch_single( - plugin_name: &str, - method: &NotificationMethod, - request: &NotificationRequest, - plugins: &PluginMap, - ctx: &ExecutionContext, - event_tx: EventSender, - ) -> Result<(), NotificationError> { - // TODO: Template resolution (currently passthrough) - // In the future: - // 1. Resolve template based on method.template - // 2. Render with minijinja - // 3. Merge with method.config - let config = method.config.clone(); - - call_named_plugin(plugins, plugin_name, config, ctx, &event_tx) - .await - .map_err(|e| { - log::warn!("Plugin {} failed: {}", plugin_name, e); - NotificationError::from(e) - }) - } - - /// Sort notification groups by priority (ascending: lower number = higher priority) - fn sort_by_priority( - groups: &HashMap>, - ) -> Vec<(u32, HashMap)> { - let mut sorted: Vec<_> = groups.iter().map(|(k, v)| (*k, v.clone())).collect(); - sorted.sort_by_key(|(priority, _)| *priority); - sorted - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::finalize::config::NotificationMethod; - use serde_yaml::Value; - use std::collections::HashMap; - - fn create_test_method(fallible: bool) -> NotificationMethod { - NotificationMethod { - fallible, - schema: Some("fallback".to_string()), - on_success: None, - on_failure: None, - on_finish_of: None, - config: Value::Mapping(serde_yaml::Mapping::new()), - } - } - - #[test] - fn test_sort_by_priority() { - let mut groups = HashMap::new(); - groups.insert(10, HashMap::new()); - groups.insert(5, HashMap::new()); - groups.insert(20, HashMap::new()); - - let sorted = NotificationCore::sort_by_priority(&groups); - assert_eq!(sorted.len(), 3); - assert_eq!(sorted[0].0, 5); - assert_eq!(sorted[1].0, 10); - assert_eq!(sorted[2].0, 20); - } - - #[test] - fn test_notification_request_serialization() { - let request = NotificationRequest { - pipeline_name: "test-pipeline".to_string(), - build_id: "build-1".to_string(), - build_status: BuildStatus::default(), - final_result: StageResult::Success, - config: NotificationConfig { - templates: HashMap::new(), - groups: HashMap::new(), - }, - }; - - let json = serde_json::to_string(&request).unwrap(); - let deserialized: NotificationRequest = serde_json::from_str(&json).unwrap(); - assert_eq!(deserialized.pipeline_name, "test-pipeline"); - } -} diff --git a/workshop-baker/src/finalize/notification/error.rs b/workshop-baker/src/finalize/notification/error.rs deleted file mode 100644 index e03926a..0000000 --- a/workshop-baker/src/finalize/notification/error.rs +++ /dev/null @@ -1,36 +0,0 @@ -// Code in this module PARTIALLY OR FULLY utilized AI Coding Agent. -use crate::finalize::error::PluginError; -use thiserror::Error; - -#[derive(Error, Debug, Clone)] -pub enum NotificationError { - #[error("Queue closed")] - QueueClosed, - - #[error("Worker dropped")] - WorkerDropped, - - #[error("All plugins failed at priority {0}")] - AllFailed(u32), - - #[error("All priorities failed")] - AllPrioritiesFailed, - - #[error("Plugin error: {0}")] - PluginError(String), - - #[error("Template error: {0}")] - TemplateError(String), - - #[error("Unknown plugin: {0}")] - UnknownPlugin(String), - - #[error("In-site notification failed: {0}")] - InsiteFailed(String), -} - -impl From for NotificationError { - fn from(e: PluginError) -> Self { - NotificationError::PluginError(e.to_string()) - } -} diff --git a/workshop-baker/src/notify.rs b/workshop-baker/src/notify.rs new file mode 100644 index 0000000..e69de29