mig-bo4e 0.1.55

//! PID field requirements — built from PID schema JSON + TOML mappings.
//!
//! Used by `validate_pid()` to check whether a populated BO4E struct
//! has all required fields for a given PID.

use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;

use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::definition::{FieldMapping, MappingDefinition};
use crate::error::MappingError;
use crate::pid_schema_index::PidSchemaIndex;
use crate::MappingEngine;

/// Complete field requirements for a single PID.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PidRequirements {
    pub pid: String,
    pub beschreibung: String,
    pub entities: Vec<EntityRequirement>,
}

/// The scope of an entity within a PID interchange.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum EntityScope {
    /// Message-level entity (e.g., SG2, SG2.SG3 — Marktteilnehmer, Kontakt).
    /// Present once per message, outside transactions.
    Message,
    /// Transaction-level entity (e.g., SG4.*, SG4.SG8.SG10 — Prozessdaten, Marktlokation).
    /// Present within each transaction.
    Transaction,
}

/// Requirements for one BO4E entity within a PID.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityRequirement {
    pub entity: String,
    pub bo4e_type: String,
    pub companion_type: Option<String>,
    pub ahb_status: String,
    pub is_array: bool,
    pub fields: Vec<FieldRequirement>,
    /// If this entity is eligible for map-keying (BTreeMap instead of Vec),
    /// the key field name and its allowed values.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub map_key: Option<EntityMapKeyInfo>,
    /// Whether this entity is at message level or transaction level.
    #[serde(default = "default_scope")]
    pub scope: EntityScope,
}

fn default_scope() -> EntityScope {
    EntityScope::Transaction
}

/// Map-key metadata for a discriminated entity.
///
/// When an array entity has a well-known discriminator field with a fixed set
/// of values, it can be represented as a `BTreeMap<K, T>` instead of `Vec<T>`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityMapKeyInfo {
    /// The BO4E field name that acts as the key (e.g., "marktrolle").
    pub field: String,
    /// The allowed key values with their codes and descriptions.
    pub values: Vec<EntityMapKeyValue>,
}

/// A single allowed value for a map key.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityMapKeyValue {
    pub code: String,
    pub name: String,
}

/// Requirements for a single field within an entity.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FieldRequirement {
    pub bo4e_name: String,
    pub ahb_status: String,
    pub is_companion: bool,
    pub field_type: String,
    pub format: Option<String>,
    pub enum_name: Option<String>,
    pub valid_codes: Vec<CodeValue>,
    /// For fields from nested child groups (e.g., SG10 under SG8):
    /// tracks the child group name and its max_reps.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub child_group: Option<ChildGroupInfo>,
}

/// Metadata about a nested child group that a field originates from.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChildGroupInfo {
    /// Source group suffix, e.g., "sg10" from "SG4.SG8.SG10".
    pub name: String,
    /// max_reps of the child group from MIG.
    pub max_reps: i32,
}

/// A valid code value with its human-readable meaning.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CodeValue {
    pub code: String,
    pub meaning: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub enum_name: Option<String>,
}

// ── Loading ────────────────────────────────────────────────────────────────

/// Load all TOML mapping definitions for a PID from the standard directory layout.
///
/// Pattern:
/// - `message_dir` — message-level definitions (unfiltered, always included)
/// - `common_dir` — common/shared definitions, filtered by PID schema
/// - `pid_dir` — PID-specific definitions (override common where matching)
pub fn load_definitions_for_pid(
    common_dir: &Path,
    pid_dir: &Path,
    message_dir: &Path,
    schema: &Value,
) -> Result<Vec<MappingDefinition>, MappingError> {
    let mut all_defs = Vec::new();

    // 1. Message-level definitions (always included, no schema filtering)
    if message_dir.exists() {
        let msg_engine = MappingEngine::load(message_dir)?;
        all_defs.extend(msg_engine.definitions().to_vec());
    }

    // 2. Common definitions filtered by schema + PID overrides
    let schema_index = PidSchemaIndex::from_json(schema);
    if common_dir.exists() && pid_dir.exists() {
        let tx_engine = MappingEngine::load_with_common(common_dir, pid_dir, &schema_index)?;
        all_defs.extend(tx_engine.definitions().to_vec());
    } else if common_dir.exists() {
        let common_engine = MappingEngine::load_common_only(common_dir, &schema_index)?;
        all_defs.extend(common_engine.definitions().to_vec());
    } else if pid_dir.exists() {
        let pid_engine = MappingEngine::load(pid_dir)?;
        all_defs.extend(pid_engine.definitions().to_vec());
    }

    Ok(all_defs)
}

// ── Building PidRequirements ───────────────────────────────────────────────

/// Intermediate builder for accumulating fields per entity.
struct EntityBuilder {
    bo4e_type: String,
    companion_type: Option<String>,
    ahb_status: String,
    is_array: bool,
    /// Shallowest source_group depth that set is_array (to prevent deeper groups from overriding)
    _shallowest_depth: usize,
    fields: BTreeMap<String, FieldRequirement>,
    /// Accumulated map_key info (set once, from source_group + schema detection).
    map_key: Option<EntityMapKeyInfo>,
    /// Source groups that contributed to this entity (for map_key detection).
    source_groups: BTreeSet<String>,
    /// Whether this entity is message-level or transaction-level.
    scope: EntityScope,
}

