feat: allow operator to use script/flow with dynselect input (#6616)

* done

* update sqlx

* fix openapi spec

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"
+}

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);
                                                                 },
                                                                 _ => {

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:

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<RestartedFrom>,
 }
 
+#[derive(Debug, Deserialize)]
+struct DynamicSelectRequest {
+    pub entrypoint_function: String,
+    pub args: Option<HashMap<String, Box<JsonRawValue>>>,
+    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<ScriptLang> },
+}
+
 pub struct QueryOrBody<D>(pub Option<D>);
 
 #[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<DB>,
+    Extension(user_db): Extension<UserDB>,
+    Path(w_id): Path<String>,
+    Query(run_query): Query<RunJobQuery>,
+    Json(request): Json<DynamicSelectRequest>,
+) -> error::Result<Response> {
+    #[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::<DynamicInput>(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<DB>,

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,

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<String, Arc<jobs::DynamicInput>> = 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);
 

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<HashMap<String, Box<RawValue>>>,
     pub args: HashMap<String, Box<RawValue>>,

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
+			)
 		})
 	}
 

frontend/src/lib/components/EditableSchemaForm.svelte

@@ -442,7 +442,7 @@
 								}
 							}}
 							helperScript={{
-								type: 'inline',
+								source: 'inline',
 								code: dynCode!,
 								lang: dynLang!
 							}}

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
 											}

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<string, any>,
+		callbacks?: Callbacks
+	): Promise<string> {
+		return abstractRun(
+			() =>
+				JobService.runDynamicSelect({
+					workspace: $workspaceStore!,
+					requestBody: { entrypoint_function, args, runnable_ref }
+				}),
+			callbacks
+		)
+	}
+
 	export async function runPreview(
 		path: string | undefined,
 		code: string,

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}
 				<div bind:clientHeight={schemaHeight}>
 					<SchemaForm
-						helperScript={runnable.hash
-							? {
-									type: 'hash',
-									hash: runnable.hash
-								}
-							: runnable.schema?.['x-windmill-dyn-select-code'] &&
-								  runnable.schema?.['x-windmill-dyn-select-lang']
-								? {
-										type: 'inline',
-										code: runnable.schema['x-windmill-dyn-select-code'] as string,
-										lang: runnable.schema['x-windmill-dyn-select-lang'] as ScriptLang
-									}
-								: undefined}
+						helperScript={{
+							source: 'deployed',
+							path: runnable.path!,
+							runnable_kind: runnable.hash ? 'script' : 'flow'
+						}}
 						prettifyHeader
 						{noVariablePicker}
 						{autofocus}

frontend/src/lib/components/ScriptEditor.svelte

@@ -751,7 +751,7 @@
 								{#key argsRender}
 									<SchemaForm
 										helperScript={{
-											type: 'inline',
+											source: 'inline',
 											code,
 											//@ts-ignore
 											lang

frontend/src/lib/utils.ts

@@ -10,7 +10,7 @@ import { deepEqual } from 'fast-equals'
 import YAML from 'yaml'
 import { type UserExt } from './stores'
 import { sendUserToast } from './toast'
-import type { Job, Script, ScriptLang } from './gen'
+import type { Job, RunnableKind, Script, ScriptLang } from './gen'
 import type { EnumType, SchemaProperty } from './common'
 import type { Schema } from './common'
 export { sendUserToast }
@@ -621,8 +621,8 @@ export namespace DynamicInput {
 	const DYN_FORMAT_PREFIX = ['dynmultiselect-', 'dynselect-']
 
 	export type HelperScript =
-		| { type: 'inline'; path?: string; lang: Script['language']; code: string }
-		| { type: 'hash'; hash: string }
+		| { source: 'deployed'; path: string; runnable_kind: RunnableKind }
+		| { source: 'inline'; code: string; lang: ScriptLang }
 
 	export const generatePythonFnTemplate = (functionName: string): string => {
 		return `