olai-codegen 0.0.1

//! Analysis module for processing protobuf metadata into code generation plans
//!
//! This module takes the raw metadata extracted from protobuf files and analyzes it
//! to create a structured plan for code generation. It handles:
//!
//! - Grouping methods by service
//! - Extracting HTTP routing information
//! - Determining parameter types and sources
//! - Planning the structure of generated code
//! - Extracting managed resources from method return types
//!
//! ## Managed Resources
//!
//! Services often manage one or more resource types. These resources are automatically
//! extracted from the return types of get, create, and update methods. For example:
//!
//! ```proto
//! message Catalog {
//!   option (google.api.resource) = {
//!     type: "example.io/Catalog"
//!     pattern: "catalogs/{catalog}"
//!     plural: "catalogs"
//!     singular: "catalog"
//!   };
//!   string name = 1;
//!   // ... other fields
//! }
//!
//! service CatalogService {
//!   rpc GetCatalog(GetCatalogRequest) returns (Catalog);
//!   rpc CreateCatalog(CreateCatalogRequest) returns (Catalog);
//!   rpc UpdateCatalog(UpdateCatalogRequest) returns (Catalog);
//! }
//! ```
//!
//! The analysis will extract that `CatalogService` manages the `Catalog` resource,
//! making this information available for subsequent code generation phases.

use std::collections::{HashMap, HashSet};

use convert_case::{Case, Casing};
use tracing::warn;

use crate::Result;
use crate::google::api::FieldBehavior;
use crate::parsing::types::BaseType;
use crate::parsing::{CodeGenMetadata, MessageField, MethodMetadata, ServiceInfo};
use crate::utils::strings;

pub(crate) use types::MethodPlanner;
pub use types::{
    BodyField, GenerationPlan, ManagedResource, MethodPlan, PathParam, QueryParam, RequestParam,
    RequestType, ResourceHierarchy, ServicePlan, SkippedMethod, extract_managed_resources,
    split_body_fields,
};

mod types;

/// Analyze collected metadata and create a generation plan.
///
/// Methods with missing HTTP annotations are excluded from the plan and recorded in
/// [`GenerationPlan::skipped_methods`] so callers can distinguish "zero methods generated"
/// from "all methods were silently dropped".
///
/// Hierarchy derivation uses a two-phase cross-service algorithm:
/// 1. Build all service plans (hierarchy empty).
/// 2. Construct a global parent map across all services, then assign depth-sorted hierarchies.
pub fn analyze_metadata(metadata: &CodeGenMetadata) -> Result<GenerationPlan> {
    // Phase 1: analyze all services without hierarchy.
    let mut plans = Vec::new();
    let mut skipped_methods = Vec::new();
    for service_info in metadata.services.values() {
        let (plan, skipped) = analyze_service(metadata, service_info)?;
        plans.push(plan);
        skipped_methods.extend(skipped);
    }

    // Phase 2: build the cross-service global parent map, then assign depth-sorted hierarchies.
    let global_map = build_global_parent_map(&plans, metadata);
    let services = plans
        .into_iter()
        .map(|mut plan| {
            plan.hierarchy = derive_ordered_hierarchy(&plan, &global_map, metadata)?;
            Ok(plan)
        })
        .collect::<Result<Vec<_>>>()?;

    Ok(GenerationPlan {
        services,
        skipped_methods,
    })
}

// ── Cross-service hierarchy resolution ────────────────────────────────────────────────────────

/// Maps `(parent_resource_type, child_resource_type)` → proto field name that carries the
/// parent identifier on the child service's List request (e.g. `"catalog_name"`).
type GlobalParentMap = HashMap<(String, String), String>;

/// Build a map of immediate-parent relationships by scanning every service's List method query
/// params for `child_type`-annotated fields.
///
/// **Key semantic filter:** only params where `child_type == this service's managed resource type`
/// are recorded. This ensures that `catalog_name` on `ListTablesRequest` (child_type = Table) is
/// recorded as `(Catalog, Table)`, and `schema_name` as `(Schema, Table)` — but when walking
/// Schema's own List method, `catalog_name` (child_type = Schema) is recorded as `(Catalog, Schema)`.
/// The chain `Catalog → Schema → Table` is then reconstructable via depth analysis.
fn build_global_parent_map(plans: &[ServicePlan], metadata: &CodeGenMetadata) -> GlobalParentMap {
    let mut map: GlobalParentMap = HashMap::new();

    for plan in plans {
        let managed_type = match plan.managed_resources.first() {
            Some(r) => &r.descriptor.r#type,
            None => continue,
        };
        if managed_type.is_empty() {
            continue;
        }

        for method in &plan.methods {
            if method.request_type != RequestType::List {
                continue;
            }
            for param in method.query_parameters() {
                let Some(ref rr) = param.resource_reference else {
                    continue;
                };
                // Only record when child_type matches this service's own managed resource type.
                if rr.child_type != *managed_type {
                    continue;
                }
                if let Some(parent_type) = resolve_parent_type_from_field(&param.name, metadata) {
                    map.entry((parent_type, managed_type.clone()))
                        .or_insert_with(|| param.name.clone());
                }
            }
        }
    }

    map
}