impl PidRequirements {
    /// Build requirements by cross-referencing PID schema JSON with TOML definitions.
    ///
    /// For each definition, resolves its `source_path` against the schema to find the
    /// group's segments, then matches each TOML field path to a schema element to
    /// extract AHB status, valid codes, format, and enum name.
    pub fn from_schema_and_definitions(schema: &Value, definitions: &[MappingDefinition]) -> Self {
        let pid = schema
            .get("pid")
            .and_then(|v| {
                v.as_u64()
                    .map(|n| n.to_string())
                    .or_else(|| v.as_str().map(|s| s.to_string()))
            })
            .unwrap_or_default();

        let beschreibung = schema
            .get("beschreibung")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();

        // Build a synthetic group JSON from root_segments for message-level defs
        let root_group_json = schema
            .get("root_segments")
            .map(|rs| serde_json::json!({ "segments": rs }));

        let mut entity_map: BTreeMap<String, EntityBuilder> = BTreeMap::new();

        for def in definitions {
            let source_path = def.meta.source_path.as_deref().unwrap_or("");

            let group_json_owned: Value;
            let group_json: &Value = if source_path.is_empty() {
                // No source_path — try deriving from source_group, else use root_segments
                if let Some(g) = resolve_schema_group_fuzzy(schema, &def.meta.source_group) {
                    group_json_owned = g;
                    &group_json_owned
                } else {
                    match &root_group_json {
                        Some(g) => g,
                        None => continue,
                    }
                }
            } else {
                match resolve_schema_group(schema, source_path) {
                    Some(g) => g,
                    None => {
                        // Try fuzzy fallback with source_group
                        if let Some(g) = resolve_schema_group_fuzzy(schema, &def.meta.source_group)
                        {
                            group_json_owned = g;
                            &group_json_owned
                        } else {
                            continue;
                        }
                    }
                }
            };

            // Skip definitions whose discriminator doesn't match the schema group.
            // This prevents common/ TOMLs (e.g., RFF+TN → transaktionsnummer) from
            // being resolved against an incompatible group (e.g., RFF+Z13 in PID 55001).
            if let Some(ref disc) = def.meta.discriminator {
                if !discriminator_matches_schema_group(disc, group_json) {
                    continue;
                }
            }

            // Get the parent group's AHB status as fallback.
            // Root-level entities (empty source_group) map message-header segments
            // (BGM, DTM) that are always mandatory per EDIFACT/AHB rules.  The PID
            // schema's root_segments don't carry ahb_status because the AHB annotates
            // them at the segment level (e.g., S_BGM AHB_Status="Muss"), which isn't
            // propagated into the schema JSON.  Treat them as "Muss" so validation
            // catches a missing Nachricht entity.
            let parent_ahb = if def.meta.source_group.is_empty() {
                "Muss"
            } else {
                group_json
                    .get("ahb_status")
                    .and_then(|v| v.as_str())
                    .unwrap_or("")
            };

            // Derive is_array from the DEPTH-1 group's max_reps only.
            // For "SG4.SG8.SG10": depth-1 is "SG8" (parts[1]).
            // For "SG4.SG6": depth-1 is "SG6" (parts[1]).
            // For "SG4": depth-1 is "SG4" itself (parts[0]).
            // Child groups (SG10 under SG8, SG6 under SG4) should not independently
            // make the entity an array — their repetitions are handled by child_group.
            let source_parts: Vec<&str> = def.meta.source_group.split('.').collect();
            let _is_depth1_group = source_parts.len() <= 2; // "SG4" or "SG4.SG8"

            let max_reps = group_json
                .get("max_reps")
                .and_then(|v| v.as_i64())
                .unwrap_or(1) as i32;

            // is_array is determined by the group's max_reps at the entity-defining
            // depth:
            // - source_group="SG2" (depth 1, message-level): uses SG2's max_reps
            //   (e.g., 2 for MS/MR variants → Marktteilnehmer is Vec)
            // - source_group="SG4.SG8" (depth 2, tx-level child): uses SG8's max_reps
            // - source_group="SG4" (depth 1, tx root): never makes entities arrays
            // - source_group="SG4.SG8.SG10" (depth 3+): child group — doesn't
            //   affect entity is_array (handled by child_group instead)
            let depth = source_parts.len();
            let is_message_level = depth == 1 && source_parts[0] != "SG4";
            let contributes_array =
                (is_message_level || depth == 2) && max_reps > 1;

            let builder = entity_map
                .entry(def.meta.entity.clone())
                .or_insert_with(|| {
                    // Determine scope: SG4-prefixed groups are transaction-level,
                    // everything else (SG2, SG2.SG3, root) is message-level.
                    let scope = if source_parts.first().map(|s| s.to_uppercase()) == Some("SG4".to_string()) {
                        EntityScope::Transaction
                    } else {
                        EntityScope::Message
                    };
                    EntityBuilder {
                        bo4e_type: def.meta.bo4e_type.clone(),
                        companion_type: None,
                        ahb_status: parent_ahb.to_string(),
                        is_array: contributes_array,
                        _shallowest_depth: depth,
                        fields: BTreeMap::new(),
                        map_key: None,
                        source_groups: BTreeSet::new(),
                        scope,
                    }
                });
            builder.source_groups.insert(def.meta.source_group.clone());

            // Only depth-2 definitions (SG4.SG8) can set is_array
            if contributes_array {
                builder.is_array = true;
            }

            if def.meta.companion_type.is_some() {
                builder.companion_type.clone_from(&def.meta.companion_type);
            }

            // Update ahb_status if we get a non-empty one
            if !parent_ahb.is_empty() && builder.ahb_status.is_empty() {
                builder.ahb_status = parent_ahb.to_string();
            }

            // Detect child group info for nested definitions (SG4.SG8.SG10 → 3 parts)
            let child_group_info = if source_parts.len() >= 3 {
                let child_name = source_parts.last().unwrap().to_lowercase();
                Some(ChildGroupInfo {
                    name: child_name,
                    max_reps,
                })
            } else {
                None
            };

            let fields_before: std::collections::BTreeSet<String> =
                builder.fields.keys().cloned().collect();

            // Process [fields]
            process_field_section(
                &def.fields,
                group_json,
                parent_ahb,
                false,
                &mut builder.fields,
            );

            // Process [companion_fields]
            if let Some(companion) = &def.companion_fields {
                process_field_section(companion, group_json, parent_ahb, true, &mut builder.fields);
            }

            // Set child_group on all newly-added fields from this definition
            if let Some(ref cg) = child_group_info {
                for (key, field) in builder.fields.iter_mut() {
                    if !fields_before.contains(key) && field.child_group.is_none() {
                        field.child_group = Some(cg.clone());
                    }
                }
            }
        }

        // Detect map_key for entities that are candidates for BTreeMap representation.
        // This post-processes the entity builders using schema group data.
        // Note: we check ALL entities, not just is_array ones, because some schemas
        // have max_reps=1 for SG2 even though it contains MS+MR (2 reps). When
        // map_key detection succeeds, we force is_array=true.
        for (entity_name, builder) in entity_map.iter_mut() {
            if builder.map_key.is_some() {
                continue;
            }
            let detected = detect_entity_map_key(entity_name, &builder.source_groups, schema);
            if let Some(mk) = detected {
                builder.map_key = Some(mk);
                builder.is_array = true; // map-keyed entities are always arrays
            }
        }

        // Convert builders to EntityRequirements
        let entities: Vec<EntityRequirement> = entity_map
            .into_iter()
            .map(|(name, builder)| EntityRequirement {
                entity: name,
                bo4e_type: builder.bo4e_type,
                companion_type: builder.companion_type,
                ahb_status: builder.ahb_status,
                is_array: builder.is_array,
                fields: builder.fields.into_values().collect(),
                map_key: builder.map_key,
                scope: builder.scope,
            })
            .collect();

        PidRequirements {
            pid,
            beschreibung,
            entities,
        }
    }
}

// ── Map-Key Detection ─────────────────────────────────────────────────────

/// Known entity→key field mappings for map-key detection.
///
/// These are hardcoded because the relationship between entity names and their
/// discriminator field names is a domain convention, not derivable from schema.
const MAP_KEY_RULES: &[(&str, &str, &str)] = &[
    // (entity_name, source_group_pattern, key_field)
    ("Marktteilnehmer", "SG2", "marktrolle"),
    ("Geschaeftspartner", "SG12", "nad_qualifier"),
    ("Kontakt", "SG3", "ctaFunctionCode"),
];

/// Detect map_key for an entity based on its source_groups and the PID schema.
///
/// For single-group patterns (e.g., SG2 with both MS/MR inside one node),
/// inspects the group's discriminator codes directly.
///
/// For split-child patterns (e.g., SG12 with separate sg12_z63, sg12_z65 nodes),
/// aggregates discriminator codes from all sibling child groups.
fn detect_entity_map_key(
    entity_name: &str,
    source_groups: &BTreeSet<String>,
    schema: &Value,
) -> Option<EntityMapKeyInfo> {
    // Find a matching rule
    let (_, sg_pattern, key_field) = MAP_KEY_RULES.iter().find(|(ent, sg_pat, _)| {
        *ent == entity_name
            && source_groups
                .iter()
                .any(|sg| sg == *sg_pat || sg.contains(sg_pat))
    })?;

    let fields = schema.get("fields")?.as_object()?;
    let sg_lower = sg_pattern.to_lowercase();

    // Strategy 1: Single-group detection (e.g., SG2 with both MS and MR)
    if let Some(group_json) = fields.get(&sg_lower) {
        if let Some(info) = detect_map_key_from_group(group_json, key_field) {
            return Some(info);
        }
    }

    // Strategy 2: Aggregate from split-child groups (e.g., sg12_z63, sg12_z65, ...)
    // For "SG12" in source_group "SG4.SG12", look in fields.sg4.children.sg12_*
    // For "SG2" at top level, look in fields.sg2_* siblings
    let info = aggregate_split_child_codes(fields, &sg_lower, key_field);
    if info.is_some() {
        return info;
    }

    // Strategy 3: For nested groups like SG4.SG12, look inside sg4's children
    for sg in source_groups {
        let parts: Vec<&str> = sg.split('.').collect();
        if parts.len() >= 2 {
            let parent_lower = parts[0].to_lowercase();
            let child_prefix = parts[1].to_lowercase();
            if child_prefix.starts_with(&sg_lower) || sg_lower.starts_with(&child_prefix) {
                // Look in parent group's children
                if let Some(parent) = find_group_in_fields(fields, &parent_lower) {
                    if let Some(children) = parent.get("children").and_then(|c| c.as_object()) {
                        let info = aggregate_codes_from_children(children, &sg_lower, key_field);
                        if info.is_some() {
                            return info;
                        }
                    }
                }
            }
        }
    }

    None
}

