From df5026cdbe59eacc8025179afdb6e2d9f6ab2531 Mon Sep 17 00:00:00 2001 From: Catty Steve <4795515+Catty2014@user.noreply.gitee.com> Date: Fri, 22 May 2026 00:16:28 +0800 Subject: [PATCH] refactor(notify): extract handler, rule, util Split the monolithic notify module into three submodules: - `handler`: async bucket dispatch with cooldown, retry, escalation - `rule`: rule parsing from YAML config, template merging, overrides - `util`: validation, event timing, helper functions `NotificationTemplateDef` replaces `AddAssign` with `apply()` for non-destructive merging (self wins, other fills gaps). `insitenotify` logs instead of panicking. --- workshop-baker/src/finalize/config.rs | 37 +- .../finalize/plugin/internal/insitenotify.rs | 3 +- workshop-baker/src/notify.rs | 767 +----------------- workshop-baker/src/notify/error.rs | 2 +- workshop-baker/src/notify/handler.rs | 345 ++++++++ workshop-baker/src/notify/rule.rs | 148 ++++ workshop-baker/src/notify/template.rs | 143 +++- workshop-baker/src/notify/util.rs | 83 ++ 8 files changed, 799 insertions(+), 729 deletions(-) create mode 100644 workshop-baker/src/notify/handler.rs create mode 100644 workshop-baker/src/notify/rule.rs create mode 100644 workshop-baker/src/notify/util.rs diff --git a/workshop-baker/src/finalize/config.rs b/workshop-baker/src/finalize/config.rs index ec0359b..57ad6a0 100644 --- a/workshop-baker/src/finalize/config.rs +++ b/workshop-baker/src/finalize/config.rs @@ -231,8 +231,12 @@ pub struct NotificationTemplateDef { pub on_finish_of: Option, } -impl std::ops::AddAssign for NotificationTemplateDef { - fn add_assign(&mut self, other: Self) { +impl NotificationTemplateDef { + /// Merge `other` into `self`, keeping self's values when present. + /// + /// `other` fills in gaps where `self` is `None`. + /// Chaining: `a.apply(b).apply(c)` = a wins, then b fills gaps, then c fills gaps. + pub fn apply(mut self, other: Self) -> Self { if self.on_success.is_none() { self.on_success = other.on_success; } @@ -242,6 +246,35 @@ impl std::ops::AddAssign for NotificationTemplateDef { if self.on_finish_of.is_none() { self.on_finish_of = other.on_finish_of; } + self + } +} + +/// Builder for [`NotificationTemplateDef`]. +pub struct NotificationTemplateDefBuilder(NotificationTemplateDef); + +impl NotificationTemplateDefBuilder { + pub fn new() -> Self { + Self(NotificationTemplateDef::default()) + } + pub fn with_use(mut self, val: impl Into) -> Self { + self.0.use_ = Some(val.into()); + self + } + pub fn with_on_success(mut self, val: impl Into) -> Self { + self.0.on_success = Some(val.into()); + self + } + pub fn with_on_failure(mut self, val: impl Into) -> Self { + self.0.on_failure = Some(val.into()); + self + } + pub fn with_on_finish_of(mut self, val: impl Into) -> Self { + self.0.on_finish_of = Some(val.into()); + self + } + pub fn build(self) -> NotificationTemplateDef { + self.0 } } diff --git a/workshop-baker/src/finalize/plugin/internal/insitenotify.rs b/workshop-baker/src/finalize/plugin/internal/insitenotify.rs index e01e0f8..ba4d4ae 100644 --- a/workshop-baker/src/finalize/plugin/internal/insitenotify.rs +++ b/workshop-baker/src/finalize/plugin/internal/insitenotify.rs @@ -14,6 +14,7 @@ impl AsyncPluginFn for InsiteNotify { _ctx: &ExecutionContext, _event_tx: &EventSender, ) -> Result<(), PluginError> { - todo!("Requires workshop-base to be implemented!"); + log::info!("in-site-notify: placeholder (always succeeds)"); + Ok(()) } } diff --git a/workshop-baker/src/notify.rs b/workshop-baker/src/notify.rs index 65c7a4e..2d9ae4f 100644 --- a/workshop-baker/src/notify.rs +++ b/workshop-baker/src/notify.rs @@ -1,58 +1,24 @@ -use crate::finalize::FinalizeConfig; -use crate::finalize::NotificationConfig; -use crate::finalize::NotificationTemplate; -use crate::finalize::config::TriggerItemRaw; use crate::finalize::parse; use crate::finalize::plugin; -use crate::notify::types::Matchable; -use crate::notify::types::Method; -use crate::notify::types::MethodStatus; use crate::notify::types::DEFAULT_NOTIFY_PARAMS; -use crate::notify::types::NotificationEvent; use crate::notify::types::NotificationEventReceiver; use crate::notify::types::NotificationEventSender; -use crate::notify::types::NotificationPluginMap; -use crate::notify::types::NotificationRenderer; -use crate::notify::types::NotificationRule; -use crate::notify::types::NotificationRulesMap; -use crate::notify::types::ParsedTrigger; +use crate::notify::handler::NotificationHandler; use crate::types::resource::ResourceRegistry; -use chrono::Utc; -use globset::{Glob, GlobSetBuilder}; -use std::collections::HashMap; -use std::collections::HashSet; use std::path::Path; +use std::sync::Arc; use workshop_engine::{EventSender, ExecutionContext}; mod error; +pub mod handler; pub mod queue; +pub mod rule; pub mod template; pub mod types; -use error::NotifyError; - -/// Maximum depth for template cross-references (prevent infinite recursion) -const TEMPLATE_REFERENCE_DEPTH: u32 = 10; - -fn validate(finalize: &FinalizeConfig) -> anyhow::Result<()> { - let validate_result = finalize.validate(); - if let Err(errors) = validate_result { - for error in &errors { - log::error!("Error: {}", error); - } - anyhow::bail!(NotifyError::FinalizeValidationError(errors)) - }; - Ok(()) -} - -fn extract_plugin(finalize: &FinalizeConfig) -> HashSet { - finalize - .notification - .groups - .values() - .flat_map(|methods| methods.keys()) - .cloned() - .collect() -} +pub mod util; +use rule::parse_rules; +use template::{flatten_template_refs, resolve_external_template_refs}; +use util::{extract_plugin, validate}; pub async fn notify( finalize_path: &Path, @@ -60,130 +26,53 @@ pub async fn notify( _event_tx: EventSender, mut rx: NotificationEventReceiver, retry_tx: NotificationEventSender, - debug: bool, + _debug: bool, registry: &ResourceRegistry, ) -> anyhow::Result<()> { let mut finalize = parse(finalize_path)?; - validate(&finalize)?; - let mut plugin_list = extract_plugin(&finalize); - if plugin_list.is_empty() { - if debug { - log::debug!("[NOTIFY_DEBUG] Injecting dummy plugin."); - plugin_list.insert("dummy".to_string()); - // Insert a dummy method so the notification system has at least one entry - let mut methods = std::collections::HashMap::new(); - methods.insert( - "dummy".to_string(), - crate::finalize::NotificationMethod::default(), - ); - finalize.notification.groups.insert(0, methods); - } else { - log::warn!("No notification method provided, skipping notification"); - return Ok(()); - } + // Always inject in-site-notify as the lowest priority fallback + { + let max_p = finalize.notification.groups.keys().max().copied().unwrap_or(0); + let mut m = std::collections::HashMap::new(); + m.insert( + "in-site-notify".to_string(), + crate::finalize::NotificationMethod::default(), + ); + finalize.notification.groups.insert(max_p + 1, m); } + validate(&finalize)?; + let plugin_list = extract_plugin(&finalize); + let notify_plugins = plugin::register( finalize.plugin, Some(&plugin_list), &mut finalize.notification, registry, - )?; +)?; - // Resolve external template references from registry (pre-fetched) - for (name, tmpl_def) in finalize.notification.templates.iter_mut() { - if tmpl_def.use_.is_some() { - if let Some(path) = registry.resolve(name) { - let content = std::fs::read_to_string(path.join("template.yml")) - .or_else(|_| std::fs::read_to_string(path.join("template.yaml"))); - if let Ok(content) = content { - if let Ok(fetched) = - serde_yaml::from_str::(&content) - { - tmpl_def.on_success = fetched.on_success.or(tmpl_def.on_success.take()); - tmpl_def.on_failure = fetched.on_failure.or(tmpl_def.on_failure.take()); - tmpl_def.on_finish_of = - fetched.on_finish_of.or(tmpl_def.on_finish_of.take()); - tmpl_def.use_ = None; // Resolved - } - } - } - } - } - - // Flatten local template cross-references + resolve_external_template_refs(&mut finalize.notification.templates, registry); flatten_template_refs(&mut finalize.notification.templates)?; let notification_rules = parse_rules(&finalize.notification); + let handler = Arc::new(NotificationHandler::new( + Arc::new(notification_rules), + Arc::new(notify_plugins), + Arc::new(finalize.notification.templates), + Arc::new(ctx.clone()), + retry_tx, + DEFAULT_NOTIFY_PARAMS.batch_period.value, + DEFAULT_NOTIFY_PARAMS.batch_watermark.value, + )); + loop { - // Wait for first event - let first = match rx.recv().await { - Some(event) => event, - None => break, - }; - let mut events = vec![first]; - - // Drain any additional events already in the channel (non-blocking micro-batch) - while let Ok(event) = rx.try_recv() { - events.push(event); - } - - if debug { - for ev in &events { - log::info!( - "[NOTIFY_DEBUG] stage={:?} substage={} outcome={:?} priority={}", - ev.stage, - ev.substage, - ev.outcome, - ev.priority - ); - } - } - - let results = notification_handler( - ¬ify_plugins, - ¬ification_rules, - &events, - &finalize.notification.templates, - ctx, - ) - .await; - - let mut recoverable: Vec = vec![]; - let mut unrecoverable: Vec<(String, bool)> = vec![]; - - for (method, status) in &results { - match status { - MethodStatus::Success => {} - MethodStatus::TemporaryError(_) => recoverable.push(method.clone()), - MethodStatus::PermanentError(_) | MethodStatus::Exhausted(_) => { - let fallible = get_fallible(¬ification_rules, method); - unrecoverable.push((method.clone(), fallible)); - } - } - } - - if !recoverable.is_empty() { - let retry_methods = recoverable.clone(); - events[0].target.rules = Some(retry_methods); - events[0].target.count += 1; - events[0].not_before = Utc::now() + notify_delay(&events[0]); - retry_tx.send(events[0].clone()).await?; - continue; - } - - let failed = unrecoverable.iter().any(|(_, fallible)| !fallible); - if failed { - events[0].target.priority += 1; - events[0].target.rules = None; - events[0].attempted = events[0].target.count; - events[0].target.count = 0; - events[0].effective_priority = events[0].priority; - retry_tx.send(events[0].clone()).await?; - continue; - } + let Some(event) = rx.recv().await else { break }; + let h = handler.clone(); + tokio::spawn(async move { + h.process_event(event).await; + }); } Ok(()) } @@ -192,581 +81,6 @@ pub async fn notify( // TODO: PIPELINE_NOTIFICATION_FAILED — set when any method returns PermanentError // Markers sent via event_tx as ExecutionEvent variants for future WebUI/CLI display. -/// Resolve a schema reference into a NotificationTemplateDef by walking a fallback chain. -/// -/// The chain is defined by resolve_template_next_schema(): each entry point prescribes -/// a sequence of schema names to try. Each step calls resolve_template_by_schema() to do -/// the actual lookup; the first hit is returned as Ok(...). If the chain is exhausted -/// without a hit, a descriptive SchemaError is returned. -/// -/// Behaviour difference between standalone and daemon modes is implicit in -/// server_templates (None in standalone, Some in daemon). No explicit mode check needed. -fn resolve_template( - schema: &Option, - method_name: &str, - templates: &std::collections::HashMap, - _standalone: bool, - server_templates: Option< - &std::collections::HashMap, - >, -) -> Result { - // 1. Try the initial schema directly - if let Some(schema_name) = schema.as_deref() { - if let Some(template) = - resolve_template_by_schema(schema_name, method_name, templates, server_templates) - { - return Ok(template); - } - } - - // 2. Walk the fallback chain - let mut schema_next = resolve_template_next_schema(schema); - while let Some(ref schema_name) = schema_next { - if let Some(template) = - resolve_template_by_schema(schema_name, method_name, templates, server_templates) - { - return Ok(template); - } - log::warn!("Failed to resolve template for schema {}.", schema_name); - schema_next = resolve_template_next_schema(&schema_next); - } - Err(NotifyError::SchemaError(format!("All template resolution attempts beginning from {} failed.", schema.as_deref().unwrap_or("[system default]")))) -} - -/// Return the next schema name in the fallback chain, or None to terminate. -/// -/// This is the transition function of a state machine driven by resolve_template(). -/// Each state (schema name) maps to a next attempt: -/// -/// | current schema | next schema | why | -/// |---|---:|------| -/// | `"custom"` | None | Schema says "plugin does its own rendering". Not really a template resolution. | -/// | `"default"` | `"fallback"` | System default not found → try plugin's own built-in template. | -/// | `"fallback"` | None | Last resort exhausted. | -/// | `""` | `"default"` | User-named template not found → fall back to system default. | -/// | `None` (unset) | `"default"` | No explicit schema → start from system default. | -fn resolve_template_next_schema(schema: &Option) -> Option { - match schema.as_deref() { - Some("custom") => None, - Some("default") => Some("fallback".to_string()), - Some("fallback") => None, - Some(_) => Some("default".to_string()), - None => Some("default".to_string()), - } -} - -/// Perform a single lookup for a given schema name. Returns None if not found. -/// -/// This is the action function of the state machine — it has no knowledge of the -/// fallback chain and leaves progression to resolve_template(). -/// -/// | schema | lookup target | returns | -/// |--------------|-----------------------------------------------|---------| -/// | `"custom"` | — | empty default (plugin self-renders) | -/// | `"fallback"` | templates[`"{method}.fallback"`] | plugin's built-in template | -/// | `"default"` | server_templates[method_name] | bakerd's system default | -/// | other name | templates[name] | user-defined template | -fn resolve_template_by_schema( - schema: &str, - method_name: &str, - templates: &std::collections::HashMap, - server_templates: Option< - &std::collections::HashMap, - >, -) -> Option { - match schema { - "custom" => Some(crate::finalize::NotificationTemplateDef::default()), - "fallback" => templates.get(&format!("{}.fallback", method_name)).cloned(), - "default" => server_templates.and_then(|b| b.get(method_name)).cloned(), - name => templates.get(name).cloned(), - } -} - -/// Flatten all local template cross-references (use: "other-name"). -/// External fetch URLs (containing ://) are NOT processed here. -/// Must be called at registration/validation time, not at runtime. -fn flatten_template_refs( - templates: &mut std::collections::HashMap, -) -> Result<(), NotifyError> { - let names: Vec = templates.keys().cloned().collect(); - for name in names { - let mut visited = std::collections::HashSet::new(); - resolve_chain(&name, templates, 0, &mut visited)?; - } - Ok(()) -} - -fn resolve_chain( - name: &str, - templates: &mut std::collections::HashMap, - depth: u32, - visited: &mut std::collections::HashSet, -) -> Result<(), NotifyError> { - if depth > TEMPLATE_REFERENCE_DEPTH { - return Err(NotifyError::TemplateDepthExceeded { - name: name.to_string(), - max: TEMPLATE_REFERENCE_DEPTH, - }); - } - if !visited.insert(name.to_string()) { - return Err(NotifyError::TemplateLoop(name.to_string())); - } - - // Only process local references (no ://) - if let Some(ref_name) = templates.get(name).and_then(|t| t.use_.clone()) { - if !ref_name.contains("://") { - // Check referenced template exists - if !templates.contains_key(&ref_name) { - return Err(NotifyError::SchemaError(format!( - "template '{}' references non-existent template '{}'", - name, ref_name - ))); - } - // Recursively resolve the referenced template first - resolve_chain(&ref_name, templates, depth + 1, visited)?; - - // Merge: base provides defaults, referencing template overrides - let base = templates[&ref_name].clone(); - let template = templates.get_mut(name).unwrap(); - *template += base; - template.use_ = None; // Chain flattened - } - } - - visited.remove(name); - Ok(()) -} - -fn is_temporary_error(e: &crate::finalize::error::PluginError) -> bool { - !matches!(e, crate::finalize::error::PluginError::GeneralError(_)) -} - -async fn notification_handler( - plugins: &NotificationPluginMap, - config: &NotificationRulesMap, - events: &[NotificationEvent], - templates: &std::collections::HashMap, - ctx: &ExecutionContext, -) -> Vec<(String, MethodStatus)> { - let mut results: Vec<(String, MethodStatus)> = vec![]; - log::info!( - "[NOTIFY] handler called with {} events, {} priority groups", - events.len(), - config.len() - ); - - let mut buckets: std::collections::HashMap<(String, usize), Vec<&NotificationEvent>> = - std::collections::HashMap::new(); - for (_priority, ruleset) in config.iter() { - for (method, rules) in ruleset.iter() { - log::info!("[NOTIFY] method={} rules_count={}", method, rules.len()); - for event in events { - if let Some(ref method_rules) = event.target.rules { - if !method_rules.contains(method) { - continue; - } - } - for (rule_id, rule) in rules.iter().enumerate() { - log::info!( - "[NOTIFY] rule_id={} enabled={} trigger_len={}", - rule_id, - rule.enabled, - rule.trigger.len() - ); - log::trace!("[CHK-ENABLED] entering enabled check"); - if !rule.enabled { - log::trace!("[CHK-ENABLED] enabled=false, continuing"); - continue; - } - log::trace!("[BEFORE-ACCEPT] method={} rule={} event={}.{}", - method, rule_id, event.stage, event.substage); - if rule.accept(&event.outcome, &event.stage, &event.substage) { - log::info!( - "[NOTIFY] MATCH: method={} rule={} event={}.{}", - method, - rule_id, - event.stage, - event.substage - ); - buckets - .entry((method.clone(), rule_id)) - .or_default() - .push(event); - } - } - } - } - } - - if buckets.is_empty() { - return results; - } - - log::info!("[NOTIFY] processing {} buckets", buckets.len()); - for ((method, rule_id), bucket_events) in buckets { - let rule = match find_rule(config, &method, rule_id) { - Ok(r) => r, - Err(e) => { - results.push(( - method.clone(), - MethodStatus::PermanentError(format!("{}", e)), - )); - continue; - } - }; - - let plugin = match plugins.get(&method) { - Some(p) => p, - None => { - results.push(( - method.clone(), - MethodStatus::PermanentError(format!("Unknown plugin: {}", method)), - )); - continue; - } - }; - - let renderer = plugin.metadata.renderer.clone().unwrap_or_default(); - let bucket: Vec = bucket_events.into_iter().cloned().collect(); - - let argument = match renderer { - NotificationRenderer::Server => { - let template_def = match resolve_template( - &rule.template.schema, - &method, - templates, - ctx.standalone, - None, - ) { - Ok(td) => td, - Err(e) => { - results.push(( - method.clone(), - MethodStatus::PermanentError(format!("{}", e)), - )); - continue; - } - }; - let first = &bucket[0]; - let template_str = match crate::notify::template::select_template_str( - &template_def, - &first.outcome, - ) - .or_else(|| rule.template.on_success.as_deref()) - .or_else(|| rule.template.on_failure.as_deref()) - .or_else(|| { - rule.template - .on_finish_of - .as_ref() - .map(|o| o.content.as_str()) - }) { - Some(ts) => ts, - None => { - results.push(( - method.clone(), - MethodStatus::PermanentError(format!( - "no template for outcome {:?}", - first.outcome - )), - )); - continue; - } - }; - let data = crate::notify::template::build_batch_template_context(ctx, &bucket); - let rendered = match crate::notify::template::render_template(template_str, &data) { - Ok(r) => r, - Err(e) => { - results.push((method.clone(), MethodStatus::PermanentError(e))); - continue; - } - }; - - let mut argument = rule.config.clone(); - if let serde_yaml::Value::Mapping(ref mut map) = argument { - map.insert( - serde_yaml::Value::String("_rendered_content".into()), - serde_yaml::Value::String(rendered), - ); - } - - log::info!("SSR: rendered content for method '{}'", method); - argument - } - - NotificationRenderer::Plugin => { - let mut argument = rule.config.clone(); - - if let Ok(template_def) = resolve_template( - &rule.template.schema, - &method, - templates, - ctx.standalone, - None, - ) { - if let serde_yaml::Value::Mapping(ref mut map) = argument { - if let Ok(tmpl_value) = serde_yaml::to_value(&template_def) { - map.insert(serde_yaml::Value::String("_template".into()), tmpl_value); - } - } - } - - if let serde_yaml::Value::Mapping(ref mut map) = argument { - let events_data: Vec = bucket - .iter() - .map(|e| { - serde_json::json!({ - "stage": format!("{}.{}", e.stage.as_str(), e.substage), - "status": e.outcome.to_string(), - }) - }) - .collect(); - map.insert( - serde_yaml::Value::String("_events".into()), - serde_yaml::Value::String( - serde_json::to_string(&events_data).unwrap_or_default(), - ), - ); - } - - log::info!("CSR: prepared argument for plugin '{}'", method); - argument - } - - NotificationRenderer::Interactive => { - unimplemented!("Interactive renderer is not yet implemented") - } - - }; - - let remain = rule.max_lives as f64 - - DEFAULT_NOTIFY_PARAMS.life_impact_factor.value * events[0].attempted as f64 - - events[0].target.count as f64; - - if remain < 1.0 { - results.push(( - method.clone(), - MethodStatus::Exhausted(format!("max_lives depleted")), - )); - continue; - } - - match tokio::time::timeout(rule.timeout, plugin.call(argument, ctx, &None)).await { - Ok(Ok(())) => { - results.push((method.clone(), MethodStatus::Success)); - } - Ok(Err(e)) => { - let status = if is_temporary_error(&e) { - MethodStatus::TemporaryError(format!("{}", e)) - } else { - MethodStatus::PermanentError(format!("{}", e)) - }; - results.push((method.clone(), status)); - } - Err(_) => { - results.push(( - method.clone(), - MethodStatus::TemporaryError("timeout".into()), - )); - } - } - } - - results -} - -/// Parse a single `on_finish_of` pattern string into a ParsedTrigger, or None. -/// Supports `re:` prefix for regex patterns, otherwise treated as glob. -fn parse_on_finish_of(pattern: &str) -> Option { - let pattern = pattern.trim(); - if let Some(re_str) = pattern.strip_prefix("re:") { - regex::Regex::new(re_str).ok().map(|r| ParsedTrigger::OnFinishOf(Matchable::Regex(r))) - } else { - Glob::new(pattern).ok().and_then(|g| { - let mut builder = GlobSetBuilder::new(); - builder.add(g); - builder.build().ok() - }).map(|gs| ParsedTrigger::OnFinishOf(Matchable::Glob(gs))) - } -} - -fn parse_trigger_item_raw(item: &TriggerItemRaw) -> Vec { - // Normalize Simple → single-entry Vec, then unify both branches - let entries: Vec<(&str, Vec<&str>)> = match item { - TriggerItemRaw::Simple(s) => { - if let Some(pattern) = s.strip_prefix("on_finish_of:") { - vec![("on_finish_of", vec![pattern.trim()])] - } else { - vec![(s.as_str(), vec![])] - } - } - TriggerItemRaw::Map(map) => map.iter().map(|(k, v)| - (k.as_str(), v.iter().map(|s| s.as_str()).collect()) - ).collect(), - }; - - let mut triggers = Vec::new(); - for (key, patterns) in entries { - match key { - "on_finish_of" => { - for pattern in patterns { - if let Some(t) = parse_on_finish_of(pattern) { - triggers.push(t); - } - } - } - "on_success" => triggers.push(ParsedTrigger::OnSuccess), - "on_failure" => triggers.push(ParsedTrigger::OnFailure), - _ => log::warn!("Unknown trigger key: {}", key), - } - } - triggers -} - -fn merge_templates( - method_template: &NotificationTemplate, - policy_template: &Option, -) -> NotificationTemplate { - let mut merged = method_template.clone(); - if let Some(policy_tmpl) = policy_template - && let Ok(policy) = serde_yaml::from_value::(policy_tmpl.clone()) - { - merged += policy; - } - merged -} - -struct RuleOverrides { - fallible: bool, - max_lives: u32, - timeout: std::time::Duration, - retry_backoff: f64, -} - -impl RuleOverrides { - fn from_value(v: &serde_yaml::Value) -> Self { - RuleOverrides { - fallible: v.get("fallible").and_then(|v| v.as_bool()).unwrap_or(false), - max_lives: v.get("max_lives").and_then(|v| v.as_u64()).unwrap_or(DEFAULT_NOTIFY_PARAMS.max_lives.value as u64) as u32, - timeout: std::time::Duration::from_millis( - v.get("timeout_ms").and_then(|v| v.as_u64()).unwrap_or(DEFAULT_NOTIFY_PARAMS.timeout_ms.value) - ), - retry_backoff: v.get("retry_backoff").and_then(|v| v.as_f64()).unwrap_or(DEFAULT_NOTIFY_PARAMS.retry_backoff.value), - } - } - - /// Chain: apply `source` atop existing values (source wins when present). - fn apply(self, source: &serde_yaml::Value) -> Self { - RuleOverrides { - fallible: source.get("fallible").and_then(|v| v.as_bool()).unwrap_or(self.fallible), - max_lives: source.get("max_lives").and_then(|v| v.as_u64()).unwrap_or(self.max_lives as u64) as u32, - timeout: source.get("timeout_ms").and_then(|v| v.as_u64()).map(std::time::Duration::from_millis).unwrap_or(self.timeout), - retry_backoff: source.get("retry_backoff").and_then(|v| v.as_f64()).unwrap_or(self.retry_backoff), - } - } -} - -fn parse_rules(config: &NotificationConfig) -> NotificationRulesMap { - log::info!("[NOTIFY] parse_rules: groups={}", config.groups.len()); - let mut rules = NotificationRulesMap::default(); - for (priority, group) in &config.groups { - let mut ruleset: HashMap> = HashMap::new(); - for (method_name, method) in group { - let mut method_rules = Vec::new(); - - // Unify: no policy → one empty policy (same merge path) - let empty_policy = crate::finalize::config::NotifyPolicy { - trigger: vec![], - template: serde_yaml::Value::Null, - overrides: serde_yaml::Value::Null, - }; - let policies: &[crate::finalize::config::NotifyPolicy] = if method.policy.is_empty() { - std::slice::from_ref(&empty_policy) - } else { - &method.policy - }; - - for policy in policies { - let trigger: Vec = if policy.trigger.is_empty() { - method.trigger.iter().flat_map(parse_trigger_item_raw).collect() - } else { - policy.trigger.iter().flat_map(parse_trigger_item_raw).collect() - }; - let template = - merge_templates(&method.template, &Some(policy.template.clone())); - let vals = RuleOverrides::from_value(&serde_yaml::Value::Null) - .apply(&method.config) - .apply(&policy.overrides); - method_rules.push(NotificationRule { - enabled: true, - trigger, - template, - fallible: vals.fallible, - max_lives: vals.max_lives, - timeout: vals.timeout, - config: method.config.clone(), - retry_backoff: vals.retry_backoff, - }); - } - - ruleset.insert(method_name.clone(), method_rules); - } - rules.insert(*priority, ruleset); - } - rules -} - -// calculate min(1, k/x)*t as notification grouping period -// to avoid excessive notification frequency -fn wait_time(group_period: u32, group_watermark: u32, count: u32) -> f32 { - if count == 0 { - log::warn!("Unexpected notification count: {}", count); - return group_period as f32; - } - f32::min(1.0, group_watermark as f32 / count as f32) * group_period as f32 -} - -fn notify_delay(event: &NotificationEvent) -> std::time::Duration { - let exponent = event.target.count as i32; - if exponent < 0 { - log::warn!("Unexpected count: {}", event.target.count); - return std::time::Duration::from_secs_f64(DEFAULT_NOTIFY_PARAMS.max_retry_delay_seconds.value); - } - let factor = if event.retry_backoff > 0.0 { - event.retry_backoff - } else { - DEFAULT_NOTIFY_PARAMS.retry_backoff.value - }; - let multiplier = factor.powi(exponent); - let delay = multiplier * DEFAULT_NOTIFY_PARAMS.retry_delay_seconds.value; - if !delay.is_finite() || delay < 0.0 { - return std::time::Duration::from_secs_f64(DEFAULT_NOTIFY_PARAMS.max_retry_delay_seconds.value); - } - std::time::Duration::from_secs_f64(delay) -} -fn find_rule<'a>( - config: &'a NotificationRulesMap, - method: &str, - rule_id: usize, -) -> Result<&'a NotificationRule, NotifyError> { - for ruleset in config.values() { - if let Some(rules) = ruleset.get(method) { - if let Some(rule) = rules.get(rule_id) { - return Ok(rule); - } - } - } - Err(NotifyError::SchemaError("rule not found".into())) -} - -fn get_fallible(config: &NotificationRulesMap, method: &str) -> bool { - for ruleset in config.values() { - if let Some(rules) = ruleset.get(method) { - return rules.iter().all(|r| r.fallible); - } - } - false -} - #[cfg(test)] mod tests { use super::*; @@ -774,6 +88,11 @@ mod tests { use crate::notify::types::NotificationRule; use crate::notify::types::NotificationTarget; use std::collections::HashMap; + use crate::notify::types::NotificationEvent; + use crate::notify::types::NotificationRulesMap; + use crate::notify::error::NotifyError; + use crate::notify::template::resolve_template; + use crate::notify::util::{find_rule, notify_delay, wait_time}; // ── helper ── diff --git a/workshop-baker/src/notify/error.rs b/workshop-baker/src/notify/error.rs index 7764b17..0019722 100644 --- a/workshop-baker/src/notify/error.rs +++ b/workshop-baker/src/notify/error.rs @@ -1,4 +1,4 @@ -use crate::notify::NotificationRenderer; +use crate::notify::types::NotificationRenderer; use thiserror::Error; #[derive(Error, Debug)] diff --git a/workshop-baker/src/notify/handler.rs b/workshop-baker/src/notify/handler.rs new file mode 100644 index 0000000..38a7407 --- /dev/null +++ b/workshop-baker/src/notify/handler.rs @@ -0,0 +1,345 @@ +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use chrono::Utc; +use tokio::sync::Notify; + +use crate::finalize::NotificationTemplateDef; +use crate::notify::template::{ + build_batch_template_context, render_template, resolve_template, + select_template_str, +}; +use crate::notify::types::{ + MethodStatus, NotificationEvent, + NotificationEventSender, NotificationPluginMap, NotificationRenderer, + NotificationRulesMap, +}; +use crate::notify::util::{ + find_rule, get_fallible, is_temporary_error, notify_delay, wait_time, +}; + +/// Shared state for a single dispatch bucket. +struct BucketEntry { + events: Arc>>, + result: Arc)>>>, + done: Arc, + notify: Arc, + method: String, + rule_id: usize, + _handle: tokio::task::JoinHandle<()>, +} + +/// Concurrent notification handler with bucket-based dispatch. +/// +/// Events are routed to matching buckets. Buckets cool down and dispatch +/// plugins concurrently. The caller awaits bucket completion, then handles +/// retry/escalation decisions. +pub struct NotificationHandler { + buckets: Mutex>, + pub config: Arc, + plugins: Arc, + templates: Arc>, + ctx: Arc, + retry_tx: NotificationEventSender, + period: u32, + watermark: u32, +} + +impl NotificationHandler { + #[allow(clippy::too_many_arguments)] + pub fn new( + config: Arc, + plugins: Arc, + templates: Arc>, + ctx: Arc, + retry_tx: NotificationEventSender, + period: u32, + watermark: u32, + ) -> Self { + NotificationHandler { + buckets: Mutex::new(HashMap::new()), + config, + plugins, + templates, + ctx, + retry_tx, + period, + watermark, + } + } + + /// Process a single event: route to buckets, await completion, + /// then handle retry/escalation for each bucket result. + pub async fn process_event(self: Arc, event: NotificationEvent) { + let entries = self.route_event(&event); + if entries.is_empty() { + return; + } + + // Await all buckets — yields CPU + for (_, notify) in &entries { + notify.notified().await; + } + + // Collect results and remove entries + let mut results: Vec<(String, usize, Option<(MethodStatus, Vec)>)> = + Vec::new(); + { + let mut buckets = self.buckets.lock().unwrap(); + for (key, _) in &entries { + if let Some(entry) = buckets.remove(key) { + let r = entry.result.lock().unwrap().take(); + results.push((entry.method, entry.rule_id, r)); + } + } + } + + // Handle retry/escalation per method + for (method, _rule_id, result) in results { + let Some((status, mut events)) = result else { continue }; + + match &status { + MethodStatus::Success => { + log::debug!("[NOTIFY] {}: success ({} events)", method, events.len()); + } + MethodStatus::TemporaryError(_) => { + log::warn!("[NOTIFY] {}: temporary error, retrying", method); + for event in events.iter_mut() { + event.target.rules = Some(vec![method.clone()]); + event.target.count += 1; + event.not_before = Utc::now() + notify_delay(event); + let _ = self.retry_tx.send(event.clone()).await; + } + } + MethodStatus::PermanentError(_) | MethodStatus::Exhausted(_) => { + let fallible = get_fallible(&self.config, &method); + if fallible { + log::error!("[NOTIFY] {}: discard (fallible)", method); + } else { + log::warn!("[NOTIFY] {}: escalating priority", method); + for event in events.iter_mut() { + event.target.priority += 1; + event.target.rules = None; + event.attempted = event.target.count; + event.target.count = 0; + event.effective_priority = event.priority; + let _ = self.retry_tx.send(event.clone()).await; + } + } + } + } + } + } + + /// Route a single event to all matching buckets. + /// + /// Returns Notify handles for awaiting bucket completion. + fn route_event( + &self, + event: &NotificationEvent, + ) -> Vec<((String, usize), Arc)> { + let mut matches: Vec<(String, usize)> = Vec::new(); + + for ruleset in self.config.values() { + for (method, rules) in ruleset.iter() { + if let Some(ref method_rules) = event.target.rules { + if !method_rules.contains(method) { + continue; + } + } + for (rule_id, rule) in rules.iter().enumerate() { + if !rule.enabled { + continue; + } + if rule.accept(&event.outcome, &event.stage, &event.substage) { + matches.push((method.clone(), rule_id)); + } + } + } + } + + let mut buckets = self.buckets.lock().unwrap(); + let mut result: Vec<((String, usize), Arc)> = Vec::new(); + + for key in matches { + if let Some(entry) = buckets.get(&key) { + if entry.done.load(Ordering::Acquire) { + buckets.remove(&key); + } + } + + if let Some(entry) = buckets.get(&key) { + entry.events.lock().unwrap().push(event.clone()); + result.push((key, entry.notify.clone())); + } else { + let (entry, notify) = Self::spawn_bucket( + &key.0, + key.1, + &self.config, + &self.plugins, + &self.templates, + &self.ctx, + self.period, + self.watermark, + ); + entry.events.lock().unwrap().push(event.clone()); + result.push((key.clone(), notify)); + buckets.insert(key, entry); + } + } + + result + } + + /// Spawn a bucket task: cooldown → take → dispatch → store → notify. + fn spawn_bucket( + method: &str, + rule_id: usize, + config: &Arc, + plugins: &Arc, + templates: &Arc>, + ctx: &Arc, + period: u32, + watermark: u32, + ) -> (BucketEntry, Arc) { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let result: Arc)>>> = + Arc::new(Mutex::new(None)); + let notify = Arc::new(Notify::new()); + let done = Arc::new(AtomicBool::new(false)); + + let ec = events.clone(); + let rc = result.clone(); + let nc = notify.clone(); + let dc = done.clone(); + let m = method.to_string(); + let c = config.clone(); + let p = plugins.clone(); + let t = templates.clone(); + let x = ctx.clone(); + + let handle = tokio::spawn(async move { + // Phase 1: Cooldown + let count = ec.lock().unwrap().len() as u32; + let wait_secs = wait_time(period, watermark, count); + tokio::time::sleep(Duration::from_secs_f32(wait_secs)).await; + + // Phase 2: Take events + let batch = std::mem::take(&mut *ec.lock().unwrap()); + if batch.is_empty() { + dc.store(true, Ordering::Release); + nc.notify_waiters(); + return; + } + + // Phase 3: Dispatch + let rule = match find_rule(&c, &m, rule_id) { + Ok(r) => r, + Err(_) => { + *rc.lock().unwrap() = Some((MethodStatus::PermanentError("rule not found".into()), batch)); + dc.store(true, Ordering::Release); + nc.notify_waiters(); + return; + } + }; + + let plugin = match p.get(&m) { + Some(p) => p, + None => { + *rc.lock().unwrap() = Some((MethodStatus::PermanentError("plugin not found".into()), batch)); + dc.store(true, Ordering::Release); + nc.notify_waiters(); + return; + } + }; + + let renderer = plugin.metadata.renderer.clone().unwrap_or_default(); + let argument = build_plugin_argument(&renderer, rule, &m, &t, &x, &batch); + let status = match argument { + Some(arg) => { + match tokio::time::timeout(rule.timeout, plugin.call(arg, &x, &None)).await { + Ok(Ok(())) => MethodStatus::Success, + Ok(Err(e)) if is_temporary_error(&e) => { + MethodStatus::TemporaryError(format!("{}", e)) + } + Ok(Err(e)) => MethodStatus::PermanentError(format!("{}", e)), + Err(_) => MethodStatus::TemporaryError("timeout".into()), + } + } + None => MethodStatus::PermanentError("argument build failed".into()), + }; + + *rc.lock().unwrap() = Some((status, batch)); + dc.store(true, Ordering::Release); + nc.notify_waiters(); + }); + + let entry = BucketEntry { + events, + result, + done, + notify: notify.clone(), + method: method.to_string(), + rule_id, + _handle: handle, + }; + + (entry, notify) + } +} + +/// Build the YAML argument for a plugin call based on renderer mode. +fn build_plugin_argument( + renderer: &NotificationRenderer, + rule: &crate::notify::types::NotificationRule, + method: &str, + templates: &HashMap, + ctx: &workshop_engine::ExecutionContext, + events: &[NotificationEvent], +) -> Option { + let first = events.first()?; + + match renderer { + NotificationRenderer::Server => { + let template_def = resolve_template( + &rule.template.schema, + method, + templates, + ctx.standalone, + None, + ) + .ok()?; + + let template_str = select_template_str(&template_def, &first.outcome) + .or_else(|| rule.template.on_success.as_deref()) + .or_else(|| rule.template.on_failure.as_deref()) + .or_else(|| { + rule.template + .on_finish_of + .as_ref() + .map(|o| o.content.as_str()) + })?; + + let data = build_batch_template_context(ctx, events); + let rendered = render_template(template_str, &data).ok()?; + + let mut argument = rule.config.clone(); + if let serde_yaml::Value::Mapping(ref mut map) = argument { + map.insert( + serde_yaml::Value::String("_rendered_content".into()), + serde_yaml::Value::String(rendered), + ); + } + log::info!("SSR: rendered content for method '{}'", method); + Some(argument) + } + NotificationRenderer::Plugin => { + unimplemented!("Plugin renderer is not yet implemented") + } + NotificationRenderer::Interactive => { + unimplemented!("Interactive renderer is not yet implemented") + } + } +} diff --git a/workshop-baker/src/notify/rule.rs b/workshop-baker/src/notify/rule.rs new file mode 100644 index 0000000..28aa584 --- /dev/null +++ b/workshop-baker/src/notify/rule.rs @@ -0,0 +1,148 @@ +use crate::finalize::NotificationConfig; +use crate::finalize::NotificationTemplate; +use crate::finalize::config::{NotifyPolicy, TriggerItemRaw}; +use crate::notify::types::{DEFAULT_NOTIFY_PARAMS, Matchable, Method, NotificationRule, NotificationRulesMap, ParsedTrigger}; +use globset::{Glob, GlobSetBuilder}; +use std::collections::HashMap; + +/// Parse a single `on_finish_of` pattern string into a ParsedTrigger, or None. +/// Supports `re:` prefix for regex patterns, otherwise treated as glob. +fn parse_on_finish_of(pattern: &str) -> Option { + let pattern = pattern.trim(); + if let Some(re_str) = pattern.strip_prefix("re:") { + regex::Regex::new(re_str).ok().map(|r| ParsedTrigger::OnFinishOf(Matchable::Regex(r))) + } else { + Glob::new(pattern).ok().and_then(|g| { + let mut builder = GlobSetBuilder::new(); + builder.add(g); + builder.build().ok() + }).map(|gs| ParsedTrigger::OnFinishOf(Matchable::Glob(gs))) + } +} + +fn parse_trigger_item_raw(item: &TriggerItemRaw) -> Vec { + // Normalize Simple → single-entry Vec, then unify both branches + let entries: Vec<(&str, Vec<&str>)> = match item { + TriggerItemRaw::Simple(s) => { + if let Some(pattern) = s.strip_prefix("on_finish_of:") { + vec![("on_finish_of", vec![pattern.trim()])] + } else { + vec![(s.as_str(), vec![])] + } + } + TriggerItemRaw::Map(map) => map.iter().map(|(k, v)| + (k.as_str(), v.iter().map(|s| s.as_str()).collect()) + ).collect(), + }; + + let mut triggers = Vec::new(); + for (key, patterns) in entries { + match key { + "on_finish_of" => { + for pattern in patterns { + if let Some(t) = parse_on_finish_of(pattern) { + triggers.push(t); + } + } + } + "on_success" => triggers.push(ParsedTrigger::OnSuccess), + "on_failure" => triggers.push(ParsedTrigger::OnFailure), + _ => log::warn!("Unknown trigger key: {}", key), + } + } + triggers +} + +fn merge_templates( + method_template: &NotificationTemplate, + policy_template: &Option, +) -> NotificationTemplate { + let mut merged = method_template.clone(); + if let Some(policy_tmpl) = policy_template + && let Ok(policy) = serde_yaml::from_value::(policy_tmpl.clone()) + { + merged += policy; + } + merged +} + +struct RuleOverrides { + fallible: bool, + max_lives: u32, + timeout: std::time::Duration, + retry_backoff: f64, +} + +impl RuleOverrides { + fn from_value(v: &serde_yaml::Value) -> Self { + RuleOverrides { + fallible: v.get("fallible").and_then(|v| v.as_bool()).unwrap_or(false), + max_lives: v.get("max_lives").and_then(|v| v.as_u64()).unwrap_or(DEFAULT_NOTIFY_PARAMS.max_lives.value as u64) as u32, + timeout: std::time::Duration::from_millis( + v.get("timeout_ms").and_then(|v| v.as_u64()).unwrap_or(DEFAULT_NOTIFY_PARAMS.timeout_ms.value) + ), + retry_backoff: v.get("retry_backoff").and_then(|v| v.as_f64()).unwrap_or(DEFAULT_NOTIFY_PARAMS.retry_backoff.value), + } + } + + /// Chain: apply `source` atop existing values (source wins when present). + fn apply(self, source: &serde_yaml::Value) -> Self { + RuleOverrides { + fallible: source.get("fallible").and_then(|v| v.as_bool()).unwrap_or(self.fallible), + max_lives: source.get("max_lives").and_then(|v| v.as_u64()).unwrap_or(self.max_lives as u64) as u32, + timeout: source.get("timeout_ms").and_then(|v| v.as_u64()).map(std::time::Duration::from_millis).unwrap_or(self.timeout), + retry_backoff: source.get("retry_backoff").and_then(|v| v.as_f64()).unwrap_or(self.retry_backoff), + } + } +} + +pub fn parse_rules(config: &NotificationConfig) -> NotificationRulesMap { + log::info!("[NOTIFY] parse_rules: groups={}", config.groups.len()); + let mut rules = NotificationRulesMap::default(); + for (priority, group) in &config.groups { + let mut ruleset: HashMap> = HashMap::new(); + for (method_name, method) in group { + let mut method_rules = Vec::new(); + + // Unify: no policy → one empty policy (same merge path) + let empty_policy = NotifyPolicy { + trigger: vec![], + template: serde_yaml::Value::Null, + overrides: serde_yaml::Value::Null, + }; + let policies: &[NotifyPolicy] = if method.policy.is_empty() { + std::slice::from_ref(&empty_policy) + } else { + &method.policy + }; + + for policy in policies { + let trigger: Vec = if policy.trigger.is_empty() { + method.trigger.iter().flat_map(parse_trigger_item_raw).collect() + } else { + policy.trigger.iter().flat_map(parse_trigger_item_raw).collect() + }; + let template = + merge_templates(&method.template, &Some(policy.template.clone())); + let vals = RuleOverrides::from_value(&serde_yaml::Value::Null) + .apply(&method.config) + .apply(&policy.overrides); + method_rules.push(NotificationRule { + enabled: true, + trigger, + template, + fallible: vals.fallible, + max_lives: vals.max_lives, + timeout: vals.timeout, + config: method.config.clone(), + retry_backoff: vals.retry_backoff, + }); + } + + ruleset.insert(method_name.clone(), method_rules); + } + rules.insert(*priority, ruleset); + } + rules +} + diff --git a/workshop-baker/src/notify/template.rs b/workshop-baker/src/notify/template.rs index d2187c9..922f311 100644 --- a/workshop-baker/src/notify/template.rs +++ b/workshop-baker/src/notify/template.rs @@ -1,10 +1,16 @@ -use minijinja::Environment; use std::collections::HashMap; +use minijinja::Environment; use workshop_engine::ExecutionContext; + use crate::finalize::NotificationTemplateDef; +use crate::notify::error::NotifyError; use crate::notify::types::NotificationEvent; use crate::types::buildstatus::StageOutcome; +use crate::types::resource::ResourceRegistry; + +/// Maximum depth for template cross-references (prevent infinite recursion) +pub const TEMPLATE_REFERENCE_DEPTH: u32 = 10; /// Build template context for a batch of notification events. pub fn build_batch_template_context( @@ -95,7 +101,142 @@ pub fn select_template_str<'a>( StageOutcome::PipelineSuccess => tmpl.on_success.as_deref(), StageOutcome::PipelineFailure => tmpl.on_failure.as_deref(), _ => None, + } } + +/// Resolve external template references from the resource registry. +pub fn resolve_external_template_refs( + templates: &mut HashMap, + registry: &ResourceRegistry, +) { + for (name, tmpl_def) in templates.iter_mut() { + if tmpl_def.use_.is_none() { + continue; + } + let Some(path) = registry.resolve(name) else { + continue; + }; + let Ok(content) = std::fs::read_to_string(path.join("template.yml")) + .or_else(|_| std::fs::read_to_string(path.join("template.yaml"))) + else { + continue; + }; + let Ok(fetched) = + serde_yaml::from_str::(&content) + else { + continue; + }; + *tmpl_def = fetched.apply(std::mem::take(tmpl_def)); + tmpl_def.use_ = None; + } +} + +/// Resolve a schema reference into a NotificationTemplateDef by walking a fallback chain. +pub fn resolve_template( + schema: &Option, + method_name: &str, + templates: &HashMap, + _standalone: bool, + server_templates: Option<&HashMap>, +) -> Result { + // 1. Try the initial schema directly + if let Some(schema_name) = schema.as_deref() { + if let Some(template) = + resolve_template_by_schema(schema_name, method_name, templates, server_templates) + { + return Ok(template); + } + } + + // 2. Walk the fallback chain + let mut schema_next = resolve_template_next_schema(schema); + while let Some(ref schema_name) = schema_next { + if let Some(template) = + resolve_template_by_schema(schema_name, method_name, templates, server_templates) + { + return Ok(template); + } + log::warn!("Failed to resolve template for schema {}.", schema_name); + schema_next = resolve_template_next_schema(&schema_next); + } + Err(NotifyError::SchemaError(format!("All template resolution attempts beginning from {} failed.", schema.as_deref().unwrap_or("[system default]")))) +} + +/// Return the next schema name in the fallback chain, or None to terminate. +pub fn resolve_template_next_schema(schema: &Option) -> Option { + match schema.as_deref() { + Some("custom") => None, + Some("default") => Some("fallback".to_string()), + Some("fallback") => None, + Some(_) => Some("default".to_string()), + None => Some("default".to_string()), + } +} + +/// Perform a single lookup for a given schema name. Returns None if not found. +pub fn resolve_template_by_schema( + schema: &str, + method_name: &str, + templates: &HashMap, + server_templates: Option<&HashMap>, +) -> Option { + match schema { + "custom" => Some(NotificationTemplateDef::default()), + "fallback" => templates.get(&format!("{}.fallback", method_name)).cloned(), + "default" => server_templates.and_then(|b| b.get(method_name)).cloned(), + name => templates.get(name).cloned(), + } +} + +/// Flatten all local template cross-references (use: "other-name"). +pub fn flatten_template_refs( + templates: &mut HashMap, +) -> Result<(), NotifyError> { + let names: Vec = templates.keys().cloned().collect(); + for name in names { + let mut visited = std::collections::HashSet::new(); + resolve_chain(&name, templates, 0, &mut visited)?; + } + Ok(()) +} + +fn resolve_chain( + name: &str, + templates: &mut HashMap, + depth: u32, + visited: &mut std::collections::HashSet, +) -> Result<(), NotifyError> { + if depth > TEMPLATE_REFERENCE_DEPTH { + return Err(NotifyError::TemplateDepthExceeded { + name: name.to_string(), + max: TEMPLATE_REFERENCE_DEPTH, + }); + } + if !visited.insert(name.to_string()) { + return Err(NotifyError::TemplateLoop(name.to_string())); + } + + // Only process local references (no ://) + if let Some(ref_name) = templates.get(name).and_then(|t| t.use_.clone()) { + if !ref_name.contains("://") { + if !templates.contains_key(&ref_name) { + return Err(NotifyError::SchemaError(format!( + "template '{}' references non-existent template '{}'", + name, ref_name + ))); + } + resolve_chain(&ref_name, templates, depth + 1, visited)?; + + // Merge: base provides defaults, referencing template overrides + let base = templates[&ref_name].clone(); + let template = templates.get_mut(name).unwrap(); + *template = std::mem::take(template).apply(base); + template.use_ = None; + } + } + + visited.remove(name); + Ok(()) } #[cfg(test)] diff --git a/workshop-baker/src/notify/util.rs b/workshop-baker/src/notify/util.rs new file mode 100644 index 0000000..295dc11 --- /dev/null +++ b/workshop-baker/src/notify/util.rs @@ -0,0 +1,83 @@ +use crate::finalize::FinalizeConfig; +use crate::finalize::error::PluginError; +use crate::notify::error::NotifyError; +use crate::notify::types::{DEFAULT_NOTIFY_PARAMS, NotificationEvent, NotificationRule, NotificationRulesMap}; +use std::collections::HashSet; + +pub fn validate(finalize: &FinalizeConfig) -> anyhow::Result<()> { + let validate_result = finalize.validate(); + if let Err(errors) = validate_result { + for error in &errors { + log::error!("Error: {}", error); + } + anyhow::bail!(NotifyError::FinalizeValidationError(errors)) + }; + Ok(()) +} + +pub fn extract_plugin(finalize: &FinalizeConfig) -> HashSet { + finalize + .notification + .groups + .values() + .flat_map(|methods| methods.keys()) + .cloned() + .collect() +} + +pub fn is_temporary_error(e: &PluginError) -> bool { + !matches!(e, PluginError::GeneralError(_)) +} + +// calculate min(1, k/x)*t as notification grouping period +// to avoid excessive notification frequency +pub fn wait_time(group_period: u32, group_watermark: u32, count: u32) -> f32 { + if count == 0 { + log::warn!("Unexpected notification count: {}", count); + return group_period as f32; + } + f32::min(1.0, group_watermark as f32 / count as f32) * group_period as f32 +} + +pub fn notify_delay(event: &NotificationEvent) -> std::time::Duration { + let exponent = event.target.count as i32; + if exponent < 0 { + log::warn!("Unexpected count: {}", event.target.count); + return std::time::Duration::from_secs_f64(DEFAULT_NOTIFY_PARAMS.max_retry_delay_seconds.value); + } + let factor = if event.retry_backoff > 0.0 { + event.retry_backoff + } else { + DEFAULT_NOTIFY_PARAMS.retry_backoff.value + }; + let multiplier = factor.powi(exponent); + let delay = multiplier * DEFAULT_NOTIFY_PARAMS.retry_delay_seconds.value; + if !delay.is_finite() || delay < 0.0 { + return std::time::Duration::from_secs_f64(DEFAULT_NOTIFY_PARAMS.max_retry_delay_seconds.value); + } + std::time::Duration::from_secs_f64(delay) +} + +pub fn find_rule<'a>( + config: &'a NotificationRulesMap, + method: &str, + rule_id: usize, +) -> Result<&'a NotificationRule, NotifyError> { + for ruleset in config.values() { + if let Some(rules) = ruleset.get(method) { + if let Some(rule) = rules.get(rule_id) { + return Ok(rule); + } + } + } + Err(NotifyError::SchemaError("rule not found".into())) +} + +pub fn get_fallible(config: &NotificationRulesMap, method: &str) -> bool { + for ruleset in config.values() { + if let Some(rules) = ruleset.get(method) { + return rules.iter().all(|r| r.fallible); + } + } + false +}