honey-biscuit-workshop/workshop-baker/src/notify.rs

use crate::finalize::FinalizeConfig;
use crate::finalize::NotificationConfig;
use crate::finalize::NotificationTemplate;
use crate::finalize::config::TriggerItemRaw;
use crate::finalize::constant::FINALIZE_STANDALONE_PLUGIN_PATH;
use crate::finalize::parse;
use crate::finalize::plugin;
use crate::utils::fetch_plugin;
use crate::notify::types::Matchable;
use crate::notify::types::Method;
use crate::notify::types::MethodStatus;
use crate::notify::types::NOTIFICATION_DEFAULT_FALLIBLE;
use crate::notify::types::NOTIFICATION_DEFAULT_MAX_LIVES;
use crate::notify::types::NOTIFICATION_DEFAULT_RETRY_BACKOFF;
use crate::notify::types::NOTIFICATION_DEFAULT_TIMEOUT_MS;
use crate::notify::types::NOTIFICATION_LIFE_IMPACT_FACTOR;
use crate::notify::types::NOTIFICATION_NOTREADY_DELAY;
use crate::notify::types::NOTIFICATION_TEMPORARY_DELAY_FACTOR;
use crate::notify::types::NOTIFICATION_TEMPORARY_DELAY_TIME;
use crate::notify::types::NOTIFICATION_TEMPORARY_MAX_TIME;
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 workshop_engine::{EventSender, ExecutionContext};
use chrono::Utc;
use globset::{Glob, GlobSetBuilder};
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::PathBuf;
use std::path::Path;

mod error;
pub mod types;
pub mod template;
pub mod interactive;
pub mod queue;
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<String> {
    finalize
        .notification
        .groups
        .values()
        .flat_map(|methods| methods.keys())
        .cloned()
        .collect()
}

pub async fn notify(
    finalize_path: &Path,
    ctx: &mut ExecutionContext,
    _event_tx: EventSender,
    mut rx: NotificationEventReceiver,
    retry_tx: NotificationEventSender,
    debug: Option<String>,
) -> anyhow::Result<()> {
    let mut finalize = parse(finalize_path)?;
    validate(&finalize)?;
    let mut plugin_list = extract_plugin(&finalize);

    let is_debug = debug.as_deref().unwrap_or("").contains("notify");

    if plugin_list.is_empty() {
        if is_debug {
            log::info!("[NOTIFY_DEBUG] Injecting dummy plugin for debug observation");
            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 {
                    template: crate::finalize::NotificationTemplate::default(),
                    trigger: vec![],
                    config: serde_yaml::Value::Mapping(serde_yaml::Mapping::new()),
                    policy: vec![],
                },
            );
            finalize.notification.groups.insert(0, methods);
        } else {
            log::warn!("No notification method provided, skipping notification");
            return Ok(());
        }
    }

    if ctx.standalone {
        fetch_plugin::fetch_plugins(
            &PathBuf::from(FINALIZE_STANDALONE_PLUGIN_PATH),
            &mut finalize.plugin,
            Some(&plugin_list),
        )
        .await?;
    }
    let notify_plugins = plugin::register(finalize.plugin, Some(&plugin_list), &mut finalize.notification)?;



    // Fetch external templates referenced by use_ (git:// / https:// URLs)
    let template_cache = std::env::temp_dir().join("workshop-templates");
    for (name, tmpl_def) in finalize.notification.templates.iter_mut() {
        if let Some(ref url) = tmpl_def.use_ {
            if url.contains("://") {
                // External fetch
                match crate::utils::fetch::fetch_resource(name, url, &template_cache).await {
                    Ok(path) => {
                        // Read the fetched template file
                        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::<crate::finalize::NotificationTemplateDef>(&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
                            }
                        }
                    }
                    Err(e) => {
                        log::error!("Failed to fetch template '{}' from '{}': {}", name, url, e);
                    }
                }
            }
            // Non-:// use_ values are local template references → handled in Part 2
        }
    }

    // Flatten local template cross-references
    flatten_template_refs(&mut finalize.notification.templates)?;

    let notification_rules = parse_rules(&finalize.notification);


    loop {
        // Wait for first event
        let first = if is_debug {
            // Debug mode: don't block forever — timeout prevents deadlock
            match tokio::time::timeout(std::time::Duration::from_secs(5), rx.recv()).await {
                Ok(Some(event)) => event,
                Ok(None) => break,
                Err(_) => {
                    log::info!("[NOTIFY_DEBUG] No events within timeout, shutting down");
                    break;
                }
            }
        } else {
            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.as_deref().unwrap_or("").contains("notify") {
            for ev in &events {
                log::info!("[NOTIFY_DEBUG] stage={:?} substage={} outcome={:?} priority={}",
                    ev.stage, ev.substage, ev.outcome, ev.priority);
            }
        }

        let results = notification_handler(&notify_plugins, &notification_rules, &events, &finalize.notification.templates, ctx).await;

        let mut na: Vec<String> = vec![];
        let mut te: Vec<String> = vec![];
        let mut pe_ex: Vec<(String, bool)> = vec![];

        for (method, status) in &results {
            match status {
                MethodStatus::Success => {}
                MethodStatus::NotAvailable(_) => na.push(method.clone()),
                MethodStatus::TemporaryError(_) => te.push(method.clone()),
                MethodStatus::PermanentError(_) | MethodStatus::Exhausted(_) => {
                    let fallible = get_fallible(&notification_rules, method);
                    pe_ex.push((method.clone(), fallible));
                }
            }
        }

        if !na.is_empty() || !te.is_empty() {
            let mut retry_methods = na.clone();
            retry_methods.extend(te.clone());
            events[0].target.rules = Some(retry_methods);
            if !te.is_empty() {
                events[0].target.count += 1;
                events[0].not_before = Utc::now() + notify_delay(&events[0]);
            } else {
                events[0].effective_priority += 1;
                events[0].not_before = Utc::now() + NOTIFICATION_NOTREADY_DELAY;
            }
            retry_tx.send(events[0].clone()).await?;
            continue;
        }

        let has_hard = pe_ex.iter().any(|(_, fallible)| !fallible);
        if has_hard {
            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;
        }
    }
    Ok(())
}

