json_eval_rs/parse_schema/

common.rs

use crate::jsoneval::path_utils;
use crate::jsoneval::table_metadata::{
    ColumnMetadata, RepeatBoundMetadata, RowMetadata, TableMetadata,
};
use crate::{LogicId, RLogic};
/// Shared utilities for schema parsing (used by both legacy and parsed implementations)
use indexmap::{IndexMap, IndexSet};
use serde_json::Map;
use serde_json::Value;
use std::sync::Arc;

/// Collect schema field pointers referenced by `$layout` elements.
///
/// This runs once during schema parsing. Runtime schema-value extraction then uses
/// an O(1) set membership check instead of re-walking a potentially large schema.
pub fn collect_layout_field_refs(value: &Value, refs: &mut IndexSet<String>) {
    fn collect_elements(elements: &Value, refs: &mut IndexSet<String>) {
        let Some(elements) = elements.as_array() else {
            return;
        };
        for element in elements {
            let Some(map) = element.as_object() else {
                continue;
            };
            if let Some(reference) = map.get("$ref").and_then(Value::as_str) {
                let pointer = path_utils::normalize_to_json_pointer(reference);
                refs.insert(pointer.trim_start_matches('#').to_string());
            }
            if let Some(children) = map.get("elements") {
                collect_elements(children, refs);
            }
        }
    }

    match value {
        Value::Object(map) => {
            if let Some(elements) = map
                .get("$layout")
                .and_then(Value::as_object)
                .and_then(|layout| layout.get("elements"))
            {
                collect_elements(elements, refs);
            }
            for child in map.values() {
                collect_layout_field_refs(child, refs);
            }
        }
        Value::Array(values) => {
            for child in values {
                collect_layout_field_refs(child, refs);
            }
        }
        _ => {}
    }
}

/// Collect $ref dependencies from a JSON value recursively
pub fn collect_refs(value: &Value, refs: &mut IndexSet<String>) {
    match value {
        Value::Object(map) => {
            if let Some(path) = map.get("$ref").and_then(Value::as_str) {
                refs.insert(path_utils::normalize_to_json_pointer(path).into_owned());
            }
            if let Some(path) = map.get("ref").and_then(Value::as_str) {
                refs.insert(path_utils::normalize_to_json_pointer(path).into_owned());
            }
            if let Some(var_val) = map.get("var") {
                match var_val {
                    Value::String(s) => {
                        refs.insert(s.clone());
                    }
                    Value::Array(arr) => {
                        if let Some(path) = arr.get(0).and_then(Value::as_str) {
                            refs.insert(path.to_string());
                        }
                    }
                    _ => {}
                }
            }
            for val in map.values() {
                collect_refs(val, refs);
            }
        }
        Value::Array(arr) => {
            for val in arr {
                collect_refs(val, refs);
            }
        }
        _ => {}
    }
}

/// Check if a value contains any actionable schema keys recursively (with depth limit for arrays)
/// used to skip large pure-data arrays during schema walking
#[inline]
pub fn has_actionable_keys(value: &Value) -> bool {
    match value {
        Value::Object(map) => {
            if map.contains_key("$evaluation")
                || map.contains_key("$table")
                || map.contains_key("dependents")
                || map.contains_key("$layout")
            {
                return true;
            }

            // Check for conditional hidden/disabled fields
            if let Some(Value::Object(condition)) = map.get("condition") {
                if condition.contains_key("hidden") || condition.contains_key("disabled") {
                    return true;
                }
            }

            // Check for rules object
            if map.contains_key("rules") {
                return true;
            }

            // Check for type="array" with items (subforms)
            if let Some(Value::String(type_str)) = map.get("type") {
                if type_str == "array" && map.contains_key("items") {
                    return true;
                }
            }

            // Check for options with URL templates
            if let Some(Value::String(url)) = map.get("url") {
                if url.contains('{') && url.contains('}') {
                    return true;
                }
            }

            map.values().any(has_actionable_keys)
        }
        Value::Array(arr) => arr.iter().take(5).any(has_actionable_keys),
        _ => false,
    }
}

