refactor!: move notification module to src/notify.rs
BREAKING CHANGE: notification system is now separate module
This commit is contained in:
Catty Steve
@@ -1,5 +0,0 @@
|
||||
pub mod core;
|
||||
pub mod error;
|
||||
|
||||
pub use core::{NotificationCore, NotificationRequest};
|
||||
pub use error::NotificationError;
|
||||
@@ -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<Result<(), NotificationError>>,
|
||||
}
|
||||
|
||||
/// Core notification dispatcher with MQ logic
|
||||
pub struct NotificationCore {
|
||||
plugins: Arc<PluginMap>,
|
||||
tx: mpsc::Sender<NotificationJob>,
|
||||
_worker: tokio::task::JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl NotificationCore {
|
||||
pub fn new(plugins: PluginMap) -> Self {
|
||||
let (tx, mut rx) = mpsc::channel::<NotificationJob>(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<PluginMap>,
|
||||
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<u32, HashMap<String, NotificationMethod>>,
|
||||
) -> Vec<(u32, HashMap<String, NotificationMethod>)> {
|
||||
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");
|
||||
}
|
||||
}
|
||||
@@ -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<PluginError> for NotificationError {
|
||||
fn from(e: PluginError) -> Self {
|
||||
NotificationError::PluginError(e.to_string())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user