/// Derive the ordered ancestor chain for a service's managed resource using the global parent map.
///
/// Returns entries sorted **root-first** (shallowest ancestor first), so iterating them gives
/// the correct left-to-right param order (e.g. `[catalog_name, schema_name]` for a Table service).
///
/// Returns `Err` if a cycle is detected in the resource hierarchy annotations.
fn derive_ordered_hierarchy(
    plan: &ServicePlan,
    global_map: &GlobalParentMap,
    metadata: &CodeGenMetadata,
) -> Result<Vec<ResourceHierarchy>> {
    let managed_type = match plan.managed_resources.first() {
        Some(r) => r.descriptor.r#type.clone(),
        None => return Ok(vec![]),
    };

    // Collect all (parent_resource_type, parent_field_name) entries for this managed resource.
    let mut ancestors: Vec<(usize, String, String)> = global_map
        .iter()
        .filter(|((_, child), _)| child == &managed_type)
        .map(|((parent_type, _), field_name)| {
            let depth = compute_depth(parent_type, global_map, &mut HashSet::new())?;
            Ok((depth, parent_type.clone(), field_name.clone()))
        })
        .collect::<Result<Vec<_>>>()?;

    if ancestors.is_empty() {
        return Ok(vec![]);
    }

    // Sort root-first (ascending depth).
    ancestors.sort_by_key(|(depth, _, _)| *depth);

    Ok(ancestors
        .into_iter()
        .map(|(_, parent_type, field_name)| {
            let parent_singular = field_name
                .strip_suffix("_name")
                .and_then(|s| metadata.resource_from_singular(s))
                .map(|_| field_name.strip_suffix("_name").unwrap().to_string());
            ResourceHierarchy {
                child_resource_type: managed_type.clone(),
                parent_resource_type: parent_type,
                parent_field_name: field_name,
                parent_singular,
            }
        })
        .collect())
}

/// Compute the depth of a resource type in the parent chain (0 = root, no parent in map).
///
/// Returns `Err` if a cycle is detected in the resource hierarchy annotations.
fn compute_depth(
    resource_type: &str,
    map: &GlobalParentMap,
    visited: &mut HashSet<String>,
) -> Result<usize> {
    if !visited.insert(resource_type.to_string()) {
        return Err(crate::Error::Build(format!(
            "Cycle detected in resource hierarchy at type: {resource_type}"
        )));
    }
    let parent = map
        .iter()
        .find(|((_, child), _)| child == resource_type)
        .map(|((parent, _), _)| parent.clone());

    match parent {
        None => Ok(0),
        Some(parent_type) => Ok(1 + compute_depth(&parent_type, map, visited)?),
    }
}