/// Find a group in the fields object, supporting both exact match and prefix variants.
fn find_group_in_fields<'a>(
    fields: &'a serde_json::Map<String, Value>,
    group_lower: &str,
) -> Option<&'a Value> {
    if let Some(v) = fields.get(group_lower) {
        return Some(v);
    }
    // Check for prefixed variants (sg4_xxx, etc.)
    let prefix = format!("{group_lower}_");
    for (k, v) in fields {
        if k.starts_with(&prefix) {
            return Some(v);
        }
    }
    None
}

/// Detect map_key from a single group's discriminator (works when >=2 codes in one node).
fn detect_map_key_from_group(group_json: &Value, key_field: &str) -> Option<EntityMapKeyInfo> {
    let disc = group_json.get("discriminator")?;
    let disc_element = disc.get("element")?.as_str()?;
    let disc_segment = disc.get("segment")?.as_str()?;

    let mut values: Vec<EntityMapKeyValue> = Vec::new();
    let mut seen_codes: BTreeSet<String> = BTreeSet::new();

    if let Some(segments) = group_json.get("segments").and_then(|s| s.as_array()) {
        for seg in segments {
            let seg_id = seg.get("id").and_then(|v| v.as_str()).unwrap_or_default();
            if seg_id != disc_segment {
                continue;
            }
            if let Some(elements) = seg.get("elements").and_then(|e| e.as_array()) {
                for el in elements {
                    let el_id = el.get("id").and_then(|v| v.as_str()).unwrap_or_default();
                    if el_id != disc_element {
                        continue;
                    }
                    collect_codes_from_element(el, &mut seen_codes, &mut values);
                }
            }
        }
    }

    // Also check discriminator.values fallback
    if let Some(disc_values) = disc.get("values").and_then(|v| v.as_array()) {
        for val in disc_values {
            if let Some(code) = val.as_str() {
                if seen_codes.insert(code.to_string()) {
                    values.push(EntityMapKeyValue {
                        code: code.to_string(),
                        name: String::new(),
                    });
                }
            }
        }
    }

    if values.len() >= 2 {
        Some(EntityMapKeyInfo {
            field: key_field.to_string(),
            values,
        })
    } else {
        None
    }
}

/// Collect code values from an element's codes array and nested components.
fn collect_codes_from_element(
    el: &Value,
    seen_codes: &mut BTreeSet<String>,
    values: &mut Vec<EntityMapKeyValue>,
) {
    if let Some(codes) = el.get("codes").and_then(|c| c.as_array()) {
        for code_obj in codes {
            if let Some(code) = code_obj.get("value").and_then(|v| v.as_str()) {
                if seen_codes.insert(code.to_string()) {
                    let name = code_obj
                        .get("name")
                        .and_then(|v| v.as_str())
                        .unwrap_or_default()
                        .to_string();
                    values.push(EntityMapKeyValue {
                        code: code.to_string(),
                        name,
                    });
                }
            }
        }
    }
    // Check components for nested codes
    if let Some(components) = el.get("components").and_then(|c| c.as_array()) {
        for comp in components {
            collect_codes_from_element(comp, seen_codes, values);
        }
    }
}

/// Aggregate discriminator codes from split child groups (e.g., sg12_z63, sg12_z65).
fn aggregate_split_child_codes(
    fields: &serde_json::Map<String, Value>,
    sg_lower: &str,
    key_field: &str,
) -> Option<EntityMapKeyInfo> {
    aggregate_codes_from_children(fields, sg_lower, key_field)
}

/// Aggregate codes from children whose key starts with `prefix_`.
fn aggregate_codes_from_children(
    children: &serde_json::Map<String, Value>,
    prefix: &str,
    key_field: &str,
) -> Option<EntityMapKeyInfo> {
    let prefix_underscore = format!("{prefix}_");
    let mut values: Vec<EntityMapKeyValue> = Vec::new();
    let mut seen_codes: BTreeSet<String> = BTreeSet::new();

    for (k, child) in children {
        if !k.starts_with(&prefix_underscore) && k != prefix {
            continue;
        }
        // Extract the qualifier suffix from the key (e.g., "z63" from "sg12_z63")
        if let Some(suffix) = k.strip_prefix(&prefix_underscore) {
            let code = suffix.to_uppercase();
            if seen_codes.insert(code.clone()) {
                // Try to get a name from the child's beschreibung or name
                let name = child
                    .get("beschreibung")
                    .or_else(|| child.get("name"))
                    .and_then(|v| v.as_str())
                    .unwrap_or_default()
                    .to_string();
                values.push(EntityMapKeyValue { code, name });
            }
        }

        // Also check discriminator codes inside the child
        if let Some(disc) = child.get("discriminator") {
            if let Some(disc_values) = disc.get("values").and_then(|v| v.as_array()) {
                for val in disc_values {
                    if let Some(code) = val.as_str() {
                        if seen_codes.insert(code.to_string()) {
                            values.push(EntityMapKeyValue {
                                code: code.to_string(),
                                name: String::new(),
                            });
                        }
                    }
                }
            }
        }
    }

    if values.len() >= 2 {
        Some(EntityMapKeyInfo {
            field: key_field.to_string(),
            values,
        })
    } else {
        None
    }
}

/// Check whether a TOML definition's discriminator is compatible with a schema group.
///
/// The schema group has a `discriminator` object with `{segment, element}` and the
/// group's segments contain `codes` arrays listing valid values. If the definition's
/// discriminator value (e.g., `TN` from `RFF.c506.d1153=TN`) is not among the schema
/// group's codes for that element, the definition doesn't belong to this PID.
///
/// Returns `true` if compatible (or if we can't determine — be permissive).
fn discriminator_matches_schema_group(disc: &str, group_json: &Value) -> bool {
    let Some((_, disc_value)) = disc.split_once('=') else {
        return true; // Can't parse — be permissive
    };
    // Strip occurrence suffix like "#0" from discriminator value
    let disc_value = disc_value.split('#').next().unwrap_or(disc_value);

    let Some(schema_disc) = group_json.get("discriminator") else {
        return true; // No schema discriminator — can't verify
    };
    let Some(disc_element) = schema_disc.get("element").and_then(|v| v.as_str()) else {
        return true;
    };

    // Find the discriminator element in the group's segments and check if disc_value
    // is among its codes
    let Some(segments) = group_json.get("segments").and_then(|v| v.as_array()) else {
        return true;
    };

    // Collect ALL codes for the discriminator element across all segments.
    // The schema may have multiple segment variants (e.g., two RFF segments:
    // one with Z13, another with TN) and we need to check across all of them.
    let mut found_element = false;
    let mut all_codes: Vec<String> = Vec::new();

    for seg in segments {
        let elements = match seg.get("elements").and_then(|v| v.as_array()) {
            Some(e) => e,
            None => continue,
        };
        for el in elements {
            let mut collect_codes = |element: &Value| -> bool {
                let Some(id) = element.get("id").and_then(|v| v.as_str()) else {
                    return false;
                };
                if id != disc_element {
                    return false;
                }
                if let Some(codes) = element.get("codes").and_then(|v| v.as_array()) {
                    for c in codes {
                        if let Some(v) = c.get("value").and_then(|v| v.as_str()) {
                            all_codes.push(v.to_string());
                        }
                    }
                }
                true
            };

            // Check top-level element
            if collect_codes(el) {
                found_element = true;
            }
            // Check components within composite
            if let Some(components) = el.get("components").and_then(|v| v.as_array()) {
                for comp in components {
                    if collect_codes(comp) {
                        found_element = true;
                    }
                }
            }
        }
    }

    if !found_element {
        return true; // Element not found in schema — be permissive
    }

    all_codes.iter().any(|c| c == disc_value)
}

// ── Schema Navigation Helpers ──────────────────────────────────────────────

