diff --git a/backend/.sqlx/query-d2fe10b3e608407147adb704a7cf3da60f22dcce6f05852de2f75c737b57907f.json b/backend/.sqlx/query-d2fe10b3e608407147adb704a7cf3da60f22dcce6f05852de2f75c737b57907f.json new file mode 100644 index 0000000000..8e693056a0 --- /dev/null +++ b/backend/.sqlx/query-d2fe10b3e608407147adb704a7cf3da60f22dcce6f05852de2f75c737b57907f.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n schema \n FROM \n flow \n WHERE \n workspace_id = $1 AND \n path = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "schema", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "d2fe10b3e608407147adb704a7cf3da60f22dcce6f05852de2f75c737b57907f" +} diff --git a/backend/src/main.rs b/backend/src/main.rs index 849c4a7774..87a148d5c3 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -909,6 +909,8 @@ Windmill Community Edition {GIT_VERSION} } } &"flow" => { + let dynamic_input_key = windmill_common::jobs::generate_dynamic_input_key(workspace_id, path); + windmill_common::DYNAMIC_INPUT_CACHE.remove(&dynamic_input_key); windmill_common::FLOW_VERSION_CACHE.remove(&key); }, _ => { diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index fb7398faec..bdc6b8e76f 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -7504,6 +7504,30 @@ paths: application/json: schema: {} + /w/{workspace}/jobs/run/dynamic_select: + post: + summary: run dynamic select helper function + operationId: runDynamicSelect + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: dynamic select request + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/DynamicInputData" + responses: + "201": + description: dynamic select job created + content: + text/plain: + schema: + type: string + format: uuid + /w/{workspace}/jobs/queue/list: get: summary: list all queued jobs @@ -18121,6 +18145,49 @@ components: required: - path + DynamicInputData: + type: object + properties: + entrypoint_function: + type: string + description: Name of the function to execute for dynamic select + args: + type: object + description: Arguments to pass to the function + runnable_ref: + type: object + oneOf: + - type: object + properties: + source: + type: string + enum: [deployed] + path: + type: string + description: Path to the deployed script or flow + runnable_kind: + $ref: "#/components/schemas/RunnableKind" + required: + - source + - path + - runnable_kind + - type: object + properties: + source: + type: string + enum: [inline] + code: + type: string + description: Code content for inline execution + language: + $ref: "#/components/schemas/ScriptLang" + required: + - source + - code + required: + - entrypoint_function + - runnable_ref + WindmillLargeFile: type: object properties: diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 5071730b51..b8d51c3020 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -33,10 +33,11 @@ use windmill_common::error::JsonResult; use windmill_common::flow_status::{JobResult, RestartedFrom}; use windmill_common::jobs::{ check_tag_available_for_workspace_internal, format_completed_job_result, format_result, - ENTRYPOINT_OVERRIDE, + DynamicInput, ENTRYPOINT_OVERRIDE, }; -use windmill_common::utils::WarnAfterExt; +use windmill_common::utils::{RunnableKind, WarnAfterExt}; use windmill_common::worker::{Connection, CLOUD_HOSTED, TMP_DIR}; +use windmill_common::DYNAMIC_INPUT_CACHE; #[cfg(all(feature = "enterprise", feature = "smtp"))] use windmill_common::{email_oss::send_email_html, server::load_smtp_config}; @@ -192,6 +193,7 @@ pub fn workspaced_service() -> Router { "/run_wait_result/preview_flow", post(run_wait_result_preview_flow), ) + .route("/run/dynamic_select", post(run_dynamic_select)) .route("/list", get(list_jobs)) .route( "/list_selected_job_groups", @@ -3473,6 +3475,22 @@ struct PreviewFlow { restarted_from: Option, } +#[derive(Debug, Deserialize)] +struct DynamicSelectRequest { + pub entrypoint_function: String, + pub args: Option>>, + pub runnable_ref: DynamicSelectRunnableRef, +} + +#[derive(Debug, Deserialize)] +#[serde(tag = "source")] +pub enum DynamicSelectRunnableRef { + #[serde(rename = "deployed")] + Deployed { path: String, runnable_kind: RunnableKind }, + #[serde(rename = "inline")] + Inline { code: String, language: Option }, +} + pub struct QueryOrBody(pub Option); #[axum::async_trait] @@ -6000,6 +6018,155 @@ async fn run_wait_result_preview_flow( return result; } +async fn run_dynamic_select( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Path(w_id): Path, + Query(run_query): Query, + Json(request): Json, +) -> error::Result { + #[cfg(feature = "enterprise")] + check_license_key_valid().await?; + + if matches!( + request.runnable_ref, + DynamicSelectRunnableRef::Inline { .. } + ) && authed.is_operator + { + return Err(error::Error::NotAuthorized( + "Operators cannot run preview jobs for security reasons".to_string(), + )); + } + + let dynamic_input: DynamicInput; + + match request.runnable_ref { + DynamicSelectRunnableRef::Deployed { path, runnable_kind } => match runnable_kind { + RunnableKind::Script => { + let mut script_args = request.args.unwrap_or_default(); + script_args.insert( + "_ENTRYPOINT_OVERRIDE".to_string(), + serde_json::value::to_raw_value(&request.entrypoint_function)?, + ); + + let push_args = PushArgsOwned { extra: None, args: script_args.clone() }; + + let (uuid, _) = run_script_by_path_inner( + authed.clone(), + db.clone(), + user_db.clone(), + w_id.clone(), + StripPath(path), + run_query.clone(), + push_args.clone(), + ) + .await?; + + return Ok((StatusCode::CREATED, uuid.to_string()).into_response()); + } + RunnableKind::Flow => { + let mut conn = user_db.clone().begin(&authed).await?; + + let dynamic_input_res = match DYNAMIC_INPUT_CACHE.get(&format!("{}:{}", w_id, path)) + { + Some(cached) => cached.as_ref().clone(), + None => { + let dynamic_input = sqlx::query_scalar!( + r#" + SELECT + schema + FROM + flow + WHERE + workspace_id = $1 AND + path = $2 + "#, + &w_id, + &path + ) + .fetch_one(&mut *conn) + .await? + .and_then(|dynamic_input| { + Some(serde_json::from_value::(dynamic_input)) + }) + .transpose()?; + + let Some(dynamic_input) = dynamic_input else { + return Err(Error::BadRequest(format!( + "Flow at path {} does not have a dynamic select schema", + path + ))); + }; + + let dynamic_input_key = + windmill_common::jobs::generate_dynamic_input_key(&w_id, &path); + DYNAMIC_INPUT_CACHE + .insert(dynamic_input_key, Arc::new(dynamic_input.clone())); + dynamic_input + } + }; + + conn.commit().await?; + + dynamic_input = dynamic_input_res; + } + }, + DynamicSelectRunnableRef::Inline { code, language } => { + dynamic_input = DynamicInput { + x_windmill_dyn_select_code: code, + x_windmill_dyn_select_lang: language.unwrap_or_default(), + }; + } + } + + let scheduled_for = run_query.get_scheduled_for(&db).await?; + let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()); + + let (uuid, tx) = push( + &db, + tx, + &w_id, + JobPayload::Code(RawCode { + hash: None, + content: dynamic_input.x_windmill_dyn_select_code, + path: None, + language: dynamic_input.x_windmill_dyn_select_lang, + lock: None, + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + }), + PushArgs::from(&request.args.unwrap_or_default()), + authed.display_username(), + &authed.email, + username_to_permissioned_as(&authed.username), + authed.token_prefix.as_deref(), + scheduled_for, + None, + None, + None, + None, + run_query.job_id, + false, + false, + None, + true, + None, + run_query.timeout, + None, + None, + Some(&authed.clone().into()), + false, + ) + .await?; + tx.commit().await?; + + Ok((StatusCode::CREATED, uuid.to_string()).into_response()) +} + pub async fn run_job_by_hash( authed: ApiAuthed, Extension(db): Extension, diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index d6b58896bc..e8c663988b 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -29,6 +29,14 @@ use crate::{ FlowVersionInfo, ScriptHashInfo, }; +#[derive(Debug, Deserialize, Clone)] +pub struct DynamicInput { + #[serde(rename = "x-windmill-dyn-select-code")] + pub x_windmill_dyn_select_code: String, + #[serde(rename = "x-windmill-dyn-select-lang")] + pub x_windmill_dyn_select_lang: ScriptLang, +} + #[derive(sqlx::Type, Serialize, Deserialize, Debug, Clone)] #[sqlx(type_name = "JOB_TRIGGER_KIND", rename_all = "lowercase")] #[serde(rename_all = "lowercase")] @@ -549,6 +557,11 @@ pub async fn script_path_to_payload<'e>( )) } +#[inline(always)] +pub fn generate_dynamic_input_key(workspace_id: &str, path: &str) -> String { + format!("{workspace_id}:{path}") +} + pub async fn get_payload_tag_from_prefixed_path( path: &str, db: &DB, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 7c391a8fee..98dac87744 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -152,6 +152,7 @@ lazy_static::lazy_static! { pub static ref DEPLOYED_SCRIPT_HASH_CACHE: Cache<(String, String), ExpiringLatestVersionId> = Cache::new(1000); pub static ref FLOW_VERSION_CACHE: Cache<(String, String), ExpiringLatestVersionId> = Cache::new(1000); + pub static ref DYNAMIC_INPUT_CACHE: Cache> = Cache::new(1000); pub static ref DEPLOYED_SCRIPT_INFO_CACHE: Cache<(String, i64), ScriptHashInfo> = Cache::new(1000); pub static ref FLOW_INFO_CACHE: Cache<(String, i64), FlowVersionInfo> = Cache::new(1000); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index fcefc458f7..487a7be632 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3508,7 +3508,7 @@ macro_rules! fetch_scalar_isolated { use sqlx::types::JsonRawValue; -#[derive(Debug, Default)] +#[derive(Debug, Clone, Default)] pub struct PushArgsOwned { pub extra: Option>>, pub args: HashMap>, diff --git a/frontend/src/lib/components/DynamicInput.svelte b/frontend/src/lib/components/DynamicInput.svelte index 61beb8d715..23d2912052 100644 --- a/frontend/src/lib/components/DynamicInput.svelte +++ b/frontend/src/lib/components/DynamicInput.svelte @@ -84,22 +84,12 @@ reject(error) } } - helperScript?.type == 'inline' - ? resultJobLoader?.runPreview( - helperScript?.path ?? 'NO_PATH', - helperScript.code, - helperScript.lang, - { ...otherArgs, filterText, _ENTRYPOINT_OVERRIDE: entrypoint }, - undefined, - undefined, - undefined, - cb - ) - : resultJobLoader?.runScriptByHash( - helperScript?.hash ?? 'NO_HASH', - { ...otherArgs, filterText, _ENTRYPOINT_OVERRIDE: entrypoint }, - cb - ) + resultJobLoader?.runDynamicInputScript( + entrypoint, + helperScript!, + { ...otherArgs, filterText, _ENTRYPOINT_OVERRIDE: entrypoint }, + cb + ) }) } diff --git a/frontend/src/lib/components/EditableSchemaForm.svelte b/frontend/src/lib/components/EditableSchemaForm.svelte index 34109cb083..297ee73ad0 100644 --- a/frontend/src/lib/components/EditableSchemaForm.svelte +++ b/frontend/src/lib/components/EditableSchemaForm.svelte @@ -442,7 +442,7 @@ } }} helperScript={{ - type: 'inline', + source: 'inline', code: dynCode!, lang: dynLang! }} diff --git a/frontend/src/lib/components/FlowPreviewContent.svelte b/frontend/src/lib/components/FlowPreviewContent.svelte index 1ac77da5a9..558a854bca 100644 --- a/frontend/src/lib/components/FlowPreviewContent.svelte +++ b/frontend/src/lib/components/FlowPreviewContent.svelte @@ -521,7 +521,7 @@ helperScript={flowStore.val.schema?.['x-windmill-dyn-select-code'] && flowStore.val.schema?.['x-windmill-dyn-select-lang'] ? { - type: 'inline', + source: 'inline', code: flowStore.val.schema['x-windmill-dyn-select-code'] as string, lang: flowStore.val.schema['x-windmill-dyn-select-lang'] as ScriptLang } diff --git a/frontend/src/lib/components/JobLoader.svelte b/frontend/src/lib/components/JobLoader.svelte index e27cf66047..89281c7619 100644 --- a/frontend/src/lib/components/JobLoader.svelte +++ b/frontend/src/lib/components/JobLoader.svelte @@ -18,7 +18,7 @@ import { onDestroy, tick, untrack } from 'svelte' import type { SupportedLanguage } from '$lib/common' import { sendUserToast } from '$lib/toast' - import { isScriptPreview } from '$lib/utils' + import { DynamicInput, isScriptPreview } from '$lib/utils' // Will be set to number if job is not a flow @@ -273,6 +273,22 @@ } } + export async function runDynamicInputScript( + entrypoint_function: string, + runnable_ref: DynamicInput.HelperScript, + args: Record, + callbacks?: Callbacks + ): Promise { + return abstractRun( + () => + JobService.runDynamicSelect({ + workspace: $workspaceStore!, + requestBody: { entrypoint_function, args, runnable_ref } + }), + callbacks + ) + } + export async function runPreview( path: string | undefined, code: string, diff --git a/frontend/src/lib/components/RunForm.svelte b/frontend/src/lib/components/RunForm.svelte index 2111531dfe..ef2dafc974 100644 --- a/frontend/src/lib/components/RunForm.svelte +++ b/frontend/src/lib/components/RunForm.svelte @@ -21,7 +21,6 @@ import { triggerableByAI } from '$lib/actions/triggerableByAI.svelte' import InputSelectedBadge from './schema/InputSelectedBadge.svelte' import { untrack } from 'svelte' - import { type ScriptLang } from '$lib/gen' let reloadArgs = $state(0) let jsonEditor: JsonInputs | undefined = $state(undefined) @@ -257,19 +256,11 @@ {#key reloadArgs}
{ return `