/// Resolve the resource type string for a proto field name by stripping `"_name"` and looking up
/// the resulting singular in `metadata`.
///
/// Returns `None` if the field name doesn't end in `"_name"` or no resource matches.
fn resolve_parent_type_from_field(field_name: &str, metadata: &CodeGenMetadata) -> Option<String> {
    field_name
        .strip_suffix("_name")
        .and_then(|s| metadata.resource_from_singular(s))
        .map(|rd| rd.r#type.clone())
}

/// Analyze a single service and create a service plan.
///
/// Returns the plan and a list of methods that were skipped due to incomplete metadata.
fn analyze_service(
    metadata: &CodeGenMetadata,
    info: &ServiceInfo,
) -> Result<(ServicePlan, Vec<SkippedMethod>)> {
    let handler_name = strings::service_to_handler_name(&info.name);
    let base_path = strings::service_to_base_path(&info.name);

    let mut method_plans = Vec::new();
    let mut skipped = Vec::new();

    for method in &info.methods {
        if let Some(method_plan) = analyze_method(metadata, method)? {
            method_plans.push(method_plan);
        } else {
            warn!(
                "Skipping method {}.{} - incomplete metadata",
                info.name, method.method_name
            );
            skipped.push(SkippedMethod {
                service_name: info.name.clone(),
                method_name: method.method_name.clone(),
                reason: "missing HTTP annotation".to_string(),
            });
        }
    }

    let managed_resources = types::extract_managed_resources(metadata, &method_plans);

    Ok((
        ServicePlan {
            service_name: info.name.clone(),
            handler_name,
            base_path,
            package: info.package.clone(),
            methods: method_plans,
            managed_resources,
            documentation: info.documentation.clone(),
            hierarchy: vec![], // filled in Phase 2 of analyze_metadata
        },
        skipped,
    ))
}

/// Analyze a single method and create a method plan.
///
/// Returns `None` if the method has incomplete metadata (e.g., missing HTTP annotation).
pub(crate) fn analyze_method(
    metadata: &CodeGenMetadata,
    method: &MethodMetadata,
) -> Result<Option<MethodPlan>> {
    let http_method = match method.http_method() {
        Some(m) => m.to_string(),
        None => {
            warn!(
                "Method {}.{} missing HTTP info",
                method.service_name, method.method_name
            );
            return Ok(None);
        }
    };

    let planner = MethodPlanner::try_new(method, metadata)?;
    let request_type = planner.request_type();
    let has_response = planner.has_response();
    let output_resource_type = planner.output_resource_type();
    let http_pattern = planner.into_http_pattern();

    let input_fields = metadata.get_message_fields(&method.input_type);
    let (path_params, query_params, body_fields) = extract_request_fields(method, &input_fields)?;

    let parameters = path_params
        .into_iter()
        .map(Into::into)
        .chain(query_params.into_iter().map(Into::into))
        .chain(body_fields.into_iter().map(Into::into))
        .collect();

    Ok(Some(MethodPlan {
        metadata: method.clone(),
        handler_function_name: method.method_name.to_case(Case::Snake),
        http_method,
        parameters,
        has_response,
        request_type,
        output_resource_type,
        http_pattern,
    }))
}

/// Extract and classify request fields from an input message into path, query, and body buckets.
///
/// - Path parameters are matched against URL template parameters and ordered accordingly.
/// - Fields annotated `OUTPUT_ONLY` are excluded entirely — they are server-generated and
///   must not appear in request extractors or client request builders.
/// - Fields matching the `body` spec (`"*"`, `""`, or a specific field name) become body fields.
/// - All remaining fields become query parameters. Fields not explicitly marked optional
///   via `UnifiedType.is_optional` are treated as required query parameters.
/// - Oneof fields are always placed in the body as optional variants.
fn extract_request_fields(
    method: &MethodMetadata,
    input_fields: &[MessageField],
) -> Result<(Vec<PathParam>, Vec<QueryParam>, Vec<BodyField>)> {
    let mut path_params = Vec::new();
    let mut query_params = Vec::new();
    let mut body_fields = Vec::new();

    let path_param_names = method.http_pattern.parameter_names();
    let body_spec = method.http_rule.body.as_str();

    // Build an O(1) lookup map for input fields by name.
    let fields_by_name: HashMap<&str, &MessageField> =
        input_fields.iter().map(|f| (f.name.as_str(), f)).collect();

    let mut processed_fields = HashSet::new();

    // Add path parameters in URL template order.
    for path_param_name in path_param_names {
        if let Some(field) = fields_by_name.get(path_param_name.as_str()) {
            // Path params are always included even if OUTPUT_ONLY (they appear in the URL, not
            // the request body), but in practice OUTPUT_ONLY fields should never be path params.
            path_params.push(PathParam {
                name: field.name.clone(),
                field_type: field.unified_type.clone(),
                documentation: field.documentation.clone(),
            });
            processed_fields.insert(field.name.as_str());
        }
    }

    // Classify remaining fields as body or query.
    for field in input_fields {
        let field_name = field.name.as_str();

        if processed_fields.contains(field_name) {
            continue;
        }

        // Skip OUTPUT_ONLY fields — they are server-generated and never provided by clients.
        if field.field_behavior.contains(&FieldBehavior::OutputOnly) {
            processed_fields.insert(field_name);
            continue;
        }

        // Oneof fields are always body fields and always optional.
        if matches!(field.unified_type.base_type, BaseType::OneOf(_)) {
            body_fields.push(BodyField {
                name: field.name.clone(),
                field_type: field.unified_type.clone().optional(),
                repeated: false,
                oneof_variants: field.oneof_variants.clone(),
                documentation: field.documentation.clone(),
            });
            processed_fields.insert(field_name);
            continue;
        }

        let is_body = match body_spec {
            "*" => true,
            "" => false,
            specific => specific == field_name,
        };

        if is_body {
            body_fields.push(BodyField {
                name: field.name.clone(),
                field_type: field.unified_type.clone(),
                repeated: field.unified_type.is_repeated,
                oneof_variants: None,
                documentation: field.documentation.clone(),
            });
        } else {
            query_params.push(QueryParam {
                name: field.name.clone(),
                field_type: field.unified_type.clone(),
                documentation: field.documentation.clone(),
                resource_reference: field.resource_reference.clone(),
            });
        }
        processed_fields.insert(field_name);
    }

    Ok((path_params, query_params, body_fields))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::google::api::{HttpRule, ResourceDescriptor, http_rule::Pattern};
    use crate::parsing::types::UnifiedType;
    use crate::parsing::{CodeGenMetadata, HttpPattern, MessageInfo, MethodMetadata, ServiceInfo};
    use std::collections::HashMap;

    fn make_metadata_with_catalog() -> CodeGenMetadata {
        let catalog_resource = ResourceDescriptor {
            r#type: "example.io/Catalog".to_string(),
            pattern: vec!["catalogs/{catalog}".to_string()],
            name_field: "name".to_string(),
            history: 0,
            plural: "catalogs".to_string(),
            singular: "catalog".to_string(),
            style: vec![],
        };
        let catalog_info = MessageInfo {
            name: "Catalog".to_string(),
            fields: vec![],
            resource_descriptor: Some(catalog_resource),
            documentation: None,
        };
        let mut messages = HashMap::new();
        messages.insert("Catalog".to_string(), catalog_info);
        CodeGenMetadata {
            messages,
            ..Default::default()
        }
    }

    fn make_get_method() -> MethodMetadata {
        MethodMetadata {
            service_name: "CatalogService".to_string(),
            method_name: "GetCatalog".to_string(),
            input_type: "GetCatalogRequest".to_string(),
            output_type: "Catalog".to_string(),
            operation: None,
            http_rule: HttpRule {
                selector: "".to_string(),
                pattern: Some(Pattern::Get("/catalogs/{name}".to_string())),
                body: "".to_string(),
                response_body: "".to_string(),
                additional_bindings: vec![],
            },
            http_pattern: HttpPattern::parse("/catalogs/{name}"),
            documentation: None,
        }
    }

    #[test]
    fn test_managed_resources_extraction() {
        let metadata = make_metadata_with_catalog();
        let service_info = ServiceInfo {
            name: "CatalogService".to_string(),
            package: "example.catalogs.v1".to_string(),
            documentation: None,
            methods: vec![make_get_method()],
        };
        let (service_plan, skipped) = analyze_service(&metadata, &service_info).unwrap();

        assert!(skipped.is_empty());
        assert_eq!(service_plan.managed_resources.len(), 1);
        assert_eq!(service_plan.managed_resources[0].type_name, "Catalog");
        assert_eq!(
            service_plan.managed_resources[0].descriptor.r#type,
            "example.io/Catalog"
        );
        assert_eq!(
            service_plan.managed_resources[0].descriptor.singular,
            "catalog"
        );
        assert_eq!(
            service_plan.managed_resources[0].descriptor.plural,
            "catalogs"
        );
    }

    #[test]
    fn test_no_duplicate_managed_resources() {
        let metadata = make_metadata_with_catalog();
        let update_method = MethodMetadata {
            service_name: "CatalogService".to_string(),
            method_name: "UpdateCatalog".to_string(),
            input_type: "UpdateCatalogRequest".to_string(),
            output_type: "Catalog".to_string(),
            operation: None,
            http_rule: HttpRule {
                selector: "".to_string(),
                pattern: Some(Pattern::Patch("/catalogs/{name}".to_string())),
                body: "*".to_string(),
                response_body: "".to_string(),
                additional_bindings: vec![],
            },
            http_pattern: HttpPattern::parse("/catalogs/{name}"),
            documentation: None,
        };
        let service_info = ServiceInfo {
            name: "CatalogService".to_string(),
            package: "example.catalogs.v1".to_string(),
            documentation: None,
            methods: vec![make_get_method(), update_method],
        };
        let (service_plan, _skipped) = analyze_service(&metadata, &service_info).unwrap();

        assert_eq!(service_plan.managed_resources.len(), 1);
        assert_eq!(service_plan.managed_resources[0].type_name, "Catalog");
    }

    #[test]
    fn test_analyze_method_missing_http_pattern_returns_none() {
        let metadata = CodeGenMetadata::default();
        let method = MethodMetadata {
            service_name: "SomeService".to_string(),
            method_name: "SomeMethod".to_string(),
            input_type: "".to_string(),
            output_type: "".to_string(),
            operation: None,
            http_rule: HttpRule {
                selector: "".to_string(),
                pattern: None,
                body: "".to_string(),
                response_body: "".to_string(),
                additional_bindings: vec![],
            },
            http_pattern: HttpPattern::parse(""),
            documentation: None,
        };
        let result = analyze_method(&metadata, &method).unwrap();
        assert!(result.is_none());
    }

    // --- extract_request_fields unit tests ---

    fn make_string_field(name: &str, optional: bool) -> MessageField {
        use crate::parsing::types::BaseType;
        MessageField {
            name: name.to_string(),
            unified_type: UnifiedType {
                base_type: BaseType::String,
                is_optional: optional,
                is_repeated: false,
            },
            documentation: None,
            oneof_variants: None,
            field_behavior: vec![],
            is_sensitive: false,
            resource_reference: None,
        }
    }

    fn make_repeated_field(name: &str) -> MessageField {
        use crate::parsing::types::BaseType;
        MessageField {
            name: name.to_string(),
            unified_type: UnifiedType {
                base_type: BaseType::String,
                is_optional: false,
                is_repeated: true,
            },
            documentation: None,
            oneof_variants: None,
            field_behavior: vec![],
            is_sensitive: false,
            resource_reference: None,
        }
    }

    fn make_method_with_pattern(pattern: Pattern, body: &str, path: &str) -> MethodMetadata {
        MethodMetadata {
            service_name: "Svc".to_string(),
            method_name: "Method".to_string(),
            input_type: "".to_string(),
            output_type: "".to_string(),
            operation: None,
            http_rule: HttpRule {
                selector: "".to_string(),
                pattern: Some(pattern),
                body: body.to_string(),
                response_body: "".to_string(),
                additional_bindings: vec![],
            },
            http_pattern: HttpPattern::parse(path),
            documentation: None,
        }
    }

    #[test]
    fn test_extract_path_params_in_url_order() {
        let method =
            make_method_with_pattern(Pattern::Get("/a/{x}/b/{y}".to_string()), "", "/a/{x}/b/{y}");
        let fields = vec![make_string_field("y", false), make_string_field("x", false)];
        let (path, query, body) = extract_request_fields(&method, &fields).unwrap();
        // Path params should be in URL order: x, y — not field declaration order
        assert_eq!(path.len(), 2);
        assert_eq!(path[0].name, "x");
        assert_eq!(path[1].name, "y");
        assert!(query.is_empty());
        assert!(body.is_empty());
    }

    #[test]
    fn test_extract_body_wildcard() {
        let method = make_method_with_pattern(Pattern::Post("/items".to_string()), "*", "/items");
        let fields = vec![
            make_string_field("name", false),
            make_string_field("description", true),
        ];
        let (path, query, body) = extract_request_fields(&method, &fields).unwrap();
        assert!(path.is_empty());
        assert!(query.is_empty());
        assert_eq!(body.len(), 2);
    }

    #[test]
    fn test_extract_specific_body_field() {
        let method = make_method_with_pattern(
            Pattern::Patch("/items/{name}".to_string()),
            "payload",
            "/items/{name}",
        );
        let fields = vec![
            make_string_field("name", false),    // path
            make_string_field("payload", false), // body (specific)
            make_string_field("extra", true),    // query
        ];
        let (path, query, body) = extract_request_fields(&method, &fields).unwrap();
        assert_eq!(path.len(), 1);
        assert_eq!(path[0].name, "name");
        assert_eq!(body.len(), 1);
        assert_eq!(body[0].name, "payload");
        assert_eq!(query.len(), 1);
        assert_eq!(query[0].name, "extra");
    }

    #[test]
    fn test_extract_no_body_spec_all_query() {
        let method = make_method_with_pattern(Pattern::Get("/items".to_string()), "", "/items");
        let fields = vec![
            make_string_field("filter", true),
            make_string_field("page_size", true),
        ];
        let (path, query, body) = extract_request_fields(&method, &fields).unwrap();
        assert!(path.is_empty());
        assert_eq!(query.len(), 2);
        assert!(body.is_empty());
    }

    #[test]
    fn test_extract_repeated_field_becomes_body_with_repeated_flag() {
        let method = make_method_with_pattern(Pattern::Post("/items".to_string()), "*", "/items");
        let fields = vec![make_repeated_field("tags")];
        let (_, _, body) = extract_request_fields(&method, &fields).unwrap();
        assert_eq!(body.len(), 1);
        assert!(body[0].repeated);
    }

    // ── Cross-service hierarchy chain resolution tests ─────────────────────────────────────────

    /// Build a three-level metadata fixture: Catalog → Schema → Table.
    ///
    /// Three services:
    /// - CatalogService: ListCatalogs (no child_type params), GetCatalog
    /// - SchemaService: ListSchemas (catalog_name with child_type = Schema), GetSchema
    /// - TableService: ListTables (catalog_name and schema_name both with child_type = Table), GetTable
    fn make_three_level_metadata() -> (CodeGenMetadata, ServiceInfo, ServiceInfo, ServiceInfo) {
        use crate::google::api::ResourceReference;
        use crate::parsing::types::BaseType;

        let mut messages = HashMap::new();

        // Catalog resource
        messages.insert(
            "Catalog".to_string(),
            MessageInfo {
                name: "Catalog".to_string(),
                fields: vec![],
                resource_descriptor: Some(ResourceDescriptor {
                    r#type: "example.io/Catalog".to_string(),
                    pattern: vec!["catalogs/{catalog}".to_string()],
                    name_field: "name".to_string(),
                    history: 0,
                    plural: "catalogs".to_string(),
                    singular: "catalog".to_string(),
                    style: vec![],
                }),
                documentation: None,
            },
        );

        // Schema resource
        messages.insert(
            "Schema".to_string(),
            MessageInfo {
                name: "Schema".to_string(),
                fields: vec![],
                resource_descriptor: Some(ResourceDescriptor {
                    r#type: "example.io/Schema".to_string(),
                    pattern: vec!["schemas/{schema}".to_string()],
                    name_field: "name".to_string(),
                    history: 0,
                    plural: "schemas".to_string(),
                    singular: "schema".to_string(),
                    style: vec![],
                }),
                documentation: None,
            },
        );

        // Table resource
        messages.insert(
            "Table".to_string(),
            MessageInfo {
                name: "Table".to_string(),
                fields: vec![],
                resource_descriptor: Some(ResourceDescriptor {
                    r#type: "example.io/Table".to_string(),
                    pattern: vec!["tables/{table}".to_string()],
                    name_field: "full_name".to_string(),
                    history: 0,
                    plural: "tables".to_string(),
                    singular: "table".to_string(),
                    style: vec![],
                }),
                documentation: None,
            },
        );

        // ListCatalogsRequest — no child_type params
        messages.insert(
            "ListCatalogsRequest".to_string(),
            MessageInfo {
                name: "ListCatalogsRequest".to_string(),
                fields: vec![],
                resource_descriptor: None,
                documentation: None,
            },
        );

        // ListSchemasRequest — catalog_name with child_type = Schema
        messages.insert(
            "ListSchemasRequest".to_string(),
            MessageInfo {
                name: "ListSchemasRequest".to_string(),
                fields: vec![MessageField {
                    name: "catalog_name".to_string(),
                    unified_type: UnifiedType {
                        base_type: BaseType::String,
                        is_optional: false,
                        is_repeated: false,
                    },
                    documentation: None,
                    oneof_variants: None,
                    field_behavior: vec![crate::google::api::FieldBehavior::Required],
                    is_sensitive: false,
                    resource_reference: Some(ResourceReference {
                        r#type: String::new(),
                        child_type: "example.io/Schema".to_string(),
                    }),
                }],
                resource_descriptor: None,
                documentation: None,
            },
        );

        // ListTablesRequest — catalog_name and schema_name, both child_type = Table (flat API)
        messages.insert(
            "ListTablesRequest".to_string(),
            MessageInfo {
                name: "ListTablesRequest".to_string(),
                fields: vec![
                    MessageField {
                        name: "catalog_name".to_string(),
                        unified_type: UnifiedType {
                            base_type: BaseType::String,
                            is_optional: false,
                            is_repeated: false,
                        },
                        documentation: None,
                        oneof_variants: None,
                        field_behavior: vec![crate::google::api::FieldBehavior::Required],
                        is_sensitive: false,
                        resource_reference: Some(ResourceReference {
                            r#type: String::new(),
                            child_type: "example.io/Table".to_string(),
                        }),
                    },
                    MessageField {
                        name: "schema_name".to_string(),
                        unified_type: UnifiedType {
                            base_type: BaseType::String,
                            is_optional: false,
                            is_repeated: false,
                        },
                        documentation: None,
                        oneof_variants: None,
                        field_behavior: vec![crate::google::api::FieldBehavior::Required],
                        is_sensitive: false,
                        resource_reference: Some(ResourceReference {
                            r#type: String::new(),
                            child_type: "example.io/Table".to_string(),
                        }),
                    },
                ],
                resource_descriptor: None,
                documentation: None,
            },
        );

        let metadata = CodeGenMetadata {
            messages,
            ..Default::default()
        };

        let catalog_svc = ServiceInfo {
            name: "CatalogService".to_string(),
            package: "example.v1".to_string(),
            documentation: None,
            methods: vec![
                MethodMetadata {
                    service_name: "CatalogService".to_string(),
                    method_name: "ListCatalogs".to_string(),
                    input_type: "ListCatalogsRequest".to_string(),
                    output_type: "ListCatalogsResponse".to_string(),
                    operation: None,
                    http_rule: HttpRule {
                        selector: "".to_string(),
                        pattern: Some(Pattern::Get("/catalogs".to_string())),
                        body: "".to_string(),
                        response_body: "".to_string(),
                        additional_bindings: vec![],
                    },
                    http_pattern: HttpPattern::parse("/catalogs"),
                    documentation: None,
                },
                MethodMetadata {
                    service_name: "CatalogService".to_string(),
                    method_name: "GetCatalog".to_string(),
                    input_type: "GetCatalogRequest".to_string(),
                    output_type: "Catalog".to_string(),
                    operation: None,
                    http_rule: HttpRule {
                        selector: "".to_string(),
                        pattern: Some(Pattern::Get("/catalogs/{name}".to_string())),
                        body: "".to_string(),
                        response_body: "".to_string(),
                        additional_bindings: vec![],
                    },
                    http_pattern: HttpPattern::parse("/catalogs/{name}"),
                    documentation: None,
                },
            ],
        };

        let schema_svc = ServiceInfo {
            name: "SchemaService".to_string(),
            package: "example.v1".to_string(),
            documentation: None,
            methods: vec![
                MethodMetadata {
                    service_name: "SchemaService".to_string(),
                    method_name: "ListSchemas".to_string(),
                    input_type: "ListSchemasRequest".to_string(),
                    output_type: "ListSchemasResponse".to_string(),
                    operation: None,
                    http_rule: HttpRule {
                        selector: "".to_string(),
                        pattern: Some(Pattern::Get("/schemas".to_string())),
                        body: "".to_string(),
                        response_body: "".to_string(),
                        additional_bindings: vec![],
                    },
                    http_pattern: HttpPattern::parse("/schemas"),
                    documentation: None,
                },
                MethodMetadata {
                    service_name: "SchemaService".to_string(),
                    method_name: "GetSchema".to_string(),
                    input_type: "GetSchemaRequest".to_string(),
                    output_type: "Schema".to_string(),
                    operation: None,
                    http_rule: HttpRule {
                        selector: "".to_string(),
                        pattern: Some(Pattern::Get("/schemas/{full_name}".to_string())),
                        body: "".to_string(),
                        response_body: "".to_string(),
                        additional_bindings: vec![],
                    },
                    http_pattern: HttpPattern::parse("/schemas/{full_name}"),
                    documentation: None,
                },
            ],
        };

        let table_svc = ServiceInfo {
            name: "TableService".to_string(),
            package: "example.v1".to_string(),
            documentation: None,
            methods: vec![
                MethodMetadata {
                    service_name: "TableService".to_string(),
                    method_name: "ListTables".to_string(),
                    input_type: "ListTablesRequest".to_string(),
                    output_type: "ListTablesResponse".to_string(),
                    operation: None,
                    http_rule: HttpRule {
                        selector: "".to_string(),
                        pattern: Some(Pattern::Get("/tables".to_string())),
                        body: "".to_string(),
                        response_body: "".to_string(),
                        additional_bindings: vec![],
                    },
                    http_pattern: HttpPattern::parse("/tables"),
                    documentation: None,
                },
                MethodMetadata {
                    service_name: "TableService".to_string(),
                    method_name: "GetTable".to_string(),
                    input_type: "GetTableRequest".to_string(),
                    output_type: "Table".to_string(),
                    operation: None,
                    http_rule: HttpRule {
                        selector: "".to_string(),
                        pattern: Some(Pattern::Get("/tables/{full_name}".to_string())),
                        body: "".to_string(),
                        response_body: "".to_string(),
                        additional_bindings: vec![],
                    },
                    http_pattern: HttpPattern::parse("/tables/{full_name}"),
                    documentation: None,
                },
            ],
        };

        (metadata, catalog_svc, schema_svc, table_svc)
    }

    /// Build service plans from three-level fixture metadata for hierarchy testing.
    fn make_plans_from_fixture() -> (Vec<ServicePlan>, CodeGenMetadata) {
        let (metadata, catalog_svc, schema_svc, table_svc) = make_three_level_metadata();
        let mut plans = Vec::new();
        for svc in &[&catalog_svc, &schema_svc, &table_svc] {
            let (plan, _) = analyze_service(&metadata, svc).unwrap();
            plans.push(plan);
        }
        (plans, metadata)
    }

    #[test]
    fn test_build_global_parent_map() {
        let (plans, metadata) = make_plans_from_fixture();
        let map = build_global_parent_map(&plans, &metadata);

        // Catalog → Schema (from ListSchemasRequest.catalog_name with child_type=Schema)
        assert_eq!(
            map.get(&(
                "example.io/Catalog".to_string(),
                "example.io/Schema".to_string()
            )),
            Some(&"catalog_name".to_string()),
            "Catalog→Schema mapping missing"
        );
        // Schema → Table (from ListTablesRequest.schema_name with child_type=Table)
        assert_eq!(
            map.get(&(
                "example.io/Schema".to_string(),
                "example.io/Table".to_string()
            )),
            Some(&"schema_name".to_string()),
            "Schema→Table mapping missing"
        );
        // Catalog → Table (flat-API artifact from ListTablesRequest.catalog_name)
        assert_eq!(
            map.get(&(
                "example.io/Catalog".to_string(),
                "example.io/Table".to_string()
            )),
            Some(&"catalog_name".to_string()),
            "Catalog→Table flat-API mapping missing"
        );
    }

    #[test]
    fn test_derive_ordered_hierarchy_three_levels() {
        let (plans, metadata) = make_plans_from_fixture();
        let map = build_global_parent_map(&plans, &metadata);
        let table_plan = plans
            .iter()
            .find(|p| p.service_name == "TableService")
            .unwrap();
        let hierarchy = derive_ordered_hierarchy(table_plan, &map, &metadata)
            .expect("no cycles in test fixture");

        assert_eq!(hierarchy.len(), 2, "expected 2 ancestors for Table");
        // Root-first: catalog_name (depth 0) before schema_name (depth 1)
        assert_eq!(hierarchy[0].parent_field_name, "catalog_name");
        assert_eq!(hierarchy[1].parent_field_name, "schema_name");
        assert_eq!(hierarchy[0].parent_resource_type, "example.io/Catalog");
        assert_eq!(hierarchy[1].parent_resource_type, "example.io/Schema");
        assert_eq!(hierarchy[0].parent_singular, Some("catalog".to_string()));
        assert_eq!(hierarchy[1].parent_singular, Some("schema".to_string()));
        // child_resource_type is the managed resource (Table) for all entries
        assert!(
            hierarchy
                .iter()
                .all(|h| h.child_resource_type == "example.io/Table")
        );
    }

    #[test]
    fn test_derive_ordered_hierarchy_two_levels() {
        let (plans, metadata) = make_plans_from_fixture();
        let map = build_global_parent_map(&plans, &metadata);
        let schema_plan = plans
            .iter()
            .find(|p| p.service_name == "SchemaService")
            .unwrap();
        let hierarchy = derive_ordered_hierarchy(schema_plan, &map, &metadata)
            .expect("no cycles in test fixture");

        assert_eq!(hierarchy.len(), 1);
        assert_eq!(hierarchy[0].parent_field_name, "catalog_name");
        assert_eq!(hierarchy[0].parent_resource_type, "example.io/Catalog");
    }

    #[test]
    fn test_derive_ordered_hierarchy_root_resource() {
        let (plans, metadata) = make_plans_from_fixture();
        let map = build_global_parent_map(&plans, &metadata);
        let catalog_plan = plans
            .iter()
            .find(|p| p.service_name == "CatalogService")
            .unwrap();
        let hierarchy = derive_ordered_hierarchy(catalog_plan, &map, &metadata)
            .expect("no cycles in test fixture");

        assert!(
            hierarchy.is_empty(),
            "root resource should have empty hierarchy"
        );
    }

    #[test]
    fn test_compute_depth_cycle_guard() {
        // A ↔ B mutual cycle: compute_depth should return Err, not panic or recurse infinitely.
        let mut map: GlobalParentMap = HashMap::new();
        map.insert(("A".to_string(), "B".to_string()), "a_name".to_string());
        map.insert(("B".to_string(), "A".to_string()), "b_name".to_string());

        assert!(compute_depth("A", &map, &mut HashSet::new()).is_err());
        assert!(compute_depth("B", &map, &mut HashSet::new()).is_err());
    }

    #[test]
    fn test_full_analyze_metadata_hierarchy_ordering() {
        let (metadata, catalog_svc, schema_svc, table_svc) = make_three_level_metadata();
        let mut services_map = HashMap::new();
        services_map.insert("CatalogService".to_string(), catalog_svc);
        services_map.insert("SchemaService".to_string(), schema_svc);
        services_map.insert("TableService".to_string(), table_svc);
        let full_metadata = CodeGenMetadata {
            messages: metadata.messages,
            services: services_map,
            ..Default::default()
        };

        let plan = analyze_metadata(&full_metadata).unwrap();
        let table_svc_plan = plan
            .services
            .iter()
            .find(|s| s.service_name == "TableService")
            .expect("TableService plan not found");

        assert_eq!(table_svc_plan.hierarchy.len(), 2);
        assert_eq!(
            table_svc_plan.hierarchy[0].parent_field_name,
            "catalog_name"
        );
        assert_eq!(table_svc_plan.hierarchy[1].parent_field_name, "schema_name");
    }
}