From 9b1c30eeff35291ad50f3ddeb64831eac88e2f66 Mon Sep 17 00:00:00 2001 From: HugoCasa Date: Fri, 16 May 2025 01:10:51 +0200 Subject: [PATCH] fix: add v1 preprocessor support to workspace preprocessor script (#5757) * fix: add v1 preprocessor support to workspace preprocessor script * nit * build --- ...cbb60a5446993217baffa83f843bc12a5ac73.json | 35 ---- ...ccc3be832b12102eb86a07d8405a4fa9287d5.json | 29 +++ backend/windmill-api/src/trigger_helpers.rs | 181 ++++++++++++------ 3 files changed, 147 insertions(+), 98 deletions(-) delete mode 100644 backend/.sqlx/query-0e296134f05593edc989c628c00cbb60a5446993217baffa83f843bc12a5ac73.json create mode 100644 backend/.sqlx/query-72916f8e490f8252e0a51b7f562ccc3be832b12102eb86a07d8405a4fa9287d5.json diff --git a/backend/.sqlx/query-0e296134f05593edc989c628c00cbb60a5446993217baffa83f843bc12a5ac73.json b/backend/.sqlx/query-0e296134f05593edc989c628c00cbb60a5446993217baffa83f843bc12a5ac73.json deleted file mode 100644 index 5b086529a6..0000000000 --- a/backend/.sqlx/query-0e296134f05593edc989c628c00cbb60a5446993217baffa83f843bc12a5ac73.json +++ /dev/null @@ -1,35 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT\n value->'preprocessor_module' IS NOT NULL as has_preprocessor,\n value->'preprocessor_module'->'value'->'input_transforms'->'wm_trigger' IS NOT NULL as is_v1_preprocessor,\n schema as \"schema: _\"\n FROM flow \n WHERE workspace_id = $1 \n AND path = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "has_preprocessor", - "type_info": "Bool" - }, - { - "ordinal": 1, - "name": "is_v1_preprocessor", - "type_info": "Bool" - }, - { - "ordinal": 2, - "name": "schema: _", - "type_info": "Json" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - null, - null, - true - ] - }, - "hash": "0e296134f05593edc989c628c00cbb60a5446993217baffa83f843bc12a5ac73" -} diff --git a/backend/.sqlx/query-72916f8e490f8252e0a51b7f562ccc3be832b12102eb86a07d8405a4fa9287d5.json b/backend/.sqlx/query-72916f8e490f8252e0a51b7f562ccc3be832b12102eb86a07d8405a4fa9287d5.json new file mode 100644 index 0000000000..5321de8ccb --- /dev/null +++ b/backend/.sqlx/query-72916f8e490f8252e0a51b7f562ccc3be832b12102eb86a07d8405a4fa9287d5.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n value->'preprocessor_module'->'value' as \"preprocessor_module: _\",\n schema as \"schema: _\"\n FROM flow \n WHERE workspace_id = $1\n AND path = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "preprocessor_module: _", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "schema: _", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null, + true + ] + }, + "hash": "72916f8e490f8252e0a51b7f562ccc3be832b12102eb86a07d8405a4fa9287d5" +} diff --git a/backend/windmill-api/src/trigger_helpers.rs b/backend/windmill-api/src/trigger_helpers.rs index 69a5a014a4..5d7112e0af 100644 --- a/backend/windmill-api/src/trigger_helpers.rs +++ b/backend/windmill-api/src/trigger_helpers.rs @@ -4,6 +4,7 @@ use serde_json::value::RawValue; use std::collections::HashMap; use windmill_common::{ error::Result, + flows::FlowModuleValue, get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path, scripts::{ScriptHash, ScriptLang}, worker::to_raw_value, @@ -38,12 +39,6 @@ struct ScriptInfo { schema: Option>, } -struct FlowInfo { - has_preprocessor: Option, - is_v1_preprocessor: Option, - schema: Option>, -} - #[derive(Debug, Deserialize)] struct PropertyDefinition { r#type: Option, @@ -104,9 +99,8 @@ async fn get_script_info( .await } -fn runnable_format_from_schema( +fn runnable_format_from_schema_without_preprocessor( trigger_kind: &TriggerKind, - has_preprocessor: bool, schema: Option>, ) -> RunnableFormat { match trigger_kind { @@ -119,7 +113,7 @@ fn runnable_format_from_schema( }) }) => { - RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor } + RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: false } } TriggerKind::Kafka | TriggerKind::Nats if schema.as_ref().is_some_and(|schema| { @@ -129,18 +123,46 @@ fn runnable_format_from_schema( .is_some_and(|properties| properties.keys().any(|key| key == "msg")) }) => { - RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor } + RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: false } } - _ => RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor }, + _ => RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor: false }, } } + +fn runnable_format_from_preprocessor_args( + args: Option>, +) -> RunnableFormat { + if let Some(args) = args { + if args.iter().any(|arg| arg.name == "wm_trigger") + || (args.len() > 0 && args.iter().all(|arg| arg.name != "event")) + { + RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: true } + } else { + RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor: true } + } + } else { + RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor: true } + } +} + +enum PreprocessorInfo { + Preprocessor { content: String, language: ScriptLang }, + NoPreprocessor { schema: Option> }, +} + +#[derive(Debug, Deserialize)] +struct FlowInfo { + preprocessor_module: Option>, + schema: Option>, +} + pub async fn get_runnable_format( runnable_id: RunnableId, workspace_id: &str, db: &DB, trigger_kind: &TriggerKind, ) -> Result { - match runnable_id { + let (key, preprocessor_info) = match runnable_id { RunnableId::FlowPath(path) => { let FlowVersionInfo { version, .. } = get_latest_flow_version_info_for_path(db, workspace_id, &path, true).await?; @@ -157,11 +179,10 @@ pub async fn get_runnable_format( let flow_info = sqlx::query_as!( FlowInfo, "SELECT - value->'preprocessor_module' IS NOT NULL as has_preprocessor, - value->'preprocessor_module'->'value'->'input_transforms'->'wm_trigger' IS NOT NULL as is_v1_preprocessor, + value->'preprocessor_module'->'value' as \"preprocessor_module: _\", schema as \"schema: _\" FROM flow - WHERE workspace_id = $1 + WHERE workspace_id = $1 AND path = $2", workspace_id, path @@ -169,18 +190,40 @@ pub async fn get_runnable_format( .fetch_one(db) .await?; - let has_preprocessor = flow_info.has_preprocessor.unwrap_or(false); - let is_v1_preprocessor = flow_info.is_v1_preprocessor.unwrap_or(false); - - let runnable_format = if has_preprocessor && is_v1_preprocessor { - RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: true } + if let Some(preprocessor_module) = flow_info.preprocessor_module { + match preprocessor_module.0 { + FlowModuleValue::RawScript { content, language, .. } => { + (key, PreprocessorInfo::Preprocessor { content, language }) + } + FlowModuleValue::Script { path, hash, .. } => { + let hash = if let Some(hash) = hash { + hash.0 + } else { + let script_hash = + get_latest_deployed_hash_for_path(db, workspace_id, &path).await?; + script_hash.hash + }; + let script_info = get_script_info(db, workspace_id, hash).await?; + ( + key, + PreprocessorInfo::Preprocessor { + content: script_info.content, + language: script_info.language, + }, + ) + } + _ => { + return Err(windmill_common::error::Error::internal_err( + "Unsupported preprocessor module".to_string(), + )); + } + } } else { - runnable_format_from_schema(trigger_kind, has_preprocessor, flow_info.schema) - }; - - RUNNABLE_FORMAT_VERSION_CACHE.insert(key, runnable_format); - - Ok(runnable_format) + ( + key, + PreprocessorInfo::NoPreprocessor { schema: flow_info.schema }, + ) + } } RunnableId::ScriptId(script_id) => { let hash = script_id.get_script_hash(workspace_id, db).await?; @@ -194,47 +237,59 @@ pub async fn get_runnable_format( let script_info = get_script_info(db, workspace_id, hash).await?; - let has_preprocessor = script_info.has_preprocessor.unwrap_or(false); - - let runnable_format = if has_preprocessor { - let args = match script_info.language { - ScriptLang::Bun - | ScriptLang::Bunnative - | ScriptLang::Deno - | ScriptLang::Nativets => { - let args = windmill_parser_ts::parse_deno_signature( - &script_info.content, - true, - false, - Some("preprocessor".to_string()), - )?; - Some(args.args) - } - ScriptLang::Python3 => { - let args = windmill_parser_py::parse_python_signature( - &script_info.content, - Some("preprocessor".to_string()), - false, - )?; - Some(args.args) - } - _ => None, - }; - - if args.is_some_and(|args| args.iter().any(|arg| arg.name == "wm_trigger")) { - RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: true } - } else { - runnable_format_from_schema(trigger_kind, has_preprocessor, script_info.schema) - } + if script_info.has_preprocessor.unwrap_or(false) { + ( + key, + PreprocessorInfo::Preprocessor { + content: script_info.content, + language: script_info.language, + }, + ) } else { - runnable_format_from_schema(trigger_kind, has_preprocessor, script_info.schema) + ( + key, + PreprocessorInfo::NoPreprocessor { schema: script_info.schema }, + ) + } + } + }; + + let runnable_format = match preprocessor_info { + PreprocessorInfo::Preprocessor { content, language } => { + let args = match language { + ScriptLang::Bun + | ScriptLang::Bunnative + | ScriptLang::Deno + | ScriptLang::Nativets => { + let args = windmill_parser_ts::parse_deno_signature( + &content, + true, + false, + Some("preprocessor".to_string()), + )?; + Some(args.args) + } + ScriptLang::Python3 => { + let args = windmill_parser_py::parse_python_signature( + &content, + Some("preprocessor".to_string()), + false, + )?; + Some(args.args) + } + _ => None, }; - RUNNABLE_FORMAT_VERSION_CACHE.insert(key, runnable_format); - - Ok(runnable_format) + runnable_format_from_preprocessor_args(args) } - } + PreprocessorInfo::NoPreprocessor { schema } => { + runnable_format_from_schema_without_preprocessor(trigger_kind, schema) + } + }; + + RUNNABLE_FORMAT_VERSION_CACHE.insert(key, runnable_format); + + Ok(runnable_format) } #[allow(dead_code)]