// TODO: PIPELINE_NOTIFICATION_EXHAUSTED — set when any method returns Exhausted
// TODO: PIPELINE_NOTIFICATION_FAILED    — set when any method returns PermanentError
//       Markers sent via event_tx as ExecutionEvent variants for WebUI/CLI display.


/// Resolve a schema reference into a concrete NotificationTemplateDef.
///
/// Resolution rules:
/// - `None` (unspecified): c/d → ws-base default → fallback; standalone → fallback
/// - `Some("custom")`: return empty default (caller uses rule.template.on_* directly)
/// - `Some("default")`: ws-base default (fallback in standalone)
/// - `Some("<name>")`: look up in templates; if not found and name == method_name,
///   fallback to `{name}.fallback`; otherwise error
fn resolve_template_content(
    schema: &Option<String>,
    method_name: &str,
    templates: &std::collections::HashMap<String, crate::finalize::NotificationTemplateDef>,
    standalone: bool,
    ws_base_templates: Option<&std::collections::HashMap<String, crate::finalize::NotificationTemplateDef>>,
) -> Result<crate::finalize::NotificationTemplateDef, NotifyError> {
    match schema.as_deref() {
        Some("custom") => {
            // Caller reads template content from rule.template.on_* directly
            Ok(crate::finalize::NotificationTemplateDef::default())
        }

        Some("default") => {
            if standalone {
                templates.get(&format!("{}.fallback", method_name))
                    .cloned()
                    .ok_or_else(|| NotifyError::SchemaError(
                        format!("{} fallback template not found in standalone mode", method_name)
                    ))
            } else {
                ws_base_templates
                    .and_then(|b| b.get("default"))
                    .cloned()
                    .ok_or_else(|| NotifyError::SchemaError(
                        "ws-base default template not found".into()
                    ))
            }
        }

        Some(name) => {
            match templates.get(name) {
                Some(tmpl) => Ok(tmpl.clone()),
                None => {
                    if name == method_name {
                        // Fallback to plugin's own template
                        templates.get(&format!("{}.fallback", name))
                            .cloned()
                            .ok_or_else(|| NotifyError::SchemaError(
                                format!("plugin '{}' has no fallback template", name)
                            ))
                    } else {
                        Err(NotifyError::SchemaError(
                            format!("template '{}' not found", name)
                        ))
                    }
                }
            }
        }

        None => {
            if standalone {
                templates.get(&format!("{}.fallback", method_name))
                    .cloned()
                    .ok_or_else(|| NotifyError::SchemaError(
                        format!("no template configured and '{}' has no fallback", method_name)
                    ))
            } else {
                // c/d: ws-base default → fallback (two-level chain)
                ws_base_templates
                    .and_then(|base| base.get("default"))
                    .or_else(|| templates.get(&format!("{}.fallback", method_name)))
                    .cloned()
                    .ok_or_else(|| NotifyError::SchemaError(
                        "no template available (ws-base default not found, no plugin fallback)".into()
                    ))
            }
        }
    }
}