/// Fuzzy-resolve a source_group like "SG4.SG8.SG10" against the schema.
///
/// Converts source_group parts to lowercase and finds ALL matching variant paths
/// in the schema tree, merging their segments into a synthetic group. This is needed
/// because the assembler merges same-ID groups (e.g., sg8_z01 + sg8_z75 → SG8),
/// and TOML mappings reference the merged result.
fn resolve_schema_group_fuzzy(schema: &Value, source_group: &str) -> Option<Value> {
    let parts: Vec<String> = source_group.split('.').map(|p| p.to_lowercase()).collect();
    if parts.is_empty() {
        return None;
    }

    let fields = schema.get("fields")?.as_object()?;

    // Collect ALL matching groups at each level and merge their segments
    let initial = find_all_matching_values(fields, &parts[0]);
    if initial.is_empty() {
        return None;
    }

    let mut current_groups: Vec<&Value> = initial;

    for part in &parts[1..] {
        let mut next_groups = Vec::new();
        for group in &current_groups {
            if let Some(children) = group.get("children").and_then(|c| c.as_object()) {
                next_groups.extend(find_all_matching_values(children, part));
            }
        }
        if next_groups.is_empty() {
            return None;
        }
        current_groups = next_groups;
    }

    // Merge all matching groups' segments into one synthetic group
    merge_groups_segments(&current_groups)
}

/// Find ALL values in a JSON object whose key matches exactly or has the given prefix followed by '_'.
fn find_all_matching_values<'a>(
    obj: &'a serde_json::Map<String, Value>,
    prefix: &str,
) -> Vec<&'a Value> {
    let mut results = Vec::new();
    // Exact match
    if let Some(v) = obj.get(prefix) {
        results.push(v);
        return results;
    }
    // Prefix match: "sg8" matches "sg8_z01", "sg8_z75", etc.
    let prefix_underscore = format!("{prefix}_");
    for (k, v) in obj {
        if k.starts_with(&prefix_underscore) {
            results.push(v);
        }
    }
    results
}

/// Merge segments from multiple schema groups into a single synthetic group.
///
/// Unions segments by tag (keeping all elements from all variants) and merges
/// ahb_status, taking the first non-empty one.
fn merge_groups_segments(groups: &[&Value]) -> Option<Value> {
    let mut merged_segments: Vec<Value> = Vec::new();
    let mut ahb_status = String::new();

    for group in groups {
        if ahb_status.is_empty() {
            if let Some(s) = group.get("ahb_status").and_then(|v| v.as_str()) {
                ahb_status = s.to_string();
            }
        }
        if let Some(segments) = group.get("segments").and_then(|v| v.as_array()) {
            for seg in segments {
                let seg_id = seg.get("id").and_then(|v| v.as_str()).unwrap_or("");
                // Check if we already have this segment tag
                if let Some(existing) = merged_segments
                    .iter_mut()
                    .find(|s| s.get("id").and_then(|v| v.as_str()).unwrap_or("") == seg_id)
                {
                    // Merge elements: add any elements/components not already present
                    merge_segment_elements(existing, seg);
                } else {
                    merged_segments.push(seg.clone());
                }
            }
        }
    }

    if merged_segments.is_empty() {
        return None;
    }

    let mut result = serde_json::json!({ "segments": merged_segments });
    if !ahb_status.is_empty() {
        result["ahb_status"] = Value::String(ahb_status);
    }
    // Propagate max_reps from source groups (take the max)
    let max_reps = groups
        .iter()
        .filter_map(|g| g.get("max_reps").and_then(|v| v.as_i64()))
        .max();
    if let Some(max_reps) = max_reps {
        result["max_reps"] = serde_json::Value::Number(max_reps.into());
    }
    Some(result)
}

/// Merge elements from `source` segment into `target` segment.
fn merge_segment_elements(target: &mut Value, source: &Value) {
    let source_elements = match source.get("elements").and_then(|v| v.as_array()) {
        Some(e) => e,
        None => return,
    };

    let target_elements = match target.get_mut("elements").and_then(|v| v.as_array_mut()) {
        Some(e) => e,
        None => return,
    };

    for src_elem in source_elements {
        // For composites, merge components (especially codes)
        if let Some(src_composite) = src_elem.get("composite").and_then(|v| v.as_str()) {
            if let Some(tgt_elem) = target_elements
                .iter_mut()
                .find(|e| e.get("composite").and_then(|v| v.as_str()) == Some(src_composite))
            {
                merge_composite_components(tgt_elem, src_elem);
            } else {
                target_elements.push(src_elem.clone());
            }
        } else if let Some(src_id) = src_elem.get("id").and_then(|v| v.as_str()) {
            // Simple element — add if not present
            let already_present = target_elements.iter().any(|e| {
                e.get("id").and_then(|v| v.as_str()) == Some(src_id) && e.get("composite").is_none()
            });
            if !already_present {
                target_elements.push(src_elem.clone());
            }
        }
    }
}

/// Merge components within a composite, unioning code values.
fn merge_composite_components(target: &mut Value, source: &Value) {
    let source_comps = match source.get("components").and_then(|v| v.as_array()) {
        Some(c) => c,
        None => return,
    };
    let target_comps = match target.get_mut("components").and_then(|v| v.as_array_mut()) {
        Some(c) => c,
        None => return,
    };

    for src_comp in source_comps {
        let src_id = src_comp.get("id").and_then(|v| v.as_str()).unwrap_or("");
        if let Some(tgt_comp) = target_comps
            .iter_mut()
            .find(|c| c.get("id").and_then(|v| v.as_str()).unwrap_or("") == src_id)
        {
            // Merge codes
            if let Some(src_codes) = src_comp.get("codes").and_then(|v| v.as_array()) {
                if let Some(tgt_codes) = tgt_comp.get_mut("codes").and_then(|v| v.as_array_mut()) {
                    for code in src_codes {
                        let code_val = code.get("value").and_then(|v| v.as_str()).unwrap_or("");
                        let already = tgt_codes.iter().any(|c| {
                            c.get("value").and_then(|v| v.as_str()).unwrap_or("") == code_val
                        });
                        if !already {
                            tgt_codes.push(code.clone());
                        }
                    }
                } else {
                    // Target has no codes, source does — add them
                    if let Some(tgt_comp_obj) = tgt_comp.as_object_mut() {
                        tgt_comp_obj.insert("codes".to_string(), Value::Array(src_codes.clone()));
                        // Also update type to "code" if source is code
                        if src_comp.get("type").and_then(|v| v.as_str()) == Some("code") {
                            tgt_comp_obj
                                .insert("type".to_string(), Value::String("code".to_string()));
                        }
                    }
                }
            }
        } else {
            target_comps.push(src_comp.clone());
        }
    }
}

/// Resolve a source_path like "sg4.sg5_z16" into the corresponding schema group JSON.
///
/// Navigates: `schema.fields.{part1}.children.{part2}.children.{part3}...`
fn resolve_schema_group<'a>(schema: &'a Value, source_path: &str) -> Option<&'a Value> {
    let parts: Vec<&str> = source_path.split('.').collect();
    if parts.is_empty() {
        return None;
    }

    let mut current = schema.get("fields")?.get(parts[0])?;

    for part in &parts[1..] {
        current = current.get("children")?.get(*part)?;
    }

    Some(current)
}

/// Process a TOML field section ([fields] or [companion_fields]) and resolve each
/// field against the schema to build FieldRequirements.
fn process_field_section(
    section: &indexmap::IndexMap<String, FieldMapping>,
    group_json: &Value,
    parent_ahb: &str,
    is_companion: bool,
    output: &mut BTreeMap<String, FieldRequirement>,
) {
    for (path, field_mapping) in section {
        let bo4e_name = match extract_target_name(field_mapping) {
            Some(name) if !name.is_empty() => name,
            _ => continue, // Skip empty targets / defaults-only
        };

        // Deduplicate by bo4e_name
        if output.contains_key(&bo4e_name) {
            continue;
        }

        let parsed = match parse_toml_field_path(path) {
            Some(p) => p,
            None => continue,
        };

        let (seg_tag, composite_or_element, component_id) = parsed;

        let field = if let Some(schema_elem) = find_schema_element(
            group_json,
            &seg_tag,
            &composite_or_element,
            component_id.as_deref(),
        ) {
            let mut req = build_field_requirement(&schema_elem, &bo4e_name, is_companion, parent_ahb);
            // Merge codes from ALL matching segments with the same element.
            // E.g., SG2 has two NAD segments (MS and MR) — merge both codes
            // into one list so the enum includes both variants.
            merge_codes_from_all_segments(
                group_json,
                &seg_tag,
                &composite_or_element,
                component_id.as_deref(),
                &mut req.valid_codes,
            );
            req
        } else {
            // Field not found in schema — create minimal field
            FieldRequirement {
                bo4e_name: bo4e_name.clone(),
                ahb_status: if parent_ahb.is_empty() {
                    "unknown".to_string()
                } else {
                    parent_ahb.to_string()
                },
                is_companion,
                field_type: "data".to_string(),
                format: None,
                enum_name: None,
                valid_codes: Vec::new(),
                child_group: None,
            }
        };

        output.insert(bo4e_name, field);
    }
}

