use pg_query::{
    NodeEnum,
    protobuf::{AlterTableType, ConstrType, RangeVar, Token},
};
use text_size::TextRange;

use crate::core::ddl::util::parse_name;

use crate::{
    core::{
        ddl::range_var_to_key,
        resolver::ResolvedRef,
        schema::{QualifiedName, SchemaModel},
    },
    parse::ParsedStatement,
};

use super::{
    emit::{emit_node, emit_range_var_direct, emit_type_name_direct, len_to_size, loc_to_size},
    lex, scope,
};

/// Emit references from DDL-specific node fields not reachable by the iterator.
///
/// Appends to `out`. Unknown or unresolvable nodes are logged at `DEBUG`.
pub fn emit_ddl_stmt(
    stmt: &ParsedStatement,
    source: &str,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    match &stmt.root {
        NodeEnum::CreateStmt(s) => emit_create_stmt_refs(s, stmt, source, model, out),
        NodeEnum::CreateFunctionStmt(s) => emit_function_refs(s, stmt, model, out),
        NodeEnum::ViewStmt(s) => emit_view_refs(s, stmt, model, out),
        NodeEnum::CreateTrigStmt(s) => emit_trigger_ref(s, stmt, source, model, out),
        NodeEnum::CreatePolicyStmt(s) => emit_policy_ref(s, stmt, model, out),
        NodeEnum::AlterTableStmt(s) => emit_alter_table_refs(s, stmt, source, model, out),
        NodeEnum::IndexStmt(s) => emit_index_ref(s, stmt, source, model, out),
        _ => {}
    }
}

fn emit_create_stmt_refs(
    stmt: &pg_query::protobuf::CreateStmt,
    parsed: &ParsedStatement,
    source: &str,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    for elt in &stmt.table_elts {
        let Some(inner) = elt.node.as_ref() else {
            continue;
        };
        match inner {
            NodeEnum::ColumnDef(col) => emit_column_def_refs(col, parsed, model, out),
            NodeEnum::Constraint(c) => emit_constraint_ref(c, None, parsed, source, model, out),
            _ => {}
        }
    }
}

fn emit_column_def_refs(
    col: &pg_query::protobuf::ColumnDef,
    parsed: &ParsedStatement,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    if let Some(tn) = col.type_name.as_ref() {
        emit_type_name_direct(tn, parsed.range, model, out);
    }
    // Inline column constraints may also carry FK references — no local_rv here.
    for cnode in &col.constraints {
        let Some(NodeEnum::Constraint(c)) = cnode.node.as_ref() else {
            continue;
        };
        // Inline FK column positions require different anchoring — out of scope.
        emit_constraint_ref_table_only(c, parsed, model, out);
    }
}

fn emit_function_refs(
    stmt: &pg_query::protobuf::CreateFunctionStmt,
    parsed: &ParsedStatement,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    if let Some(tn) = stmt.return_type.as_ref() {
        emit_type_name_direct(tn, parsed.range, model, out);
        emit_return_type_as_table(tn, parsed, model, out);
    }
    emit_param_type_refs(&stmt.parameters, parsed, model, out);
    super::body::emit_function_body_refs(stmt, model, out);
}

fn emit_param_type_refs(
    parameters: &[pg_query::protobuf::Node],
    parsed: &ParsedStatement,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    use pg_query::protobuf::FunctionParameterMode;
    for node in parameters {
        let Some(NodeEnum::FunctionParameter(p)) = node.node.as_ref() else {
            continue;
        };
        if FunctionParameterMode::try_from(p.mode) == Ok(FunctionParameterMode::FuncParamTable) {
            continue;
        }
        if let Some(tn) = p.arg_type.as_ref() {
            emit_type_name_direct(tn, parsed.range, model, out);
        }
    }
}

// emits a table/view reference when RETURNS names one (not a type)
fn emit_return_type_as_table(
    tn: &pg_query::protobuf::TypeName,
    parsed: &ParsedStatement,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    let Some((schema, name)) = parse_name(&tn.names) else {
        return;
    };
    if schema.as_deref() == Some("pg_catalog") {
        return;
    }
    let qn = QualifiedName::new(schema.as_deref(), &name);
    if !model.tables.contains_key(&qn) && !model.views.contains_key(&qn) {
        return;
    }
    let abs_start = parsed.range.start() + loc_to_size(tn.location);
    let abs_end = abs_start + len_to_size(name.len());
    out.push(ResolvedRef {
        symbol: qn.to_string(),
        range: TextRange::new(abs_start, abs_end),
        statement_range: parsed.range,
    });
}