/// 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<String, crate::finalize::NotificationTemplateDef>,
) -> Result<(), NotifyError> {
    let names: Vec<String> = 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<String, crate::finalize::NotificationTemplateDef>,
    depth: u32,
    visited: &mut std::collections::HashSet<String>,
) -> 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 tmpl = templates.get_mut(name).unwrap();
            if tmpl.on_success.is_none() { tmpl.on_success = base.on_success.clone(); }
            if tmpl.on_failure.is_none() { tmpl.on_failure = base.on_failure.clone(); }
            if tmpl.on_finish_of.is_none() { tmpl.on_finish_of = base.on_finish_of.clone(); }
            tmpl.use_ = None; // Chain flattened
        }
    }

    visited.remove(name);
    Ok(())
}

fn is_plugin_unavailable(e: &crate::finalize::error::PluginError) -> bool {
    matches!(e, crate::finalize::error::PluginError::PluginUnavailable { .. })
}

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<String, crate::finalize::NotificationTemplateDef>,
    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());
            eprintln!("[CHK-ENABLED] entering enabled check");
                    if !rule.enabled {
            eprintln!("[CHK-ENABLED] enabled=false, continuing");
                        continue;
                    }
            eprintln!("[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<NotificationEvent> = bucket_events.into_iter().cloned().collect();

        let argument = match renderer {
            NotificationRenderer::Server => {
                let template_def = match resolve_template_content(
                    &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_content(
                    &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<serde_json::Value> = 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 => {
                let template_data =
                    crate::notify::template::build_batch_template_context(ctx, &bucket);
                let mut server =
                    crate::notify::interactive::InteractiveServer::new(template_data);
                let port = match server.start().await {
                    Ok(p) => p,
                    Err(e) => {
                        results.push((method.clone(), MethodStatus::TemporaryError(format!("Failed to start IR server: {}", e))));
                        continue;
                    }
                };

                let mut argument = rule.config.clone();
                if let serde_yaml::Value::Mapping(ref mut map) = argument {
                    map.insert(
                        serde_yaml::Value::String("HBW_IR_PORT".into()),
                        serde_yaml::Value::Number(port.into()),
                    );
                }

                log::info!(
                    "IR: server started on port {} for method '{}'",
                    port, method
                );

                let remain = rule.max_lives as f64
                    - NOTIFICATION_LIFE_IMPACT_FACTOR * 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"))));
                    server.shutdown().await;
                    continue;
                }

                let status = match tokio::time::timeout(rule.timeout, plugin.call(argument, ctx, &None)).await {
                    Ok(Ok(())) => MethodStatus::Success,
                    Ok(Err(e)) => {
                        if is_plugin_unavailable(&e) {
                            MethodStatus::NotAvailable(format!("{}", e))
                        } else if is_temporary_error(&e) {
                            MethodStatus::TemporaryError(format!("{}", e))
                        } else {
                            MethodStatus::PermanentError(format!("{}", e))
                        }
                    }
                    Err(_) => MethodStatus::TemporaryError("timeout".into()),
                };
                server.shutdown().await;
                results.push((method.clone(), status));
                continue;
            }
        };

        let remain = rule.max_lives as f64
            - NOTIFICATION_LIFE_IMPACT_FACTOR * 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_plugin_unavailable(&e) {
                    MethodStatus::NotAvailable(format!("{}", e))
                } else 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
}

fn parse_trigger_item_raw(item: &TriggerItemRaw) -> Vec<ParsedTrigger> {
    match item {
        TriggerItemRaw::Simple(s) => match s.as_str() {
            "on_success" => vec![ParsedTrigger::OnSuccess],
            "on_failure" => vec![ParsedTrigger::OnFailure],
            s if s.starts_with("on_finish_of:") => {
                let pattern = s.trim_start_matches("on_finish_of:").trim();
                let mut builder = GlobSetBuilder::new();
                builder.add(Glob::new(pattern).expect("Invalid glob pattern"));
                let glob_set = builder.build().expect("Failed to compile glob set");
                vec![ParsedTrigger::OnFinishOf(Matchable::Glob(glob_set))]
            }
            _ => {
                log::warn!("Unknown trigger string: {}", s);
                vec![]
            }
        },
        TriggerItemRaw::Map(map) => {
            let mut triggers = Vec::new();
            for (key, patterns) in map {
                match key.as_str() {
                    "on_finish_of" => {
                        for pattern in patterns {
                            let mut builder = GlobSetBuilder::new();
                            builder.add(Glob::new(pattern).expect("Invalid glob pattern"));
                            let glob_set = builder.build().expect("Failed to compile glob set");
                            triggers.push(ParsedTrigger::OnFinishOf(Matchable::Glob(glob_set)));
                        }
                    }
                    "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<serde_yaml::Value>,
) -> NotificationTemplate {
    let mut merged = method_template.clone();
    if let Some(policy_tmpl) = policy_template
        && let Ok(policy_template) =
            serde_yaml::from_value::<NotificationTemplate>(policy_tmpl.clone())
    {
        if policy_template.schema.is_some() {
            merged.schema = policy_template.schema;
        }
        if policy_template.on_success.is_some() {
            merged.on_success = policy_template.on_success;
        }
        if policy_template.on_failure.is_some() {
            merged.on_failure = policy_template.on_failure;
        }
        if policy_template.on_finish_of.is_some() {
            merged.on_finish_of = policy_template.on_finish_of;
        }
    }
    merged
}

fn extract_overrides(overrides: &serde_yaml::Value) -> (bool, u32, std::time::Duration, f64) {
    let fallible = overrides
        .get("fallible")
        .and_then(|v| v.as_bool())
        .unwrap_or(NOTIFICATION_DEFAULT_FALLIBLE);
    let max_lives = overrides
        .get("max_lives")
        .and_then(|v| v.as_u64())
        .unwrap_or(NOTIFICATION_DEFAULT_MAX_LIVES as u64) as u32;
    let timeout_ms = overrides
        .get("timeout_ms")
        .and_then(|v| v.as_u64())
        .unwrap_or(NOTIFICATION_DEFAULT_TIMEOUT_MS);
    let retry_backoff = overrides
        .get("retry_backoff")
        .and_then(|v| v.as_f64())
        .unwrap_or(NOTIFICATION_DEFAULT_RETRY_BACKOFF);
    (
        fallible,
        max_lives,
        std::time::Duration::from_millis(timeout_ms),
        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<Method, Vec<NotificationRule>> = HashMap::new();
        for (method_name, method) in group {
            let mut method_rules: Vec<NotificationRule> = Vec::new();
            if method.policy.is_empty() {
                let trigger: Vec<ParsedTrigger> = method
                    .trigger
                    .iter()
                    .flat_map(parse_trigger_item_raw)
                    .collect();
                let template = method.template.clone();
                let (fallible, max_lives, timeout, retry_backoff) =
                    extract_overrides(&method.config);
                method_rules.push(NotificationRule {
                    enabled: true,
                    trigger,
                    template,
                    fallible,
                    max_lives,
                    timeout,
                    config: method.config.clone(),
                    retry_backoff,
                });
            } else {
                for policy in &method.policy {
                    let trigger: Vec<ParsedTrigger> = 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 method_overrides = extract_overrides(&method.config);
                    let policy_overrides = extract_overrides(&policy.overrides);
                    let fallible = if policy.overrides.get("fallible").is_some() {
                        policy_overrides.0
                    } else {
                        method_overrides.0
                    };
                    let max_lives = if policy.overrides.get("max_lives").is_some() {
                        policy_overrides.1
                    } else {
                        method_overrides.1
                    };
                    let timeout = if policy.overrides.get("timeout_ms").is_some() {
                        policy_overrides.2
                    } else {
                        method_overrides.2
                    };
                    let retry_backoff = if policy.overrides.get("retry_backoff").is_some() {
                        policy_overrides.3
                    } else {
                        method_overrides.3
                    };
                    method_rules.push(NotificationRule {
                        enabled: true,
                        trigger,
                        template,
                        fallible,
                        max_lives,
                        timeout,
                        config: method.config.clone(),
                        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(NOTIFICATION_TEMPORARY_MAX_TIME);
    }
    let factor = if event.retry_backoff > 0.0 { event.retry_backoff } else { NOTIFICATION_TEMPORARY_DELAY_FACTOR as f64 };
    let multiplier = factor.powi(exponent);
    let delay = multiplier * NOTIFICATION_TEMPORARY_DELAY_TIME;
    if !delay.is_finite() || delay < 0.0 {
        return std::time::Duration::from_secs_f64(NOTIFICATION_TEMPORARY_MAX_TIME);
    }
    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::*;
    use crate::finalize::NotificationTemplateDef;
    use crate::notify::types::NotificationRule;
    use crate::notify::types::NotificationTarget;
    use std::collections::HashMap;

    // ── helper ──

    fn make_tmpl(on_success: &str, on_failure: &str) -> NotificationTemplateDef {
        NotificationTemplateDef {
            use_: None,
            on_success: if on_success.is_empty() { None } else { Some(on_success.into()) },
            on_failure: if on_failure.is_empty() { None } else { Some(on_failure.into()) },
            on_finish_of: None,
        }
    }

    fn make_tmpl_with_use(use_ref: &str) -> NotificationTemplateDef {
        NotificationTemplateDef {
            use_: Some(use_ref.into()),
            on_success: None,
            on_failure: None,
            on_finish_of: None,
        }
    }

    // ═══════════════════════════════════════════
    // resolve_template_content
    // ═══════════════════════════════════════════

    #[test]
    fn resolve_none_cd_default_found() {
        let mut ws = HashMap::new();
        ws.insert("default".into(), make_tmpl("OK", "FAIL"));
        let templates = HashMap::new();
        let result = resolve_template_content(&None, "mail", &templates, false, Some(&ws));
        assert!(result.is_ok());
        assert_eq!(result.unwrap().on_success.unwrap(), "OK");
    }

    #[test]
    fn resolve_none_cd_default_missing_fallback() {
        let ws: HashMap<String, NotificationTemplateDef> = HashMap::new();
        let mut templates = HashMap::new();
        templates.insert("mail.fallback".into(), make_tmpl("FALLBACK", ""));
        let result = resolve_template_content(&None, "mail", &templates, false, Some(&ws));
        assert!(result.is_ok());
        assert_eq!(result.unwrap().on_success.unwrap(), "FALLBACK");
    }

    #[test]
    fn resolve_none_cd_no_default_no_fallback_error() {
        let ws: HashMap<String, NotificationTemplateDef> = HashMap::new();
        let templates = HashMap::new();
        let result = resolve_template_content(&None, "mail", &templates, false, Some(&ws));
        assert!(result.is_err());
        assert!(format!("{}", result.unwrap_err()).contains("no template available"));
    }

    #[test]
    fn resolve_none_standalone_fallback() {
        let mut templates = HashMap::new();
        templates.insert("mail.fallback".into(), make_tmpl("STANDALONE", ""));
        let result = resolve_template_content(&None, "mail", &templates, true, None);
        assert!(result.is_ok());
        assert_eq!(result.unwrap().on_success.unwrap(), "STANDALONE");
    }

    #[test]
    fn resolve_none_standalone_no_fallback_error() {
        let result = resolve_template_content(&None, "mail", &HashMap::new(), true, None);
        assert!(result.is_err());
    }

    #[test]
    fn resolve_custom() {
        let result = resolve_template_content(
            &Some("custom".into()), "mail", &HashMap::new(), false, None,
        );
        assert!(result.is_ok());
        // "custom" returns an empty default — caller reads rule.template.on_* directly
    }

    #[test]
    fn resolve_default_standalone_fallback() {
        let mut templates = HashMap::new();
        templates.insert("mail.fallback".into(), make_tmpl("FB", ""));
        let result = resolve_template_content(
            &Some("default".into()), "mail", &templates, true, None,
        );
        assert!(result.is_ok());
        assert_eq!(result.unwrap().on_success.unwrap(), "FB");
    }

    #[test]
    fn resolve_default_standalone_no_fallback_error() {
        let result = resolve_template_content(
            &Some("default".into()), "mail", &HashMap::new(), true, None,
        );
        assert!(result.is_err());
    }

    #[test]
    fn resolve_default_cd_wsbase() {
        let mut ws = HashMap::new();
        ws.insert("default".into(), make_tmpl("WS_OK", ""));
        let result = resolve_template_content(
            &Some("default".into()), "mail", &HashMap::new(), false, Some(&ws),
        );
        assert!(result.is_ok());
        assert_eq!(result.unwrap().on_success.unwrap(), "WS_OK");
    }

    #[test]
    fn resolve_default_cd_no_wsbase_error() {
        let result = resolve_template_content(
            &Some("default".into()), "mail", &HashMap::new(), false, None,
        );
        assert!(result.is_err());
    }

    #[test]
    fn resolve_named_found() {
        let mut templates = HashMap::new();
        templates.insert("sms-tmpl1".into(), make_tmpl("NAMED", ""));
        let result = resolve_template_content(
            &Some("sms-tmpl1".into()), "mail", &templates, false, None,
        );
        assert!(result.is_ok());
        assert_eq!(result.unwrap().on_success.unwrap(), "NAMED");
    }

    #[test]
    fn resolve_named_not_found_method_match_fallback() {
        let mut templates = HashMap::new();
        templates.insert("mail.fallback".into(), make_tmpl("FB", ""));
        let result = resolve_template_content(
            &Some("mail".into()), "mail", &templates, false, None,
        );
        assert!(result.is_ok());
        assert_eq!(result.unwrap().on_success.unwrap(), "FB");
    }

    #[test]
    fn resolve_named_not_found_no_match_error() {
        let result = resolve_template_content(
            &Some("ghost".into()), "mail", &HashMap::new(), false, None,
        );
        assert!(result.is_err());
    }

    // ═══════════════════════════════════════════
    // flatten_template_refs
    // ═══════════════════════════════════════════

    #[test]
    fn flatten_no_refs_unchanged() {
        let mut templates = HashMap::new();
        templates.insert("a".into(), make_tmpl("A_OK", "A_FAIL"));
        let result = flatten_template_refs(&mut templates);
        assert!(result.is_ok());
        assert_eq!(templates["a"].on_success.as_deref(), Some("A_OK"));
        assert!(templates["a"].use_.is_none());
    }

    #[test]
    fn flatten_single_ref_merged() {
        let mut templates = HashMap::new();
        templates.insert("base".into(), make_tmpl("BASE_OK", "BASE_FAIL"));
        templates.insert("child".into(), make_tmpl_with_use("base"));
        let result = flatten_template_refs(&mut templates);
        assert!(result.is_ok());
        assert_eq!(templates["child"].on_success.as_deref(), Some("BASE_OK"));
        assert_eq!(templates["child"].on_failure.as_deref(), Some("BASE_FAIL"));
        assert!(templates["child"].use_.is_none());
    }

    #[test]
    fn flatten_child_overrides_parent() {
        let mut templates = HashMap::new();
        templates.insert("base".into(), make_tmpl("BASE_OK", "BASE_FAIL"));
        templates.insert("child".into(), NotificationTemplateDef {
            use_: Some("base".into()),
            on_success: Some("MY_OK".into()),
            on_failure: None,
            on_finish_of: None,
        });
        let result = flatten_template_refs(&mut templates);
        assert!(result.is_ok());
        assert_eq!(templates["child"].on_success.as_deref(), Some("MY_OK"));
        assert_eq!(templates["child"].on_failure.as_deref(), Some("BASE_FAIL"));
    }

    #[test]
    fn flatten_chain_a_b_c() {
        let mut templates = HashMap::new();
        templates.insert("c".into(), make_tmpl("C_OK", "C_FAIL"));
        templates.insert("b".into(), make_tmpl_with_use("c"));
        templates.insert("a".into(), make_tmpl_with_use("b"));
        let result = flatten_template_refs(&mut templates);
        assert!(result.is_ok());
        assert_eq!(templates["a"].on_success.as_deref(), Some("C_OK"));
        assert_eq!(templates["b"].on_success.as_deref(), Some("C_OK"));
    }

    #[test]
    fn flatten_cycle_detected() {
        let mut templates = HashMap::new();
        templates.insert("x".into(), make_tmpl_with_use("y"));
        templates.insert("y".into(), make_tmpl_with_use("x"));
        let result = flatten_template_refs(&mut templates);
        assert!(result.is_err());
        match result.unwrap_err() {
            NotifyError::TemplateLoop(_) => {}
            other => panic!("expected TemplateLoop, got {:?}", other),
        }
    }

    #[test]
    fn flatten_self_ref_detected() {
        let mut templates = HashMap::new();
        templates.insert("self".into(), make_tmpl_with_use("self"));
        let result = flatten_template_refs(&mut templates);
        assert!(result.is_err());
        match result.unwrap_err() {
            NotifyError::TemplateLoop(_) => {}
            other => panic!("expected TemplateLoop, got {:?}", other),
        }
    }

    #[test]
    fn flatten_nonexistent_ref_error() {
        let mut templates = HashMap::new();
        templates.insert("orphan".into(), make_tmpl_with_use("ghost"));
        let result = flatten_template_refs(&mut templates);
        assert!(result.is_err());
    }

    #[test]
    fn flatten_external_url_preserved() {
        let mut templates = HashMap::new();
        templates.insert("ext".into(), NotificationTemplateDef {
            use_: Some("git://example.com/repo@v1".into()),
            on_success: None,
            on_failure: None,
            on_finish_of: None,
        });
        let result = flatten_template_refs(&mut templates);
        assert!(result.is_ok());
        // External URLs are NOT flattened — use_ preserved
        assert_eq!(templates["ext"].use_.as_deref(), Some("git://example.com/repo@v1"));
    }

    // ═══════════════════════════════════════════
    // wait_time
    // ═══════════════════════════════════════════

    #[test]
    fn wait_time_zero_count() {
        let w = wait_time(30, 5, 0);
        assert_eq!(w, 30.0);
    }

    #[test]
    fn wait_time_at_watermark() {
        let w = wait_time(30, 5, 5);
        assert_eq!(w, 30.0);
    }

    #[test]
    fn wait_time_below_watermark() {
        let w = wait_time(30, 5, 2);
        assert!((w - 30.0).abs() < 0.01);
    }

    #[test]
    fn wait_time_above_watermark() {
        let w = wait_time(30, 5, 10);
        assert!((w - 15.0).abs() < 0.01);
    }

    #[test]
    fn wait_time_far_above_watermark() {
        let w = wait_time(30, 5, 100);
        assert!((w - 1.5).abs() < 0.01);
    }

    // ═══════════════════════════════════════════
    // notify_delay
    // ═══════════════════════════════════════════

    fn make_event(count: u32, retry_backoff: f64) -> NotificationEvent {
        NotificationEvent {
            id: uuid::Uuid::new_v4(),
            stage: crate::types::buildstatus::StagePhase::Bake,
            substage: "build".into(),
            outcome: crate::types::buildstatus::StageOutcome::PipelineSuccess,
            not_before: chrono::Utc::now(),
            priority: 0,
            effective_priority: 0,
            attempted: 0,
            retry_backoff,
            target: NotificationTarget { priority: 0, rules: None, count },
        }
    }

    #[test]
    fn notify_delay_default_factor() {
        let event = make_event(1, 2.0);
        let delay = notify_delay(&event);
        assert!(delay.as_secs_f64() > 0.0);
    }

    #[test]
    fn notify_delay_larger_factor() {
        let event1 = make_event(1, 2.0);
        let event2 = make_event(1, 4.0);
        let d1 = notify_delay(&event1).as_secs_f64();
        let d2 = notify_delay(&event2).as_secs_f64();
        assert!(d2 > d1, "larger retry_backoff should produce longer delay");
    }

    // ═══════════════════════════════════════════
    // find_rule
    // ═══════════════════════════════════════════

    #[test]
    fn find_rule_found() {
        use crate::finalize::NotificationTemplate;
        let rule = NotificationRule {
            enabled: true,
            trigger: vec![],
            template: NotificationTemplate {
                schema: None,
                on_success: None,
                on_failure: None,
                on_finish_of: None,
            },
            fallible: false,
            max_lives: 3,
            timeout: std::time::Duration::from_secs(30),
            config: serde_yaml::Value::Null,
            retry_backoff: 2.0,
        };
        let mut rules_map: NotificationRulesMap = HashMap::new();
        rules_map.insert(0, {
            let mut m: HashMap<String, Vec<NotificationRule>> = HashMap::new();
            m.insert("mail".into(), vec![rule]);
            m
        });
        let result = find_rule(&rules_map, "mail", 0);
        assert!(result.is_ok());
    }

    #[test]
    fn find_rule_not_found() {
        let rules_map: NotificationRulesMap = HashMap::new();
        let result = find_rule(&rules_map, "mail", 0);
        assert!(result.is_err());
    }
}