/// Extract the target field name from a FieldMapping.
fn extract_target_name(fm: &FieldMapping) -> Option<String> {
    match fm {
        FieldMapping::Simple(s) => {
            if s.is_empty() {
                None
            } else {
                Some(s.clone())
            }
        }
        FieldMapping::Structured(s) => {
            if s.target.is_empty() {
                None
            } else {
                Some(s.target.clone())
            }
        }
        FieldMapping::Nested(_) => None,
    }
}

/// Parse a TOML field path into (segment_tag, composite_or_element, component).
///
/// Examples:
/// - "loc.c517.d3225" -> ("loc", "c517", Some("d3225"))
/// - "nad.d3035" -> ("nad", "d3035", None)
/// - "cav[Z91].c889.d7111" -> ("cav", "c889", Some("d7111"))
fn parse_toml_field_path(path: &str) -> Option<(String, String, Option<String>)> {
    let parts: Vec<&str> = path.split('.').collect();
    if parts.len() < 2 || parts.len() > 3 {
        return None;
    }

    // Strip qualifier from first part: "cav[Z91]" -> "cav"
    let seg_tag = parts[0].split('[').next().unwrap_or(parts[0]).to_string();

    let second = parts[1].to_string();

    let third = if parts.len() == 3 {
        Some(parts[2].to_string())
    } else {
        None
    };

    Some((seg_tag, second, third))
}

/// Merge codes from ALL matching segments into the `valid_codes` list.
///
/// When a group has multiple segments with the same tag (e.g., two NAD segments
/// for MS and MR in SG2), `find_schema_element` only returns the first match.
/// This function finds all matches and adds any codes not already in the list.
fn merge_codes_from_all_segments(
    group_json: &Value,
    seg_tag: &str,
    composite_or_element: &str,
    component_id: Option<&str>,
    valid_codes: &mut Vec<CodeValue>,
) {
    let Some(segments) = group_json.get("segments").and_then(|v| v.as_array()) else {
        return;
    };
    let seg_tag_upper = seg_tag.to_uppercase();

    let matching_segs: Vec<&Value> = segments
        .iter()
        .filter(|s| {
            s.get("id")
                .and_then(|v| v.as_str())
                .map(|id| id.to_uppercase() == seg_tag_upper)
                .unwrap_or(false)
        })
        .collect();

    // Collect existing code values for deduplication
    let existing: std::collections::HashSet<String> =
        valid_codes.iter().map(|c| c.code.clone()).collect();

    for seg in &matching_segs {
        // Try to find the same element in this segment
        if let Some(elem) = find_element_in_segment(seg, composite_or_element, component_id) {
            if let Some(codes) = elem.get("codes").and_then(|v| v.as_array()) {
                for code_val in codes {
                    let code = code_val
                        .get("value")
                        .and_then(|v| v.as_str())
                        .unwrap_or("");
                    if !code.is_empty() && !existing.contains(code) {
                        let meaning = code_val
                            .get("name")
                            .and_then(|v| v.as_str())
                            .unwrap_or("")
                            .to_string();
                        let enum_name = code_val
                            .get("enum_name")
                            .or_else(|| code_val.get("enum"))
                            .and_then(|v| v.as_str())
                            .map(|s| s.to_string());
                        valid_codes.push(CodeValue {
                            code: code.to_string(),
                            meaning,
                            enum_name,
                        });
                    }
                }
            }
        }
    }
}

/// Find a specific element within a single segment (helper for merge).
///
/// Handles both named (d3035, c517.d3225) and numeric (0, 1.0) paths.
fn find_element_in_segment(
    seg: &Value,
    composite_or_element: &str,
    component_id: Option<&str>,
) -> Option<Value> {
    let elements = seg.get("elements")?.as_array()?;

    // Check if numeric path
    let is_numeric = composite_or_element
        .chars()
        .all(|c| c.is_ascii_digit());

    if is_numeric {
        let elem_idx: usize = composite_or_element.parse().ok()?;
        // Find element at this index
        let elem = elements
            .iter()
            .find(|e| e.get("index").and_then(|v| v.as_u64()) == Some(elem_idx as u64))?;

        if let Some(comp_str) = component_id {
            if let Ok(sub_idx) = comp_str.parse::<usize>() {
                // Numeric component sub_index
                let components = elem.get("components")?.as_array()?;
                return components
                    .iter()
                    .find(|c| {
                        c.get("sub_index").and_then(|v| v.as_u64()) == Some(sub_idx as u64)
                    })
                    .cloned();
            }
        }
        // Simple element — return itself
        if elem.get("composite").is_none() {
            return Some(elem.clone());
        }
        // Composite without component — return first component
        let components = elem.get("components")?.as_array()?;
        return components.first().cloned();
    }

    let normalized_id = normalize_edifact_id(composite_or_element);
    let is_composite = normalized_id.starts_with('C');

    if is_composite {
        for elem in elements {
            let cid = elem.get("composite").and_then(|v| v.as_str()).unwrap_or("");
            if cid.to_uppercase() == normalized_id {
                if let Some(comp_id) = component_id {
                    let comp_normalized = normalize_edifact_id(comp_id);
                    let components = elem.get("components")?.as_array()?;
                    return components
                        .iter()
                        .find(|c| {
                            c.get("id")
                                .and_then(|v| v.as_str())
                                .map(|id| id.to_uppercase() == comp_normalized)
                                .unwrap_or(false)
                        })
                        .cloned();
                }
                return Some(elem.clone());
            }
        }
    } else {
        for elem in elements {
            if elem.get("composite").is_some() {
                continue;
            }
            let eid = elem.get("id").and_then(|v| v.as_str()).unwrap_or("");
            if eid.to_uppercase() == normalized_id {
                return Some(elem.clone());
            }
        }
    }
    None
}

/// Find a schema element matching a TOML field path within a group's segments.
///
/// Handles EDIFACT ID paths (`c517.d3225`), numeric paths (`2.0`),
/// and simple elements (`d3035` or `0`).
fn find_schema_element(
    group_json: &Value,
    seg_tag: &str,
    composite_or_element: &str,
    component_id: Option<&str>,
) -> Option<Value> {
    let segments = group_json.get("segments")?.as_array()?;
    let seg_tag_upper = seg_tag.to_uppercase();

    // Find matching segment(s) by id
    let matching_segs: Vec<&Value> = segments
        .iter()
        .filter(|s| {
            s.get("id")
                .and_then(|v| v.as_str())
                .map(|id| id.to_uppercase() == seg_tag_upper)
                .unwrap_or(false)
        })
        .collect();

    if matching_segs.is_empty() {
        return None;
    }

    // Check if this is a numeric index path (e.g., "2.0" instead of "c517.d3225")
    let is_numeric = composite_or_element.chars().all(|c| c.is_ascii_digit());

    if is_numeric {
        return find_schema_element_by_index(&matching_segs, composite_or_element, component_id);
    }

    let normalized_id = normalize_edifact_id(composite_or_element);
    let is_composite = normalized_id.starts_with('C');

    for seg in &matching_segs {
        let elements = seg.get("elements")?.as_array()?;

        if is_composite {
            for elem in elements {
                let composite_id = elem.get("composite").and_then(|v| v.as_str()).unwrap_or("");
                if composite_id.to_uppercase() == normalized_id {
                    if let Some(comp_id) = component_id {
                        // Check if component is numeric sub_index
                        let comp_stripped = strip_ordinal_suffix(comp_id);
                        if comp_stripped.chars().all(|c| c.is_ascii_digit()) {
                            let sub_idx: usize = comp_stripped.parse().ok()?;
                            if let Some(components) =
                                elem.get("components").and_then(|v| v.as_array())
                            {
                                return components
                                    .iter()
                                    .find(|c| {
                                        c.get("sub_index").and_then(|v| v.as_u64())
                                            == Some(sub_idx as u64)
                                    })
                                    .cloned();
                            }
                        }
                        // Named component ID
                        let comp_normalized = normalize_edifact_id(comp_id);
                        let ordinal = extract_ordinal(comp_id);
                        if let Some(components) = elem.get("components").and_then(|v| v.as_array())
                        {
                            let mut count = 0usize;
                            for component in components {
                                let cid =
                                    component.get("id").and_then(|v| v.as_str()).unwrap_or("");
                                if cid.to_uppercase() == comp_normalized {
                                    if count == ordinal {
                                        return Some(component.clone());
                                    }
                                    count += 1;
                                }
                            }
                        }
                    } else {
                        return Some(elem.clone());
                    }
                }
            }
        } else {
            // Simple data element ID like "d3227" -> normalized "3227"
            let ordinal = extract_ordinal(composite_or_element);
            let mut count = 0usize;
            for elem in elements {
                if elem.get("composite").is_some() {
                    continue;
                }
                let eid = elem.get("id").and_then(|v| v.as_str()).unwrap_or("");
                if eid.to_uppercase() == normalized_id {
                    if count == ordinal {
                        return Some(elem.clone());
                    }
                    count += 1;
                }
            }
        }
    }

    None
}