fn emit_view_refs(
    stmt: &pg_query::protobuf::ViewStmt,
    parsed: &ParsedStatement,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    let Some(query) = stmt.query.as_deref() else {
        return;
    };
    let scope = query
        .node
        .as_ref()
        .map(|n| scope::build_scope(n, model))
        .unwrap_or_default();
    emit_node(query, parsed.range, &scope, model, out);
}

// pre-registers the target table so bare column refs resolve without a FROM clause
fn build_table_scope(
    rv: Option<&pg_query::protobuf::RangeVar>,
    model: &SchemaModel,
) -> scope::ScopeStack {
    let mut stack = scope::ScopeStack::default();
    let Some(rv) = rv else { return stack };
    let qn = crate::core::ddl::range_var_to_key(rv);
    if model.tables.contains_key(&qn) || model.views.contains_key(&qn) {
        stack.from_scope.push(qn);
    }
    stack
}

fn emit_trigger_ref(
    stmt: &pg_query::protobuf::CreateTrigStmt,
    parsed: &ParsedStatement,
    source: &str,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    let Some(rv) = stmt.relation.as_ref() else {
        tracing::debug!("CreateTrigStmt has no relation, skipping");
        return;
    };
    emit_range_var_direct(rv, parsed.range, model, out);
    if let Some(when) = stmt.when_clause.as_deref() {
        let scope = build_table_scope(stmt.relation.as_ref(), model);
        emit_node(when, parsed.range, &scope, model, out);
    }
    emit_trigger_func_ref(parsed, source, model, out);
}

fn emit_trigger_func_ref(
    parsed: &ParsedStatement,
    source: &str,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    let stmt_text = &source[parsed.range];
    let Some((sym_str, range)) = lex::function_name_after_execute(stmt_text, parsed.range.start())
    else {
        return;
    };
    let qn = QualifiedName::from_symbol_str(&sym_str);
    if model.functions.contains_key(&qn) {
        out.push(ResolvedRef {
            symbol: sym_str,
            range,
            statement_range: parsed.range,
        });
    }
}

fn emit_alter_table_refs(
    stmt: &pg_query::protobuf::AlterTableStmt,
    parsed: &ParsedStatement,
    source: &str,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    for cmd_node in &stmt.cmds {
        let Some(NodeEnum::AlterTableCmd(cmd)) = cmd_node.node.as_ref() else {
            continue;
        };
        if AlterTableType::try_from(cmd.subtype) != Ok(AlterTableType::AtAddConstraint) {
            continue;
        }
        let Some(def_node) = cmd.def.as_ref() else {
            continue;
        };
        let Some(NodeEnum::Constraint(c)) = def_node.node.as_ref() else {
            continue;
        };
        emit_constraint_ref(c, stmt.relation.as_ref(), parsed, source, model, out);
    }
}

// inline column constraints: emit table ref only; column position anchoring is deferred
fn emit_constraint_ref_table_only(
    c: &pg_query::protobuf::Constraint,
    parsed: &ParsedStatement,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    if ConstrType::try_from(c.contype) != Ok(ConstrType::ConstrForeign) {
        return;
    }
    let Some(rv) = c.pktable.as_ref() else {
        tracing::debug!("FK Constraint has no pktable, skipping");
        return;
    };
    emit_range_var_direct(rv, parsed.range, model, out);
}

// local_rv is the ALTER TABLE target; None skips fk_attrs column emission
fn emit_constraint_ref(
    c: &pg_query::protobuf::Constraint,
    local_rv: Option<&RangeVar>,
    parsed: &ParsedStatement,
    source: &str,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    if ConstrType::try_from(c.contype) != Ok(ConstrType::ConstrForeign) {
        return;
    }
    let Some(rv) = c.pktable.as_ref() else {
        tracing::debug!("FK Constraint has no pktable, skipping");
        return;
    };
    emit_range_var_direct(rv, parsed.range, model, out);

    let stmt_text = &source[parsed.range];
    let tokens = match pg_query::scan(stmt_text) {
        Ok(r) => r.tokens,
        Err(e) => {
            tracing::debug!("FK scan failed: {e}");
            return;
        }
    };

    emit_fk_attrs(c, local_rv, parsed, &tokens, model, out);
    emit_pk_attrs(c, parsed, &tokens, model, out);
}

