Implement cached schema & variant materialized views infrastructure

This change establishes the foundation for materialized views (MVs) that
provide cached access to schema and variant data from uninstalled packages.
The implementation uses the MV building pattern established by the change set
MVs and supports graceful fallback chains in API endpoints.

Key components:
- CachedSchema and CachedDefaultVariant MV types for schema metadata access
- Deployment MV discovery system with direct key lookup in content-
  addressable store vs O(nm) traversal through layers
- JoinSet task scheduling for parallel MV building in edda-server
- Luminork API endpoints updated with MV-first lookups and DAL fallbacks
- Function collection system with error handling for variant data assembly

The MV system enables efficient access to schema information without
requiring full package installation, supporting API integrations and MCP
servers.

Author: jhelwig

Cargo.lock

@@ -6,6 +6,7 @@ rust_library(
         "//lib/dal:dal",
         "//lib/si-frontend-mv-types-rs:si-frontend-mv-types",
         "//lib/si-id:si-id",
+        "//lib/si-pkg:si-pkg",
         "//lib/telemetry-rs:telemetry",
         "//third-party/rust:remain",
         "//third-party/rust:serde_json",

lib/dal-materialized-views/BUCK

@@ -14,6 +14,7 @@ remain = { workspace = true }
 serde_json = { workspace = true }
 si-frontend-mv-types = { path = "../../lib/si-frontend-mv-types-rs" }
 si-id = { path = "../../lib/si-id" }
+si-pkg = { path = "../../lib/si-pkg" }
 telemetry = { path = "../../lib/telemetry-rs" }
 thiserror = { workspace = true }
 

lib/dal-materialized-views/Cargo.toml

@@ -0,0 +1,37 @@
+use si_id::FuncId;
+use si_pkg::HasUniqueId;
+use telemetry::prelude::*;
+
+pub mod schema;
+pub mod schemas;
+
+/// Collects function IDs from SiPkg function types, handling invalid IDs gracefully.
+/// Logs warnings for function IDs that cannot be parsed as ULIDs and skips them.
+pub(crate) fn collect_function_ids<T, E, I, F>(
+    get_functions: F,
+    collector: &mut Vec<FuncId>,
+    schema_id: &str,
+    variant_id: &str,
+    func_type: &str,
+) -> Result<(), E>
+where
+    F: FnOnce() -> Result<I, E>,
+    I: IntoIterator<Item = T>,
+    T: HasUniqueId,
+    E: std::error::Error,
+{
+    for func in get_functions()? {
+        if let Some(unique_id) = func.unique_id() {
+            match unique_id.parse::<FuncId>() {
+                Ok(func_id) => collector.push(func_id),
+                Err(e) => {
+                    debug!(
+                        "Skipping invalid function ID '{}' in {} for schema {} variant {}: {}. Expected ULID format.",
+                        unique_id, func_type, schema_id, variant_id, e
+                    );
+                }
+            }
+        }
+    }
+    Ok(())
+}

lib/dal-materialized-views/src/cached.rs

@@ -0,0 +1,73 @@
+use dal::{
+    DalContext,
+    SchemaId,
+    SchemaVariantId,
+    cached_module::CachedModule,
+};
+use si_frontend_mv_types::cached_schema::CachedSchema as CachedSchemaMv;
+use telemetry::prelude::*;
+
+#[instrument(
+    name = "dal_materialized_views.cached_schema",
+    level = "debug",
+    skip_all
+)]
+pub async fn assemble(ctx: DalContext, id: SchemaId) -> crate::Result<CachedSchemaMv> {
+    let mut module = CachedModule::find_latest_for_schema_id(&ctx, id)
+        .await?
+        .ok_or_else(|| crate::Error::Schema(dal::SchemaError::UninstalledSchemaNotFound(id)))?;
+
+    // Get the SiPkg data to extract variant information
+    let si_pkg = module.si_pkg(&ctx).await?;
+    let schemas = si_pkg.schemas()?;
+    let schema = schemas
+        .into_iter()
+        .next()
+        .ok_or_else(|| crate::Error::Schema(dal::SchemaError::UninstalledSchemaNotFound(id)))?;
+
+    let variants = schema.variants()?;
+
+    // Find the default variant based on the schema data
+    let schema_data = schema
+        .data()
+        .ok_or_else(|| crate::Error::Schema(dal::SchemaError::UninstalledSchemaNotFound(id)))?;
+
+    let default_variant =
+        if let Some(default_variant_unique_id) = schema_data.default_schema_variant() {
+            // Find the variant with the matching unique_id
+            variants
+                .iter()
+                .find(|v| v.unique_id() == Some(default_variant_unique_id))
+                .or_else(|| variants.first()) // Fallback to first variant
+        } else {
+            // No default specified, use first variant
+            variants.first()
+        }
+        .ok_or_else(|| crate::Error::Schema(dal::SchemaError::UninstalledSchemaNotFound(id)))?;
+
+    // Extract the variant unique_id and convert to SchemaVariantId
+    let default_variant_id: SchemaVariantId = default_variant
+        .unique_id()
+        .ok_or_else(|| crate::Error::Schema(dal::SchemaError::UninstalledSchemaNotFound(id)))?
+        .parse::<SchemaVariantId>()
+        .map_err(|_| crate::Error::Schema(dal::SchemaError::UninstalledSchemaNotFound(id)))?;
+
+    // Collect all variant IDs from their unique_ids
+    let mut variant_ids = Vec::new();
+    for variant in &variants {
+        if let Some(unique_id) = variant.unique_id() {
+            if let Ok(variant_id) = unique_id.parse::<SchemaVariantId>() {
+                variant_ids.push(variant_id);
+            }
+        }
+    }
+
+    Ok(CachedSchemaMv::new(
+        id,
+        module.schema_name,
+        default_variant_id,
+        variant_ids,
+    ))
+}
+
+pub mod variant;