/// Compute forward/normal column partitions with transitive closure
///
/// This function identifies which columns have forward references (dependencies on later columns)
/// and separates them from normal columns for proper evaluation order.
pub fn compute_column_partitions(columns: &[ColumnMetadata]) -> (Vec<usize>, Vec<usize>) {
    use std::collections::HashSet;

    // Build set of all forward-referencing column names (direct + transitive)
    let mut fwd_cols = HashSet::new();
    for col in columns {
        if col.has_forward_ref {
            fwd_cols.insert(col.name.as_ref());
        }
    }

    // Transitive closure: any column that depends on forward columns is also forward
    loop {
        let mut changed = false;
        for col in columns {
            if !fwd_cols.contains(col.name.as_ref()) {
                // Check if this column depends on any forward column
                for dep in col.dependencies.iter() {
                    // Strip $ prefix from dependency name for comparison
                    let dep_name = dep.trim_start_matches('$');
                    if fwd_cols.contains(dep_name) {
                        fwd_cols.insert(col.name.as_ref());
                        changed = true;
                        break;
                    }
                }
            }
        }
        // Stop when no more changes
        if !changed {
            break;
        }
    }

    // Separate into forward and normal indices
    let mut forward_indices = Vec::new();
    let mut normal_indices = Vec::new();

    for (idx, col) in columns.iter().enumerate() {
        if fwd_cols.contains(col.name.as_ref()) {
            forward_indices.push(idx);
        } else {
            normal_indices.push(idx);
        }
    }

    let forward_sorted = toposort_column_indices(columns, forward_indices);
    let normal_sorted = toposort_column_indices(columns, normal_indices);

    (forward_sorted, normal_sorted)
}

/// Topologically sort column indices within a partition based on intra-row dependencies.
/// Uses Kahn's algorithm with cycle fallback to guarantee acyclic dependencies evaluate in causal order.
fn toposort_column_indices(columns: &[ColumnMetadata], indices: Vec<usize>) -> Vec<usize> {
    if indices.len() <= 1 {
        return indices;
    }

    let n = indices.len();
    let mut name_to_sub_idx = std::collections::HashMap::with_capacity(n);
    for (sub_idx, &col_idx) in indices.iter().enumerate() {
        name_to_sub_idx.insert(columns[col_idx].name.as_ref(), sub_idx);
    }

    let mut in_degree = vec![0usize; n];
    let mut adj = vec![Vec::new(); n];

    for (u, &col_idx) in indices.iter().enumerate() {
        for dep in columns[col_idx].dependencies.iter() {
            if dep.starts_with('$') {
                let dep_name = dep.trim_start_matches('$');
                if let Some(&v) = name_to_sub_idx.get(dep_name) {
                    if v != u {
                        // v must evaluate before u: edge v -> u
                        adj[v].push(u);
                        in_degree[u] += 1;
                    }
                }
            }
        }
    }

    let mut queue = std::collections::VecDeque::new();
    for (i, &deg) in in_degree.iter().enumerate() {
        if deg == 0 {
            queue.push_back(i);
        }
    }

    let mut sorted = Vec::with_capacity(n);
    let mut visited = vec![false; n];

    while let Some(v) = queue.pop_front() {
        visited[v] = true;
        sorted.push(indices[v]);
        for &u in &adj[v] {
            in_degree[u] -= 1;
            if in_degree[u] == 0 {
                queue.push_back(u);
            }
        }
    }

    // Cycle fallback: append any unvisited columns in original order
    if sorted.len() < n {
        for (i, &was_visited) in visited.iter().enumerate() {
            if !was_visited {
                sorted.push(indices[i]);
            }
        }
    }

    sorted
}