fn emit_fk_attrs(
    c: &pg_query::protobuf::Constraint,
    local_rv: Option<&RangeVar>,
    parsed: &ParsedStatement,
    tokens: &[pg_query::protobuf::ScanToken],
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    let Some(rv) = local_rv else { return };
    let local_qn = range_var_to_key(rv);
    let Some(local_table) = model.tables.get(&local_qn) else {
        return;
    };
    let fk_names = string_attrs(&c.fk_attrs);
    if fk_names.is_empty() {
        return;
    }
    let fk_ranges = lex::idents_in_parens_after_tokens(tokens, parsed.range.start(), Token::Key);
    for (name, range) in fk_names.iter().zip(fk_ranges.iter()) {
        if local_table.columns.contains_key(name.as_str()) {
            out.push(ResolvedRef {
                symbol: format!("{local_qn}#{name}"),
                range: *range,
                statement_range: parsed.range,
            });
        }
    }
}

fn emit_pk_attrs(
    c: &pg_query::protobuf::Constraint,
    parsed: &ParsedStatement,
    tokens: &[pg_query::protobuf::ScanToken],
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    let Some(pktable_rv) = c.pktable.as_ref() else {
        return;
    };
    let pk_qn = range_var_to_key(pktable_rv);
    let Some(pk_table) = model.tables.get(&pk_qn) else {
        return;
    };
    let pk_names = string_attrs(&c.pk_attrs);
    if pk_names.is_empty() {
        return;
    }
    let pk_ranges =
        lex::idents_in_parens_after_tokens(tokens, parsed.range.start(), Token::References);
    for (name, range) in pk_names.iter().zip(pk_ranges.iter()) {
        if pk_table.columns.contains_key(name.as_str()) {
            out.push(ResolvedRef {
                symbol: format!("{pk_qn}#{name}"),
                range: *range,
                statement_range: parsed.range,
            });
        }
    }
}

fn string_attrs(nodes: &[pg_query::protobuf::Node]) -> Vec<String> {
    nodes
        .iter()
        .filter_map(|n| n.node.as_ref())
        .filter_map(|n| {
            if let NodeEnum::String(s) = n {
                Some(s.sval.clone())
            } else {
                None
            }
        })
        .collect()
}

fn emit_index_ref(
    stmt: &pg_query::protobuf::IndexStmt,
    parsed: &ParsedStatement,
    source: &str,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    let Some(rv) = stmt.relation.as_ref() else {
        tracing::debug!("IndexStmt has no relation, skipping");
        return;
    };
    emit_range_var_direct(rv, parsed.range, model, out);
    emit_index_column_refs(stmt, rv, parsed, source, model, out);
}

fn emit_index_column_refs(
    stmt: &pg_query::protobuf::IndexStmt,
    rv: &RangeVar,
    parsed: &ParsedStatement,
    source: &str,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    let table_qn = range_var_to_key(rv);
    let Some(table) = model.tables.get(&table_qn) else {
        return;
    };

    let named_params: Vec<&str> = stmt
        .index_params
        .iter()
        .filter_map(|n| n.node.as_ref())
        .filter_map(|n| {
            if let NodeEnum::IndexElem(e) = n {
                Some(e.name.as_str())
            } else {
                None
            }
        })
        .filter(|name| !name.is_empty())
        .collect();

    if named_params.is_empty() {
        return;
    }

    let stmt_text = &source[parsed.range];
    let col_ranges = lex::idents_in_parens_after(stmt_text, parsed.range.start(), Token::On);

    for (col_name, range) in named_params.iter().zip(col_ranges.iter()) {
        if table.columns.contains_key(*col_name) {
            out.push(ResolvedRef {
                symbol: format!("{table_qn}#{col_name}"),
                range: *range,
                statement_range: parsed.range,
            });
        }
    }
}

