use pg_query::{NodeEnum, protobuf::Node};
use text_size::TextRange;

use crate::core::schema::{QualifiedName, SchemaModel, ViewDef};

use super::tables::range_var_to_key;

pub(super) fn handle_view(
    stmt: &pg_query::protobuf::ViewStmt,
    range: TextRange,
    model: &mut SchemaModel,
) {
    let Some(rel) = &stmt.view else {
        tracing::warn!("ViewStmt has no view relation, skipping");
        return;
    };
    let key = range_var_to_key(rel);
    let source_tables = stmt
        .query
        .as_deref()
        .map_or_else(Vec::new, extract_range_vars);
    model.views.insert(
        key,
        ViewDef {
            range,
            doc: None,
            source_tables,
        },
    );
}

fn extract_range_vars(query: &Node) -> Vec<QualifiedName> {
    let mut out: Vec<QualifiedName> = Vec::new();
    walk_node(query, &mut |node| {
        if let NodeEnum::RangeVar(rv) = node {
            let schema = if rv.schemaname.is_empty() {
                None
            } else {
                Some(rv.schemaname.as_str())
            };
            let qn = QualifiedName::new(schema, &rv.relname);
            if !out.contains(&qn) {
                out.push(qn);
            }
        }
    });
    out
}

fn walk_node<F: FnMut(&NodeEnum)>(node: &Node, f: &mut F) {
    if let Some(inner) = node.node.as_ref() {
        walk_all_nodes(inner, f);
    }
}

fn walk_all_nodes<F: FnMut(&NodeEnum)>(root: &NodeEnum, f: &mut F) {
    f(root);
    walk_children(root, f);
}

fn walk_children<F: FnMut(&NodeEnum)>(node: &NodeEnum, f: &mut F) {
    match node {
        NodeEnum::SelectStmt(s) => {
            walk_slice(&s.from_clause, f);
            walk_slice(&s.target_list, f);
            if let Some(w) = s.where_clause.as_deref() {
                walk_node(w, f);
            }
            if let Some(wc) = &s.with_clause {
                walk_slice(&wc.ctes, f);
            }
            if let Some(l) = s.larg.as_ref() {
                walk_all_nodes(&NodeEnum::SelectStmt(l.clone()), f);
            }
            if let Some(r) = s.rarg.as_ref() {
                walk_all_nodes(&NodeEnum::SelectStmt(r.clone()), f);
            }
        }
        NodeEnum::JoinExpr(j) => {
            if let Some(l) = j.larg.as_deref() {
                walk_node(l, f);
            }
            if let Some(r) = j.rarg.as_deref() {
                walk_node(r, f);
            }
        }
        NodeEnum::RangeSubselect(rs) => {
            if let Some(q) = rs.subquery.as_deref() {
                walk_node(q, f);
            }
        }
        NodeEnum::CommonTableExpr(c) => {
            if let Some(q) = c.ctequery.as_deref() {
                walk_node(q, f);
            }
        }
        _ => {}
    }
}

fn walk_slice<F: FnMut(&NodeEnum)>(nodes: &[Node], f: &mut F) {
    for n in nodes {
        walk_node(n, f);
    }
}

#[cfg(test)]
mod tests {
    use crate::{
        core::schema::{QualifiedName, SchemaModel},
        parse::split_and_parse,
    };

    fn walk(sql: &str) -> SchemaModel {
        let (stmts, _) = split_and_parse(sql).expect("must parse");
        let mut model = SchemaModel::default();
        super::super::walk_ddl(&stmts, &mut model);
        model
    }

    #[test]
    fn create_view_inserts_view_def() {
        let model = walk(concat!(
            "CREATE TABLE public.items (id bigint NOT NULL);",
            "CREATE VIEW public.v1 AS SELECT id FROM public.items;",
        ));
        assert!(
            model
                .views
                .contains_key(&QualifiedName::new(Some("public"), "v1"))
        );
    }

    #[test]
    fn view_source_tables_populated_from_single_table() {
        let model = walk(concat!(
            "CREATE TABLE public.users (id bigint NOT NULL, name text);",
            "CREATE VIEW public.active_users AS SELECT id, name FROM public.users WHERE id > 0;",
        ));
        let key = QualifiedName::new(Some("public"), "active_users");
        let view = model.views.get(&key).expect("view must exist");
        let expected = QualifiedName::new(Some("public"), "users");
        assert!(
            view.source_tables.contains(&expected),
            "source_tables must contain public.users; got {:?}",
            view.source_tables,
        );
    }

    #[test]
    fn view_source_tables_populated_from_join() {
        let model = walk(concat!(
            "CREATE TABLE public.users (id bigint NOT NULL);",
            "CREATE TABLE public.orders (id bigint NOT NULL, user_id bigint NOT NULL);",
            "CREATE VIEW public.user_orders AS ",
            "  SELECT u.id, o.id AS order_id FROM public.users u JOIN public.orders o ON u.id = o.user_id;",
        ));
        let key = QualifiedName::new(Some("public"), "user_orders");
        let view = model.views.get(&key).expect("view must exist");
        let users = QualifiedName::new(Some("public"), "users");
        let orders = QualifiedName::new(Some("public"), "orders");
        assert!(
            view.source_tables.contains(&users),
            "source_tables must contain public.users; got {:?}",
            view.source_tables,
        );
        assert!(
            view.source_tables.contains(&orders),
            "source_tables must contain public.orders; got {:?}",
            view.source_tables,
        );
    }

    #[test]
    fn view_source_tables_deduplicates() {
        let model = walk(concat!(
            "CREATE TABLE public.items (id bigint NOT NULL);",
            "CREATE VIEW public.doubled AS ",
            "  SELECT a.id FROM public.items a JOIN public.items b ON a.id = b.id;",
        ));
        let key = QualifiedName::new(Some("public"), "doubled");
        let view = model.views.get(&key).expect("view must exist");
        let items = QualifiedName::new(Some("public"), "items");
        let count = view.source_tables.iter().filter(|t| *t == &items).count();
        assert_eq!(
            count, 1,
            "deduplication must yield exactly one items entry; got {:?}",
            view.source_tables
        );
    }
}