pub fn walk_schema(
    value: &Value,
    path: &str,
    engine: &mut RLogic,
    evaluations: &mut IndexMap<String, LogicId>,
    tables: &mut IndexMap<String, Value>,
    deps: &mut IndexMap<String, IndexSet<String>>,
    value_fields: &mut Vec<String>,
    layout_paths: &mut Vec<String>,
    dependents: &mut IndexMap<String, Vec<crate::DependentItem>>,
    options_templates: &mut Vec<(String, String, String)>,
    subforms: &mut Vec<(String, serde_json::Map<String, Value>, Value)>,
    fields_with_rules: &mut Vec<String>,
    conditional_hidden_fields: &mut Vec<String>,
    conditional_readonly_fields: &mut Vec<String>,
) -> Result<(), String> {
    match value {
        Value::Object(map) => {
            // Check for $evaluation
            if let Some(evaluation) = map.get("$evaluation") {
                let key = path.to_string();
                let logic_value = evaluation.get("logic").unwrap_or(evaluation);
                let logic_id = engine
                    .compile(logic_value)
                    .map_err(|e| format!("failed to compile evaluation at {key}: {e}"))?;
                evaluations.insert(key.clone(), logic_id);

                // Collect dependencies with smart table inheritance
                let mut refs: IndexSet<String> = engine
                    .get_referenced_vars(&logic_id)
                    .unwrap_or_default()
                    .into_iter()
                    .map(|dep| path_utils::canonicalize_schema_path(&dep).into_owned())
                    .filter(|dep| {
                        // Filter out simple column references (e.g., "/INSAGE_YEAR", "/PREM_PP")
                        // These are FINDINDEX/MATCH column names, not actual data dependencies
                        // Real dependencies have multiple path segments (e.g., "/illustration/properties/...")
                        // Update: allow top-level fields only if they are system paths or deeper paths
                        dep.matches('/').count() > 1 || dep.starts_with("/$")
                    })
                    .collect();
                let mut extra_refs = IndexSet::new();
                collect_refs(logic_value, &mut extra_refs);
                if !extra_refs.is_empty() {
                    refs.extend(extra_refs.into_iter());
                }

                // For table dependencies, inherit parent table path instead of individual rows
                let refs: IndexSet<String> = refs
                    .into_iter()
                    .filter_map(|dep| {
                        // If dependency is a table row (contains /$table/), inherit table path
                        if let Some(table_idx) = dep.find("/$table/") {
                            let table_path = &dep[..table_idx];
                            Some(table_path.to_string())
                        } else {
                            Some(dep.to_string())
                        }
                    })
                    .collect();

                if !refs.is_empty() {
                    deps.insert(key.clone(), refs);
                }
            }

            // Check for $table
            if let Some(table) = map.get("$table") {
                let key = path.to_string();

                let rows = table.clone();
                let datas = map
                    .get("$datas")
                    .cloned()
                    .unwrap_or_else(|| Value::Array(vec![]));
                let skip = map.get("$skip").cloned().unwrap_or(Value::Bool(false));
                let clear = map.get("$clear").cloned().unwrap_or(Value::Bool(false));

                let mut table_entry = Map::new();
                table_entry.insert("rows".to_string(), rows);
                table_entry.insert("datas".to_string(), datas);
                table_entry.insert("skip".to_string(), skip);
                table_entry.insert("clear".to_string(), clear);

                tables.insert(key, Value::Object(table_entry));
            }

            // Check for $layout with elements
            if let Some(layout_obj) = map.get("$layout") {
                if let Some(Value::Array(_)) = layout_obj.get("elements") {
                    let layout_elements_path = format!("{}/$layout/elements", path);
                    layout_paths.push(layout_elements_path);
                }
            }

            // Check for rules object - collect field path for efficient validation
            if map.contains_key("rules") && !path.is_empty() && !path.starts_with("#/$") {
                // Convert JSON pointer path to dotted notation for validation
                // E.g., "#/properties/form/properties/name" -> "form.name"
                let field_path = path
                    .trim_start_matches('#')
                    .replace("/properties/", ".")
                    .trim_start_matches('/')
                    .trim_start_matches('.')
                    .to_string();

                if !field_path.is_empty() && !field_path.starts_with("$") {
                    fields_with_rules.push(field_path);
                }
            }

            // Check for options with URL templates
            if let Some(Value::String(url)) = map.get("url") {
                // Check if URL contains template pattern {variable}
                if url.contains('{') && url.contains('}') {
                    // Convert to JSON pointer format for evaluated_schema access
                    let url_path = path_utils::normalize_to_json_pointer(&format!("{}/url", path))
                        .into_owned();
                    let params_path =
                        path_utils::normalize_to_json_pointer(&format!("{}/params", path))
                            .into_owned();
                    options_templates.push((url_path, url.clone(), params_path));
                }
            }

            // Check for array fields with items (subforms)
            if let Some(Value::String(type_str)) = map.get("type") {
                if type_str == "array" {
                    if let Some(items) = map.get("items") {
                        // Store subform info for later creation (after walk completes)
                        subforms.push((path.to_string(), map.clone(), items.clone()));
                    }
                }
            }

            // Check for conditional hidden/disabled fields
            if let Some(Value::Object(condition)) = map.get("condition") {
                // Hidden
                if condition.contains_key("hidden") {
                    conditional_hidden_fields.push(path.to_string());
                }
                // Disabled (Read Only) - only relevant if it has a value enforce
                if condition.contains_key("disabled") && map.contains_key("value") {
                    conditional_readonly_fields.push(path.to_string());
                }
            }

            // Check for dependents array
            if let Some(Value::Array(dependents_arr)) = map.get("dependents") {
                let mut dependent_items = Vec::new();

                for (dep_idx, dep_item) in dependents_arr.iter().enumerate() {
                    if let Value::Object(dep_obj) = dep_item {
                        if let Some(Value::String(ref_path)) = dep_obj.get("$ref") {
                            // Process clear - compile if it's an $evaluation
                            let clear_val = if let Some(clear) = dep_obj.get("clear") {
                                if let Value::Object(clear_obj) = clear {
                                    if clear_obj.contains_key("$evaluation") {
                                        // Compile and store the evaluation
                                        let clear_eval = clear_obj.get("$evaluation").unwrap();
                                        let clear_key =
                                            format!("{}/dependents/{}/clear", path, dep_idx);
                                        let logic_id = engine.compile(clear_eval).map_err(|e| {
                                            format!(
                                                "Failed to compile dependent clear at {}: {}",
                                                clear_key, e
                                            )
                                        })?;
                                        evaluations.insert(clear_key.clone(), logic_id);
                                        // Replace with eval key reference
                                        Some(Value::String(clear_key))
                                    } else {
                                        Some(clear.clone())
                                    }
                                } else {
                                    Some(clear.clone())
                                }
                            } else {
                                None
                            };

                            // Process value - compile if it's an $evaluation
                            let value_val = if let Some(value) = dep_obj.get("value") {
                                if let Value::Object(value_obj) = value {
                                    if value_obj.contains_key("$evaluation") {
                                        // Compile and store the evaluation
                                        let value_eval = value_obj.get("$evaluation").unwrap();
                                        let value_key =
                                            format!("{}/dependents/{}/value", path, dep_idx);
                                        let logic_id = engine.compile(value_eval).map_err(|e| {
                                            format!(
                                                "Failed to compile dependent value at {}: {}",
                                                value_key, e
                                            )
                                        })?;
                                        evaluations.insert(value_key.clone(), logic_id);
                                        // Replace with eval key reference
                                        Some(Value::String(value_key))
                                    } else {
                                        Some(value.clone())
                                    }
                                } else {
                                    Some(value.clone())
                                }
                            } else {
                                None
                            };

                            dependent_items.push(crate::DependentItem {
                                ref_path: ref_path.clone(),
                                clear: clear_val,
                                value: value_val,
                            });
                        }
                    }
                }

                if !dependent_items.is_empty() {
                    dependents.insert(path.to_string(), dependent_items);
                }
            }

            // Recurse into children
            Ok(for (key, val) in map {
                // Skip special evaluation and dependents keys from recursion (already processed above)
                if key == "$evaluation"
                    || key == "dependents"
                    || (key == "items" && map.get("type").and_then(Value::as_str) == Some("array"))
                {
                    continue;
                }

                let next_path = if path == "#" {
                    format!("#/{key}")
                } else {
                    format!("{path}/{key}")
                };

                // Check if this is a "value" field
                // Allow $params but exclude other special $ paths like $layout, $items, etc.
                let is_excluded_special_path = next_path.contains("/$layout/")
                    || next_path.contains("/$items/")
                    || next_path.contains("/$options/")
                    || next_path.contains("/$dependents/")
                    || next_path.contains("/$rules/");

                if key == "value" && !is_excluded_special_path {
                    value_fields.push(next_path.clone());
                }

                // Recurse into all children (including $ keys like $table, $datas, etc.)
                walk_schema(
                    val,
                    &next_path,
                    engine,
                    evaluations,
                    tables,
                    deps,
                    value_fields,
                    layout_paths,
                    dependents,
                    options_templates,
                    subforms,
                    fields_with_rules,
                    conditional_hidden_fields,
                    conditional_readonly_fields,
                )?;
            })
        }
        Value::Array(arr) => {
            // Skip large arrays that contain no actionable schema keys.
            // This avoids recursively walking pure-data arrays (e.g., table rows in $params).
            // We specifically avoid skipping layout elements which can be large.
            let is_layout_array = path.contains("$layout") || path.contains("elements");
            if !is_layout_array && arr.len() > 10 && !has_actionable_keys(value) {
                return Ok(());
            }
            Ok(for (index, item) in arr.iter().enumerate() {
                let next_path = if path == "#" {
                    format!("#/{index}")
                } else {
                    format!("{path}/{index}")
                };
                walk_schema(
                    item,
                    &next_path,
                    engine,
                    evaluations,
                    tables,
                    deps,
                    value_fields,
                    layout_paths,
                    dependents,
                    options_templates,
                    subforms,
                    fields_with_rules,
                    conditional_hidden_fields,
                    conditional_readonly_fields,
                )?;
            })
        }
        _ => Ok(()),
    }
}
pub fn collect_table_dependencies(
    tables: &IndexMap<String, Value>,
    dependencies: &mut IndexMap<String, IndexSet<String>>,
) {
    for (table_key, _) in tables.iter() {
        let mut table_deps = IndexSet::new();

        let table_data_prefix = path_utils::normalize_to_json_pointer(table_key)
            .replace("/properties/", "/")
            .trim_start_matches('#')
            .to_string();
        let table_data_prefix_slash = format!("{}/", table_data_prefix);

        for (eval_key, deps) in dependencies.iter() {
            let is_child = eval_key.len() > table_key.len()
                && eval_key.starts_with(table_key.as_str())
                && eval_key.as_bytes().get(table_key.len()) == Some(&b'/');

            if is_child {
                for dep in deps {
                    let dep_data_path = path_utils::normalize_to_json_pointer(dep)
                        .replace("/properties/", "/")
                        .trim_start_matches('#')
                        .to_string();

                    if dep_data_path == table_data_prefix
                        || dep_data_path.starts_with(&table_data_prefix_slash)
                    {
                        continue;
                    }

                    let is_params_dep = dep.contains("$params");
                    let is_inline_system = !is_params_dep
                        && !dep.contains("$context")
                        && (dep.starts_with("/$") || dep.starts_with('$'));
                    if is_inline_system {
                        continue;
                    }
                    table_deps.insert(dep.clone());
                }
            }
        }

        if !table_deps.is_empty() {
            dependencies.insert(table_key.clone(), table_deps);
        }
    }
}

