feat(luminork): Add ability to see upgrade_available and upgrade schema

For an installed schema, we should see if there's an upgrade available
and if so, add the ability to upgrade the schema

Author: stack72

lib/luminork-server/src/service/v1.rs

@@ -159,6 +159,7 @@ pub use schemas::{
     },
     unlock_schema::UnlockedSchemaV1Response,
     update_schema_variant::UpdateSchemaVariantV1Request,
+    upgrade_schema::UpgradeSchemaResponse,
 };
 pub use search::{
     SearchV1Request,
@@ -221,6 +222,7 @@ pub use crate::api_types::func_run::v1::{
         schemas::create_codegen::create_variant_codegen,
         schemas::create_management::create_variant_management,
         schemas::update_schema_variant::update_schema_variant,
+        schemas::upgrade_schema::upgrade_schema,
         funcs::get_func_run::get_func_run,
         funcs::get_func::get_func,
         funcs::update_func::update_func,

lib/luminork-server/src/service/v1/schemas/create_schema.rs

@@ -93,6 +93,7 @@ pub async fn create_schema(
         default_variant_id: created_schema_variant.id,
         variant_ids: variants.into_iter().map(|v| v.id).collect(),
         schema_id: schema.id(),
+        upgrade_available: None,
     }))
 }
 

lib/luminork-server/src/service/v1/schemas/find_schema.rs

@@ -24,6 +24,7 @@ use utoipa::{
 use super::{
     SchemaError,
     SchemaResult,
+    check_schema_upgrade_available,
 };
 use crate::extract::{
     PosthogEventTracker,
@@ -133,6 +134,12 @@ pub async fn find_schema(
         },
     };
 
+    let upgrade_available = if installed {
+        check_schema_upgrade_available(ctx, schema_id).await?
+    } else {
+        None
+    };
+
     tracker.track(
         ctx,
         "api_find_schema",
@@ -146,6 +153,7 @@ pub async fn find_schema(
         schema_id,
         category,
         installed,
+        upgrade_available,
     }))
 }
 
@@ -160,6 +168,8 @@ pub struct FindSchemaV1Response {
     pub category: Option<String>,
     #[schema(value_type = bool)]
     pub installed: bool,
+    #[schema(value_type = Option<bool>)]
+    pub upgrade_available: Option<bool>,
 }
 
 enum SchemaReference {

lib/luminork-server/src/service/v1/schemas/get_schema.rs

@@ -23,6 +23,7 @@ use super::{
     SchemaResponseV1,
     SchemaResult,
     SchemaV1RequestPath,
+    check_schema_upgrade_available,
 };
 use crate::extract::{
     PosthogEventTracker,
@@ -58,6 +59,7 @@ pub async fn get_schema(
     if let Ok(Some(schema)) = Schema::get_by_id_opt(ctx, schema_id).await {
         let default_variant_id = Schema::default_variant_id(ctx, schema_id).await?;
         let variants = SchemaVariant::list_for_schema(ctx, schema_id).await?;
+        let upgrade_available = check_schema_upgrade_available(ctx, schema_id).await?;
 
         tracker.track(
             ctx,
@@ -75,6 +77,7 @@ pub async fn get_schema(
             name: schema.name,
             default_variant_id,
             variant_ids: variants.into_iter().map(|v| v.id).collect_vec(),
+            upgrade_available,
         }));
     }
 
@@ -101,6 +104,7 @@ pub async fn get_schema(
                     name: cached_schema.name,
                     default_variant_id: cached_schema.default_variant_id,
                     variant_ids: cached_schema.variant_ids,
+                    upgrade_available: None, // Uninstalled schema
                 }));
             }
         }

lib/luminork-server/src/service/v1/schemas/mod.rs

@@ -45,6 +45,7 @@ use si_frontend_mv_types::{
     management::ManagementFuncKind,
     prop_schema::PropSchemaV1 as CachedPropSchemaV1,
 };