/// Find a schema element by numeric index path (e.g., "2.0" = element at index 2, component at sub_index 0).
fn find_schema_element_by_index(
    matching_segs: &[&Value],
    element_index_str: &str,
    component_sub_index_str: Option<&str>,
) -> Option<Value> {
    let elem_idx: usize = element_index_str.parse().ok()?;

    for seg in matching_segs {
        let elements = match seg.get("elements").and_then(|v| v.as_array()) {
            Some(e) => e,
            None => continue,
        };

        let elem = match elements
            .iter()
            .find(|e| e.get("index").and_then(|v| v.as_u64()) == Some(elem_idx as u64))
        {
            Some(e) => e,
            None => continue,
        };

        if let Some(comp_str) = component_sub_index_str {
            let comp_stripped = strip_ordinal_suffix(comp_str);
            if let Ok(sub_idx) = comp_stripped.parse::<usize>() {
                if let Some(components) = elem.get("components").and_then(|v| v.as_array()) {
                    return components
                        .iter()
                        .find(|c| {
                            c.get("sub_index").and_then(|v| v.as_u64()) == Some(sub_idx as u64)
                        })
                        .cloned();
                }
            }
        }

        return Some(elem.clone());
    }

    None
}

/// Build a FieldRequirement from a resolved schema element.
fn build_field_requirement(
    schema_elem: &Value,
    bo4e_name: &str,
    is_companion: bool,
    _parent_ahb: &str,
) -> FieldRequirement {
    // Only use the element's own AHB status — do NOT inherit from parent group.
    // In EDIFACT AHB, each data element has its own status. Elements without
    // explicit ahb_status are implicitly optional within the group (e.g., NAD
    // C080 sub-components for namenszusatz, additional address lines).
    let ahb_status = schema_elem
        .get("ahb_status")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();

    let field_type = schema_elem
        .get("type")
        .and_then(|v| v.as_str())
        .unwrap_or("data")
        .to_string();

    let format = schema_elem
        .get("format")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string());

    let enum_name = schema_elem
        .get("codes")
        .and_then(|v| v.as_array())
        .and_then(|codes| {
            codes
                .first()
                .and_then(|c| c.get("enum").and_then(|v| v.as_str()))
                .map(to_pascal_case)
        });

    let valid_codes = extract_valid_codes(schema_elem);

    FieldRequirement {
        bo4e_name: bo4e_name.to_string(),
        ahb_status,
        is_companion,
        field_type,
        format,
        enum_name,
        valid_codes,
        child_group: None,
    }
}