pub fn categorize_evaluations(
    sorted_evaluations: &[Vec<String>],
    evaluations: &IndexMap<String, crate::LogicId>,
    tables: &IndexMap<String, Value>,
) -> (Vec<String>, Vec<String>) {
    let batched_keys: IndexSet<String> = sorted_evaluations.iter().flatten().cloned().collect();

    let mut rules_evaluations = Vec::new();
    let mut others_evaluations = Vec::new();

    for eval_key in evaluations.keys() {
        if batched_keys.contains(eval_key) {
            continue;
        }

        if tables.iter().any(|(key, _)| eval_key.starts_with(key)) {
            continue;
        }

        if eval_key.contains("/$params/") {
            continue;
        }

        if eval_key.contains("/rules/") {
            rules_evaluations.push(eval_key.clone());
        } else if !eval_key.contains("/dependents/") {
            others_evaluations.push(eval_key.clone());
        }
    }

    (rules_evaluations, others_evaluations)
}

pub fn process_value_fields(
    value_fields: Vec<String>,
    tables: &IndexMap<String, Value>,
) -> Vec<String> {
    let mut value_evaluations = Vec::new();

    for path in value_fields {
        if value_evaluations.contains(&path) {
            continue;
        }

        if path.contains("/$params/") || tables.iter().any(|(key, _)| path.starts_with(key)) {
            continue;
        }

        value_evaluations.push(path);
    }

    value_evaluations
}

