use crate::{PipelineDefinition, PipelineTemplateCandidate, pipeline_definition_from_value};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;

/// Portable persisted pipeline template record shared by SQL adapters and API responses.
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
pub struct PipelineTemplateRecord {
    pub id: String,
    pub template_name: String,
    pub definition: Value,
    pub metadata: Value,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

/// Request body for saving a managed pipeline template through admin surfaces.
#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
pub struct SavePipelineTemplateRequest {
    pub definition: Value,
    #[serde(default)]
    pub metadata: Option<Value>,
}

/// Builds the named-definition lookup map used by runtime pipeline resolution.
pub fn pipeline_definition_map_from_records(
    records: Vec<PipelineTemplateRecord>,
) -> Result<HashMap<String, PipelineDefinition>, String> {
    let mut definitions: HashMap<String, PipelineDefinition> =
        HashMap::with_capacity(records.len());
    for record in records {
        let (template_name, definition) = parse_record_definition(record)?;
        definitions.insert(template_name, definition);
    }
    Ok(definitions)
}

/// Converts persisted records into runtime template candidates for summary merging.
pub fn runtime_template_candidates_from_records(
    records: Vec<PipelineTemplateRecord>,
) -> Result<Vec<PipelineTemplateCandidate>, String> {
    let mut runtime_templates: Vec<PipelineTemplateCandidate> = Vec::with_capacity(records.len());
    for record in records {
        let (template_name, definition) = parse_record_definition(record)?;
        runtime_templates.push(PipelineTemplateCandidate {
            name: template_name,
            definition,
            source: "database".to_string(),
            mutable: true,
        });
    }
    Ok(runtime_templates)
}

fn parse_record_definition(
    record: PipelineTemplateRecord,
) -> Result<(String, PipelineDefinition), String> {
    let PipelineTemplateRecord {
        template_name,
        definition,
        ..
    } = record;
    let parsed = pipeline_definition_from_value(definition)
        .map_err(|err| format!("template '{template_name}': {err}"))?;
    Ok((template_name, parsed))
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::TimeZone;
    use serde_json::json;

    fn sample_record(template_name: &str, definition: Value) -> PipelineTemplateRecord {
        let timestamp = Utc
            .with_ymd_and_hms(2025, 1, 1, 0, 0, 0)
            .single()
            .expect("valid timestamp");
        PipelineTemplateRecord {
            id: format!("{template_name}-id"),
            template_name: template_name.to_string(),
            definition,
            metadata: json!({ "managed_by": "test" }),
            created_at: timestamp,
            updated_at: timestamp,
        }
    }

    #[test]
    fn pipeline_definition_map_from_records_parses_persisted_templates() {
        let definitions = pipeline_definition_map_from_records(vec![sample_record(
            "copy_events",
            json!({
                "source": { "table_name": "events_source" },
                "sink": { "table_name": "events_sink" }
            }),
        )])
        .expect("records should parse");

        let definition = definitions
            .get("copy_events")
            .expect("template definition should exist");
        assert_eq!(definition.source.table_name, "events_source");
        assert_eq!(definition.sink.table_name, "events_sink");
    }

    #[test]
    fn runtime_template_candidates_from_records_marks_database_templates_mutable() {
        let runtime_templates = runtime_template_candidates_from_records(vec![sample_record(
            "copy_events",
            json!({
                "source": { "table_name": "events_source" },
                "sink": { "table_name": "events_sink" }
            }),
        )])
        .expect("records should parse");

        assert_eq!(runtime_templates.len(), 1);
        assert_eq!(runtime_templates[0].name, "copy_events");
        assert_eq!(runtime_templates[0].source, "database");
        assert!(runtime_templates[0].mutable);
    }

    #[test]
    fn record_conversion_errors_include_template_name_context() {
        let error = pipeline_definition_map_from_records(vec![sample_record(
            "broken_template",
            json!({
                "source": { "table_name": "" },
                "sink": { "table_name": "events_sink" }
            }),
        )])
        .expect_err("invalid record should fail");

        assert!(error.contains("template 'broken_template'"));
        assert!(error.contains("source.table_name is required"));
    }
}
