use serde::Deserialize;
use serde_json::Value;
use std::collections::HashMap;
use std::error::Error;
use std::fs;
use std::path::Path;

use crate::{PipelineDefinition, PipelineRequest, SinkConfig, SourceConfig, TransformConfig};

#[derive(Debug, Deserialize)]
#[serde(rename_all = "snake_case")]
struct PipelineEntry {
    name: String,
    source: SourceConfig,
    transform: Option<TransformConfig>,
    sink: SinkConfig,
}

#[derive(Debug, Deserialize)]
struct PipelinesFile {
    pipelines: Vec<PipelineEntry>,
}

/// Loads named pipeline definitions from a YAML file into a registry map.
pub fn load_registry_from_path<P: AsRef<Path>>(
    path: P,
) -> Result<HashMap<String, PipelineDefinition>, Box<dyn Error + Send + Sync>> {
    let content = fs::read_to_string(path.as_ref())?;
    let file: PipelinesFile = serde_yaml::from_str(&content)?;
    let mut registry = HashMap::new();
    for entry in file.pipelines {
        let definition = PipelineDefinition {
            source: entry.source,
            transform: entry.transform,
            sink: entry.sink,
        };
        registry.insert(entry.name, definition);
    }
    Ok(registry)
}

/// Validates that the minimum required pipeline fields are present.
pub fn validate_pipeline_definition(definition: &PipelineDefinition) -> Result<(), String> {
    if definition.source.table_name.trim().is_empty() {
        return Err("source.table_name is required".to_string());
    }
    if definition.sink.table_name.trim().is_empty() {
        return Err("sink.table_name is required".to_string());
    }
    Ok(())
}

/// Parses and validates a pipeline definition from arbitrary JSON.
pub fn pipeline_definition_from_value(value: Value) -> Result<PipelineDefinition, String> {
    let definition: PipelineDefinition = serde_json::from_value(value)
        .map_err(|err| format!("invalid pipeline definition: {err}"))?;
    validate_pipeline_definition(&definition)?;
    Ok(definition)
}

/// Resolves a request body into the effective pipeline definition.
pub fn resolve_definition(
    request: &PipelineRequest,
    registry: &HashMap<String, PipelineDefinition>,
    database_templates: &HashMap<String, PipelineDefinition>,
) -> Result<PipelineDefinition, String> {
    let base = if let Some(ref name) = request.pipeline {
        registry
            .get(name)
            .or_else(|| database_templates.get(name))
            .cloned()
            .ok_or_else(|| format!("unknown pipeline '{name}'"))?
    } else {
        PipelineDefinition::default()
    };

    let definition = PipelineDefinition {
        source: request.source.clone().unwrap_or(base.source),
        transform: request.transform.clone().or(base.transform),
        sink: request.sink.clone().unwrap_or(base.sink),
    };
    validate_pipeline_definition(&definition)?;
    Ok(definition)
}

/// Picks the explicit pipeline client override or falls back to the request client.
pub fn resolve_client_name(override_client: Option<&str>, fallback: &str) -> String {
    override_client
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or(fallback)
        .to_string()
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use std::time::{SystemTime, UNIX_EPOCH};

    #[test]
    fn validate_pipeline_definition_requires_source_and_sink_tables() {
        let missing_source = PipelineDefinition {
            source: SourceConfig::default(),
            transform: None,
            sink: SinkConfig {
                table_name: "sink".to_string(),
                client: None,
            },
        };
        let missing_sink = PipelineDefinition {
            source: SourceConfig {
                table_name: "source".to_string(),
                ..Default::default()
            },
            transform: None,
            sink: SinkConfig::default(),
        };

        assert_eq!(
            validate_pipeline_definition(&missing_source),
            Err("source.table_name is required".to_string())
        );
        assert_eq!(
            validate_pipeline_definition(&missing_sink),
            Err("sink.table_name is required".to_string())
        );
    }

    #[test]
    fn pipeline_definition_from_value_validates_required_fields() {
        let error = pipeline_definition_from_value(json!({
            "source": { "table_name": "" },
            "sink": { "table_name": "events_sink" }
        }))
        .expect_err("definition should fail");

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

    #[test]
    fn resolve_definition_merges_template_and_inline_overrides() {
        let registry = HashMap::from([(
            "copy_events".to_string(),
            PipelineDefinition {
                source: SourceConfig {
                    table_name: "events_source".to_string(),
                    limit: Some(10),
                    ..Default::default()
                },
                transform: None,
                sink: SinkConfig {
                    table_name: "events_sink".to_string(),
                    client: None,
                },
            },
        )]);

        let resolved = resolve_definition(
            &crate::PipelineRequest {
                pipeline: Some("copy_events".to_string()),
                source: Some(SourceConfig {
                    table_name: "events_source".to_string(),
                    limit: Some(25),
                    ..Default::default()
                }),
                transform: Some(TransformConfig {
                    group_by: Some("workspace_id".to_string()),
                    ..Default::default()
                }),
                sink: None,
                dry_run: Some(true),
            },
            &registry,
            &HashMap::new(),
        )
        .expect("definition should resolve");

        assert_eq!(resolved.source.limit, Some(25));
        assert_eq!(resolved.sink.table_name, "events_sink");
        assert_eq!(
            resolved
                .transform
                .as_ref()
                .and_then(|transform| transform.group_by.as_deref()),
            Some("workspace_id")
        );
    }

    #[test]
    fn resolve_client_name_prefers_override() {
        let client = resolve_client_name(Some("analytics_client"), "default_client");
        assert_eq!(client, "analytics_client");
    }

    #[test]
    fn resolve_client_name_falls_back_to_default() {
        let client = resolve_client_name(Some("  "), "default_client");
        assert_eq!(client, "default_client");
    }

    #[test]
    fn load_registry_from_path_reads_yaml_templates() {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system time")
            .as_nanos();
        let temp_path = std::env::temp_dir().join(format!("athena-pipelines-{unique}.yaml"));

        fs::write(
            &temp_path,
            r#"
pipelines:
  - name: copy_events
    source:
      table_name: events_source
      limit: 10
    sink:
      table_name: events_sink
"#,
        )
        .expect("write temp yaml");

        let registry = load_registry_from_path(&temp_path).expect("registry should load");
        let definition = registry.get("copy_events").expect("template entry");
        assert_eq!(definition.source.table_name, "events_source");
        assert_eq!(definition.sink.table_name, "events_sink");

        let _ = fs::remove_file(temp_path);
    }
}