pub fn compile_table_metadata(
    evaluations: &IndexMap<String, crate::LogicId>,
    engine: &crate::RLogic,
    eval_key: &str,
    table: &Value,
) -> Result<TableMetadata, String> {
    let rows = table
        .get("rows")
        .and_then(|v| v.as_array())
        .ok_or("table missing rows")?;
    let empty_datas = Vec::new();
    let datas = table
        .get("datas")
        .and_then(|v| v.as_array())
        .unwrap_or(&empty_datas);

    // Pre-compile data plans with Arc sharing
    let mut data_plans = Vec::with_capacity(datas.len());
    for (idx, entry) in datas.iter().enumerate() {
        let Some(name) = entry.get("name").and_then(|v| v.as_str()) else {
            continue;
        };
        let logic_path = format!("{eval_key}/$datas/{idx}/data");
        let logic = evaluations.get(&logic_path).copied();
        let literal = entry.get("data").map(|v| Arc::new(v.clone()));
        data_plans.push((Arc::from(name), logic, literal));
    }

    // Pre-compile row plans with dependency analysis
    let mut row_plans = Vec::with_capacity(rows.len());
    for (row_idx, row_val) in rows.iter().enumerate() {
        let Some(row_obj) = row_val.as_object() else {
            continue;
        };

        if let Some(repeat_arr) = row_obj.get("$repeat").and_then(|v| v.as_array()) {
            if repeat_arr.len() == 3 {
                let start_logic_path = format!("{eval_key}/$table/{row_idx}/$repeat/0");
                let end_logic_path = format!("{eval_key}/$table/{row_idx}/$repeat/1");
                let start_logic = evaluations.get(&start_logic_path).copied();
                let end_logic = evaluations.get(&end_logic_path).copied();

                let start_literal = Arc::new(repeat_arr.get(0).cloned().unwrap_or(Value::Null));
                let end_literal = Arc::new(repeat_arr.get(1).cloned().unwrap_or(Value::Null));

                if let Some(template) = repeat_arr.get(2).and_then(|v| v.as_object()) {
                    let mut columns = Vec::with_capacity(template.len());
                    for (col_name, col_val) in template {
                        let col_eval_path =
                            format!("{eval_key}/$table/{row_idx}/$repeat/2/{col_name}");
                        let logic = evaluations.get(&col_eval_path).copied();
                        let literal = if logic.is_none() {
                            Some(col_val.clone())
                        } else {
                            None
                        };

                        // Extract dependencies ONCE at parse time (not during evaluation)
                        let (dependencies, has_forward_ref) = if let Some(logic_id) = logic {
                            let deps = engine
                                .get_referenced_vars(&logic_id)
                                .unwrap_or_default()
                                .into_iter()
                                .filter(|v| {
                                    v.starts_with('$') && v != "$iteration" && v != "$threshold"
                                })
                                .collect();
                            let has_fwd = engine.has_forward_reference(&logic_id);
                            (deps, has_fwd)
                        } else {
                            (Vec::new(), false)
                        };

                        columns.push(ColumnMetadata::new(
                            col_name,
                            logic,
                            literal,
                            dependencies,
                            has_forward_ref,
                        ));
                    }

                    // Pre-compute forward column propagation (transitive closure)
                    let (forward_cols, normal_cols) = compute_column_partitions(&columns);

                    row_plans.push(RowMetadata::Repeat {
                        start: RepeatBoundMetadata {
                            logic: start_logic,
                            literal: start_literal,
                        },
                        end: RepeatBoundMetadata {
                            logic: end_logic,
                            literal: end_literal,
                        },
                        columns: columns.into(),
                        forward_cols: forward_cols.into(),
                        normal_cols: normal_cols.into(),
                    });
                    continue;
                }
            }
        }

        // Static row
        let mut columns = Vec::with_capacity(row_obj.len());
        for (col_name, col_val) in row_obj {
            if col_name == "$repeat" {
                continue;
            }
            let col_eval_path = format!("{eval_key}/$table/{row_idx}/{col_name}");
            let logic = evaluations.get(&col_eval_path).copied();
            let literal = if logic.is_none() {
                Some(col_val.clone())
            } else {
                None
            };

            // Extract dependencies ONCE at parse time
            let (dependencies, has_forward_ref) = if let Some(logic_id) = logic {
                let deps = engine
                    .get_referenced_vars(&logic_id)
                    .unwrap_or_default()
                    .into_iter()
                    .filter(|v| v.starts_with('$') && v != "$iteration" && v != "$threshold")
                    .collect();
                let has_fwd = engine.has_forward_reference(&logic_id);
                (deps, has_fwd)
            } else {
                (Vec::new(), false)
            };

            columns.push(ColumnMetadata::new(
                col_name,
                logic,
                literal,
                dependencies,
                has_forward_ref,
            ));
        }
        row_plans.push(RowMetadata::Static {
            columns: columns.into(),
        });
    }

    // Pre-compile skip/clear logic
    let skip_logic = evaluations.get(&format!("{eval_key}/$skip")).copied();
    let skip_literal = table.get("skip").and_then(Value::as_bool).unwrap_or(false);
    let clear_logic = evaluations.get(&format!("{eval_key}/$clear")).copied();
    let clear_literal = table.get("clear").and_then(Value::as_bool).unwrap_or(false);

    Ok(TableMetadata {
        data_plans: data_plans.into(),
        row_plans: row_plans.into(),
        skip_logic,
        skip_literal,
        clear_logic,
        clear_literal,
    })
}
pub fn build_reffed_by(
    dependencies: &IndexMap<String, IndexSet<String>>,
) -> IndexMap<String, Vec<String>> {
    let mut reffed_by: IndexMap<String, Vec<String>> = IndexMap::new();

    for (eval_path, deps) in dependencies.iter() {
        if eval_path.ends_with("/condition/hidden") {
            let subject_path = eval_path[..eval_path.len() - 17].to_string();

            for dep in deps {
                let normalized_dep = path_utils::normalize_to_json_pointer(dep)
                    .replace("/properties/", "/")
                    .trim_start_matches('#')
                    .to_string();

                let dep_key = if normalized_dep.starts_with('/') {
                    normalized_dep
                } else {
                    format!("/{}", normalized_dep)
                };

                reffed_by
                    .entry(dep_key)
                    .or_insert_with(Vec::new)
                    .push(subject_path.clone());
            }
        }
    }

    reffed_by
}

