feat(backend): resume from owner directly in flow status viewer (#1042)

* foo

* progress

backend/Cargo.lock

@@ -4095,7 +4095,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-api"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "anyhow",
  "argon2",
@@ -4146,7 +4146,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-api-client"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "base64",
  "chrono",
@@ -4161,7 +4161,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-audit"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "chrono",
  "serde",
@@ -4174,7 +4174,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-common"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "anyhow",
  "axum",
@@ -4198,7 +4198,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "serde",
  "serde_json",
@@ -4206,7 +4206,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-bash"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "anyhow",
  "itertools",
@@ -4220,7 +4220,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-go"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "anyhow",
  "itertools",
@@ -4232,7 +4232,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-py"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "anyhow",
  "itertools",
@@ -4247,7 +4247,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-ts"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "anyhow",
  "deno_core",
@@ -4261,7 +4261,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-queue"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "anyhow",
  "chrono",
@@ -4284,7 +4284,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-worker"
-version = "1.56.0"
+version = "1.56.1"
 dependencies = [
  "anyhow",
  "async-recursion",

backend/windmill-api/openapi.yaml

@@ -2931,6 +2931,23 @@ paths:
               schema:
                 $ref: "#/components/schemas/Job"
 
+  # /w/{workspace}/jobs/flow/current_state/{id}:
+  #   get:
+  #     summary: get flow current step state
+  #     operationId: getJob
+  #     tags:
+  #       - job
+  #     parameters:
+  #       - $ref: "#/components/parameters/WorkspaceId"
+  #       - $ref: "#/components/parameters/JobId"
+  #     responses:
+  #       "200":
+  #         description: state details
+  #         content:
+  #           application/json:
+  #             schema:
+  #               type: string
+
   /w/{workspace}/jobs/getupdate/{id}:
     get:
       summary: get job updates
@@ -3156,6 +3173,29 @@ paths:
               schema:
                 type: string
 
+  /w/{workspace}/jobs/flow/resume/{id}:
+    post:
+      summary: resume a job for a suspended flow as an owner
+      operationId: resumeSuspendedJobAsOwner
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/JobId"
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+      responses:
+        "201":
+          description: job resumed
+          content:
+            text/plain:
+              schema:
+                type: string
+
   /w/{workspace}/jobs/cancel/{id}/{resume_id}/{signature}:
     get:
       summary: cancel a job for a suspended flow

backend/windmill-api/src/jobs.rs

@@ -63,6 +63,7 @@ pub fn workspaced_service() -> Router {
         .route("/completed/get_result/:id", get(get_completed_job_result))
         .route("/completed/delete/:id", post(delete_completed_job))
         .route("/get/:id", get(get_job))
+        .route("/flow/resume/:id", post(resume_suspended_job_as_owner))
         .route("/getupdate/:id", get(get_job_update))
         .route(
             "/job_signature/:job_id/:resume_id",
@@ -454,6 +455,25 @@ async fn list_jobs(
     Ok(Json(jobs.into_iter().map(From::from).collect()))
 }
 
+pub async fn resume_suspended_job_as_owner(
+    authed: Authed,
+    Extension(db): Extension<DB>,
+    Path((w_id, job_id)): Path<(String, Uuid)>,
+    QueryOrBody(value): QueryOrBody<serde_json::Value>,
+) -> error::Result<StatusCode> {
+    let value = value.unwrap_or(serde_json::Value::Null);
+    let mut tx = db.begin().await?;
+
+    let flow = get_suspended_flow_info(job_id, &mut tx).await?;
+
+    insert_resume_job(0, job_id, &flow, value, Some(authed.username), &mut tx).await?;
+
+    resume_immediately_if_relevant(flow, job_id, &mut tx).await?;
+
+    tx.commit().await?;
+    Ok(StatusCode::CREATED)
+}
+
 pub async fn resume_suspended_job(
     /* unauthed */
     Extension(db): Extension<DB>,
@@ -472,18 +492,7 @@ pub async fn resume_suspended_job(
     }
     mac.verify_slice(hex::decode(secret)?.as_ref())
         .map_err(|_| anyhow::anyhow!("Invalid signature"))?;
-    let flow = sqlx::query!(
-        r#"
-        SELECT id, flow_status, suspend
-        FROM queue
-        WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1)
-        FOR UPDATE
-        "#,
-        job_id,
-    )
-    .fetch_optional(&mut tx)
-    .await?
-    .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?;
+    let flow = get_suspended_flow_info(job_id, &mut tx).await?;
 
     let exists = sqlx::query_scalar!(
         r#"
@@ -499,6 +508,55 @@ pub async fn resume_suspended_job(
         return Err(anyhow::anyhow!("resume request already sent").into());
     }
 
+    insert_resume_job(resume_id, job_id, &flow, value, approver.approver, &mut tx).await?;
+
+    resume_immediately_if_relevant(flow, job_id, &mut tx).await?;
+
+    tx.commit().await?;
+    Ok(StatusCode::CREATED)
+}
+
+/* If the flow is currently waiting to be resumed (`FlowStatusModule::WaitingForEvents`)
+ * the suspend column must be set to the number of resume messages waited on.
+ *
+ * The flow's queue row is locked in this transaction because to avoid race conditions around
+ * the suspend column.
+ * That is, a job needs one event but it hasn't arrived, a worker counts zero events before
+ * entering WaitingForEvents.  Then this message arrives but the job isn't in WaitingForEvents
+ * yet so the suspend counter isn't updated.  Then the job enters WaitingForEvents expecting
+ * one event to arrive based on the count that is no longer correct. */
+async fn resume_immediately_if_relevant<'c>(
+    flow: FlowInfo,
+    job_id: Uuid,
+    tx: &mut Transaction<'c, Postgres>,
+) -> error::Result<()> {
+    Ok(
+        if let Some(suspend) = (0 < flow.suspend).then(|| flow.suspend - 1) {
+            let status =
+                serde_json::from_value::<FlowStatus>(flow.flow_status.context("no flow status")?)
+                    .context("deserialize flow status")?;
+            if matches!(status.current_step(), Some(FlowStatusModule::WaitingForEvents { job, .. }) if job == &job_id)
+            {
+                sqlx::query!(
+                    "UPDATE queue SET suspend = $1 WHERE id = $2",
+                    suspend,
+                    flow.id,
+                )
+                .execute(tx)
+                .await?;
+            }
+        },
+    )
+}
+
+async fn insert_resume_job<'c>(
+    resume_id: u32,
+    job_id: Uuid,
+    flow: &FlowInfo,
+    value: serde_json::Value,
+    approver: Option<String>,
+    tx: &mut Transaction<'c, Postgres>,
+) -> error::Result<()> {
     sqlx::query!(
         r#"
         INSERT INTO resume_job
@@ -510,38 +568,38 @@ pub async fn resume_suspended_job(
         job_id,
         flow.id,
         value,
-        approver.approver
+        approver
     )
-    .execute(&mut tx)
+    .execute(tx)
     .await?;
+    Ok(())
+}
 
-    /* If the flow is currently waiting to be resumed (`FlowStatusModule::WaitingForEvents`)
-     * the suspend column must be set to the number of resume messages waited on.
-     *
-     * The flow's queue row is locked in this transaction because to avoid race conditions around
-     * the suspend column.
-     * That is, a job needs one event but it hasn't arrived, a worker counts zero events before
-     * entering WaitingForEvents.  Then this message arrives but the job isn't in WaitingForEvents
-     * yet so the suspend counter isn't updated.  Then the job enters WaitingForEvents expecting
-     * one event to arrive based on the count that is no longer correct. */
-    if let Some(suspend) = (0 < flow.suspend).then(|| flow.suspend - 1) {
-        let status =
-            serde_json::from_value::<FlowStatus>(flow.flow_status.context("no flow status")?)
-                .context("deserialize flow status")?;
-        if matches!(status.current_step(), Some(FlowStatusModule::WaitingForEvents { job, .. }) if job == &job_id)
-        {
-            sqlx::query!(
-                "UPDATE queue SET suspend = $1 WHERE id = $2",
-                suspend,
-                flow.id,
-            )
-            .execute(&mut tx)
-            .await?;
-        }
-    }
+#[derive(sqlx::FromRow)]
+struct FlowInfo {
+    id: Uuid,
+    flow_status: Option<serde_json::Value>,
+    suspend: i32,
+}
 
-    tx.commit().await?;
-    Ok(StatusCode::CREATED)
+async fn get_suspended_flow_info<'c>(
+    job_id: Uuid,
+    tx: &mut Transaction<'c, Postgres>,
+) -> error::Result<FlowInfo> {
+    let flow = sqlx::query_as!(
+        FlowInfo,
+        r#"
+        SELECT id, flow_status, suspend
+        FROM queue
+        WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1)
+        FOR UPDATE
+        "#,
+        job_id,
+    )
+    .fetch_optional(tx)
+    .await?
+    .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?;
+    Ok(flow)
 }
 
 pub async fn cancel_suspended_job(
@@ -1439,17 +1497,21 @@ async fn get_completed_job(
     Ok(Json(job))
 }
 
-// async fn get_flow_current_step_state(
-//     Extension(db): Extension<DB>,
-//     Path((w_id, id)): Path<(String, Uuid)>,
-// ) -> error::JsonResult<String> {
-//     let x = sqlx::query!("
-//     SELECT (flow_status->'step')::integer as step, jsonb_array_length(flow_status->'modules') as len
-//         FROM queue WHERE id = $1 AND workspace_id = $2",
-//     id, w_id)
-//     .fetch_optional(&db).await?;
-//     Ok(Json(String::new()))
-// }
+async fn get_flow_current_step_state(
+    Extension(db): Extension<DB>,
+    Path((w_id, id)): Path<(String, Uuid)>,
+) -> error::JsonResult<String> {
+    let x = sqlx::query!(
+        "
+    SELECT (flow_status->'step')::integer as step, jsonb_array_length(flow_status->'modules') as len
+        FROM queue WHERE id = $1 AND workspace_id = $2",
+        id,
+        w_id
+    )
+    .fetch_optional(&db)
+    .await?;
+    Ok(Json(String::new()))
+}
 
 async fn get_completed_job_result(
     Extension(db): Extension<DB>,

backend/windmill-queue/src/jobs.rs

@@ -453,7 +453,7 @@ pub async fn push<'c>(
         {
             let mut modules = flow.modules.clone();
             modules.push(FlowModule {
-                id: "".to_string(),
+                id: format!("{}-v", flow.modules[flow.modules.len() - 1].id),
                 value: FlowModuleValue::Identity,
                 input_transforms: HashMap::new(),
                 stop_after_if: None,

frontend/src/lib/components/FlowPreviewContent.svelte

@@ -17,6 +17,7 @@
 	let capturePayload: CapturePayload
 	export let previewMode: 'upTo' | 'whole'
 	export let open: boolean
+	export let is_owner: boolean = false
 
 	export let jobId: string | undefined = undefined
 	export let job: Job | undefined = undefined
@@ -48,6 +49,7 @@
 				return m
 			})
 	}
+
 	function extractFlow(previewMode: 'upTo' | 'whole'): Flow {
 		if (previewMode === 'whole') {
 			return $flowStore
@@ -156,7 +158,7 @@
 		/>
 		<div class="h-full pt-4 grow">
 			{#if jobId}
-				<FlowStatusViewer bind:flowState={$flowStateStore} {jobId} bind:job />
+				<FlowStatusViewer bind:is_owner bind:flowState={$flowStateStore} {jobId} bind:job />
 			{:else}
 				<div class="italic text-gray-500 h-full grow"> Flow status will be displayed here </div>
 			{/if}

frontend/src/lib/components/FlowStatusViewer.svelte

@@ -1,10 +1,10 @@
 <script lang="ts">
 	import { FlowStatusModule, Job, JobService } from '$lib/gen'
-	import { workspaceStore } from '$lib/stores'
+	import { userStore, workspaceStore } from '$lib/stores'
 	import FlowJobResult from './FlowJobResult.svelte'
 	import FlowPreviewStatus from './preview/FlowPreviewStatus.svelte'
 	import Icon from 'svelte-awesome'
-	import { faChevronDown, faChevronUp, faHourglassHalf } from '@fortawesome/free-solid-svg-icons'
+	import { faChevronDown, faChevronUp } from '@fortawesome/free-solid-svg-icons'
 	import { createEventDispatcher } from 'svelte'
 	import { onDestroy } from 'svelte'
 	import type { FlowState } from './flows/flowState'
@@ -13,8 +13,11 @@
 	import Tabs from './common/tabs/Tabs.svelte'
 	import { FlowGraph, type GraphModuleState } from './graph'
 	import ModuleStatus from './ModuleStatus.svelte'
-	import { displayDate, truncateRev } from '$lib/utils'
+	import { displayDate, isOwner, truncateRev } from '$lib/utils'
 	import JobArgs from './JobArgs.svelte'
+	import autosize from 'svelte-autosize'
+	import Tooltip from './Tooltip.svelte'
+	import SimpleEditor from './SimpleEditor.svelte'
 
 	const dispatch = createEventDispatcher()
 
@@ -34,6 +37,8 @@
 	let localFlowModuleStates: Record<string, GraphModuleState> = {}
 	export let retry_status: Record<string, number> = {}
 
+	export let is_owner = false
+
 	let selectedNode: string | undefined = undefined
 
 	let jobResults: any[] = []
@@ -133,11 +138,12 @@
 
 	$: job && dispatch('jobsLoaded', job)
 
-	function updateJobId() {
+	async function updateJobId() {
 		if (jobId !== job?.id) {
 			retry_status = {}
 			localFlowModuleStates = {}
-			loadJobInProgress()
+			await loadJobInProgress()
+			job?.script_path && loadOwner(job.script_path)
 		}
 	}
 
@@ -149,7 +155,13 @@
 		timeout && clearTimeout(timeout)
 	})
 
+	async function loadOwner(path: string) {
+		is_owner = await isOwner(path, $userStore!, $workspaceStore!)
+	}
+
 	let selected: 'graph' | 'sequence' = 'graph'
+
+	let payload: string = '"a test payload in json"'
 </script>
 
 {#if job}
@@ -168,6 +180,41 @@
 					<div class="w-full h-full">
 						<FlowJobResult result={job.result} logs={job.logs ?? ''} />
 					</div>
+				{:else if job.flow_status?.modules?.[job?.flow_status?.step].type === FlowStatusModule.type.WAITING_FOR_EVENTS}
+					<div class="w-full h-full mt-2 text-sm text-gray-600">
+						<p>Waiting for approval from the previous step</p>
+						<div>
+							{#if is_owner}
+								<div class="flex flex-row gap-2 mt-2">
+									<div>
+										<Button
+											color="green"
+											variant="border"
+											on:click={async () =>
+												await JobService.resumeSuspendedJobAsOwner({
+													workspace: $workspaceStore ?? '',
+													id: job?.flow_status?.modules?.[job?.flow_status?.step - 1]?.job ?? '',
+													requestBody: JSON.parse(payload)
+												})}
+											>Resume <Tooltip
+												>Since you are an owner of this flow, you can send resume events without
+												necessarily knowing the resume id sent by the approval step</Tooltip
+											></Button
+										>
+									</div>
+									<div class="w-full border rounded-lg border-gray-600 p-2">
+										<SimpleEditor automaticLayout lang="json" bind:code={payload} autoHeight />
+									</div>
+									<Tooltip
+										>The payload is optional, it is passed to the following step through the
+										`resume` variable</Tooltip
+									>
+								</div>
+							{:else}
+								You cannot resume the job without the resume id since you are not an owner of {job.script_path}
+							{/if}
+						</div>
+					</div>
 				{:else if job.logs}
 					<div class="text-xs p-4 bg-gray-50 overflow-auto max-h-80 border">
 						<pre class="w-full">{job.logs}</pre>

frontend/src/lib/components/InputTransformForm.svelte

@@ -16,7 +16,6 @@
 	import type { InputTransform } from '$lib/gen'
 	import TemplateEditor from './TemplateEditor.svelte'
 	import Tooltip from './Tooltip.svelte'
-	import { escape } from 'svelte/internal'
 
 	export let schema: Schema
 	export let arg: InputTransform | any
@@ -28,7 +27,8 @@
 	export let variableEditor: VariableEditor | undefined = undefined
 	export let itemPicker: ItemPicker | undefined = undefined
 
-	export let monaco: SimpleEditor | undefined = undefined
+	let monaco: SimpleEditor | undefined = undefined
+	let monacoTemplate: TemplateEditor | undefined = undefined
 	let argInput: ArgInput | undefined = undefined
 
 	let inputCat: InputCat = 'object'
@@ -73,6 +73,7 @@
 		if (isStaticTemplate(inputCat)) {
 			arg.value = `\$\{${rawValue}}`
 			setPropertyType(arg.value)
+			monacoTemplate?.setCode(arg.value)
 		} else {
 			arg.expr = getDefaultExpr(undefined, previousModuleId, rawValue)
 			arg.type = 'javascript'
@@ -86,6 +87,7 @@
 			focusProp(argName, 'append', (path) => {
 				const toAppend = `\$\{${path}}`
 				arg.value = `${arg.value ?? ''}${toAppend}`
+				monacoTemplate?.setCode(arg.value)
 				setPropertyType(arg.value)
 				argInput?.focus()
 				return false
@@ -219,8 +221,13 @@
 			</span>
 		{/if}
 		{#if isStaticTemplate(inputCat) && propertyType == 'static'}
-			<div class="py-1">
-				<TemplateEditor {extraLib} on:focus={onFocus} bind:code={arg.value} />
+			<div class="py-1 rounded border border-1 border-gray-500">
+				<TemplateEditor
+					bind:this={monacoTemplate}
+					{extraLib}
+					on:focus={onFocus}
+					bind:code={arg.value}
+				/>
 			</div>
 		{:else if propertyType === undefined || propertyType == 'static'}
 			<ArgInput

frontend/src/lib/components/TemplateEditor.svelte

@@ -401,6 +401,19 @@
 		}
 	}
 
+	export function insertAtCursor(code: string): void {
+		if (editor) {
+			editor.trigger('keyboard', 'type', { text: code })
+		}
+	}
+
+	export function setCode(ncode: string): void {
+		code = ncode
+		if (editor) {
+			editor.setValue(ncode)
+		}
+	}
+
 	export function getCode(): string {
 		return editor?.getValue() ?? ''
 	}

frontend/src/lib/components/flows/content/FlowModuleScript.svelte

@@ -6,6 +6,7 @@
 
 	let code: string
 	let language: 'deno' | 'python3' | 'go' | 'bash'
+	let description: string
 
 	async function loadCode(path: string) {
 		const script = await getScriptByPath(path!)

frontend/src/lib/components/flows/header/FlowPreviewButtons.svelte

@@ -30,6 +30,8 @@
 			'failure'
 		].includes($selectedId) ||
 		$selectedId?.includes('branch')
+
+	let is_owner = false
 </script>
 
 <div class="flex flex-row-reverse justify-between items-center gap-x-2">
@@ -65,6 +67,7 @@
 
 <Drawer bind:open={previewOpen} size="75%">
 	<FlowPreviewContent
+		bind:is_owner
 		open={previewOpen}
 		bind:previewMode
 		bind:job

frontend/src/lib/components/flows/previousResults.ts

@@ -125,17 +125,17 @@ export function getStepPropPicker(
 
 
 
-	if (approvers && ((previousModule?.suspend?.required_events ?? 0) > 0)) {
+	if (pickableProperties.hasResume) {
 		pickableProperties["approvers"] = "The list of approvers"
 	}
 
 	return {
-		extraLib: buildExtraLib(flowInput, priorIds),
+		extraLib: buildExtraLib(flowInput, priorIds, previousModule?.suspend != undefined),
 		pickableProperties
 	}
 }
 
-export function buildExtraLib(flowInput: Record<string, any>, results: Record<string, any>): string {
+export function buildExtraLib(flowInput: Record<string, any>, results: Record<string, any>, resume: boolean): string {
 	return `
 /**
 * get variable (including secret) at path
@@ -163,6 +163,18 @@ declare const params: any;
  * result by id
  */
 declare const results = ${JSON.stringify(results)};
+
+${resume ? `
+/**
+ * resume payload
+ */
+declare const resume: any
+
+/**
+ * The list of approvers separated by ,
+ */
+declare const approvers: string
+` : ''}
 `
 
 }

frontend/src/lib/components/propertyPicker/PropPicker.svelte

@@ -129,7 +129,8 @@
 						pureViewer={!$propPickerConfig}
 						json={{
 							resume: 'The resume payload',
-							resumes: 'All resume payloads from all approvers'
+							resumes: 'All resume payloads from all approvers',
+							approvers: 'The list of approvers'
 						}}
 						on:select={(e) => {
 							dispatch('select', `${e.detail}`)

frontend/src/lib/utils.ts

@@ -474,7 +474,8 @@ export function scriptPathToHref(path: string): string {
 export async function getScriptByPath(path: string): Promise<{
 	content: string
 	language: SupportedLanguage
-	schema: any
+	schema: any,
+	description: string
 }> {
 	if (path.startsWith('hub/')) {
 		const { content, language, schema } = await ScriptService.getHubScriptByPath({ path })
@@ -482,7 +483,8 @@ export async function getScriptByPath(path: string): Promise<{
 		return {
 			content,
 			language: language as SupportedLanguage,
-			schema
+			schema,
+			description: ''
 		}
 	} else {
 		const script = await ScriptService.getScriptByPath({
@@ -492,7 +494,8 @@ export async function getScriptByPath(path: string): Promise<{
 		return {
 			content: script.content,
 			language: script.language,
-			schema: script.schema
+			schema: script.schema,
+			description: script.description
 		}
 	}
 }