fn emit_policy_ref(
    stmt: &pg_query::protobuf::CreatePolicyStmt,
    parsed: &ParsedStatement,
    model: &SchemaModel,
    out: &mut Vec<ResolvedRef>,
) {
    let Some(rv) = stmt.table.as_ref() else {
        tracing::debug!("CreatePolicyStmt has no table, skipping");
        return;
    };
    emit_range_var_direct(rv, parsed.range, model, out);
    let scope = build_table_scope(stmt.table.as_ref(), model);
    for expr in [stmt.qual.as_deref(), stmt.with_check.as_deref()]
        .into_iter()
        .flatten()
    {
        emit_node(expr, parsed.range, &scope, model, out);
    }
}

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

    const BASE_SCHEMA: &str = concat!(
        "CREATE TABLE public.users (id bigint NOT NULL, email text NOT NULL, created_at timestamptz);",
        "CREATE TABLE public.orders (id bigint NOT NULL, user_id bigint NOT NULL, total numeric(10,2));",
        "CREATE TYPE public.order_status AS ENUM ('pending', 'shipped', 'delivered');",
    );

    fn base_model() -> SchemaModel {
        let (stmts, _) = split_and_parse(BASE_SCHEMA).expect("base schema must parse");
        let mut model = SchemaModel::default();
        walk_ddl(&stmts, &mut model);
        model
    }

    fn ddl_refs(sql: &str) -> Vec<ResolvedRef> {
        let model = base_model();
        let (stmts, _) = split_and_parse(sql).expect("sql must parse");
        let mut out = Vec::new();
        for stmt in &stmts {
            emit_ddl_stmt(stmt, sql, &model, &mut out);
        }
        out
    }

    fn self_schema_refs(sql: &str) -> Vec<ResolvedRef> {
        let (stmts, _) = split_and_parse(sql).expect("sql must parse");
        let mut model = SchemaModel::default();
        walk_ddl(&stmts, &mut model);
        let mut out = Vec::new();
        for stmt in &stmts {
            emit_ddl_stmt(stmt, sql, &model, &mut out);
        }
        out
    }

    fn has_sym(refs: &[ResolvedRef], sym: &str) -> bool {
        refs.iter().any(|r| r.symbol == sym)
    }

    #[test]
    fn fk_inline_constraint_emits_ref() {
        let refs = ddl_refs(
            "CREATE TABLE public.r (id bigint NOT NULL, user_id bigint REFERENCES public.users(id));",
        );
        assert!(
            has_sym(&refs, "public/users"),
            "FK must emit ref; got {refs:?}"
        );
    }

    #[test]
    fn column_enum_type_emits_ref() {
        let refs = ddl_refs(
            "CREATE TABLE public.s (id bigint NOT NULL, status public.order_status NOT NULL);",
        );
        assert!(
            has_sym(&refs, "public/order_status"),
            "enum type must emit ref; got {refs:?}"
        );
    }

    #[test]
    fn function_body_emits_table_ref() {
        let refs = ddl_refs(
            "CREATE FUNCTION public.f(p bigint) RETURNS bigint LANGUAGE sql AS $$ SELECT id FROM public.orders WHERE id = p $$;",
        );
        assert!(
            has_sym(&refs, "public/orders"),
            "function body must emit ref; got {refs:?}"
        );
    }

    #[test]
    fn trigger_relation_emits_ref() {
        let refs = ddl_refs(
            "CREATE TRIGGER t AFTER INSERT ON public.users FOR EACH ROW EXECUTE FUNCTION public.f();",
        );
        assert!(
            has_sym(&refs, "public/users"),
            "trigger must emit ref; got {refs:?}"
        );
    }

    #[test]
    fn policy_table_emits_ref() {
        let refs = ddl_refs("CREATE POLICY p ON public.users USING (id = 1);");
        assert!(
            has_sym(&refs, "public/users"),
            "policy must emit ref; got {refs:?}"
        );
    }

    #[test]
    fn alter_table_add_fk_constraint_emits_ref() {
        let refs = ddl_refs(concat!(
            "CREATE TABLE public.items (id bigint NOT NULL, order_id bigint NOT NULL);",
            "ALTER TABLE public.items ADD CONSTRAINT items_order_id_fkey ",
            "FOREIGN KEY (order_id) REFERENCES public.orders(id);",
        ));
        assert!(
            has_sym(&refs, "public/orders"),
            "ALTER TABLE ADD FK must emit ref to referenced table; got {refs:?}"
        );
    }

    #[test]
    fn create_index_emits_table_ref() {
        let refs = ddl_refs("CREATE INDEX idx ON public.orders (total);");
        assert!(
            has_sym(&refs, "public/orders"),
            "CREATE INDEX must emit ref to table; got {refs:?}"
        );
    }

    #[test]
    fn create_unique_index_emits_table_ref() {
        let refs = ddl_refs("CREATE UNIQUE INDEX idx ON public.users (email);");
        assert!(
            has_sym(&refs, "public/users"),
            "CREATE UNIQUE INDEX must emit ref to table; got {refs:?}"
        );
    }

    #[test]
    fn function_param_user_type_emits_occurrence() {
        let refs = ddl_refs(
            "CREATE FUNCTION public.f(p_status public.order_status) RETURNS bigint LANGUAGE sql AS $$ SELECT 1 $$;",
        );
        assert!(
            has_sym(&refs, "public/order_status"),
            "function param with user type must emit reference occurrence; got {refs:?}",
        );
    }

    #[test]
    fn function_with_regular_in_param_emits_occurrence_for_that_param() {
        let refs = ddl_refs(
            "CREATE FUNCTION public.f(p public.order_status) RETURNS bigint LANGUAGE sql AS $$ SELECT 1 $$;",
        );
        let param_refs: Vec<_> = refs
            .iter()
            .filter(|r| r.symbol == "public/order_status")
            .collect();
        assert!(
            !param_refs.is_empty(),
            "IN param with user type must emit occurrence; got {refs:?}",
        );
    }

    #[test]
    fn view_body_table_ref_emits_occurrence() {
        let refs = ddl_refs("CREATE VIEW public.v AS SELECT id FROM public.users;");
        assert!(
            has_sym(&refs, "public/users"),
            "view body must emit table ref; got {refs:?}"
        );
    }

    #[test]
    fn view_body_column_ref_emits_occurrence() {
        let refs = ddl_refs("CREATE VIEW public.v AS SELECT id FROM public.users;");
        assert!(
            has_sym(&refs, "public/users#id"),
            "view body must emit column ref; got {refs:?}"
        );
    }

    #[test]
    fn view_body_join_emits_both_table_refs() {
        let refs = ddl_refs(concat!(
            "CREATE VIEW public.v AS ",
            "SELECT u.id FROM public.users u JOIN public.orders o ON u.id = o.user_id;",
        ));
        assert!(
            has_sym(&refs, "public/users"),
            "view body join must emit users ref; got {refs:?}"
        );
        assert!(
            has_sym(&refs, "public/orders"),
            "view body join must emit orders ref; got {refs:?}"
        );
    }

    #[test]
    fn view_body_aliased_table_resolves_column() {
        let refs = ddl_refs("CREATE VIEW public.v AS SELECT u.id FROM public.users u;");
        assert!(
            has_sym(&refs, "public/users#id"),
            "aliased table col ref must resolve; got {refs:?}"
        );
    }

    const FUNC_SCHEMA: &str = concat!(
        "CREATE TABLE public.docs (id bigint NOT NULL);",
        "CREATE FUNCTION public.update_updated_at_column() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RETURN NEW; END $$;",
    );

    fn func_model() -> SchemaModel {
        let (stmts, _) = split_and_parse(FUNC_SCHEMA).expect("func schema must parse");
        let mut model = SchemaModel::default();
        walk_ddl(&stmts, &mut model);
        model
    }

    fn func_refs(trigger_sql: &str) -> Vec<ResolvedRef> {
        let model = func_model();
        let full_sql = format!("{FUNC_SCHEMA}{trigger_sql}");
        let (stmts, _) = split_and_parse(&full_sql).expect("sql must parse");
        let mut out = Vec::new();
        for stmt in &stmts {
            emit_ddl_stmt(stmt, &full_sql, &model, &mut out);
        }
        out
    }

    #[test]
    fn trigger_execute_function_emits_func_ref() {
        let refs = func_refs(concat!(
            "CREATE TRIGGER docs_updated_at BEFORE UPDATE ON public.docs ",
            "FOR EACH ROW EXECUTE FUNCTION public.update_updated_at_column();",
        ));
        assert!(
            has_sym(&refs, "public/update_updated_at_column"),
            "trigger must emit ref to EXECUTE FUNCTION target; got {refs:?}"
        );
    }

    #[test]
    fn trigger_execute_function_not_in_model_skipped() {
        let refs = func_refs(concat!(
            "CREATE TRIGGER docs_updated_at BEFORE UPDATE ON public.docs ",
            "FOR EACH ROW EXECUTE FUNCTION public.nonexistent_func();",
        ));
        assert!(
            !has_sym(&refs, "public/nonexistent_func"),
            "unknown function must not emit ref; got {refs:?}"
        );
    }

    #[test]
    fn trigger_no_execute_function_no_crash() {
        // A degenerate trigger definition without the standard EXECUTE clause —
        // the parser still accepts it. Must not panic.
        let refs = ddl_refs(
            "CREATE TRIGGER t AFTER INSERT ON public.users FOR EACH ROW EXECUTE FUNCTION public.f();",
        );
        // public.f is not in base_model, so no func ref — but also no panic.
        assert!(
            has_sym(&refs, "public/users"),
            "table ref must still emit; got {refs:?}"
        );
    }

    #[test]
    fn trigger_with_no_when_clause_emits_only_table_ref() {
        let refs = ddl_refs(
            "CREATE TRIGGER t AFTER INSERT ON public.users FOR EACH ROW EXECUTE FUNCTION public.f();",
        );
        assert!(
            has_sym(&refs, "public/users"),
            "trigger must emit table ref; got {refs:?}"
        );
    }

    #[test]
    fn trigger_when_clause_column_ref_attempted() {
        let refs = self_schema_refs(concat!(
            "CREATE TABLE public.t (active boolean);",
            "CREATE TRIGGER trg AFTER UPDATE ON public.t ",
            "FOR EACH ROW WHEN (NEW.active = true) EXECUTE FUNCTION public.f();",
        ));
        assert!(
            has_sym(&refs, "public/t"),
            "trigger table ref must still emit; got {refs:?}"
        );
    }

    #[test]
    fn policy_no_expressions_emits_only_table_ref() {
        let refs = ddl_refs("CREATE POLICY p ON public.users AS PERMISSIVE FOR ALL TO PUBLIC;");
        assert!(
            has_sym(&refs, "public/users"),
            "policy must emit table ref; got {refs:?}"
        );
    }

    #[test]
    fn policy_using_column_ref_emits_occurrence() {
        let refs = self_schema_refs(concat!(
            "CREATE TABLE public.t (user_id bigint);",
            "CREATE POLICY p ON public.t USING (user_id = 1);",
        ));
        assert!(
            has_sym(&refs, "public/t#user_id"),
            "policy USING must emit column ref; got {refs:?}"
        );
    }

    #[test]
    fn policy_with_check_column_ref_emits_occurrence() {
        let refs = self_schema_refs(concat!(
            "CREATE TABLE public.t (user_id bigint);",
            "CREATE POLICY p ON public.t WITH CHECK (user_id = 1);",
        ));
        assert!(
            has_sym(&refs, "public/t#user_id"),
            "policy WITH CHECK must emit column ref; got {refs:?}"
        );
    }

    #[test]
    fn index_single_column_emits_column_ref() {
        let sql = concat!(
            "CREATE TABLE public.t (id bigint);",
            "CREATE INDEX idx ON public.t (id);",
        );
        let refs = self_schema_refs(sql);
        assert!(
            has_sym(&refs, "public/t#id"),
            "single-column index must emit column ref; got {refs:?}"
        );
    }

    #[test]
    fn index_multi_column_emits_all_column_refs() {
        let sql = concat!(
            "CREATE TABLE public.t (a bigint, b bigint);",
            "CREATE INDEX idx ON public.t (a, b);",
        );
        let refs = self_schema_refs(sql);
        assert!(
            has_sym(&refs, "public/t#a"),
            "must emit ref for first col; got {refs:?}"
        );
        assert!(
            has_sym(&refs, "public/t#b"),
            "must emit ref for second col; got {refs:?}"
        );
    }

    #[test]
    fn index_column_not_in_table_skipped() {
        let sql = concat!(
            "CREATE TABLE public.t (id bigint);",
            "CREATE INDEX idx ON public.t (nonexistent);",
        );
        let refs = self_schema_refs(sql);
        assert!(
            !has_sym(&refs, "public/t#nonexistent"),
            "column not in table must not emit ref; got {refs:?}"
        );
    }

    #[test]
    fn index_expression_column_skipped() {
        // Expression index elements have an empty `name` field. No panic expected.
        let sql = concat!(
            "CREATE TABLE public.t (id bigint);",
            "CREATE INDEX idx ON public.t ((id + 1));",
        );
        // Must not panic; ref count is irrelevant.
        let refs = self_schema_refs(sql);
        assert!(
            !has_sym(&refs, "public/t#"),
            "expression element must not emit empty-name ref; got {refs:?}"
        );
    }

    #[test]
    fn index_column_name_in_index_name_resolves_correctly() {
        // "idx_token" contains "token" in the index name. Only the column inside
        // the parens after ON must match.
        let sql = concat!(
            "CREATE TABLE public.t (token bigint);",
            "CREATE INDEX idx_token ON public.t (token);",
        );
        let refs = self_schema_refs(sql);
        let col_refs: Vec<_> = refs
            .iter()
            .filter(|r| r.symbol == "public/t#token")
            .collect();
        assert_eq!(
            col_refs.len(),
            1,
            "exactly one column ref for token; got {refs:?}"
        );
    }

    const FK_SCHEMA: &str = concat!(
        "CREATE TABLE public.users (id bigint NOT NULL, name text NOT NULL);",
        "CREATE TABLE public.orders (id bigint NOT NULL, user_id bigint NOT NULL);",
    );

    fn fk_model() -> SchemaModel {
        let (stmts, _) = split_and_parse(FK_SCHEMA).expect("fk schema must parse");
        let mut model = SchemaModel::default();
        walk_ddl(&stmts, &mut model);
        model
    }

    fn fk_refs(alter_sql: &str) -> Vec<ResolvedRef> {
        let model = fk_model();
        let full_sql = format!("{FK_SCHEMA}{alter_sql}");
        let (stmts, _) = split_and_parse(&full_sql).expect("sql must parse");
        let mut out = Vec::new();
        for stmt in &stmts {
            emit_ddl_stmt(stmt, &full_sql, &model, &mut out);
        }
        out
    }

    #[test]
    fn alter_table_fk_emits_fk_col_ref() {
        let refs = fk_refs(concat!(
            "ALTER TABLE public.orders ADD CONSTRAINT orders_user_id_fkey ",
            "FOREIGN KEY (user_id) REFERENCES public.users(id);",
        ));
        assert!(
            has_sym(&refs, "public/orders#user_id"),
            "FK local column must emit ref; got {refs:?}"
        );
    }

    #[test]
    fn alter_table_fk_emits_pk_col_ref() {
        let refs = fk_refs(concat!(
            "ALTER TABLE public.orders ADD CONSTRAINT orders_user_id_fkey ",
            "FOREIGN KEY (user_id) REFERENCES public.users(id);",
        ));
        assert!(
            has_sym(&refs, "public/users#id"),
            "FK referenced column must emit ref; got {refs:?}"
        );
    }

    #[test]
    fn alter_table_fk_col_not_in_local_table_skipped() {
        let refs = fk_refs(concat!(
            "ALTER TABLE public.orders ADD CONSTRAINT orders_bad_fkey ",
            "FOREIGN KEY (nonexistent) REFERENCES public.users(id);",
        ));
        assert!(
            !has_sym(&refs, "public/orders#nonexistent"),
            "unknown fk col must not emit ref; got {refs:?}"
        );
    }

    #[test]
    fn alter_table_fk_pk_col_not_in_ref_table_skipped() {
        let refs = fk_refs(concat!(
            "ALTER TABLE public.orders ADD CONSTRAINT orders_bad_fkey ",
            "FOREIGN KEY (user_id) REFERENCES public.users(nonexistent);",
        ));
        assert!(
            !has_sym(&refs, "public/users#nonexistent"),
            "unknown pk col must not emit ref; got {refs:?}"
        );
    }
}