/// Extract valid code values from a schema element's "codes" array.
fn extract_valid_codes(element: &Value) -> Vec<CodeValue> {
    element
        .get("codes")
        .and_then(|v| v.as_array())
        .map(|codes| {
            codes
                .iter()
                .filter_map(|c| {
                    let code = c.get("value").and_then(|v| v.as_str())?.to_string();
                    let meaning = c
                        .get("name")
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .to_string();
                    let enum_name = c
                        .get("enum")
                        .and_then(|v| v.as_str())
                        .map(|s| s.to_string());
                    Some(CodeValue {
                        code,
                        meaning,
                        enum_name,
                    })
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Convert SCREAMING_SNAKE_CASE to PascalCase.
///
/// "HAUSHALTSKUNDE_ENWG" -> "HaushaltskundeEnwg"
fn to_pascal_case(s: &str) -> String {
    s.split('_')
        .map(|part| {
            let mut chars = part.chars();
            match chars.next() {
                Some(c) => {
                    let upper = c.to_uppercase().to_string();
                    let lower: String = chars.map(|c| c.to_ascii_lowercase()).collect();
                    format!("{upper}{lower}")
                }
                None => String::new(),
            }
        })
        .collect()
}

/// Normalize an EDIFACT ID path component to match schema IDs.
///
/// TOML paths use lowercase prefixed IDs: `c517`, `d3227`.
/// Schema JSON uses: `"C517"` for composites, `"3227"` for data elements (no D prefix).
fn normalize_edifact_id(id: &str) -> String {
    let stripped = strip_ordinal_suffix(id);
    let upper = stripped.to_uppercase();
    // Composite IDs start with C followed by digits
    if upper.starts_with('C') && upper.len() > 1 && upper[1..].chars().all(|c| c.is_ascii_digit()) {
        return upper;
    }
    // Data element IDs: strip 'D' prefix if present -> "D3227" -> "3227"
    if upper.starts_with('D') && upper.len() > 1 && upper[1..].chars().all(|c| c.is_ascii_digit()) {
        return upper[1..].to_string();
    }
    upper
}

/// Strip ordinal suffix from an EDIFACT ID: "c556_2" -> "c556", "d3036_2" -> "d3036".
fn strip_ordinal_suffix(id: &str) -> &str {
    if let Some(pos) = id.rfind('_') {
        let suffix = &id[pos + 1..];
        if !suffix.is_empty() && suffix.chars().all(|c| c.is_ascii_digit()) {
            return &id[..pos];
        }
    }
    id
}

/// Extract ordinal from a suffixed ID: "d3036_2" -> 1 (0-indexed), "d3036" -> 0.
fn extract_ordinal(id: &str) -> usize {
    if let Some(pos) = id.rfind('_') {
        let suffix = &id[pos + 1..];
        if let Ok(n) = suffix.parse::<usize>() {
            // Ordinal suffixes are 2-based (_2 = 2nd occurrence = index 1)
            return n.saturating_sub(1);
        }
    }
    0
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::Path;

    #[test]
    fn test_to_pascal_case() {
        assert_eq!(to_pascal_case("HAUSHALTSKUNDE"), "Haushaltskunde");
        assert_eq!(to_pascal_case("HAUSHALTSKUNDE_ENWG"), "HaushaltskundeEnwg");
        assert_eq!(to_pascal_case("EMAIL"), "Email");
    }

    #[test]
    fn test_parse_toml_field_path() {
        let (tag, comp, elem) = parse_toml_field_path("loc.c517.d3225").unwrap();
        assert_eq!(tag, "loc");
        assert_eq!(comp, "c517");
        assert_eq!(elem.as_deref(), Some("d3225"));

        let (tag, elem, none) = parse_toml_field_path("nad.d3035").unwrap();
        assert_eq!(tag, "nad");
        assert_eq!(elem, "d3035");
        assert!(none.is_none());

        let (tag, comp, elem) = parse_toml_field_path("cav[Z91].c889.d7111").unwrap();
        assert_eq!(tag, "cav");
        assert_eq!(comp, "c889");
        assert_eq!(elem.as_deref(), Some("d7111"));
    }

    #[test]
    fn test_normalize_edifact_id() {
        assert_eq!(normalize_edifact_id("c517"), "C517");
        assert_eq!(normalize_edifact_id("d3225"), "3225");
        assert_eq!(normalize_edifact_id("d3036_2"), "3036");
        assert_eq!(normalize_edifact_id("c556_2"), "C556");
    }

    #[test]
    fn test_extract_ordinal() {
        assert_eq!(extract_ordinal("d3036"), 0);
        assert_eq!(extract_ordinal("d3036_2"), 1);
        assert_eq!(extract_ordinal("c556_3"), 2);
    }

    #[test]
    fn test_from_schema_and_definitions_pid_55001() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Schema file not found, skipping test");
            return;
        }

        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let common_dir = base.join("common");
        let pid_dir = base.join("pid_55001");
        let message_dir = base.join("message");

        let defs = load_definitions_for_pid(&common_dir, &pid_dir, &message_dir, &schema).unwrap();
        assert!(!defs.is_empty(), "should have loaded definitions");

        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Basic metadata
        assert_eq!(reqs.pid, "55001");
        assert_eq!(reqs.beschreibung, "Anmeldung verb. MaLo");
        assert!(!reqs.entities.is_empty(), "should have entities");

        // Prozessdaten entity with pruefidentifikator field
        let prozessdaten = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Prozessdaten")
            .expect("Prozessdaten entity should exist");

        let pruefid = prozessdaten
            .fields
            .iter()
            .find(|f| f.bo4e_name == "pruefidentifikator")
            .expect("pruefidentifikator field should exist");
        assert_eq!(pruefid.ahb_status, "X");
        assert_eq!(pruefid.field_type, "code");
        assert!(
            pruefid.valid_codes.iter().any(|c| c.code == "55001"),
            "should have 55001 as valid code"
        );

        // Marktlokation entity should exist with marktlokationsId field
        let marktlokation = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Marktlokation")
            .expect("Marktlokation entity should exist");
        assert!(
            marktlokation
                .fields
                .iter()
                .any(|f| f.bo4e_name == "marktlokationsId"),
            "Marktlokation should have marktlokationsId field"
        );

        // SG10 CCI/CAV data is now merged into parent entities (no separate Zuordnung).
        // Check that Marktlokation has haushaltskunde code field (from Z01 SG10).
        let malo_code_fields: Vec<_> = marktlokation
            .fields
            .iter()
            .filter(|f| f.field_type == "code" && f.valid_codes.len() >= 2)
            .collect();
        assert!(
            !malo_code_fields.is_empty(),
            "Marktlokation should have code fields with >= 2 valid codes (haushaltskunde), fields: {:?}",
            marktlokation
                .fields
                .iter()
                .map(|f| (&f.bo4e_name, &f.field_type, f.valid_codes.len()))
                .collect::<Vec<_>>()
        );

        // Geschaeftspartner should be detected as array (PID 55001 has SG12_Z04)
        let geschaeftspartner = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Geschaeftspartner");
        // PID 55001 may have only 1 SG12 variant, so is_array depends on schema
        // At minimum, the entity should exist
        assert!(
            geschaeftspartner.is_some(),
            "Geschaeftspartner entity should exist"
        );
    }

    #[test]
    fn test_extract_valid_codes_includes_enum_name() {
        let element = serde_json::json!({
            "codes": [
                {"value": "ZF9", "name": "Der Code ist anzuwenden wenn...", "enum": "ENFG_VERRINGERUNG_UMLAGE"},
                {"value": "ZG0", "name": "Der Code ist anzuwenden wenn...", "enum": "ENFG_KEINE_VERRINGERUNG_UMLAGE"},
                {"value": "ZG1", "name": "Der Code ist anzuwenden wenn..."}
            ]
        });

        let codes = extract_valid_codes(&element);
        assert_eq!(codes.len(), 3);

        assert_eq!(codes[0].code, "ZF9");
        assert_eq!(
            codes[0].enum_name.as_deref(),
            Some("ENFG_VERRINGERUNG_UMLAGE")
        );

        assert_eq!(codes[1].code, "ZG0");
        assert_eq!(
            codes[1].enum_name.as_deref(),
            Some("ENFG_KEINE_VERRINGERUNG_UMLAGE")
        );

        assert_eq!(codes[2].code, "ZG1");
        assert_eq!(codes[2].enum_name, None, "missing enum should be None");
    }

    #[test]
    fn test_enum_name_roundtrips_through_pid_requirements() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Schema file not found, skipping test");
            return;
        }

        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let defs = load_definitions_for_pid(
            &base.join("common"),
            &base.join("pid_55001"),
            &base.join("message"),
            &schema,
        )
        .unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Find any code field with valid_codes — its enum_name should be populated
        let code_field = reqs
            .entities
            .iter()
            .flat_map(|e| &e.fields)
            .find(|f| f.field_type == "code" && !f.valid_codes.is_empty())
            .expect("should have at least one code field");

        // At least one code should have enum_name set (schema has "enum" keys)
        let has_enum_name = code_field.valid_codes.iter().any(|c| c.enum_name.is_some());
        assert!(
            has_enum_name,
            "code field {:?} should have at least one code with enum_name set, codes: {:?}",
            code_field.bo4e_name, code_field.valid_codes
        );
    }

    #[test]
    fn test_pid_55001_is_array_from_max_reps() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Schema file not found, skipping test");
            return;
        }

        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let defs = load_definitions_for_pid(
            &base.join("common"),
            &base.join("pid_55001"),
            &base.join("message"),
            &schema,
        )
        .unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // ProduktpaketDaten comes from SG8 with max_reps=99999 → should be array
        let produktpaket = reqs
            .entities
            .iter()
            .find(|e| e.entity == "ProduktpaketDaten");
        assert!(produktpaket.is_some(), "ProduktpaketDaten should exist");
        assert!(
            produktpaket.unwrap().is_array,
            "ProduktpaketDaten should be is_array=true (SG8 max_reps=99999)"
        );

        // Geschaeftspartner should still be is_array (SG12 max_reps from MIG)
        let gp = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Geschaeftspartner");
        if let Some(gp) = gp {
            // SG12 max_reps depends on the MIG; verify it's set correctly
            assert!(
                gp.is_array,
                "Geschaeftspartner should be is_array (SG12 repeats)"
            );
        }

        // Fields from SG10 (child group) should have child_group metadata
        let produktpaket = reqs
            .entities
            .iter()
            .find(|e| e.entity == "ProduktpaketDaten")
            .unwrap();
        let sg10_field = produktpaket.fields.iter().find(|f| {
            f.bo4e_name.contains("produktMerkmal") || f.bo4e_name.contains("produktWert")
        });
        assert!(
            sg10_field.is_some(),
            "ProduktpaketDaten should have SG10-sourced fields"
        );
        let cg = &sg10_field.unwrap().child_group;
        assert!(
            cg.is_some(),
            "SG10-sourced field should have child_group set, field: {:?}",
            sg10_field.unwrap().bo4e_name
        );
        assert_eq!(cg.as_ref().unwrap().name, "sg10");
    }

    #[test]
    fn test_entity_map_key_info_serde_roundtrip() {
        let info = EntityMapKeyInfo {
            field: "marktrolle".to_string(),
            values: vec![
                EntityMapKeyValue {
                    code: "MS".to_string(),
                    name: "Sender".to_string(),
                },
                EntityMapKeyValue {
                    code: "MR".to_string(),
                    name: "Recipient".to_string(),
                },
            ],
        };
        let json = serde_json::to_string(&info).unwrap();
        let roundtripped: EntityMapKeyInfo = serde_json::from_str(&json).unwrap();
        assert_eq!(roundtripped.field, "marktrolle");
        assert_eq!(roundtripped.values.len(), 2);
        assert_eq!(roundtripped.values[0].code, "MS");
        assert_eq!(roundtripped.values[1].code, "MR");
    }

    #[test]
    fn test_entity_requirement_map_key_skip_serializing_if_none() {
        let req = EntityRequirement {
            entity: "Test".to_string(),
            bo4e_type: "Test".to_string(),
            companion_type: None,
            ahb_status: "X".to_string(),
            is_array: false,
            fields: vec![],
            map_key: None,
            scope: EntityScope::Transaction,
        };
        let json = serde_json::to_string(&req).unwrap();
        assert!(
            !json.contains("map_key"),
            "map_key should be omitted when None: {json}"
        );
    }

    #[test]
    fn test_entity_requirement_map_key_serialized_when_present() {
        let req = EntityRequirement {
            entity: "Marktteilnehmer".to_string(),
            bo4e_type: "Marktteilnehmer".to_string(),
            companion_type: None,
            ahb_status: "X".to_string(),
            is_array: true,
            fields: vec![],
            map_key: Some(EntityMapKeyInfo {
                field: "marktrolle".to_string(),
                values: vec![
                    EntityMapKeyValue {
                        code: "MS".to_string(),
                        name: "Sender".to_string(),
                    },
                    EntityMapKeyValue {
                        code: "MR".to_string(),
                        name: "Recipient".to_string(),
                    },
                ],
            }),
            scope: EntityScope::Transaction,
        };
        let json = serde_json::to_string(&req).unwrap();
        assert!(
            json.contains("map_key"),
            "map_key should be present: {json}"
        );
        assert!(
            json.contains("marktrolle"),
            "field name should be present: {json}"
        );
        assert!(json.contains("MS"), "MS code should be present: {json}");
        assert!(json.contains("MR"), "MR code should be present: {json}");

        // Roundtrip
        let roundtripped: EntityRequirement = serde_json::from_str(&json).unwrap();
        let mk = roundtripped.map_key.unwrap();
        assert_eq!(mk.field, "marktrolle");
        assert_eq!(mk.values.len(), 2);
    }

    #[test]
    fn test_pid_55001_marktteilnehmer_has_map_key() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Skipping: schema not found");
            return;
        }
        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let common_dir = base.join("common");
        let pid_dir = base.join("pid_55001");
        let message_dir = base.join("message");

        let defs = load_definitions_for_pid(&common_dir, &pid_dir, &message_dir, &schema).unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Marktteilnehmer should have a map_key with marktrolle (MS + MR)
        let mt = reqs.entities.iter().find(|e| e.entity == "Marktteilnehmer");
        assert!(mt.is_some(), "should have Marktteilnehmer entity");
        let mt = mt.unwrap();
        assert!(mt.is_array, "Marktteilnehmer should be an array");
        assert!(
            mt.map_key.is_some(),
            "Marktteilnehmer should have map_key for marktrolle"
        );
        let mk = mt.map_key.as_ref().unwrap();
        assert_eq!(mk.field, "marktrolle");
        assert!(
            mk.values.len() >= 2,
            "Should have at least MS and MR, got {} values",
            mk.values.len()
        );
        let codes: Vec<&str> = mk.values.iter().map(|v| v.code.as_str()).collect();
        assert!(
            codes.contains(&"MS"),
            "Should contain MS code, got: {:?}",
            codes
        );
        assert!(
            codes.contains(&"MR"),
            "Should contain MR code, got: {:?}",
            codes
        );
    }

    #[test]
    fn test_detect_entity_map_key_geschaeftspartner_sg12() {
        // PID 55013 has 7 SG12 variants under SG4: sg12_z63, sg12_z65..sg12_z70
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55013_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Skipping: schema not found");
            return;
        }
        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let common_dir = base.join("common");
        let pid_dir = base.join("pid_55013");
        let message_dir = base.join("message");

        let defs = load_definitions_for_pid(&common_dir, &pid_dir, &message_dir, &schema).unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Geschaeftspartner should be an array entity with a map_key for nad_qualifier
        let gp = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Geschaeftspartner");
        assert!(gp.is_some(), "should have Geschaeftspartner entity");
        let gp = gp.unwrap();
        assert!(gp.is_array, "Geschaeftspartner should be an array");
        assert!(
            gp.map_key.is_some(),
            "Geschaeftspartner should have map_key for nad_qualifier"
        );
        let mk = gp.map_key.as_ref().unwrap();
        assert_eq!(mk.field, "nad_qualifier");

        // PID 55013 has 7 NAD qualifiers: Z63, Z65, Z66, Z67, Z68, Z69, Z70
        let codes: BTreeSet<&str> = mk.values.iter().map(|v| v.code.as_str()).collect();
        let expected = ["Z63", "Z65", "Z66", "Z67", "Z68", "Z69", "Z70"];
        for code in &expected {
            assert!(
                codes.contains(code),
                "Should contain {code} code, got: {:?}",
                codes
            );
        }
        assert_eq!(
            codes.len(),
            expected.len(),
            "Should have exactly {} codes, got {:?}",
            expected.len(),
            codes
        );
    }

    /// Regression test for issue #52: validate PID 55001 FV2604 requirements
    /// to catch false positives (transaktionsnummer, Kontakt, namenszusatz).
    #[test]
    fn test_issue_52_pid_55001_fv2604_false_positives() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2604/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("FV2604 schema not found, skipping");
            return;
        }

        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2604/UTILMD_Strom"
        ));
        let common_dir = base.join("common");
        let pid_dir = base.join("pid_55001");
        let message_dir = base.join("message");

        let defs = load_definitions_for_pid(&common_dir, &pid_dir, &message_dir, &schema).unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Issue #52 Q1: transaktionsnummer should NOT be in PID 55001 requirements.
        // PID 55001 SG6 only has RFF+Z13, not RFF+TN. The common _21_rff_tn.toml
        // discriminator (RFF.c506.d1153=TN) should be filtered out.
        let prozessdaten = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Prozessdaten")
            .expect("Prozessdaten should exist");
        assert!(
            !prozessdaten
                .fields
                .iter()
                .any(|f| f.bo4e_name == "transaktionsnummer"),
            "transaktionsnummer should NOT be in PID 55001 (SG6 has only RFF+Z13, not RFF+TN)"
        );

        // Issue #52 Q2: Kontakt is a message-level entity (SG2.SG3).
        // When validate_pid receives only transaction-level data, it should not
        // require Kontakt. Verify it's scoped as Message.
        let kontakt = reqs.entities.iter().find(|e| e.entity == "Kontakt");
        if let Some(kontakt) = kontakt {
            assert_eq!(
                kontakt.scope,
                EntityScope::Message,
                "Kontakt should be Message-scoped (SG2.SG3), not Transaction"
            );
        }

        // Verify that validate_pid_json_transaction skips message-level entities
        let tx_errors = crate::pid_validation::validate_pid_json_transaction(
            &serde_json::json!({}),
            &reqs,
        );
        assert!(
            !tx_errors.iter().any(|e| matches!(
                e,
                crate::pid_validation::PidValidationError::MissingEntity { entity, .. }
                if entity == "Kontakt"
            )),
            "Transaction-scoped validation should NOT report missing Kontakt"
        );

        // Issue #52 Q3: namenszusatz1/namenszusatz2 should NOT be required.
        // These C080 components (sub_index 2,3) have no explicit ahb_status
        // in the schema and should not inherit "X" from the parent group.
        let ansprechpartner = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Ansprechpartner")
            .expect("Ansprechpartner should exist");
        for name in &["namenszusatz1", "namenszusatz2"] {
            if let Some(field) = ansprechpartner.fields.iter().find(|f| f.bo4e_name == *name) {
                assert!(
                    !matches!(field.ahb_status.as_str(), "X" | "Muss" | "Soll"),
                    "{name} should NOT be required (has no explicit AHB status in schema), \
                     but got ahb_status={:?}",
                    field.ahb_status
                );
            }
        }

        // Issue #52 Q4: haushaltskunde IS legitimately required.
        let marktlokation = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Marktlokation")
            .expect("Marktlokation should exist");
        let haushaltskunde = marktlokation
            .fields
            .iter()
            .find(|f| f.bo4e_name == "haushaltskunde");
        assert!(
            haushaltskunde.is_some(),
            "haushaltskunde field should exist on Marktlokation"
        );
    }
}