+use si_pkg::SiPkgError;
 use thiserror::Error;
 use utoipa::{
     self,
@@ -67,6 +68,7 @@ pub mod list_schemas;
 pub mod search_schemas;
 pub mod unlock_schema;
 pub mod update_schema_variant;
+pub mod upgrade_schema;
 
 #[remain::sorted]
 #[derive(Debug, Error)]
@@ -87,8 +89,12 @@ pub enum SchemaError {
     MaterializedViews(#[from] dal_materialized_views::Error),
     #[error("schema missing asset func id: {0}")]
     MissingVariantFunc(SchemaVariantId),
+    #[error("module error: {0}")]
+    Module(#[from] dal::module::ModuleError),
     #[error("changes not permitted on HEAD change set")]
     NotPermittedOnHead,
+    #[error("pkg error: {0}")]
+    Pkg(#[from] dal::pkg::PkgError),
     #[error("prop error: {0}")]
     Prop(#[from] Box<PropError>),
     #[error("schema error: {0}")]
@@ -103,10 +109,18 @@ pub enum SchemaError {
     SchemaVariantNotFound(SchemaVariantId),
     #[error("schema variant {0} not a variant for the schema {1} error")]
     SchemaVariantNotMemberOfSchema(SchemaId, SchemaVariantId),
+    #[error("sipkg error: {0}")]
+    SiPkg(#[from] SiPkgError),
     #[error("slow runtime error: {0}")]
     SlowRuntime(#[from] dal::slow_rt::SlowRuntimeError),
     #[error("transactions error: {0}")]
     Transactions(#[from] TransactionsError),
+    #[error("Cannot upgrade schema {0} - has unlocked variants that need to be locked first")]
+    UnlockedVariantFoundForSchema(SchemaId),
+    #[error("No cached module found for schema {0}")]
+    UpgradableModuleNotFound(SchemaId),
+    #[error("No schema variants imported for schema {0}")]
+    UpgradeFailed(SchemaId),
     #[error("validation error: {0}")]
     Validation(String),
     #[error("variant authoring error: {0}")]
@@ -144,6 +158,10 @@ impl crate::service::v1::common::ErrorIntoResponse for SchemaError {
             SchemaError::SchemaNotFound(_) => (StatusCode::NOT_FOUND, self.to_string()),
             SchemaError::SchemaNotFoundByName(_) => (StatusCode::NOT_FOUND, self.to_string()),
             SchemaError::SchemaVariantNotFound(_) => (StatusCode::NOT_FOUND, self.to_string()),
+            SchemaError::UpgradableModuleNotFound(_) => (StatusCode::NOT_FOUND, self.to_string()),
+            SchemaError::UnlockedVariantFoundForSchema(_) => {
+                (StatusCode::PRECONDITION_FAILED, self.to_string())
+            }
             SchemaError::SchemaVariantNotMemberOfSchema(_, _) => {
                 (StatusCode::PRECONDITION_REQUIRED, self.to_string())
             }
@@ -187,6 +205,7 @@ pub fn routes() -> Router<AppState> {
             Router::new()
                 .route("/", get(get_schema::get_schema))
                 .route("/unlock", post(unlock_schema::unlock_schema))
+                .route("/upgrade", post(upgrade_schema::upgrade_schema))
                 .nest(
                     "/variant",
                     Router::new()
@@ -632,6 +651,8 @@ pub struct GetSchemaV1Response {
     pub default_variant_id: SchemaVariantId,
     #[schema(value_type = Vec<String>, example = json!(["01H9ZQD35JPMBGHH69BT0Q79VZ", "01H9ZQD35JPMBGHH69BT0Q79VY"]))]
     pub variant_ids: Vec<SchemaVariantId>,
+    #[schema(value_type = Option<bool>)]
+    pub upgrade_available: Option<bool>,
 }
 
 #[derive(Serialize, Debug, ToSchema)]
@@ -698,21 +719,58 @@ pub struct SchemaResponse {
     pub installed: bool,
 }
 
+/// Check if an upgrade is available for a given schema.
+///
+/// Returns:
+/// - `Some(true)` if an upgrade is available (installed hash is in past hashes)
+/// - `Some(false)` if no upgrade is available (hashes match or no newer version)
+/// - `None` if upgrade check is not applicable (no cached module or no installed module)
+pub async fn check_schema_upgrade_available(
+    ctx: &DalContext,
+    schema_id: SchemaId,
+) -> SchemaResult<Option<bool>> {
+    // Get the latest cached module
+    let Some(cached_module) = CachedModule::find_latest_for_schema_id(ctx, schema_id).await? else {
+        return Ok(None);
+    };
+
+    // Get the installed module
+    let Some(installed_module) =
+        dal::module::Module::find_for_module_schema_id(ctx, schema_id.into()).await?
+    else {
+        return Ok(None);
+    };
+
+    // Check if hashes differ
+    if cached_module.latest_hash == installed_module.root_hash() {
+        return Ok(Some(false));
+    }
+
+    // Check if installed hash is in past hashes (means upgrade available)
+    let past_hashes = CachedModule::list_for_schema_id(ctx, schema_id)
+        .await?
+        .iter()
+        .map(|cm| cm.latest_hash.to_owned())
+        .collect::<HashSet<String>>();
+
+    Ok(Some(past_hashes.contains(installed_module.root_hash())))
+}
+
 pub async fn get_full_schema_list(ctx: &DalContext) -> SchemaResult<Vec<SchemaResponse>> {
     let schema_ids = dal::Schema::list_ids(ctx).await?;
     let installed_schema_ids: HashSet<_> = schema_ids.iter().collect();
 
     // Get cached modules with their metadata
     let cached_modules = CachedModule::latest_modules(ctx).await?;
-    // Create a map of schema ID to cached module data
     let mut cached_module_map: HashMap<SchemaId, CachedModule> = HashMap::new();
     for module in cached_modules {
         cached_module_map.insert(module.schema_id, module);
     }
 
     // Combine both sources to create a complete list
     let mut all_schemas: Vec<SchemaResponse> = Vec::new();
-    // First add installed schemas from Schema::list_ids
+
+    // First add installed schemas
     for schema_id in &schema_ids {
         if let Some(module) = cached_module_map.get(schema_id) {
             // Schema is both installed and in cache

lib/luminork-server/src/service/v1/schemas/upgrade_schema.rs

@@ -0,0 +1,108 @@
+use axum::{
+    Json,
+    extract::Path,
+};
+use dal::{
+    Schema,
+    SchemaVariant,
+    cached_module::CachedModule,
+    pkg::{
+        ImportOptions,
+        import_pkg_from_pkg,
+    },
+};
+use serde::{
+    Deserialize,
+    Serialize,
+};
+use serde_json::json;
+use utoipa::ToSchema;
+
+use super::{
+    SchemaError,
+    SchemaResult,
+    SchemaV1RequestPath,
+};
+use crate::extract::{
+    PosthogEventTracker,
+    change_set::ChangeSetDalContext,
+};
+
+#[derive(Deserialize, Serialize, Debug, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct UpgradeSchemaResponse {
+    pub success: bool,
+}
+
+#[utoipa::path(
+      post,
+      path = "/v1/w/{workspace_id}/change-sets/{change_set_id}/schemas/{schema_id}/upgrade",
+      params(
+          ("workspace_id" = String, Path, description = "Workspace identifier"),
+          ("change_set_id" = String, Path, description = "Change Set identifier"),
+          ("schema_id" = String, Path, description = "Schema identifier"),
+      ),
+      tag = "schemas",
+      summary = "Upgrade a schema to the latest available version",
+      responses(
+          (status = 200, description = "Schema upgraded successfully", body = UpgradeSchemaResponse),
+          (status = 401, description = "Unauthorized - Invalid or missing token"),
+          (status = 404, description = "Schema not found"),
+          (status = 404, description = "No cached module found for schema"),
+          (status = 412, description = "Precondition failed - schema has unlocked variants or not installed"),
+          (status = 500, description = "Internal server error", body = crate::service::v1::common::ApiError)
+      )
+  )]
+pub async fn upgrade_schema(
+    ChangeSetDalContext(ref ctx): ChangeSetDalContext,
+    tracker: PosthogEventTracker,
+    Path(SchemaV1RequestPath { schema_id }): Path<SchemaV1RequestPath>,
+) -> SchemaResult<Json<UpgradeSchemaResponse>> {
+    let schema_exists_locally = Schema::exists_locally(ctx, schema_id).await?;
+    if !schema_exists_locally {
+        return Err(SchemaError::SchemaNotFound(schema_id));
+    }
+
+    // Check if there's an  unlocked variant
+    if SchemaVariant::get_unlocked_for_schema(ctx, schema_id)
+        .await?
+        .is_some()
+    {
+        return Err(SchemaError::UnlockedVariantFoundForSchema(schema_id));
+    }
+
+    let Some(mut cached_module) = CachedModule::find_latest_for_schema_id(ctx, schema_id).await?
+    else {
+        return Err(SchemaError::UpgradableModuleNotFound(schema_id));
+    };
+
+    let si_pkg = cached_module.si_pkg(ctx).await?;
+    let metadata = si_pkg.metadata()?;
+
+    let (_, schema_variant_ids, _) = import_pkg_from_pkg(
+        ctx,
+        &si_pkg,
+        Some(ImportOptions {
+            schema_id: Some(schema_id.into()),
+            ..Default::default()
+        }),
+    )
+    .await?;
+
+    if schema_variant_ids.is_empty() {
+        return Err(SchemaError::UpgradeFailed(schema_id));
+    }
+
+    tracker.track(
+        ctx,
+        "api_upgrade_schema",
+        json!({
+            "schema_id": schema_id,
+            "pkg_name": metadata.name(),
+        }),
+    );
+
+    ctx.commit().await?;
+
+    Ok(Json(UpgradeSchemaResponse { success: true }))
+}