mig-assembly 0.8.0

//! Order group repetitions by MIG variant.
//!
//! The MIG defines same-ID group variants in a fixed order, told apart by the
//! qualifier of their entry segment: SG2 `NAD+MS` before SG2 `NAD+MR`, SG1
//! `RFF+Z13` before `RFF+TN`, … A tree built from BO4E (the reverse mapping)
//! holds the repetitions of such a group in the order the BO4E JSON happened to
//! list them — and BO4E carries no ordering information, so that order must not
//! reach the EDIFACT.
//!
//! [`sort_repetitions_by_mig_variant`] reorders every group of a tree by the
//! MIG position of the variant each repetition belongs to. The order is derived
//! from the MIG alone. Repetitions of the same variant keep their relative
//! order: several FTX lines, positions, or time-series values of one variant are
//! data, and so is the order of codes a single variant allows (e.g. MSCONS
//! `QTY+220` / `QTY+67`), which the MIG does not prescribe.
//!
//! This runs on reverse-mapped trees only. The disassembler itself stays
//! faithful to the tree it is given, so parse → assemble → disassemble keeps
//! reproducing the input byte for byte.

use mig_types::schema::mig::{EntryQualifier, MigSchema, MigSegmentGroup};

use crate::assembler::{AssembledGroup, AssembledGroupInstance, AssembledTree};

/// Stable-sort the repetitions of every group in `tree` (recursively) by MIG
/// variant order.
///
/// `keep_order_of`: a top-level group whose repetitions are left in tree order
/// (the transaction group — the order of transactions is data); its children
/// are still sorted.
///
/// Before sorting, a child group sitting under a repetition whose MIG variant
/// does not define it is moved to the sibling repetition the MIG places it
/// under ([`preferred_parent_repetition`]). The reverse mapping puts children
/// it cannot link to a parent under the first repetition it built — which is
/// JSON array order again (e.g. SG10 data of `SEQ+Z80` under a `SEQ+Z85`
/// repetition because that one came first in the array). Placements the MIG
/// allows are left alone: they come from the BO4E structure or the mappings.
pub fn sort_repetitions_by_mig_variant(
    tree: &mut AssembledTree,
    mig: &MigSchema,
    keep_order_of: Option<&str>,
) {
    sort_groups(&mut tree.groups, &mig.segment_groups, keep_order_of);
}

/// Whether variant `v` defines a nested group `child_group_id`.
fn defines_child(v: &Variant<'_>, child_group_id: &str) -> bool {
    match &v.qualifier {
        Some(q) if !v.def.variant_entry_qualifiers.is_empty() => {
            q.nested_group_ids.iter().any(|id| id == child_group_id)
        }
        _ => v.def.nested_groups.iter().any(|g| g.id == child_group_id),
    }
}

/// Whether child repetition `child` of group `child_group_id` fits under a
/// parent repetition of variant `v`: the variant defines the child group and —
/// where the MIG lists that child group's variants for this parent variant —
/// one of them matches the child's entry qualifier.
fn fits_under(v: &Variant<'_>, child_group_id: &str, child: &AssembledGroupInstance) -> bool {
    if !defines_child(v, child_group_id) {
        return false;
    }
    if !v.def.variant_entry_qualifiers.is_empty() {
        // A merged parent keeps only which child groups each variant has.
        return true;
    }
    let child_defs: Vec<&MigSegmentGroup> = v
        .def
        .nested_groups
        .iter()
        .filter(|g| g.id == child_group_id)
        .collect();
    variant_of(child, &variants_of(&child_defs)).is_some()
}