lib/dal-materialized-views/src/cached/schema.rs

@@ -0,0 +1,200 @@
+use dal::{
+    DalContext,
+    SchemaVariantId,
+    cached_module::CachedModule,
+    schema::variant::DEFAULT_SCHEMA_VARIANT_COLOR,
+};
+use si_frontend_mv_types::cached_schema_variant::CachedSchemaVariant as CachedSchemaVariantMv;
+use si_id::FuncId;
+use si_pkg::{
+    SiPkgSchema,
+    SiPkgSchemaVariant,
+};
+use telemetry::prelude::*;
+
+use crate::cached::collect_function_ids;
+
+/// Data structure for assembled variant information
+pub(crate) struct AssembledVariantData {
+    pub variant_id: SchemaVariantId,
+    pub display_name: String,
+    pub category: String,
+    pub color: String,
+    pub description: Option<String>,
+    pub link: Option<String>,
+    pub asset_func_id: FuncId,
+    pub variant_func_ids: Vec<FuncId>,
+}
+
+/// Assembles variant data from schema and variant components
+pub(crate) async fn assemble_variant_data(
+    schema: &SiPkgSchema<'_>,
+    variant: &SiPkgSchemaVariant<'_>,
+    variant_id: SchemaVariantId,
+) -> crate::Result<AssembledVariantData> {
+    // Get variant data
+    let variant_data = variant.data().ok_or_else(|| {
+        crate::Error::SchemaVariant(dal::SchemaVariantError::NotFound(variant_id))
+    })?;
+
+    // Get the asset func unique_id and convert to FuncId
+    let asset_func_id = variant_data
+        .func_unique_id()
+        .parse::<FuncId>()
+        .map_err(|_| crate::Error::SchemaVariant(dal::SchemaVariantError::NotFound(variant_id)))?;
+
+    // Get category from schema data
+    let category = schema
+        .data
+        .as_ref()
+        .map(|d| d.category())
+        .unwrap_or("Component") // default category
+        .to_string();
+
+    // Get display name from variant spec
+    let variant_spec = variant.to_spec().await?;
+    let display_name = variant_spec
+        .data
+        .as_ref()
+        .and_then(|d| d.display_name.as_ref())
+        .map(|d| d.to_string())
+        .unwrap_or_else(|| schema.name().to_string()); // fallback to schema name
+
+    // Collect all function IDs attached to this variant
+    let mut variant_func_ids = Vec::with_capacity(100); // Pre-allocate based on typical size
+    let schema_id_str = schema.unique_id().unwrap_or("unknown");
+    let variant_id_str = variant_id.to_string();
+
+    // All function types use the same helper with closure syntax
+    collect_function_ids(
+        || variant.leaf_functions(),
+        &mut variant_func_ids,
+        schema_id_str,
+        &variant_id_str,
+        "leaf function",
+    )?;
+    collect_function_ids(
+        || variant.action_funcs(),
+        &mut variant_func_ids,
+        schema_id_str,
+        &variant_id_str,
+        "action function",
+    )?;
+    collect_function_ids(
+        || variant.auth_funcs(),
+        &mut variant_func_ids,
+        schema_id_str,
+        &variant_id_str,
+        "auth function",
+    )?;
+    collect_function_ids(
+        || variant.management_funcs(),
+        &mut variant_func_ids,
+        schema_id_str,
+        &variant_id_str,
+        "management function",
+    )?;
+    collect_function_ids(
+        || variant.si_prop_funcs(),
+        &mut variant_func_ids,
+        schema_id_str,
+        &variant_id_str,
+        "si prop function",
+    )?;
+    collect_function_ids(
+        || variant.root_prop_funcs(),
+        &mut variant_func_ids,
+        schema_id_str,
+        &variant_id_str,
+        "root prop function",
+    )?;
+
+    // Remove duplicates efficiently, and ensure we have stable output for the MV.
+    variant_func_ids.sort_unstable();
+    variant_func_ids.dedup();
+
+    Ok(AssembledVariantData {
+        variant_id,
+        display_name,
+        category,
+        color: variant_data
+            .color()
+            .unwrap_or(DEFAULT_SCHEMA_VARIANT_COLOR)
+            .to_string(),
+        description: variant_data.description().map(|d| d.to_string()),
+        link: variant_data.link().map(|l| l.to_string()),
+        asset_func_id,
+        variant_func_ids,
+    })
+}
+
+#[instrument(
+    name = "dal_materialized_views.cached_schema_variant",
+    level = "debug",
+    skip_all
+)]
+pub async fn assemble(
+    ctx: DalContext,
+    id: SchemaVariantId,
+) -> crate::Result<CachedSchemaVariantMv> {
+    // Find the cached module containing this variant by storing the module info
+    for mut module in CachedModule::latest_modules(&ctx).await? {
+        let si_pkg = module.si_pkg(&ctx).await?;
+        let schemas = si_pkg.schemas()?;
+        for schema in schemas {
+            let variants = schema.variants()?;
+            for variant in variants {
+                if let Some(unique_id) = variant.unique_id() {
+                    if let Ok(variant_id) = unique_id.parse::<SchemaVariantId>() {
+                        if variant_id == id {
+                            // Found the variant, assemble its data
+                            let assembled_data =
+                                assemble_variant_data(&schema, &variant, id).await?;
+
+                            // Determine if this variant is the default variant for the schema
+                            let is_default_variant = if let Some(schema_data) = schema.data.as_ref()
+                            {
+                                if let Some(default_variant_unique_id) =
+                                    schema_data.default_schema_variant()
+                                {
+                                    // Check if this variant's unique_id matches the default
+                                    variant.unique_id() == Some(default_variant_unique_id)
+                                } else {
+                                    // No default specified in schema, check if this is the first variant
+                                    let all_variants = schema.variants()?;
+                                    all_variants.first().map(|first| first.unique_id())
+                                        == variant.unique_id().map(Some)
+                                }
+                            } else {
+                                // No schema data, assume first variant is default
+                                let all_variants = schema.variants()?;
+                                all_variants.first().map(|first| first.unique_id())
+                                    == variant.unique_id().map(Some)
+                            };
+
+                            return Ok(CachedSchemaVariantMv::new(
+                                assembled_data.variant_id,
+                                assembled_data.display_name,
+                                assembled_data.category,
+                                assembled_data.color,
+                                true, // is_locked - cached modules are locked until installed
+                                assembled_data.description,
+                                assembled_data.link,
+                                assembled_data.asset_func_id,
+                                assembled_data.variant_func_ids,
+                                is_default_variant,
+                            ));
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    // Variant not found
+    Err(crate::Error::SchemaVariant(
+        dal::SchemaVariantError::NotFound(id),
+    ))
+}
+
+pub mod default;