use athena_gateway::normalize_column_name;
use serde_json::{Value, json};
use std::collections::HashMap;

use crate::{SourceConfig, TransformConfig};

/// Builds the gateway-style fetch JSON body for a pipeline source/transform pair.
pub fn build_fetch_body(
    source: &SourceConfig,
    transform: Option<&TransformConfig>,
    force_snake: bool,
) -> Value {
    let table_name = source
        .view_name
        .as_deref()
        .unwrap_or(source.table_name.as_str());
    let mut body = json!({
        "table_name": table_name,
    });
    if let Some(ref view_name) = source.view_name {
        body["view_name"] = json!(view_name);
    }
    if let Some(ref columns) = source.columns {
        body["columns"] = json!(columns);
    }
    if let Some(ref conditions) = source.conditions {
        body["conditions"] = json!(
            conditions
                .iter()
                .map(|condition| {
                    let eq_column = if force_snake {
                        normalize_column_name(&condition.eq_column, true)
                    } else {
                        condition.eq_column.clone()
                    };
                    json!({ "eq_column": eq_column, "eq_value": condition.eq_value.clone() })
                })
                .collect::<Vec<_>>()
        );
    }
    if let Some(limit) = source.limit {
        body["limit"] = json!(limit);
    }
    if let Some(transform) = transform {
        if let Some(ref group_by) = transform.group_by {
            body["group_by"] = json!(if force_snake {
                normalize_column_name(group_by, true)
            } else {
                group_by.clone()
            });
        }
        if let Some(ref time_granularity) = transform.time_granularity {
            body["time_granularity"] = json!(time_granularity);
        }
        if let Some(ref aggregation_column) = transform.aggregation_column {
            body["aggregation_column"] = json!(if force_snake {
                normalize_column_name(aggregation_column, true)
            } else {
                aggregation_column.clone()
            });
        }
        if let Some(ref aggregation_strategy) = transform.aggregation_strategy {
            body["aggregation_strategy"] = json!(aggregation_strategy);
        }
        if let Some(dedup) = transform.aggregation_dedup {
            body["aggregation_dedup"] = json!(dedup);
        }
    }

    body
}

/// Rewrites fetched row keys according to configured sink aliases.
pub fn apply_column_aliases(
    rows: Vec<Value>,
    aliases: Option<&HashMap<String, String>>,
) -> Vec<Value> {
    let Some(alias_map) = aliases else {
        return rows;
    };
    if alias_map.is_empty() {
        return rows;
    }

    rows.into_iter()
        .map(|row| {
            let Some(object) = row.as_object() else {
                return row;
            };
            let mut rewritten = serde_json::Map::with_capacity(object.len());
            for (key, value) in object {
                let next_key = alias_map.get(key).cloned().unwrap_or_else(|| key.clone());
                rewritten.insert(next_key, value.clone());
            }
            Value::Object(rewritten)
        })
        .collect()
}

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

    #[test]
    fn build_fetch_body_normalizes_requested_columns_when_forced() {
        let body = build_fetch_body(
            &SourceConfig {
                table_name: "events".to_string(),
                view_name: None,
                columns: Some(vec!["workspaceId".to_string()]),
                conditions: Some(vec![ConditionEntry {
                    eq_column: "workspaceId".to_string(),
                    eq_value: "acme".to_string(),
                }]),
                limit: Some(5),
                client: None,
            },
            Some(&TransformConfig {
                group_by: Some("createdAt".to_string()),
                aggregation_column: Some("amountCents".to_string()),
                ..Default::default()
            }),
            true,
        );

        assert_eq!(body["table_name"], "events");
        assert_eq!(body["conditions"][0]["eq_column"], "workspace_id");
        assert_eq!(body["group_by"], "created_at");
        assert_eq!(body["aggregation_column"], "amount_cents");
    }

    #[test]
    fn apply_column_aliases_rewrites_object_keys() {
        let rows = vec![json!({
            "email": "test@example.com",
            "created_at": "2026-01-01"
        })];
        let aliases = HashMap::from([
            ("email".to_string(), "user_email".to_string()),
            ("created_at".to_string(), "createdAt".to_string()),
        ]);

        let out = apply_column_aliases(rows, Some(&aliases));
        let first = out.first().and_then(Value::as_object).expect("row object");
        assert!(first.contains_key("user_email"));
        assert!(first.contains_key("createdAt"));
        assert!(!first.contains_key("email"));
        assert!(!first.contains_key("created_at"));
    }
}