/// Move child repetitions out of parent repetitions whose MIG variant cannot
/// hold them, into the first parent repetition (in MIG variant order) that can.
/// A child repetition no parent repetition can hold stays where it is.
fn rehome_misplaced_children(group: &mut AssembledGroup, variants: &[Variant<'_>]) {
    if group.repetitions.len() < 2 {
        return;
    }
    let reps_variant: Vec<Option<&Variant<'_>>> = group
        .repetitions
        .iter()
        .map(|rep| variant_of(rep, variants))
        .collect();
    let mut moves: Vec<(usize, String, AssembledGroupInstance)> = Vec::new();
    for r in 0..group.repetitions.len() {
        let Some(v) = reps_variant[r] else {
            continue;
        };
        for cg in &mut group.repetitions[r].child_groups {
            let id = cg.group_id.clone();
            let mut i = 0;
            while i < cg.repetitions.len() {
                let child = &cg.repetitions[i];
                let target = if fits_under(v, &id, child) {
                    None
                } else {
                    reps_variant
                        .iter()
                        .enumerate()
                        .filter_map(|(t, tv)| {
                            tv.filter(|tv| fits_under(tv, &id, child))
                                .map(|tv| (tv.rank, t))
                        })
                        .min()
                        .map(|(_, t)| t)
                };
                match target {
                    Some(t) => moves.push((t, id.clone(), cg.repetitions.remove(i))),
                    None => i += 1,
                }
            }
        }
        group.repetitions[r]
            .child_groups
            .retain(|cg| !cg.repetitions.is_empty());
    }
    for (target, id, child) in moves {
        put_child(
            &mut group.repetitions[target],
            AssembledGroup {
                group_id: id,
                repetitions: vec![child],
            },
        );
    }
}

fn put_child(rep: &mut AssembledGroupInstance, child: AssembledGroup) {
    match rep
        .child_groups
        .iter_mut()
        .find(|g| g.group_id == child.group_id)
    {
        Some(existing) => existing.repetitions.extend(child.repetitions),
        None => rep.child_groups.push(child),
    }
}

/// One orderable variant: its rank, the MIG definition its repetitions are
/// shaped by, and the qualifier identifying it (`None`: no coded entry element).
struct Variant<'a> {
    rank: usize,
    def: &'a MigSegmentGroup,
    qualifier: Option<EntryQualifier>,
}

fn variants_of<'a>(defs: &[&'a MigSegmentGroup]) -> Vec<Variant<'a>> {
    let mut out = Vec::new();
    for def in defs {
        if def.variant_entry_qualifiers.is_empty() {
            out.push(Variant {
                rank: out.len(),
                def,
                qualifier: def.entry_qualifier(),
            });
        } else {
            // A merged definition: one rank per merged variant, all shaped by
            // the merged definition.
            for q in &def.variant_entry_qualifiers {
                out.push(Variant {
                    rank: out.len(),
                    def,
                    qualifier: Some(q.clone()),
                });
            }
        }
    }
    out
}

/// The variant a repetition belongs to: the first (in MIG order) whose
/// qualifier its entry segment carries; else the first variant without a coded
/// entry element; else `None`.
fn variant_of<'v, 'a>(
    rep: &AssembledGroupInstance,
    variants: &'v [Variant<'a>],
) -> Option<&'v Variant<'a>> {
    variants
        .iter()
        .find(|v| {
            v.qualifier.as_ref().is_some_and(|q| {
                // Find the entry segment by tag: a reverse-built repetition
                // holds its segments in definition order, which need not start
                // with the entry segment.
                rep.segments
                    .iter()
                    .find(|s| s.tag.eq_ignore_ascii_case(&q.tag))
                    .is_some_and(|s| q.matches(&s.elements))
            })
        })
        .or_else(|| variants.iter().find(|v| v.qualifier.is_none()))
}

/// Where a child group without an explicit parent belongs: the repetition of
/// `parent` whose MIG variant defines a nested `child_group_id` and comes first
/// in MIG variant order (ties: the earlier repetition). When no repetition's
/// variant defines the child group, the repetition first in MIG variant order.
/// `None` only for a group without repetitions.
///
/// The reverse mapping uses this for message-level children the BO4E JSON does
/// not nest inside a parent object (e.g. a top-level `ansprechpartner` for the
/// SG2 `NAD+MS` contact): which `Marktteilnehmer` comes first in the JSON array
/// is no information, but the MIG says which variant holds the contact.
pub fn preferred_parent_repetition(
    parent: &AssembledGroup,
    parent_defs: &[MigSegmentGroup],
    child_group_id: &str,
) -> Option<usize> {
    let same_id: Vec<&MigSegmentGroup> = parent_defs
        .iter()
        .filter(|d| d.id == parent.group_id)
        .collect();
    let variants = variants_of(&same_id);
    // Key per repetition: (does not define the child, variant rank, index).
    parent
        .repetitions
        .iter()
        .enumerate()
        .map(|(i, rep)| match variant_of(rep, &variants) {
            Some(v) => (!defines_child(v, child_group_id), v.rank, i),
            None => (true, usize::MAX, i),
        })
        .min()
        .map(|(_, _, i)| i)
}

