From d1fda55691ac35e1c47ceec086e847f46b326e66 Mon Sep 17 00:00:00 2001 From: Alexander Petric Date: Fri, 20 Dec 2024 21:47:45 +0100 Subject: [PATCH] feat: interactive slack approvals (#4942) * feat: interactive slack approvals * move form creation logic to backend * adding python client * polish messages * date time picker and default values * Initial commit * Initial commit * refactor * cleanup * cleanup * cleanup * cleanup, treating numbers/integers/booleans * adding slack to ts dependencies * sqlx prepare * gitignore, adding package.lock * ellipsis comments * Update typescript-client/client.ts Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com> * refactor to move everything serverside + modal * sqlx prep * reverting slack dependency * python client update * fix build errors * Update pyproject.toml --------- Co-authored-by: Ruben Fiszel Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com> --- ...c486137d4fe79906e22842b711f4a9379b8c8.json | 67 + backend/windmill-api/openapi.yaml | 36 + backend/windmill-api/src/jobs.rs | 19 +- backend/windmill-api/src/lib.rs | 5 +- backend/windmill-api/src/slack_approvals.rs | 1133 +++++++++++++++++ python-client/build.sh | 25 +- python-client/dev.nu | 4 +- python-client/wmill/wmill/client.py | 53 + typescript-client/README_DEV.md | 3 +- typescript-client/build.jsr.sh | 2 +- typescript-client/build.sh | 15 +- typescript-client/client.js | 2 +- typescript-client/client.ts | 68 +- 13 files changed, 1403 insertions(+), 29 deletions(-) create mode 100644 backend/.sqlx/query-bb6141ad0e93986b38ccdf4d027c486137d4fe79906e22842b711f4a9379b8c8.json create mode 100644 backend/windmill-api/src/slack_approvals.rs diff --git a/backend/.sqlx/query-bb6141ad0e93986b38ccdf4d027c486137d4fe79906e22842b711f4a9379b8c8.json b/backend/.sqlx/query-bb6141ad0e93986b38ccdf4d027c486137d4fe79906e22842b711f4a9379b8c8.json new file mode 100644 index 0000000000..da89afe0f7 --- /dev/null +++ b/backend/.sqlx/query-bb6141ad0e93986b38ccdf4d027c486137d4fe79906e22842b711f4a9379b8c8.json @@ -0,0 +1,67 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n queue.job_kind AS \"job_kind: JobKind\",\n queue.script_hash AS \"script_hash: ScriptHash\",\n queue.raw_flow AS \"raw_flow: sqlx::types::Json>\",\n completed_job.parent_job AS \"parent_job: Uuid\"\n FROM queue\n JOIN completed_job ON completed_job.parent_job = queue.id\n WHERE completed_job.id = $1 AND completed_job.workspace_id = $2\n LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "job_kind: JobKind", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlescriptflow", + "flowscript", + "flownode", + "appscript" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "script_hash: ScriptHash", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "raw_flow: sqlx::types::Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "parent_job: Uuid", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false, + true, + true, + true + ] + }, + "hash": "bb6141ad0e93986b38ccdf4d027c486137d4fe79906e22842b711f4a9379b8c8" +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 64c79ff154..a074501335 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -6943,6 +6943,42 @@ paths: - resume - cancel + /w/{workspace}/jobs/slack_approval/{id}: + get: + summary: generate interactive slack approval for suspended job + operationId: getSlackApprovalPayload + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/JobId" + - name: approver + in: query + schema: + type: string + - name: message + in: query + schema: + type: string + - name: slack_resource_path + in: query + required: true + schema: + type: string + - name: channel_id + in: query + required: true + schema: + type: string + - name: flow_step_id + in: query + required: true + schema: + type: string + responses: + "200": + description: Interactive slack approval message sent successfully + /w/{workspace}/jobs_u/resume/{id}/{resume_id}/{signature}: get: summary: resume a job for a suspended flow diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 4db486dcc5..d73c8d59fb 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -2160,7 +2160,7 @@ pub struct SuspendedJobFlow { pub approvers: Vec, } -#[derive(Deserialize)] +#[derive(Deserialize, Debug)] pub struct QueryApprover { pub approver: Option, } @@ -2373,11 +2373,11 @@ fn create_signature( } #[allow(non_snake_case)] -#[derive(Serialize)] +#[derive(Serialize, Debug)] pub struct ResumeUrls { - approvalPage: String, - cancel: String, - resume: String, + pub approvalPage: String, + pub cancel: String, + pub resume: String, } fn build_resume_url( @@ -2397,6 +2397,14 @@ pub async fn get_resume_urls( Extension(db): Extension, Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>, Query(approver): Query, +) -> error::JsonResult { + get_resume_urls_internal(Extension(db), Path((w_id, job_id, resume_id)), Query(approver)).await +} + +pub async fn get_resume_urls_internal( + Extension(db): Extension, + Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>, + Query(approver): Query, ) -> error::JsonResult { let key = get_workspace_key(&w_id, &db).await?; let signature = create_signature(key, job_id, resume_id, approver.approver.clone())?; @@ -5561,3 +5569,4 @@ async fn delete_completed_job<'a>( let response = Json(cj).into_response(); Ok(response) } + diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 96a0f7e3f6..3cb499af79 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -28,7 +28,7 @@ use crate::{ use anyhow::Context; use argon2::Argon2; use axum::extract::DefaultBodyLimit; -use axum::{middleware::from_extractor, routing::get, Extension, Router}; +use axum::{middleware::from_extractor, routing::get, routing::post, Extension, Router}; use db::DB; use http::HeaderValue; use reqwest::Client; @@ -109,6 +109,7 @@ mod websocket_triggers; mod workers; mod workspaces; mod workspaces_ee; +mod slack_approvals; mod workspaces_export; mod workspaces_extra; @@ -415,6 +416,8 @@ pub async fn run_server( "/w/:workspace_id/jobs_u", jobs::workspace_unauthed_service().layer(cors.clone()), ) + .route("/slack", post(slack_approvals::slack_app_callback_handler)) + .route("/w/:workspace_id/jobs/slack_approval/:job_id", get(slack_approvals::request_slack_approval)) .nest( "/w/:workspace_id/resources_u", resources::public_service().layer(cors.clone()), diff --git a/backend/windmill-api/src/slack_approvals.rs b/backend/windmill-api/src/slack_approvals.rs new file mode 100644 index 0000000000..7bd19ce848 --- /dev/null +++ b/backend/windmill-api/src/slack_approvals.rs @@ -0,0 +1,1133 @@ +use axum::{ + extract::{Form, Path, Query}, + Extension, +}; +use hyper::StatusCode; +use serde::{Deserialize, Serialize}; +use serde_json::value::{RawValue, Value}; + +use sqlx::types::Uuid; +use std::{collections::HashMap, str::FromStr}; +use windmill_common::error::{self, Error}; + +use regex::Regex; +use reqwest::Client; + +use crate::db::{ApiAuthed, DB}; +use crate::jobs::{ + cancel_suspended_job, get_resume_urls_internal, resume_suspended_job, QueryApprover, + QueryOrBody, ResumeUrls, +}; + +use windmill_common::{ + jobs::JobKind, + scripts::ScriptHash, + variables::{build_crypt, decrypt_value_with_mc}, +}; + +#[derive(Deserialize, Debug)] +pub struct SlackFormData { + payload: String, +} + +#[derive(Deserialize, Debug, Serialize)] +struct Container { + message_ts: String, + channel_id: String, +} + +#[derive(Deserialize, Debug)] +struct Payload { + actions: Option>, + view: Option, + trigger_id: Option, + #[serde(rename = "type")] + r#type: String, + container: Option, +} + +#[derive(Deserialize, Debug)] +struct View { + state: Option, + private_metadata: Option, +} + +#[derive(Deserialize, Debug)] +struct Action { + value: Option, + action_id: String, +} + +#[derive(Deserialize, Debug)] +struct State { + values: HashMap>, +} + +#[derive(Deserialize, Debug)] +#[serde(tag = "type", rename_all = "snake_case")] +enum ValueInput { + PlainTextInput { value: Option }, + Datepicker { selected_date: Option }, + Timepicker { selected_time: Option }, + StaticSelect { selected_option: Option }, + RadioButtons { selected_option: Option }, + Checkboxes { selected_options: Option> }, +} + +#[derive(Deserialize, Debug)] +struct SelectedOption { + value: String, +} + +#[derive(Debug, Deserialize, Serialize)] +struct ResumeSchema { + schema: Schema, +} + +#[derive(Debug, Deserialize)] +struct ResumeFormRow { + resume_form: Option, + hide_cancel: Option, +} + +#[derive(Debug, Deserialize, Serialize)] +struct Schema { + order: Vec, + required: Vec, + properties: HashMap, +} + +#[derive(Debug, Deserialize, Serialize)] +struct ResumeFormField { + #[serde(rename = "type")] + r#type: String, + format: Option, + default: Option, + description: Option, + title: Option, + #[serde(rename = "enum")] + r#enum: Option>, + #[serde(rename = "enumLabels")] + enum_labels: Option>, + nullable: Option, +} + +#[derive(Deserialize)] +pub struct QueryMessage { + message: Option, +} + +#[derive(Deserialize)] +pub struct QueryResourcePath { + slack_resource_path: Option, +} + +#[derive(Deserialize)] +pub struct QueryChannelId { + channel_id: Option, +} + +#[derive(Deserialize)] +pub struct QueryFlowStepId { + flow_step_id: Option, +} + +#[derive(Deserialize, Debug)] +struct ModalActionValue { + w_id: String, + job_id: String, + path: String, + approver: Option, + message: Option, + flow_step_id: Option, +} + +#[derive(Deserialize, Debug)] +struct PrivateMetadata { + resume_url: String, + resource_path: String, + container: Container, +} + +pub async fn slack_app_callback_handler( + authed: Option, + Extension(db): Extension, + Form(form_data): Form, +) -> Result { + tracing::debug!("Form data: {:#?}", form_data); + let payload: Payload = serde_json::from_str(&form_data.payload)?; + tracing::debug!("Payload: {:#?}", payload); + + match payload.r#type.as_str() { + "view_submission" => { + //print the container + handle_submission(authed, db, &payload, "resume").await? + } + "view_closed" => handle_submission(authed, db, &payload, "cancel").await?, + _ => { + if let Some(actions) = payload.actions.as_ref() { + if let Some(action) = actions.first() { + match action.action_id.as_str() { + "open_modal" => { + let trigger_id = payload.trigger_id.as_deref().ok_or_else(|| { + Error::BadRequest("No trigger_id found in payload.".to_string()) + })?; + + let value_str = action.value.as_ref().ok_or_else(|| { + Error::BadRequest("No action value found".to_string()) + })?; + + let parsed_value: ModalActionValue = serde_json::from_str(value_str) + .map_err(|_| { + Error::BadRequest("Invalid JSON in action value".to_string()) + })?; + + let w_id = &parsed_value.w_id; + let path = &parsed_value.path; + let approver = parsed_value.approver.as_deref(); + let message = parsed_value.message.as_deref(); + let job_id = Uuid::parse_str(&parsed_value.job_id)?; + let flow_step_id = parsed_value.flow_step_id.as_deref(); + + let slack_token = get_slack_token(&db, path, w_id).await?; + let client = Client::new(); + let container = payload.container.ok_or_else(|| { + Error::BadRequest("No container found.".to_string()) + })?; + + open_modal_with_blocks( + &client, + slack_token.as_str(), + trigger_id, + db, + w_id, + path, + job_id, + approver, + message, + flow_step_id, + container, + ) + .await + .map_err(|e| { + windmill_common::error::Error::BadRequest(e.to_string()) + })?; + } + _ => println!("Unknown action_id: {}", action.action_id), + } + } + } else { + tracing::debug!("Unkown Slack Action!"); + } + } + } + + Ok(StatusCode::OK) +} + +pub async fn request_slack_approval( + _authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, job_id)): Path<(String, Uuid)>, + Query(approver): Query, + Query(message): Query, + Query(slack_resource_path): Query, + Query(channel_id): Query, + Query(flow_step_id): Query, +) -> Result { + let slack_resource_path = match slack_resource_path.slack_resource_path { + Some(path) => path, + None => { + return Err(windmill_common::error::Error::BadRequest( + "slack_resource_path is required".to_string(), + )) + } + }; + + let channel_id = match channel_id.channel_id { + Some(id) => id, + None => { + return Err(windmill_common::error::Error::BadRequest( + "Slack channel_id is required".to_string(), + )) + } + }; + + let flow_step_id = match flow_step_id.flow_step_id { + Some(id) => id, + None => { + return Err(windmill_common::error::Error::BadRequest( + "Slack flow_step_id is required".to_string(), + )) + } + }; + + let slack_token = get_slack_token(&db, slack_resource_path.as_str(), &w_id).await?; + let client = Client::new(); + + // Optional fields + let approver_str = approver.approver.as_deref(); + let message_str = message.message.as_deref(); + + tracing::debug!("Approver: {:?}", approver_str); + tracing::debug!("Message: {:?}", message_str); + tracing::debug!("W ID: {:?}", w_id); + tracing::debug!("Slack Resource Path: {:?}", slack_resource_path); + tracing::debug!("Channel ID: {:?}", channel_id); + + // Use approver_str and message_str in the function call + send_slack_message( + &client, + slack_token.as_str(), + channel_id.as_str(), + &w_id, + job_id, + &slack_resource_path, + approver_str, + message_str, + flow_step_id.as_str(), + ) + .await + .map_err(|e| windmill_common::error::Error::BadRequest(e.to_string()))?; + + Ok(StatusCode::OK) +} + +async fn handle_submission( + authed: Option, + db: DB, + payload: &Payload, + action: &str, +) -> Result<(), Error> { + let view = payload + .view + .as_ref() + .ok_or_else(|| Error::BadRequest("No view found in payload.".to_string()))?; + + let state = view + .state + .as_ref() + .ok_or_else(|| Error::BadRequest("No state found in view.".to_string()))?; + + let values = &state.values; + tracing::debug!("Values: {:#?}", values); + + let state_values = parse_state_values(state); + let state_json = serde_json::to_value(state_values).unwrap_or_else(|_| serde_json::json!({})); + + let private_metadata = view + .private_metadata + .as_ref() + .ok_or_else(|| Error::BadRequest("No private metadata found.".to_string()))?; + + tracing::debug!("Private Metadata: {}", private_metadata); + let private_metadata: PrivateMetadata = serde_json::from_str(private_metadata)?; + let resume_url = private_metadata.resume_url; + let resource_path = private_metadata.resource_path; + let container: Container = private_metadata.container; + + // Use regex to extract information from private_metadata + let re = Regex::new(r"/api/w/(?P[^/]+)/jobs_u/(?Presume|cancel)/(?P[^/]+)/(?P[^/]+)/(?P[a-fA-F0-9]+)(?:\?approver=(?P[^&]+))?").unwrap(); + let captures = re.captures(resume_url.as_str()).ok_or_else(|| { + tracing::error!("Resume URL does not match the pattern."); + Error::BadRequest("Invalid URL format.".to_string()) + })?; + + let (w_id, job_id, resume_id, secret, approver) = ( + captures.name("w_id").map_or("", |m| m.as_str()), + captures.name("job_id").map_or("", |m| m.as_str()), + captures.name("resume_id").map_or("", |m| m.as_str()), + captures.name("secret").map_or("", |m| m.as_str()), + captures.name("approver").map(|m| m.as_str().to_string()), + ); + + let approver = QueryApprover { approver: approver }; + + // Convert job_id and resume_id to appropriate types + let job_uuid = Uuid::from_str(job_id) + .map_err(|_| Error::BadRequest("Invalid job ID format.".to_string()))?; + + let resume_id_parsed = resume_id + .parse::() + .map_err(|_| Error::BadRequest("Invalid resume ID format.".to_string()))?; + + // Call the appropriate function based on the action + let res = if action == "resume" { + resume_suspended_job( + authed, + Extension(db.clone()), + Path(( + w_id.to_string(), + job_uuid, + resume_id_parsed, + secret.to_string(), + )), + Query(approver), + QueryOrBody(Some(state_json)), + ) + .await + } else { + cancel_suspended_job( + authed, + Extension(db.clone()), + Path(( + w_id.to_string(), + job_uuid, + resume_id_parsed, + secret.to_string(), + )), + Query(approver), + QueryOrBody(Some(state_json)), + ) + .await + }; + tracing::debug!("Resume job action result: {:#?}", res); + let slack_token = get_slack_token(&db, &resource_path, &w_id).await?; + update_original_slack_message(action, slack_token, container).await?; + Ok(()) +} + +async fn transform_schemas( + text: &str, + properties: Option<&HashMap>, + urls: &ResumeUrls, + order: Option<&Vec>, + required: Option<&Vec>, +) -> Result { + tracing::debug!("Resume urls: {:#?}", urls); + + let mut blocks = vec![serde_json::json!({ + "type": "section", + "text": { + "type": "mrkdwn", + "text": format!("{}\n<{}|Flow suspension details>", text, urls.approvalPage), + } + })]; + + if let Some(properties) = properties { + for key in order.unwrap() { + if let Some(schema) = properties.get(key) { + let is_required = required.unwrap().contains(key); + let input_block = create_input_block(key, schema, is_required); + match input_block { + serde_json::Value::Array(arr) => blocks.extend(arr), + _ => blocks.push(input_block), + } + } + } + } + + Ok(serde_json::Value::Array(blocks)) +} + +fn create_input_block(key: &str, schema: &ResumeFormField, required: bool) -> serde_json::Value { + let placeholder = schema + .description + .as_deref() + .filter(|desc| !desc.is_empty()) + .unwrap_or("Select an option"); + + let title = schema.title.as_deref().unwrap_or(key); + let title_with_required = if required { + format!("{}*", title) + } else { + title.to_string() + }; + + // Handle boolean type + if schema.r#type == "boolean" { + let initial_value = schema + .default + .as_ref() + .and_then(|default| default.as_bool()) + .unwrap_or(false); + + let mut element = serde_json::json!({ + "type": "checkboxes", + "optional": !required, + "options": [{ + "text": { + "type": "plain_text", + "text": title_with_required, + "emoji": true + }, + "value": "true" + }], + "action_id": key + }); + + if initial_value { + element["initial_options"] = serde_json::json!([ + { + "text": { + "type": "plain_text", + "text": title_with_required, + "emoji": true + }, + "value": "true" + } + ]); + } + + return serde_json::json!({ + "type": "input", + "element": element, + "label": { + "type": "plain_text", + "text": title_with_required, + "emoji": true + } + }); + } + + // Handle date-time format + if schema.r#type == "string" && schema.format.as_deref() == Some("date-time") { + let now = chrono::Local::now(); + let current_date = now.format("%Y-%m-%d").to_string(); + let current_time = now.format("%H:%M").to_string(); + + let (default_date, default_time) = if let Some(default) = &schema.default { + if let Ok(parsed_date) = chrono::DateTime::parse_from_rfc3339(default.as_str().unwrap()) + { + ( + parsed_date.format("%Y-%m-%d").to_string(), + parsed_date.format("%H:%M").to_string(), + ) + } else { + (current_date.clone(), current_time.clone()) + } + } else { + (current_date.clone(), current_time.clone()) + }; + + return serde_json::json!([ + { + "type": "input", + "optional": !required, + "element": { + "type": "datepicker", + "initial_date": &default_date, + "placeholder": { + "type": "plain_text", + "text": "Select a date", + "emoji": true + }, + "action_id": format!("{}_date", key) + }, + "label": { + "type": "plain_text", + "text": title_with_required, + "emoji": true + } + }, + { + "type": "input", + "optional": !required, + "element": { + "type": "timepicker", + "initial_time": &default_time, + "placeholder": { + "type": "plain_text", + "text": "Select time", + "emoji": true + }, + "action_id": format!("{}_time", key) + }, + "label": { + "type": "plain_text", + "text": " ", + "emoji": true + } + } + ]); + } + + // Handle enum type + if let Some(enums) = &schema.r#enum { + let initial_option = schema.default.as_ref().and_then(|default_value| { + enums + .iter() + .find(|enum_value| enum_value == &default_value) + .map(|enum_value| { + serde_json::json!({ + "text": { + "type": "plain_text", + "text": schema.enum_labels.as_ref() + .and_then(|labels| labels.get(enum_value)) + .unwrap_or(enum_value), + "emoji": true + }, + "value": enum_value + }) + }) + }); + + let mut element = serde_json::json!({ + "type": "static_select", + "placeholder": { + "type": "plain_text", + "text": placeholder, + "emoji": true, + }, + "options": enums.iter().map(|enum_value| { + serde_json::json!({ + "text": { + "type": "plain_text", + "text": schema.enum_labels.as_ref() + .and_then(|labels| labels.get(enum_value)) + .unwrap_or(enum_value), + "emoji": true + }, + "value": enum_value + }) + }).collect::>(), + "action_id": key + }); + + if let Some(option) = initial_option { + element["initial_option"] = option; + } + + serde_json::json!({ + "type": "input", + "optional": !required, + "element": element, + "label": { + "type": "plain_text", + "text": title_with_required, + "emoji": true + } + }) + } else if schema.r#type == "number" || schema.r#type == "integer" { + // Handle number and integer types + let initial_value = schema + .default + .as_ref() + .and_then(|default| default.as_f64()) + .unwrap_or(0.0); + + let action_id_suffix = if schema.r#type == "number" { + "_type_number" + } else { + "_type_integer" + }; + + serde_json::json!({ + "type": "input", + "optional": !required, + "element": { + "type": "plain_text_input", + "action_id": format!("{}{}", key, action_id_suffix), + "initial_value": initial_value.to_string() + }, + "label": { + "type": "plain_text", + "text": title_with_required, + "emoji": true + } + }) + } else { + // Handle other types as string + let initial_value = schema + .default + .as_ref() + .and_then(|default| default.as_str()) + .unwrap_or(""); + + serde_json::json!({ + "type": "input", + "optional": !required, + "element": { + "type": "plain_text_input", + "action_id": key, + "initial_value": initial_value + }, + "label": { + "type": "plain_text", + "text": title_with_required, + "emoji": true + } + }) + } +} + +fn parse_state_values(state: &State) -> HashMap { + state + .values + .iter() + .flat_map(|(_, inputs)| { + inputs.iter().filter_map(|(action_id, input)| { + if action_id.ends_with("_date") { + process_datetime_inputs(action_id, input, &state) + } else { + process_non_datetime_inputs(action_id, input) + } + }) + }) + .collect() +} + +fn process_datetime_inputs( + action_id: &str, + input: &ValueInput, + state: &State, +) -> Option<(String, serde_json::Value)> { + let base_key = action_id.strip_suffix("_date").unwrap(); + let time_key = format!("{}_time", base_key); + + if let ValueInput::Datepicker { selected_date: Some(date) } = input { + let matching_time = state + .values + .values() + .flat_map(|inputs| inputs.get(&time_key)) + .find_map(|time_input| match time_input { + ValueInput::Timepicker { selected_time: Some(time) } => Some(time), + _ => None, + }); + + return matching_time.map(|time| { + ( + base_key.to_string(), + serde_json::json!(format!("{}T{}:00.000Z", date, time)), + ) + }); + } + None +} + +fn process_non_datetime_inputs( + action_id: &str, + input: &ValueInput, +) -> Option<(String, serde_json::Value)> { + match input { + // Handle number and integer types first + ValueInput::PlainTextInput { value } + if action_id.ends_with("_type_number") || action_id.ends_with("_type_integer") => + { + let base_action_id = action_id + .trim_end_matches("_type_number") + .trim_end_matches("_type_integer"); + value + .as_ref() + .and_then(|v| { + v.as_str().and_then(|s| s.parse::().ok()).map(|num| { + if action_id.ends_with("_type_integer") { + ( + base_action_id.to_string(), + serde_json::json!(num.floor() as i64), + ) + } else { + (base_action_id.to_string(), serde_json::json!(num)) + } + }) + }) + .or_else(|| { + tracing::error!("Failed to parse value to number: {:?}", value); + None + }) + } + + // Plain text input: Extracts the text value + ValueInput::PlainTextInput { value } => value + .as_ref() + .map(|v| (action_id.to_string(), v.clone().into())), + + // Static select: Extracts the selected option's value + ValueInput::StaticSelect { selected_option } => selected_option + .as_ref() + .map(|so| (action_id.to_string(), serde_json::json!(so.value))), + + // Radio buttons: Extracts the selected option's value + ValueInput::RadioButtons { selected_option } => selected_option + .as_ref() + .map(|so| (action_id.to_string(), serde_json::json!(so.value))), + + // Checkboxes: Convert single "true/false" string to boolean true/false + ValueInput::Checkboxes { selected_options } => selected_options.as_ref().map(|options| { + let is_true = options.iter().any(|opt| opt.value == "true"); + (action_id.to_string(), serde_json::json!(is_true)) + }), + + // Default case: Unsupported types return `None` + _ => None, + } +} + +async fn get_slack_token(db: &DB, slack_resource_path: &str, w_id: &str) -> anyhow::Result { + let slack_token = match sqlx::query!( + "SELECT value, is_secret FROM variable WHERE path = $1", + slack_resource_path + ) + .fetch_optional(db) + .await? + { + Some(row) => row, + None => { + return Err(anyhow::anyhow!("No slack token found")); + } + }; + + if slack_token.is_secret { + let mc = build_crypt(&db, w_id).await?; + let bot_token = decrypt_value_with_mc(slack_token.value, mc).await?; + Ok(bot_token) + } else { + Ok(slack_token.value) + } +} + +// Sends a Slack message with a button that opens a modal +async fn send_slack_message( + client: &Client, + bot_token: &str, + channel_id: &str, + w_id: &str, + job_id: Uuid, + resource_path: &str, + approver: Option<&str>, + message: Option<&str>, + flow_step_id: &str, +) -> Result> { + let url = "https://slack.com/api/chat.postMessage"; + + let mut value = serde_json::json!({ + "w_id": w_id, + "job_id": job_id, + "path": resource_path, + "channel": channel_id, + "flow_step_id": flow_step_id, + }); + + if let Some(approver) = approver { + value["approver"] = serde_json::json!(approver); + } + + if let Some(message) = message { + value["message"] = serde_json::json!(message); + } + + let payload = serde_json::json!({ + "channel": channel_id, + "text": "A flow has been suspended. Please approve or reject the flow.", + "blocks": [ + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "A flow has been suspended. Please approve or reject the flow." + } + }, + { + "type": "actions", + "elements": [ + { + "type": "button", + "text": { + "type": "plain_text", + "text": "View" + }, + "action_id": "open_modal", + "value": value.to_string() + } + ] + } + ] + }); + + tracing::debug!("Payload: {:?}", payload); + + let response = client + .post(url) + .bearer_auth(bot_token) + .json(&payload) + .send() + .await?; + + if response.status().is_success() { + tracing::info!("Interactive Slack approval message sent successfully!"); + tracing::debug!("Response: {:#?}", response.text().await?); + } else { + tracing::error!( + "Failed to send interactive Slack approval message: {}", + response.text().await? + ); + } + + Ok(StatusCode::OK) +} + +async fn get_modal_blocks( + db: DB, + w_id: &str, + job_id: Uuid, + resume_id: u32, + approver: Option<&str>, + message: Option<&str>, + trigger_id: &str, + flow_step_id: Option<&str>, + resource_path: &str, + container: Container, +) -> Result, windmill_common::error::Error> { + let res = get_resume_urls_internal( + axum::Extension(db.clone()), + Path((w_id.to_string(), job_id, resume_id)), + Query(QueryApprover { approver: approver.map(|a| a.to_string()) }), + ) + .await?; + + let urls = res.0; + + tracing::debug!("Job ID: {:?}", job_id); + + let (job_kind, script_hash, raw_flow, parent_job_id) = sqlx::query!( + "SELECT + queue.job_kind AS \"job_kind: JobKind\", + queue.script_hash AS \"script_hash: ScriptHash\", + queue.raw_flow AS \"raw_flow: sqlx::types::Json>\", + completed_job.parent_job AS \"parent_job: Uuid\" + FROM queue + JOIN completed_job ON completed_job.parent_job = queue.id + WHERE completed_job.id = $1 AND completed_job.workspace_id = $2 + LIMIT 1", + job_id, + &w_id + ) + .fetch_optional(&db) + .await + .map_err(|e| error::Error::BadRequest(e.to_string()))? + .ok_or_else(|| error::Error::BadRequest("This workflow is no longer running and has either already timed out or been cancelled or completed.".to_string())) + .map(|r| (r.job_kind, r.script_hash, r.raw_flow, r.parent_job))?; + + let flow_data = match windmill_common::cache::job::fetch_flow(&db, job_kind, script_hash).await + { + Ok(data) => data, + Err(_) => { + if let Some(parent_job_id) = parent_job_id.as_ref() { + windmill_common::cache::job::fetch_preview_flow(&db, parent_job_id, raw_flow) + .await? + } else { + return Err(error::Error::BadRequest( + "This workflow is no longer running and has either already timed out or been cancelled or completed.".to_string(), + )); + } + } + }; + + let flow_value = &flow_data.flow; + let flow_step_id = flow_step_id.unwrap_or(""); + let module = flow_value.modules.iter().find(|m| m.id == flow_step_id); + + tracing::debug!("Module: {:#?}", module); + let schema = module.and_then(|module| { + module.suspend.as_ref().map(|suspend| ResumeFormRow { + resume_form: suspend.resume_form.clone(), + hide_cancel: suspend.hide_cancel, + }) + }); + + let message_str = + message.unwrap_or("*A workflow has been suspended and is waiting for approval:*\n"); + + tracing::debug!("Schema: {:#?}", schema); + + if let Some(resume_schema) = schema { + let hide_cancel = resume_schema.hide_cancel.unwrap_or(false); + + if let Some(schema_obj) = resume_schema.resume_form { + let inner_schema: ResumeSchema = + serde_json::from_value(schema_obj.clone()).map_err(|e| { + tracing::error!("Failed to deserialize form schema: {:?}", e); + Error::BadRequest( + "Failed to deserialize resume form schema! Unsupported form field used." + .to_string(), + ) + })?; + + let blocks = transform_schemas( + message_str, + Some(&inner_schema.schema.properties), + &urls, + Some(&inner_schema.schema.order), + Some(&inner_schema.schema.required), + ) + .await?; + + tracing::debug!("Slack Blocks: {:#?}", blocks); + return Ok(axum::Json(construct_payload( + blocks, + hide_cancel, + trigger_id, + &urls.resume, + resource_path, + container, + ))); + } else { + tracing::debug!("No suspend form found!"); + let blocks = transform_schemas(message_str, None, &urls, None, None).await?; + return Ok(axum::Json(construct_payload( + blocks, + hide_cancel, + trigger_id, + &urls.resume, + resource_path, + container, + ))); + } + } else { + Err(Error::BadRequest( + "No approval form schema found.".to_string(), + )) + } +} + +fn construct_payload( + blocks: serde_json::Value, + hide_cancel: bool, + trigger_id: &str, + resume_url: &str, + resource_path: &str, + container: Container, +) -> serde_json::Value { + let mut view = serde_json::json!({ + "type": "modal", + "callback_id": "submit_form", + "notify_on_close": true, + "title": { + "type": "plain_text", + "text": "Worfklow Suspended" + }, + "blocks": blocks, + "submit": { + "type": "plain_text", + "text": "Resume Workflow" + }, + "private_metadata": serde_json::json!({ "resume_url": resume_url, "resource_path": resource_path, "container": container }).to_string(), + }); + + if !hide_cancel { + view["close"] = serde_json::json!({ + "type": "plain_text", + "text": "Cancel Workflow" + }); + } + + serde_json::json!({ + "trigger_id": trigger_id, + "view": view + }) +} + +async fn open_modal_with_blocks( + client: &Client, + bot_token: &str, + trigger_id: &str, + db: DB, + w_id: &str, + resource_path: &str, + job_id: Uuid, + approver: Option<&str>, + message: Option<&str>, + flow_step_id: Option<&str>, + container: Container, +) -> Result<(), Box> { + let resume_id = rand::random::(); + let blocks_json = match get_modal_blocks( + db, + w_id, + job_id, + resume_id, + approver, + message, + trigger_id, + flow_step_id, + resource_path, + container, + ) + .await + { + Ok(blocks) => blocks, + Err(e) => axum::Json(serde_json::json!({ + "trigger_id": trigger_id, + "view": { + "type": "modal", + "title": { "type": "plain_text", "text": "Error" }, + "blocks": [ { "type": "section", "text": { "type": "mrkdwn", "text": e.to_string() } } ] + } + })), + }; + + let blocks = &*blocks_json; + + tracing::debug!("Blocks: {:#?}", blocks); + + let url = "https://slack.com/api/views.open"; + + let response = client + .post(url) + .bearer_auth(bot_token) + .json(&blocks) + .send() + .await?; + + if response.status().is_success() { + tracing::info!("Slack modal opened successfully!"); + } else { + tracing::error!("Failed to open Slack modal: {}", response.text().await?); + } + + Ok(()) +} + +async fn update_original_slack_message( + action: &str, + token: String, + container: Container, +) -> Result<(), Error> { + let message = if action == "resume" { + "\n\n*Workflow has been resumed!* :white_check_mark:" + } else { + "\n\n*Workflow has been canceled!* :x:" + }; + + let final_blocks = vec![serde_json::json!({ + "type": "section", + "text": { + "type": "mrkdwn", + "text": message + } + })]; + + let payload = serde_json::json!({ + "channel": container.channel_id, + "ts": container.message_ts, + "text": message, + "blocks": final_blocks, + "mrkdwn": true // Enable markdown to support emojis + }); + + let client = Client::new(); + + let response = client + .post("https://slack.com/api/chat.update") + .bearer_auth(token) // Use the token for authentication + .header("Content-Type", "application/json") + .json(&payload) + .send() + .await + .map_err(|e| Error::from(anyhow::Error::new(e)))?; + + if response.status().is_success() { + tracing::debug!("Slack message updated successfully!"); + } else { + tracing::error!( + "Failed to update Slack message. Status: {}, Response: {:?}", + response.status(), + response + .text() + .await + .map_err(|e| Error::from(anyhow::Error::new(e)))? + ); + } + + Ok(()) +} diff --git a/python-client/build.sh b/python-client/build.sh index 90195376e2..6c9a723606 100755 --- a/python-client/build.sh +++ b/python-client/build.sh @@ -6,7 +6,13 @@ cp ../backend/windmill-api/openapi.yaml openapi/openapi.yaml npx @redocly/openapi-cli@latest bundle openapi/openapi.yaml > openapi-bundled.yaml -sed -z 's/FlowModuleValue:/FlowModuleValue2:/' openapi-bundled.yaml > openapi-decycled.yaml +if [[ "$OSTYPE" == "darwin"* ]]; then + # sed -z is not supported on macOS, use perl instead + perl -0777 -pe 's/FlowModuleValue:/FlowModuleValue2:/g' openapi-bundled.yaml > openapi-decycled.yaml +else + sed -z 's/FlowModuleValue:/FlowModuleValue2:/' openapi-bundled.yaml > openapi-decycled.yaml +fi + echo " FlowModuleValue: {}" >> openapi-decycled.yaml npx @redocly/openapi-cli@latest bundle openapi-decycled.yaml --ext json -d > openapi-deref.json @@ -20,9 +26,19 @@ rm -rf openapi/ rm openapi* cp LICENSE windmill-api/ -sed -i '5 i license = "Apache-2.0"' windmill-api/pyproject.toml -sed -i 's/authors = \[\]/authors = \["Ruben Fiszel "\]/g' windmill-api/pyproject.toml +# Check if running on macOS +if [[ "$OSTYPE" == "darwin"* ]]; then + # macOS version + sed -i '' '5 i\ +license = "Apache-2.0"' windmill-api/pyproject.toml + sed -i '' 's/authors = \[\]/\nauthors = \["Ruben Fiszel "\]/g' windmill-api/pyproject.toml +else + # Linux version + sed -i '5 i license = "Apache-2.0"' windmill-api/pyproject.toml + + sed -i 's/authors = \[\]/authors = \["Ruben Fiszel "\]/g' windmill-api/pyproject.toml +fi echo "# Autogenerated Windmill OpenApi Client" >> windmill-api/README.md.tmp echo "This is the raw autogenerated api client. You are most likely more interested \ @@ -33,8 +49,7 @@ user friendly experience. We use \ echo "" >> windmill-api/README.md.tmp - -head -n -13 windmill-api/README.md >> windmill-api/README.md.tmp +tail -r windmill-api/README.md | tail -n +14 | tail -r >> windmill-api/README.md.tmp mv windmill-api/README.md.tmp windmill-api/README.md cd windmill-api && poetry build diff --git a/python-client/dev.nu b/python-client/dev.nu index 52d68a7bec..11b92503de 100755 --- a/python-client/dev.nu +++ b/python-client/dev.nu @@ -1,6 +1,6 @@ #! /usr/bin/env nu -let cache = "/tmp/windmill/cache/pip/" +let cache = "/tmp/windmill/cache/python_311/" # Clean cache def "main clean" [] { @@ -42,7 +42,7 @@ def main [ rm -rf ($cache ++ wmill*/wmill/*) # Copy files from local ./dist to every wm-client version in cache - ls /tmp/windmill/cache/pip/wmill* | each { + ls /tmp/windmill/cache/python_311/wmill* | each { |i| let path = $i | get name; diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index c57e181a0f..a251635502 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -623,6 +623,46 @@ class Windmill: params={"approver": approver}, ).json() + def request_interactive_slack_approval( + self, + slack_resource_path: str, + channel_id: str, + message: str = None, + approver: str = None, + ) -> None: + """ + Request interactive Slack approval + :param slack_resource_path: Slack resource path + :param channel_id: Slack channel + :param message: Message to send to Slack + :param approver: Approver name + """ + workspace = self.workspace + flow_job_id = os.environ.get("WM_FLOW_JOB_ID") + + if not flow_job_id: + raise Exception( + "You can't use 'request_interactive_slack_approval' function in a standalone script or flow step preview. Please use it in a flow or a flow preview." + ) + + # Only include non-empty parameters + params = {} + if message: + params["message"] = message + if approver: + params["approver"] = approver + if slack_resource_path: + params["slack_resource_path"] = slack_resource_path + if channel_id: + params["channel_id"] = channel_id + if os.environ.get("WM_FLOW_STEP_ID"): + params["flow_step_id"] = os.environ.get("WM_FLOW_STEP_ID") + + self.get( + f"/w/{workspace}/jobs/slack_approval/{os.environ.get('WM_JOB_ID', 'NO_JOB_ID')}", + params=params, + ) + def username_to_email(self, username: str) -> str: """ Get email from workspace username @@ -972,6 +1012,19 @@ def get_state_path() -> str: def get_resume_urls(approver: str = None) -> dict: return _client.get_resume_urls(approver) +@init_global_client +def request_interactive_slack_approval( + slack_resource_path: str, + channel_id: str, + message: str = None, + approver: str = None, +) -> None: + return _client.request_interactive_slack_approval( + slack_resource_path=slack_resource_path, + channel_id=channel_id, + message=message, + approver=approver, + ) @init_global_client def cancel_running() -> dict: diff --git a/typescript-client/README_DEV.md b/typescript-client/README_DEV.md index 5e2481d978..671f7c915b 100644 --- a/typescript-client/README_DEV.md +++ b/typescript-client/README_DEV.md @@ -1,8 +1,7 @@ # Generate windmill-client bundle ```bash -./node_modules/.bin/esbuild src/index.ts --b -undle --outfile=windmill.js --format=esm +./node_modules/.bin/esbuild src/index.ts --bundle --outfile=windmill.js --format=esm --platform=node ``` # Generate d.ts bundle diff --git a/typescript-client/build.jsr.sh b/typescript-client/build.jsr.sh index b347febb3c..cdcd68f975 100755 --- a/typescript-client/build.jsr.sh +++ b/typescript-client/build.jsr.sh @@ -14,5 +14,5 @@ cp "${script_dirpath}/s3Types.ts" "${script_dirpath}/src/" echo "" >> "${script_dirpath}/src/index.ts" echo 'export type { S3Object, DenoS3LightClientSettings } from "./s3Types";' >> "${script_dirpath}/src/index.ts" echo "" >> "${script_dirpath}/src/index.ts" -echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, task, runScript, runScriptAsync, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail } from "./client";' >> "${script_dirpath}/src/index.ts" +echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, task, runScript, runScriptAsync, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval} from "./client";' >> "${script_dirpath}/src/index.ts" diff --git a/typescript-client/build.sh b/typescript-client/build.sh index 436f2a526b..6c80fb5677 100755 --- a/typescript-client/build.sh +++ b/typescript-client/build.sh @@ -22,10 +22,15 @@ const baseUrl = getEnv("BASE_INTERNAL_URL") ?? getEnv("BASE_URL") ?? "http://loc const baseUrlApi = (baseUrl ?? '') + "/api"; EOF -sed -i 's/WITH_CREDENTIALS: false/WITH_CREDENTIALS: true/g' src/core/OpenAPI.ts -sed -i 's/TOKEN: undefined/TOKEN: getEnv("WM_TOKEN")/g' src/core/OpenAPI.ts -sed -i "s/BASE: '\/api'/BASE: baseUrlApi/g" src/core/OpenAPI.ts - +if [[ "$OSTYPE" == "darwin"* ]]; then + sed -i '' 's/WITH_CREDENTIALS: false/WITH_CREDENTIALS: true/g' src/core/OpenAPI.ts + sed -i '' 's/TOKEN: undefined/TOKEN: getEnv("WM_TOKEN")/g' src/core/OpenAPI.ts + sed -i '' "s/BASE: '\/api'/BASE: baseUrlApi/g" src/core/OpenAPI.ts +else + sed -i 's/WITH_CREDENTIALS: false/WITH_CREDENTIALS: true/g' src/core/OpenAPI.ts + sed -i 's/TOKEN: undefined/TOKEN: getEnv("WM_TOKEN")/g' src/core/OpenAPI.ts + sed -i "s/BASE: '\/api'/BASE: baseUrlApi/g" src/core/OpenAPI.ts +fi @@ -34,4 +39,4 @@ cp "${script_dirpath}/s3Types.ts" "${script_dirpath}/src/" echo "" >> "${script_dirpath}/src/index.ts" echo 'export type { S3Object, DenoS3LightClientSettings } from "./s3Types";' >> "${script_dirpath}/src/index.ts" echo "" >> "${script_dirpath}/src/index.ts" -echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, task, runScript, runScriptAsync, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail } from "./client";' >> "${script_dirpath}/src/index.ts" +echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, task, runScript, runScriptAsync, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval } from "./client";' >> "${script_dirpath}/src/index.ts" diff --git a/typescript-client/client.js b/typescript-client/client.js index 5a586411da..01dddd9614 100644 --- a/typescript-client/client.js +++ b/typescript-client/client.js @@ -9,7 +9,7 @@ var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, ge }); }; Object.defineProperty(exports, "__esModule", { value: true }); -exports.usernameToEmail = exports.uint8ArrayToBase64 = exports.base64ToUint8Array = exports.getIdToken = exports.getResumeEndpoints = exports.getResumeUrls = exports.writeS3File = exports.loadS3FileStream = exports.loadS3File = exports.denoS3LightClientSettings = exports.databaseUrlFromResource = exports.setVariable = exports.getVariable = exports.getState = exports.getInternalState = exports.getFlowUserState = exports.setFlowUserState = exports.setState = exports.setInternalState = exports.setResource = exports.getStatePath = exports.resolveDefaultResource = exports.runScriptAsync = exports.task = exports.getResultMaybe = exports.getResult = exports.waitJob = exports.runScript = exports.getRootJobId = exports.getResource = exports.getWorkspace = exports.setClient = exports.SHARED_FOLDER = exports.WorkspaceService = exports.UserService = exports.SettingsService = exports.ScheduleService = exports.ScriptService = exports.VariableService = exports.ResourceService = exports.JobService = exports.GroupService = exports.GranularAclService = exports.FlowService = exports.AuditService = exports.AdminService = void 0; +exports.requestInteractiveSlackApproval = exports.usernameToEmail = exports.uint8ArrayToBase64 = exports.base64ToUint8Array = exports.getIdToken = exports.getResumeEndpoints = exports.getResumeUrls = exports.writeS3File = exports.loadS3FileStream = exports.loadS3File = exports.denoS3LightClientSettings = exports.databaseUrlFromResource = exports.setVariable = exports.getVariable = exports.getState = exports.getInternalState = exports.getFlowUserState = exports.setFlowUserState = exports.setState = exports.setInternalState = exports.setResource = exports.getStatePath = exports.resolveDefaultResource = exports.runScriptAsync = exports.task = exports.getResultMaybe = exports.getResult = exports.waitJob = exports.runScript = exports.getRootJobId = exports.getResource = exports.getWorkspace = exports.setClient = exports.SHARED_FOLDER = exports.WorkspaceService = exports.UserService = exports.SettingsService = exports.ScheduleService = exports.ScriptService = exports.VariableService = exports.ResourceService = exports.JobService = exports.GroupService = exports.GranularAclService = exports.FlowService = exports.AuditService = exports.AdminService = void 0; const index_1 = require("./index"); const index_2 = require("./index"); var index_3 = require("./index"); diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 1fa88fca1b..8f00e4f157 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -395,15 +395,15 @@ export async function setState(state: any): Promise { */ export async function setProgress(percent: number, jobId?: any): Promise { const workspace = getWorkspace(); - let flowId = getEnv("WM_FLOW_JOB_ID"); + let flowId = getEnv("WM_FLOW_JOB_ID"); // If jobId specified we need to find if there is a parent/flow if (jobId) { const job = await JobService.getJob({ id: jobId ?? "NO_JOB_ID", workspace, - noLogs: true - }); + noLogs: true, + }); // Could be actual flowId or undefined flowId = job.parent_job; @@ -415,22 +415,22 @@ export async function setProgress(percent: number, jobId?: any): Promise { requestBody: { // In case user inputs float, it should be converted to int percent: Math.floor(percent), - flow_job_id: (flowId == "") ? undefined : flowId, - } + flow_job_id: flowId == "" ? undefined : flowId, + }, }); } /** * Get the progress * @param jobId? Job to get progress from - * @returns Optional clamped between 0 and 100 progress value + * @returns Optional clamped between 0 and 100 progress value */ export async function getProgress(jobId?: any): Promise { // TODO: Delete or set to 100 completed job metrics return await MetricsService.getJobProgress({ id: jobId ?? getEnv("WM_JOB_ID") ?? "NO_JOB_ID", workspace: getWorkspace(), - }); + }); } /** @@ -846,3 +846,57 @@ export async function usernameToEmail(username: string): Promise { const workspace = getWorkspace(); return await UserService.usernameToEmail({ username, workspace }); } + +interface SlackApprovalOptions { + slackResourcePath: string; + channelId: string; + message?: string; + approver?: string; +} + +export async function requestInteractiveSlackApproval({ + slackResourcePath, + channelId, + message, + approver, +}: SlackApprovalOptions): Promise { + const workspace = getWorkspace(); + const flowJobId = getEnv("WM_FLOW_JOB_ID"); + + if (!flowJobId) { + throw new Error( + "You can't use this function in a standalone script or flow step preview. Please use it in a flow or a flow preview." + ); + } + + const flowStepId = getEnv("WM_FLOW_STEP_ID"); + if (!flowStepId) { + throw new Error("This function can only be called as a flow step"); + } + + // Only include non-empty parameters + const params: { + approver?: string; + message?: string; + slackResourcePath: string; + channelId: string; + flowStepId: string; + } = { + slackResourcePath, + channelId, + flowStepId, + }; + + if (message) { + params.message = message; + } + if (approver) { + params.approver = approver; + } + + await JobService.getSlackApprovalPayload({ + workspace, + ...params, + id: getEnv("WM_JOB_ID") ?? "NO_JOB_ID", + }); +}