use crate::PipelineDefinition;
use std::collections::{HashMap, HashSet};

/// Summary returned when listing named pipeline templates.
#[derive(Debug, Clone, serde::Serialize, PartialEq, Eq)]
pub struct PipelineTemplateSummary {
    pub name: String,
    pub source_table: String,
    pub sink_table: String,
    pub has_transform: bool,
    pub source: String,
    pub mutable: bool,
    pub source_client: Option<String>,
    pub sink_client: Option<String>,
    pub definition: PipelineDefinition,
}

/// Portable template candidate used when merging runtime-managed templates with boot registry entries.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PipelineTemplateCandidate {
    pub name: String,
    pub definition: PipelineDefinition,
    pub source: String,
    pub mutable: bool,
}

/// Result of merging boot templates with runtime-provided templates.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PipelineTemplateSummaryMerge {
    pub templates: Vec<PipelineTemplateSummary>,
    pub shadowed_template_names: Vec<String>,
}

/// Normalizes a pipeline template name for route and admin request handling.
pub fn normalize_pipeline_template_name(template_name: &str) -> Result<String, String> {
    let value = template_name.trim();
    if value.is_empty() {
        Err("The template name must not be empty.".to_string())
    } else {
        Ok(value.to_string())
    }
}

/// Returns whether a runtime-managed template name would shadow a boot-registry template.
pub fn pipeline_template_name_conflicts_with_registry(
    template_name: &str,
    registry: &HashMap<String, PipelineDefinition>,
) -> bool {
    registry
        .keys()
        .any(|name| name.eq_ignore_ascii_case(template_name))
}

/// Builds the API-facing template summary for a named pipeline definition.
pub fn build_template_summary(
    name: &str,
    definition: &PipelineDefinition,
    source: &str,
    mutable: bool,
) -> PipelineTemplateSummary {
    PipelineTemplateSummary {
        name: name.to_string(),
        source_table: definition.source.table_name.clone(),
        sink_table: definition.sink.table_name.clone(),
        has_transform: definition.transform.is_some(),
        source: source.to_string(),
        mutable,
        source_client: definition.source.client.clone(),
        sink_client: definition.sink.client.clone(),
        definition: definition.clone(),
    }
}

/// Merges boot-registry templates with runtime-provided templates.
///
/// Boot templates win case-insensitively so runtime-managed entries cannot shadow
/// YAML-defined templates. The returned summaries are sorted by name.
pub fn merge_template_summaries(
    registry: &HashMap<String, PipelineDefinition>,
    runtime_templates: &[PipelineTemplateCandidate],
) -> PipelineTemplateSummaryMerge {
    let mut templates: Vec<PipelineTemplateSummary> = registry
        .iter()
        .map(|(name, def)| build_template_summary(name, def, "yaml", false))
        .collect();
    let mut known_names: HashSet<String> = registry
        .keys()
        .map(|name| name.to_ascii_lowercase())
        .collect();
    let mut shadowed_template_names: Vec<String> = Vec::new();

    for template in runtime_templates {
        let folded_name = template.name.to_ascii_lowercase();
        if known_names.contains(&folded_name) {
            shadowed_template_names.push(template.name.clone());
            continue;
        }

        known_names.insert(folded_name);
        templates.push(build_template_summary(
            &template.name,
            &template.definition,
            &template.source,
            template.mutable,
        ));
    }

    templates.sort_by(|left, right| left.name.to_lowercase().cmp(&right.name.to_lowercase()));
    shadowed_template_names.sort_by_key(|name| name.to_ascii_lowercase());

    PipelineTemplateSummaryMerge {
        templates,
        shadowed_template_names,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{SinkConfig, SourceConfig, TransformConfig};

    fn sample_definition(source_table: &str, sink_table: &str) -> PipelineDefinition {
        PipelineDefinition {
            source: SourceConfig {
                table_name: source_table.to_string(),
                client: Some("source_client".to_string()),
                ..Default::default()
            },
            transform: Some(TransformConfig {
                group_by: Some("workspace_id".to_string()),
                ..Default::default()
            }),
            sink: SinkConfig {
                table_name: sink_table.to_string(),
                client: Some("sink_client".to_string()),
            },
        }
    }

    #[test]
    fn normalize_pipeline_template_name_rejects_empty_values() {
        assert_eq!(
            normalize_pipeline_template_name("   "),
            Err("The template name must not be empty.".to_string())
        );
    }

    #[test]
    fn normalize_pipeline_template_name_trims_surrounding_whitespace() {
        assert_eq!(
            normalize_pipeline_template_name("  monthly_rollup  "),
            Ok("monthly_rollup".to_string())
        );
    }

    #[test]
    fn pipeline_template_name_conflicts_with_registry_is_case_insensitive() {
        let registry = HashMap::from([(
            "Boot_Example".to_string(),
            sample_definition("events_source", "events_sink"),
        )]);

        assert!(pipeline_template_name_conflicts_with_registry(
            "boot_example",
            &registry
        ));
        assert!(pipeline_template_name_conflicts_with_registry(
            "BOOT_EXAMPLE",
            &registry
        ));
        assert!(!pipeline_template_name_conflicts_with_registry(
            "other_template",
            &registry
        ));
    }

    #[test]
    fn merge_template_summaries_preserves_boot_templates_and_reports_shadowed_runtime_entries() {
        let registry = HashMap::from([(
            "boot_example".to_string(),
            sample_definition("events_source", "events_sink"),
        )]);
        let runtime_templates = vec![
            PipelineTemplateCandidate {
                name: "boot_example".to_string(),
                definition: sample_definition("shadow_source", "shadow_sink"),
                source: "database".to_string(),
                mutable: true,
            },
            PipelineTemplateCandidate {
                name: "runtime_copy".to_string(),
                definition: sample_definition("runtime_source", "runtime_sink"),
                source: "database".to_string(),
                mutable: true,
            },
        ];

        let merged = merge_template_summaries(&registry, &runtime_templates);

        assert_eq!(merged.shadowed_template_names, vec!["boot_example"]);
        assert_eq!(merged.templates.len(), 2);
        assert_eq!(merged.templates[0].name, "boot_example");
        assert_eq!(merged.templates[0].source, "yaml");
        assert_eq!(merged.templates[0].mutable, false);
        assert_eq!(merged.templates[1].name, "runtime_copy");
        assert_eq!(merged.templates[1].source, "database");
        assert_eq!(merged.templates[1].mutable, true);
    }
}