/// Stable-sort the repetitions that belong to a known variant by variant
/// rank, within the slots they occupy. A repetition matching no variant (e.g.
/// a qualifier the PID's AHB does not list, kept from the input) stays in its
/// slot: the MIG says nothing about where it goes.
fn sort_matched_in_place(reps: &mut Vec<AssembledGroupInstance>, variants: &[Variant<'_>]) {
    let ranks: Vec<Option<usize>> = reps
        .iter()
        .map(|rep| variant_of(rep, variants).map(|v| v.rank))
        .collect();
    let slots: Vec<usize> = (0..reps.len()).filter(|&i| ranks[i].is_some()).collect();
    let mut order = slots.clone();
    order.sort_by_key(|&i| ranks[i]);
    if order == slots {
        return;
    }
    let mut taken: Vec<Option<AssembledGroupInstance>> = reps.drain(..).map(Some).collect();
    let mut placed: Vec<Option<AssembledGroupInstance>> = vec![None; 0];
    placed.resize_with(taken.len(), || None);
    for (slot, src) in slots.iter().zip(&order) {
        placed[*slot] = taken[*src].take();
    }
    for (i, rep) in taken.into_iter().enumerate() {
        if rep.is_some() {
            placed[i] = rep;
        }
    }
    reps.extend(placed.into_iter().map(|r| r.expect("every slot filled")));
}

fn sort_groups(groups: &mut [AssembledGroup], defs: &[MigSegmentGroup], keep: Option<&str>) {
    for group in groups.iter_mut() {
        let same_id: Vec<&MigSegmentGroup> =
            defs.iter().filter(|d| d.id == group.group_id).collect();
        if same_id.is_empty() {
            continue;
        }
        let variants = variants_of(&same_id);

        if keep != Some(group.group_id.as_str()) && variants.len() > 1 {
            rehome_misplaced_children(group, &variants);
            sort_matched_in_place(&mut group.repetitions, &variants);
        }

        for rep in &mut group.repetitions {
            let def = variant_of(rep, &variants).map_or(same_id[0], |v| v.def);
            sort_groups(&mut rep.child_groups, &def.nested_groups, None);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::assembler::AssembledSegment;
    use crate::test_support::make_mig_group;
    use mig_types::schema::common::CodeDefinition;
    use mig_types::schema::mig::MigDataElement;

    fn coded_group(
        id: &str,
        tag: &str,
        codes: &[&str],
        nested: Vec<MigSegmentGroup>,
    ) -> MigSegmentGroup {
        let mut g = make_mig_group(id, vec![tag], nested);
        g.segments[0].data_elements.push(MigDataElement {
            id: "3035".into(),
            name: String::new(),
            description: None,
            status_std: None,
            status_spec: None,
            format_std: None,
            format_spec: None,
            codes: codes
                .iter()
                .map(|c| CodeDefinition {
                    value: c.to_string(),
                    name: String::new(),
                    description: None,
                })
                .collect(),
            position: 0,
        });
        g
    }

    fn rep(tag: &str, qualifier: &str, value: &str) -> AssembledGroupInstance {
        rep_with_children(tag, qualifier, value, vec![])
    }

    fn rep_with_children(
        tag: &str,
        qualifier: &str,
        value: &str,
        child_groups: Vec<AssembledGroup>,
    ) -> AssembledGroupInstance {
        AssembledGroupInstance {
            segments: vec![AssembledSegment {
                tag: tag.into(),
                elements: vec![vec![qualifier.into()], vec![value.into()]],
                mig_number: None,
                segment_number: None,
            }],
            child_groups,
            entry_mig_number: None,
            variant_mig_numbers: vec![],
            skipped_segments: vec![],
            skipped_positions: vec![],
        }
    }

    fn mig(groups: Vec<MigSegmentGroup>) -> MigSchema {
        MigSchema {
            message_type: "TEST".into(),
            variant: None,
            version: String::new(),
            publication_date: String::new(),
            author: String::new(),
            format_version: String::new(),
            source_file: String::new(),
            segments: vec![],
            segment_groups: groups,
        }
    }

    fn tree(groups: Vec<AssembledGroup>) -> AssembledTree {
        AssembledTree {
            segments: vec![],
            groups,
            post_group_start: 0,
            inter_group_segments: Default::default(),
        }
    }

    fn values(group: &AssembledGroup) -> Vec<String> {
        group
            .repetitions
            .iter()
            .map(|r| {
                format!(
                    "{}:{}",
                    r.segments[0].elements[0][0], r.segments[0].elements[1][0]
                )
            })
            .collect()
    }

    #[test]
    fn separate_variants_sort_in_mig_order_keeping_equal_qualifiers_in_order() {
        let schema = mig(vec![
            coded_group("SG2", "NAD", &["MS"], vec![]),
            coded_group("SG2", "NAD", &["MR"], vec![]),
        ]);
        let mut t = tree(vec![AssembledGroup {
            group_id: "SG2".into(),
            repetitions: vec![
                rep("NAD", "MR", "1"),
                rep("NAD", "MS", "2"),
                rep("NAD", "MR", "3"),
            ],
        }]);
        sort_repetitions_by_mig_variant(&mut t, &schema, None);
        assert_eq!(values(&t.groups[0]), ["MS:2", "MR:1", "MR:3"]);
    }

    #[test]
    fn merged_variants_sort_by_recorded_qualifier_order() {
        let mut merged = make_mig_group("SG2", vec!["NAD"], vec![]);
        merged.variant_entry_qualifiers = ["MR", "MS"]
            .iter()
            .map(|c| EntryQualifier {
                tag: "NAD".into(),
                element: 0,
                component: 0,
                codes: vec![c.to_string()],
                nested_group_ids: vec![],
            })
            .collect();
        let schema = mig(vec![merged]);
        let mut t = tree(vec![AssembledGroup {
            group_id: "SG2".into(),
            repetitions: vec![rep("NAD", "MS", "1"), rep("NAD", "MR", "2")],
        }]);
        sort_repetitions_by_mig_variant(&mut t, &schema, None);
        assert_eq!(values(&t.groups[0]), ["MR:2", "MS:1"]);
    }

    #[test]
    fn child_without_parent_goes_to_the_first_variant_defining_it() {
        let contact = make_mig_group("SG3", vec!["CTA"], vec![]);
        let refs = make_mig_group("SG4", vec!["RFF"], vec![]);
        let schema_groups = vec![
            coded_group("SG2", "NAD", &["MR"], vec![refs.clone()]),
            coded_group("SG2", "NAD", &["MS"], vec![contact, refs]),
        ];
        let parent = AssembledGroup {
            group_id: "SG2".into(),
            repetitions: vec![rep("NAD", "MS", "1"), rep("NAD", "MR", "2")],
        };
        assert_eq!(
            preferred_parent_repetition(&parent, &schema_groups, "SG3"),
            Some(0)
        );
        assert_eq!(
            preferred_parent_repetition(&parent, &schema_groups, "SG4"),
            Some(1)
        );
        // No variant defines it: the repetition first in MIG order (MR).
        assert_eq!(
            preferred_parent_repetition(&parent, &schema_groups, "SG9"),
            Some(1)
        );

        // Merged definition: the same answers from the recorded variant list.
        let mut merged = make_mig_group(
            "SG2",
            vec!["NAD"],
            vec![
                make_mig_group("SG3", vec!["CTA"], vec![]),
                make_mig_group("SG4", vec!["RFF"], vec![]),
            ],
        );
        merged.variant_entry_qualifiers = [("MR", vec!["SG4"]), ("MS", vec!["SG3", "SG4"])]
            .into_iter()
            .map(|(c, nested)| EntryQualifier {
                tag: "NAD".into(),
                element: 0,
                component: 0,
                codes: vec![c.into()],
                nested_group_ids: nested.into_iter().map(String::from).collect(),
            })
            .collect();
        let merged_groups = vec![merged];
        assert_eq!(
            preferred_parent_repetition(&parent, &merged_groups, "SG3"),
            Some(0)
        );
        assert_eq!(
            preferred_parent_repetition(&parent, &merged_groups, "SG4"),
            Some(1)
        );
    }

    #[test]
    fn child_under_a_variant_without_it_moves_to_the_variant_defining_it() {
        let schema = mig(vec![coded_group(
            "SG4",
            "IDE",
            &["24"],
            vec![
                coded_group("SG8", "SEQ", &["Z85"], vec![]),
                coded_group(
                    "SG8",
                    "SEQ",
                    &["Z80"],
                    vec![make_mig_group("SG10", vec!["CCI"], vec![])],
                ),
            ],
        )]);
        let sg10 = AssembledGroup {
            group_id: "SG10".into(),
            repetitions: vec![rep("CCI", "Z19", "x")],
        };
        let sg8 = AssembledGroup {
            group_id: "SG8".into(),
            repetitions: vec![
                rep_with_children("SEQ", "Z80", "1", vec![]),
                rep_with_children("SEQ", "Z85", "1", vec![sg10]),
            ],
        };
        let mut t = tree(vec![AssembledGroup {
            group_id: "SG4".into(),
            repetitions: vec![rep_with_children("IDE", "24", "t", vec![sg8])],
        }]);
        sort_repetitions_by_mig_variant(&mut t, &schema, Some("SG4"));
        let sg8 = &t.groups[0].repetitions[0].child_groups[0];
        assert_eq!(values(sg8), ["Z85:1", "Z80:1"]);
        assert!(sg8.repetitions[0].child_groups.is_empty());
        assert_eq!(values(&sg8.repetitions[1].child_groups[0]), ["Z19:x"]);
    }

    #[test]
    fn unmatched_repetitions_keep_their_slot() {
        let schema = mig(vec![
            coded_group("SG1", "RFF", &["Z30"], vec![]),
            coded_group("SG1", "RFF", &["Z13"], vec![]),
        ]);
        let mut t = tree(vec![AssembledGroup {
            group_id: "SG1".into(),
            repetitions: vec![
                rep("RFF", "Z13", "1"),
                rep("RFF", "ACW", "2"),
                rep("RFF", "Z30", "3"),
            ],
        }]);
        sort_repetitions_by_mig_variant(&mut t, &schema, None);
        assert_eq!(values(&t.groups[0]), ["Z30:3", "ACW:2", "Z13:1"]);
    }

    #[test]
    fn codes_of_one_variant_keep_data_order() {
        let schema = mig(vec![coded_group("SG10", "QTY", &["220", "67"], vec![])]);
        let mut t = tree(vec![AssembledGroup {
            group_id: "SG10".into(),
            repetitions: vec![
                rep("QTY", "67", "1"),
                rep("QTY", "220", "2"),
                rep("QTY", "67", "3"),
            ],
        }]);
        sort_repetitions_by_mig_variant(&mut t, &schema, None);
        assert_eq!(values(&t.groups[0]), ["67:1", "220:2", "67:3"]);
    }

    #[test]
    fn kept_group_is_not_reordered_but_its_children_are() {
        let schema = mig(vec![
            coded_group(
                "SG4",
                "IDE",
                &["24"],
                vec![
                    coded_group("SG12", "NAD", &["Z07"], vec![]),
                    coded_group("SG12", "NAD", &["Z08"], vec![]),
                ],
            ),
            coded_group("SG4", "IDE", &["Z01"], vec![]),
        ]);
        let children = AssembledGroup {
            group_id: "SG12".into(),
            repetitions: vec![rep("NAD", "Z08", "a"), rep("NAD", "Z07", "b")],
        };
        let mut t = tree(vec![AssembledGroup {
            group_id: "SG4".into(),
            repetitions: vec![
                rep("IDE", "Z01", "1"),
                rep_with_children("IDE", "24", "2", vec![children]),
            ],
        }]);
        sort_repetitions_by_mig_variant(&mut t, &schema, Some("SG4"));
        assert_eq!(values(&t.groups[0]), ["Z01:1", "24:2"]);
        assert_eq!(
            values(&t.groups[0].repetitions[1].child_groups[0]),
            ["Z07:b", "Z08:a"]
        );
    }
}