pub fn build_dep_formula_triggers(
    dependents_evaluations: &IndexMap<String, Vec<crate::DependentItem>>,
    evaluations: &IndexMap<String, crate::LogicId>,
    engine: &crate::RLogic,
) -> IndexMap<String, Vec<(String, usize)>> {
    let mut triggers: IndexMap<String, Vec<(String, usize)>> = IndexMap::new();

    for (source_path, dep_items) in dependents_evaluations.iter() {
        for (dep_idx, dep_item) in dep_items.iter().enumerate() {
            let formula_keys: Vec<String> = [
                dep_item
                    .value
                    .as_ref()
                    .and_then(|v| v.as_str())
                    .map(|s| s.to_string()),
                dep_item
                    .clear
                    .as_ref()
                    .and_then(|v| v.as_str())
                    .map(|s| s.to_string()),
            ]
            .into_iter()
            .flatten()
            .filter(|k| k.contains("/dependents/"))
            .collect();

            for formula_key in formula_keys {
                let logic_id = match evaluations.get(&formula_key).copied() {
                    Some(id) => id,
                    None => continue,
                };

                let refs = engine.get_referenced_vars(&logic_id).unwrap_or_default();

                for dep_ref in refs {
                    let normalized = path_utils::normalize_to_json_pointer(&dep_ref)
                        .replace("/properties/", "/")
                        .trim_start_matches('#')
                        .to_string();
                    let dep_key = if normalized.starts_with('/') {
                        normalized
                    } else {
                        format!("/{}", normalized)
                    };

                    if dep_key.starts_with("/$") {
                        continue;
                    }

                    if dep_key.matches('/').count() <= 1 {
                        continue;
                    }

                    let source_data = path_utils::normalize_to_json_pointer(source_path)
                        .replace("/properties/", "/")
                        .trim_start_matches('#')
                        .to_string();
                    let source_data_key = if source_data.starts_with('/') {
                        source_data
                    } else {
                        format!("/{}", source_data)
                    };
                    if dep_key == source_data_key {
                        continue;
                    }

                    let target_data = path_utils::normalize_to_json_pointer(&dep_item.ref_path)
                        .replace("/properties/", "/")
                        .trim_start_matches('#')
                        .to_string();
                    let target_data_key = if target_data.starts_with('/') {
                        target_data
                    } else {
                        format!("/{}", target_data)
                    };
                    if dep_key == target_data_key {
                        continue;
                    }

                    let pair = (source_path.clone(), dep_idx);
                    let entry = triggers.entry(dep_key).or_insert_with(Vec::new);
                    if !entry.contains(&pair) {
                        entry.push(pair);
                    }
                }
            }
        }
    }

    for sources in triggers.values_mut() {
        sources.sort();
        // Since we check contains above and we sort, it's pretty clean.
    }

    triggers
}