//! HoloLake 原生组合执行内核。
//!
//! 外来文件格式不进入这里。内核只接受登记过的模块、受限配方和当前登录账号的
//! HoloLake 原生对象；人类投影只是同一执行结果的不同视图，没有源数据所有权。

use ring::digest::{digest, SHA256};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::time::{SystemTime, UNIX_EPOCH};
use tauri::AppHandle;
use uuid::Uuid;

const REGISTRY_SCHEMA: &str = "hololake.composition-module-registry/v1";
const OBJECT_SCHEMA: &str = "hololake.native-object/v1";
const RECIPE_SCHEMA: &str = "hololake.composition-recipe/v1";
const PROJECTION_SCHEMA: &str = "hololake.human-projection/v1";
const MAX_ROWS: usize = 5_000;
const MAX_VIEWS: usize = 5;

#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum CompositionDimension {
    Source,
    TopLevelFolder,
}

#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum CompositionMeasure {
    DocumentCount,
    TotalBytes,
    DuplicateCount,
}

#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq, PartialOrd, Ord)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum ProjectionView {
    Dashboard,
    Comparison,
    VerticalBar,
    Classification,
    Table,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ExecuteCompositionInput {
    pub dimension: CompositionDimension,
    pub measure: CompositionMeasure,
    pub views: Vec<ProjectionView>,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ModuleDescriptor {
    pub module_id: &'static str,
    pub display_name: &'static str,
    pub kind: &'static str,
    pub input_schema: Option<&'static str>,
    pub output_schema: &'static str,
    pub deterministic: bool,
    pub authority: &'static str,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ModuleRegistry {
    pub schema: &'static str,
    pub state: &'static str,
    pub modules: Vec<ModuleDescriptor>,
    pub arbitrary_script_allowed: bool,
    pub unregistered_module_allowed: bool,
    pub direct_projection_write_allowed: bool,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct NativeColumn {
    pub column_id: &'static str,
    pub title: &'static str,
    pub value_type: &'static str,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct NativeKnowledgeRow {
    pub row_id: String,
    pub source: String,
    pub path: String,
    pub title: String,
    pub top_level_folder: String,
    pub size_bytes: u64,
    pub duplicate_count: usize,
    pub updated_at_unix_ms: u64,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct NativeObject {
    pub schema: &'static str,
    pub object_id: String,
    pub title: &'static str,
    pub columns: Vec<NativeColumn>,
    pub rows: Vec<NativeKnowledgeRow>,
    pub row_count: usize,
    pub truncated: bool,
    pub source_receipt: &'static str,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CompositionNode {
    pub node_id: String,
    pub module_id: String,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CompositionEdge {
    pub from: String,
    pub to: String,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CompositionRecipe {
    pub schema: &'static str,
    pub recipe_id: &'static str,
    pub title: &'static str,
    pub nodes: Vec<CompositionNode>,
    pub edges: Vec<CompositionEdge>,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ProjectionGroup {
    pub key: String,
    pub label: String,
    pub document_count: usize,
    pub total_bytes: u64,
    pub duplicate_count: usize,
    pub measure_value: u64,
    pub share: f64,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ProjectionMetrics {
    pub raw_document_count: usize,
    pub unique_document_count: usize,
    pub duplicate_document_count: usize,
    pub total_bytes: u64,
    pub group_count: usize,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct NativeCompositionProjection {
    pub schema: &'static str,
    pub state: &'static str,
    pub execution_id: String,
    pub executed_at_unix_ms: u64,
    pub source_is_real_account_data: bool,
    pub read_only: bool,
    pub dimension: CompositionDimension,
    pub measure: CompositionMeasure,
    pub views: Vec<ProjectionView>,
    pub native_object: NativeObject,
    pub groups: Vec<ProjectionGroup>,
    pub metrics: ProjectionMetrics,
    pub recipe: CompositionRecipe,
    pub data_sha256: String,
    pub receipt_sha256: String,
}

fn registry() -> ModuleRegistry {
    let modules = vec![
        descriptor(
            "HLC-SOURCE-KNOWLEDGE-CATALOG",
            "知识目录数据源",
            "SOURCE",
            None,
            OBJECT_SCHEMA,
        ),
        descriptor(
            "HLC-CLASSIFY-SOURCE",
            "按来源分类",
            "TRANSFORM",
            Some(OBJECT_SCHEMA),
            "hololake.classified-object/v1",
        ),
        descriptor(
            "HLC-CLASSIFY-TOP-FOLDER",
            "按一级目录分类",
            "TRANSFORM",
            Some(OBJECT_SCHEMA),
            "hololake.classified-object/v1",
        ),
        descriptor(
            "HLC-AGGREGATE-DOCUMENT-COUNT",
            "统计文档数",
            "TRANSFORM",
            Some("hololake.classified-object/v1"),
            "hololake.composition-result/v1",
        ),
        descriptor(
            "HLC-AGGREGATE-TOTAL-BYTES",
            "统计数据量",
            "TRANSFORM",
            Some("hololake.classified-object/v1"),
            "hololake.composition-result/v1",
        ),
        descriptor(
            "HLC-AGGREGATE-DUPLICATE-COUNT",
            "统计重复数",
            "TRANSFORM",
            Some("hololake.classified-object/v1"),
            "hololake.composition-result/v1",
        ),
        descriptor(
            "HLC-PROJECT-DASHBOARD",
            "仪表盘投影",
            "PROJECTION",
            Some("hololake.composition-result/v1"),
            PROJECTION_SCHEMA,
        ),
        descriptor(
            "HLC-PROJECT-COMPARISON",
            "对比投影",
            "PROJECTION",
            Some("hololake.composition-result/v1"),
            PROJECTION_SCHEMA,
        ),
        descriptor(
            "HLC-PROJECT-VERTICAL-BAR",
            "柱状图投影",
            "PROJECTION",
            Some("hololake.composition-result/v1"),
            PROJECTION_SCHEMA,
        ),
        descriptor(
            "HLC-PROJECT-CLASSIFICATION",
            "分类投影",
            "PROJECTION",
            Some("hololake.composition-result/v1"),
            PROJECTION_SCHEMA,
        ),
        descriptor(
            "HLC-PROJECT-TABLE",
            "表格投影",
            "PROJECTION",
            Some("hololake.composition-result/v1"),
            PROJECTION_SCHEMA,
        ),
    ];
    ModuleRegistry {
        schema: REGISTRY_SCHEMA,
        state: "READY",
        modules,
        arbitrary_script_allowed: false,
        unregistered_module_allowed: false,
        direct_projection_write_allowed: false,
    }
}

fn descriptor(
    module_id: &'static str,
    display_name: &'static str,
    kind: &'static str,
    input_schema: Option<&'static str>,
    output_schema: &'static str,
) -> ModuleDescriptor {
    ModuleDescriptor {
        module_id,
        display_name,
        kind,
        input_schema,
        output_schema,
        deterministic: true,
        authority: "CURRENT_AUTHENTICATED_ACCOUNT_READ_ONLY",
    }
}

pub fn get_native_composition_module_registry() -> ModuleRegistry {
    registry()
}

pub async fn execute_knowledge_native_composition(
    app: AppHandle,
    input: ExecuteCompositionInput,
) -> Result<NativeCompositionProjection, String> {
    crate::module_package_runtime::require_active_module_adapter(
        &app,
        "HLP-MOD-LOCAL-NATIVE-COMPOSITION-0001",
        "native-composition-v1",
    )?;
    validate_input(&input)?;
    let snapshot = crate::knowledge_base::get_knowledge_snapshot(app).await?;
    execute_snapshot(snapshot, input)
}

fn validate_input(input: &ExecuteCompositionInput) -> Result<(), String> {
    if input.views.is_empty() || input.views.len() > MAX_VIEWS {
        return Err("HOLOLAKE_COMPOSITION_VIEW_COUNT_INVALID".into());
    }
    let unique = input.views.iter().copied().collect::<BTreeSet<_>>();
    if unique.len() != input.views.len() {
        return Err("HOLOLAKE_COMPOSITION_DUPLICATE_VIEW".into());
    }
    Ok(())
}

fn execute_snapshot(
    snapshot: crate::knowledge_base::KnowledgeSnapshot,
    input: ExecuteCompositionInput,
) -> Result<NativeCompositionProjection, String> {
    if snapshot.documents.len() > MAX_ROWS {
        return Err("HOLOLAKE_COMPOSITION_ROW_LIMIT_EXCEEDED".into());
    }
    let rows = snapshot
        .documents
        .iter()
        .enumerate()
        .map(|(index, document)| NativeKnowledgeRow {
            row_id: format!("knowledge-row-{}", index + 1),
            source: document.source.to_string(),
            path: document.path.clone(),
            title: document.title.clone(),
            top_level_folder: top_level_folder(&document.path),
            size_bytes: document.size_bytes,
            duplicate_count: document.duplicate_count,
            updated_at_unix_ms: document.updated_at_unix_ms.min(u64::MAX as u128) as u64,
        })
        .collect::<Vec<_>>();
    let data_sha256 = sha256_json(&rows)?;
    let native_object = NativeObject {
        schema: OBJECT_SCHEMA,
        object_id: format!("knowledge-catalog-{}", &data_sha256[..16]),
        title: "当前账号知识目录",
        columns: native_columns(),
        row_count: rows.len(),
        rows,
        truncated: snapshot.truncated,
        source_receipt: "CURRENT_AUTHENTICATED_ACCOUNT_KNOWLEDGE_SNAPSHOT",
    };
    let mut aggregates = BTreeMap::<String, (usize, u64, usize)>::new();
    for row in &native_object.rows {
        let key = match input.dimension {
            CompositionDimension::Source => source_label(&row.source),
            CompositionDimension::TopLevelFolder => row.top_level_folder.clone(),
        };
        let entry = aggregates.entry(key).or_default();
        entry.0 += 1;
        entry.1 = entry.1.saturating_add(row.size_bytes);
        entry.2 = entry.2.saturating_add(row.duplicate_count);
    }
    let total_measure = aggregates
        .values()
        .map(|value| measure_value(*value, input.measure))
        .sum::<u64>();
    let mut groups = aggregates
        .into_iter()
        .map(|(label, value)| ProjectionGroup {
            key: stable_key(&label),
            label,
            document_count: value.0,
            total_bytes: value.1,
            duplicate_count: value.2,
            measure_value: measure_value(value, input.measure),
            share: if total_measure == 0 {
                0.0
            } else {
                measure_value(value, input.measure) as f64 / total_measure as f64
            },
        })
        .collect::<Vec<_>>();
    groups.sort_by(|left, right| {
        right
            .measure_value
            .cmp(&left.measure_value)
            .then_with(|| left.label.cmp(&right.label))
    });
    let recipe = compile_recipe(&input);
    validate_recipe(&recipe, &registry())?;
    let metrics = ProjectionMetrics {
        raw_document_count: snapshot.raw_document_count,
        unique_document_count: snapshot.unique_document_count,
        duplicate_document_count: snapshot.duplicate_document_count,
        total_bytes: native_object.rows.iter().map(|row| row.size_bytes).sum(),
        group_count: groups.len(),
    };
    let executed_at_unix_ms = now_unix_ms();
    let receipt_material = serde_json::to_vec(&(
        &data_sha256,
        &recipe.nodes,
        &recipe.edges,
        &groups,
        executed_at_unix_ms,
    ))
    .map_err(|error| format!("HOLOLAKE_COMPOSITION_RECEIPT_SERIALIZE_FAILED: {error}"))?;
    Ok(NativeCompositionProjection {
        schema: PROJECTION_SCHEMA,
        state: "EXECUTED",
        execution_id: format!("HLC-EXEC-{}", Uuid::new_v4()),
        executed_at_unix_ms,
        source_is_real_account_data: true,
        read_only: true,
        dimension: input.dimension,
        measure: input.measure,
        views: input.views,
        native_object,
        groups,
        metrics,
        recipe,
        data_sha256,
        receipt_sha256: sha256(&receipt_material),
    })
}

fn native_columns() -> Vec<NativeColumn> {
    vec![
        NativeColumn {
            column_id: "title",
            title: "标题",
            value_type: "TEXT",
        },
        NativeColumn {
            column_id: "source",
            title: "来源",
            value_type: "TEXT",
        },
        NativeColumn {
            column_id: "topLevelFolder",
            title: "一级分类",
            value_type: "TEXT",
        },
        NativeColumn {
            column_id: "sizeBytes",
            title: "数据量",
            value_type: "INTEGER",
        },
        NativeColumn {
            column_id: "duplicateCount",
            title: "重复数",
            value_type: "INTEGER",
        },
        NativeColumn {
            column_id: "updatedAtUnixMs",
            title: "更新时间",
            value_type: "TIME",
        },
    ]
}

fn compile_recipe(input: &ExecuteCompositionInput) -> CompositionRecipe {
    let classify = match input.dimension {
        CompositionDimension::Source => "HLC-CLASSIFY-SOURCE",
        CompositionDimension::TopLevelFolder => "HLC-CLASSIFY-TOP-FOLDER",
    };
    let aggregate = match input.measure {
        CompositionMeasure::DocumentCount => "HLC-AGGREGATE-DOCUMENT-COUNT",
        CompositionMeasure::TotalBytes => "HLC-AGGREGATE-TOTAL-BYTES",
        CompositionMeasure::DuplicateCount => "HLC-AGGREGATE-DUPLICATE-COUNT",
    };
    let mut nodes = vec![
        CompositionNode {
            node_id: "source".into(),
            module_id: "HLC-SOURCE-KNOWLEDGE-CATALOG".into(),
        },
        CompositionNode {
            node_id: "classify".into(),
            module_id: classify.into(),
        },
        CompositionNode {
            node_id: "aggregate".into(),
            module_id: aggregate.into(),
        },
    ];
    let mut edges = vec![
        CompositionEdge {
            from: "source".into(),
            to: "classify".into(),
        },
        CompositionEdge {
            from: "classify".into(),
            to: "aggregate".into(),
        },
    ];
    for (index, view) in input.views.iter().enumerate() {
        let node_id = format!("projection-{}", index + 1);
        nodes.push(CompositionNode {
            node_id: node_id.clone(),
            module_id: view_module_id(*view).into(),
        });
        edges.push(CompositionEdge {
            from: "aggregate".into(),
            to: node_id,
        });
    }
    CompositionRecipe {
        schema: RECIPE_SCHEMA,
        recipe_id: "HLC-RECIPE-KNOWLEDGE-OVERVIEW-001",
        title: "知识空间结构组合",
        nodes,
        edges,
    }
}

fn validate_recipe(
    recipe: &CompositionRecipe,
    module_registry: &ModuleRegistry,
) -> Result<(), String> {
    let descriptors = module_registry
        .modules
        .iter()
        .map(|item| (item.module_id, item))
        .collect::<BTreeMap<_, _>>();
    let nodes = recipe
        .nodes
        .iter()
        .map(|item| (item.node_id.as_str(), item))
        .collect::<BTreeMap<_, _>>();
    if nodes.len() != recipe.nodes.len() {
        return Err("HOLOLAKE_COMPOSITION_DUPLICATE_NODE".into());
    }
    for node in &recipe.nodes {
        if !descriptors.contains_key(node.module_id.as_str()) {
            return Err("HOLOLAKE_COMPOSITION_MODULE_NOT_REGISTERED".into());
        }
    }
    let mut indegree = recipe
        .nodes
        .iter()
        .map(|item| (item.node_id.as_str(), 0usize))
        .collect::<BTreeMap<_, _>>();
    let mut outgoing = BTreeMap::<&str, Vec<&str>>::new();
    for edge in &recipe.edges {
        let from = nodes
            .get(edge.from.as_str())
            .ok_or("HOLOLAKE_COMPOSITION_EDGE_NODE_UNKNOWN")?;
        let to = nodes
            .get(edge.to.as_str())
            .ok_or("HOLOLAKE_COMPOSITION_EDGE_NODE_UNKNOWN")?;
        let from_descriptor = descriptors
            .get(from.module_id.as_str())
            .ok_or("HOLOLAKE_COMPOSITION_MODULE_NOT_REGISTERED")?;
        let to_descriptor = descriptors
            .get(to.module_id.as_str())
            .ok_or("HOLOLAKE_COMPOSITION_MODULE_NOT_REGISTERED")?;
        if to_descriptor.input_schema != Some(from_descriptor.output_schema) {
            return Err("HOLOLAKE_COMPOSITION_PORT_TYPE_MISMATCH".into());
        }
        *indegree
            .get_mut(to.node_id.as_str())
            .ok_or("HOLOLAKE_COMPOSITION_EDGE_NODE_UNKNOWN")? += 1;
        outgoing
            .entry(from.node_id.as_str())
            .or_default()
            .push(to.node_id.as_str());
    }
    let mut queue = indegree
        .iter()
        .filter_map(|(id, degree)| (*degree == 0).then_some(*id))
        .collect::<VecDeque<_>>();
    let mut visited = 0usize;
    while let Some(node) = queue.pop_front() {
        visited += 1;
        for target in outgoing.get(node).into_iter().flatten() {
            let degree = indegree
                .get_mut(target)
                .ok_or("HOLOLAKE_COMPOSITION_EDGE_NODE_UNKNOWN")?;
            *degree -= 1;
            if *degree == 0 {
                queue.push_back(target);
            }
        }
    }
    if visited != recipe.nodes.len() {
        return Err("HOLOLAKE_COMPOSITION_CYCLE_REJECTED".into());
    }
    Ok(())
}

fn view_module_id(view: ProjectionView) -> &'static str {
    match view {
        ProjectionView::Dashboard => "HLC-PROJECT-DASHBOARD",
        ProjectionView::Comparison => "HLC-PROJECT-COMPARISON",
        ProjectionView::VerticalBar => "HLC-PROJECT-VERTICAL-BAR",
        ProjectionView::Classification => "HLC-PROJECT-CLASSIFICATION",
        ProjectionView::Table => "HLC-PROJECT-TABLE",
    }
}

fn measure_value(value: (usize, u64, usize), measure: CompositionMeasure) -> u64 {
    match measure {
        CompositionMeasure::DocumentCount => value.0 as u64,
        CompositionMeasure::TotalBytes => value.1,
        CompositionMeasure::DuplicateCount => value.2 as u64,
    }
}

fn top_level_folder(path: &str) -> String {
    let normalized = path.replace('\\', "/");
    let parts = normalized
        .split('/')
        .filter(|part| !part.is_empty())
        .collect::<Vec<_>>();
    // “导入/导入批次名”是转译外壳，不是人类真正要比较的知识分类。
    let candidate = if parts.first() == Some(&"导入") {
        parts.get(2).copied().unwrap_or("根页面")
    } else {
        parts.first().copied().unwrap_or("未分类")
    };
    if candidate.ends_with(".md") || candidate.ends_with(".markdown") || candidate.ends_with(".txt")
    {
        "根目录".into()
    } else {
        candidate.into()
    }
}

fn source_label(source: &str) -> String {
    match source {
        "native" => "光湖原生知识".into(),
        "legacy" => "待迁移知识".into(),
        other => other.into(),
    }
}

fn stable_key(value: &str) -> String {
    sha256(value.as_bytes())[..16].to_string()
}
fn sha256_json<T: Serialize>(value: &T) -> Result<String, String> {
    serde_json::to_vec(value)
        .map(|bytes| sha256(&bytes))
        .map_err(|error| format!("HOLOLAKE_COMPOSITION_DATA_SERIALIZE_FAILED: {error}"))
}
fn sha256(bytes: &[u8]) -> String {
    digest(&SHA256, bytes)
        .as_ref()
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}
fn now_unix_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|value| value.as_millis().min(u64::MAX as u128) as u64)
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::knowledge_base::{KnowledgeDocumentSummary, KnowledgeSnapshot};

    fn snapshot() -> KnowledgeSnapshot {
        KnowledgeSnapshot {
            schema: "hololake.native-knowledge-base/v1",
            state: "READY",
            native_root: "/private/account".into(),
            legacy_available: false,
            legacy_root: None,
            documents: vec![
                KnowledgeDocumentSummary {
                    source: "native",
                    path: "课程/第一课.md".into(),
                    title: "第一课".into(),
                    updated_at_unix_ms: 1,
                    size_bytes: 100,
                    content_sha256: "a".into(),
                    duplicate_count: 1,
                },
                KnowledgeDocumentSummary {
                    source: "native",
                    path: "课程/第二课.md".into(),
                    title: "第二课".into(),
                    updated_at_unix_ms: 2,
                    size_bytes: 300,
                    content_sha256: "b".into(),
                    duplicate_count: 0,
                },
                KnowledgeDocumentSummary {
                    source: "native",
                    path: "记录.md".into(),
                    title: "记录".into(),
                    updated_at_unix_ms: 3,
                    size_bytes: 50,
                    content_sha256: "c".into(),
                    duplicate_count: 0,
                },
            ],
            raw_document_count: 4,
            unique_document_count: 3,
            duplicate_document_count: 1,
            truncated: false,
        }
    }

    #[test]
    fn real_native_rows_feed_all_projection_views_from_one_result() {
        let result = execute_snapshot(
            snapshot(),
            ExecuteCompositionInput {
                dimension: CompositionDimension::TopLevelFolder,
                measure: CompositionMeasure::TotalBytes,
                views: vec![
                    ProjectionView::Dashboard,
                    ProjectionView::VerticalBar,
                    ProjectionView::Table,
                ],
            },
        )
        .unwrap();
        assert_eq!(result.state, "EXECUTED");
        assert_eq!(result.native_object.row_count, 3);
        assert_eq!(result.metrics.total_bytes, 450);
        assert_eq!(result.groups[0].label, "课程");
        assert_eq!(result.groups[0].measure_value, 400);
        assert!(result.source_is_real_account_data);
        assert!(result.read_only);
    }

    #[test]
    fn duplicate_or_empty_projection_requests_fail_closed() {
        assert!(validate_input(&ExecuteCompositionInput {
            dimension: CompositionDimension::Source,
            measure: CompositionMeasure::DocumentCount,
            views: vec![]
        })
        .is_err());
        assert!(validate_input(&ExecuteCompositionInput {
            dimension: CompositionDimension::Source,
            measure: CompositionMeasure::DocumentCount,
            views: vec![ProjectionView::Table, ProjectionView::Table]
        })
        .is_err());
    }

    #[test]
    fn imported_wrapper_folders_do_not_flatten_real_knowledge_categories() {
        assert_eq!(
            top_level_folder("导入/光湖语言世界/AI技能包/索引.md"),
            "AI技能包"
        );
        assert_eq!(top_level_folder("导入/光湖语言世界/首页.md"), "根目录");
        assert_eq!(top_level_folder("课程/第一课.md"), "课程");
    }

    #[test]
    fn unknown_modules_type_mismatches_and_cycles_fail_closed() {
        let mut unknown = compile_recipe(&ExecuteCompositionInput {
            dimension: CompositionDimension::Source,
            measure: CompositionMeasure::DocumentCount,
            views: vec![ProjectionView::Table],
        });
        unknown.nodes[0].module_id = "UNKNOWN".into();
        assert_eq!(
            validate_recipe(&unknown, &registry()).unwrap_err(),
            "HOLOLAKE_COMPOSITION_MODULE_NOT_REGISTERED"
        );

        let mut mismatch = compile_recipe(&ExecuteCompositionInput {
            dimension: CompositionDimension::Source,
            measure: CompositionMeasure::DocumentCount,
            views: vec![ProjectionView::Table],
        });
        mismatch.edges[0] = CompositionEdge {
            from: "aggregate".into(),
            to: "classify".into(),
        };
        assert_eq!(
            validate_recipe(&mismatch, &registry()).unwrap_err(),
            "HOLOLAKE_COMPOSITION_PORT_TYPE_MISMATCH"
        );

        let cycle_registry = ModuleRegistry {
            schema: REGISTRY_SCHEMA,
            state: "TEST",
            modules: vec![
                descriptor("LOOP-A", "A", "TRANSFORM", Some("loop/v1"), "loop/v1"),
                descriptor("LOOP-B", "B", "TRANSFORM", Some("loop/v1"), "loop/v1"),
            ],
            arbitrary_script_allowed: false,
            unregistered_module_allowed: false,
            direct_projection_write_allowed: false,
        };
        let cycle = CompositionRecipe {
            schema: RECIPE_SCHEMA,
            recipe_id: "TEST-CYCLE",
            title: "cycle",
            nodes: vec![
                CompositionNode {
                    node_id: "a".into(),
                    module_id: "LOOP-A".into(),
                },
                CompositionNode {
                    node_id: "b".into(),
                    module_id: "LOOP-B".into(),
                },
            ],
            edges: vec![
                CompositionEdge {
                    from: "a".into(),
                    to: "b".into(),
                },
                CompositionEdge {
                    from: "b".into(),
                    to: "a".into(),
                },
            ],
        };
        assert_eq!(
            validate_recipe(&cycle, &cycle_registry).unwrap_err(),
            "HOLOLAKE_COMPOSITION_CYCLE_REJECTED"
        );
    }
}