From 05ba0a06d012e29dfaa0fcfb8c2ff206ea9f4b4f Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 25 Jul 2022 10:03:52 +0200 Subject: [PATCH] feat: hub flows integration --- backend/openapi.yaml | 128 +++++++-- backend/src/{flow.rs => flows.rs} | 54 +++- backend/src/jobs.rs | 18 +- backend/src/lib.rs | 5 +- backend/src/scripts.rs | 119 ++------ backend/src/utils.rs | 46 ++- backend/src/worker_flow.rs | 8 +- backend/src/workspaces.rs | 2 +- frontend/package.json | 2 +- frontend/src/lib/components/Editor.svelte | 32 ++- .../src/lib/components/FlowBuilder.svelte | 8 +- frontend/src/lib/components/FlowEditor.svelte | 265 +++++++++--------- frontend/src/lib/components/FlowViewer.svelte | 49 +++- frontend/src/lib/components/Modal.svelte | 82 +++--- frontend/src/lib/components/Path.svelte | 1 + .../src/lib/components/SchemaViewer.svelte | 2 +- frontend/src/lib/components/flows/utils.ts | 2 +- frontend/src/lib/utils.ts | 23 +- frontend/src/routes/flows.svelte | 102 ++++++- frontend/src/routes/flows/add.svelte | 21 +- .../src/routes/flows/get/[...path].svelte | 14 - frontend/src/routes/scripts.svelte | 18 +- frontend/src/routes/scripts/add.svelte | 3 +- frontend/src/routes/user/login@user.svelte | 6 +- 24 files changed, 648 insertions(+), 362 deletions(-) rename backend/src/{flow.rs => flows.rs} (90%) diff --git a/backend/openapi.yaml b/backend/openapi.yaml index c2419564e3..4ca91a7349 100644 --- a/backend/openapi.yaml +++ b/backend/openapi.yaml @@ -1509,26 +1509,99 @@ paths: content: application/json: schema: - type: array - items: - type: object - properties: - id: - type: number - summary: - type: string - app: - type: string - approved: - type: boolean - is_trigger: - type: boolean - required: - - id - - summary - - app - - approved - - is_trigger + type: object + properties: + asks: + type: array + items: + type: object + properties: + id: + type: number + ask_id: + type: number + summary: + type: string + app: + type: string + approved: + type: boolean + is_trigger: + type: boolean + votes: + type: number + views: + type: number + required: + - id + - ask_id + - summary + - app + - approved + - is_trigger + - views + - votes + + /flows/hub/list: + get: + summary: list all available hub flows + operationId: listHubFlows + tags: + - flow + responses: + "200": + description: hub flows list + content: + application/json: + schema: + type: object + properties: + flows: + type: array + items: + type: object + properties: + id: + type: number + flow_id: + type: number + summary: + type: string + apps: + type: array + items: + type: string + approved: + type: boolean + votes: + type: number + + required: + - id + - flow_id + - summary + - apps + - approved + - votes + + /flows/hub/get/{id}: + get: + summary: get hub flow by id + operationId: getHubFlowById + tags: + - flow + parameters: + - $ref: "#/components/parameters/PathId" + responses: + "200": + description: flow + content: + application/json: + schema: + type: object + properties: + flow: + $ref: "#/components/schemas/OpenFlow" /scripts/hub/get/{path}: get: @@ -3933,6 +4006,21 @@ components: - archived - extra_perms + OpenFlow: + type: object + properties: + summary: + type: string + description: + type: string + value: + $ref: "#/components/schemas/FlowValue" + schema: + type: object + required: + - summary + - value + FlowValue: type: object properties: diff --git a/backend/src/flow.rs b/backend/src/flows.rs similarity index 90% rename from backend/src/flow.rs rename to backend/src/flows.rs index e1571f0caf..8f2683f55a 100644 --- a/backend/src/flow.rs +++ b/backend/src/flows.rs @@ -7,10 +7,11 @@ use std::collections::HashMap; +use reqwest::Client; use sql_builder::prelude::*; use axum::{ - extract::{Extension, Path, Query}, + extract::{Extension, Host, Path, Query}, routing::{get, post}, Json, Router, }; @@ -21,11 +22,11 @@ use sqlx::{FromRow, Postgres, Transaction}; use crate::{ audit::{audit_log, ActionKind}, db::{UserDB, DB}, - error::{self, Error, JsonResult, Result}, + error::{self, to_anyhow, Error, JsonResult, Result}, jobs::RawCode, scripts::Schema, users::Authed, - utils::{Pagination, StripPath}, + utils::{http_get_from_hub, list_elems_from_hub, Pagination, StripPath}, }; pub fn workspaced_service() -> Router { @@ -38,6 +39,12 @@ pub fn workspaced_service() -> Router { .route("/exists/*path", get(exists_flow_by_path)) } +pub fn global_service() -> Router { + Router::new() + .route("/hub/list", get(list_hub_flows)) + .route("/hub/get/:id", get(get_hub_flow_by_id)) +} + #[derive(FromRow, Serialize)] pub struct Flow { pub workspace_id: String, @@ -162,6 +169,47 @@ async fn list_flows( Ok(Json(rows)) } +async fn list_hub_flows( + Authed { + email, username, .. + }: Authed, + Extension(http_client): Extension, + Host(host): Host, +) -> JsonResult { + let flows = list_elems_from_hub( + http_client, + "https://hub.windmill.dev/searchFlowData?approved=true", + email, + username, + host, + ) + .await?; + Ok(Json(flows)) +} + +pub async fn get_hub_flow_by_id( + Authed { + email, username, .. + }: Authed, + Path(id): Path, + Extension(http_client): Extension, + Host(host): Host, +) -> JsonResult { + let value = http_get_from_hub( + http_client, + &format!("https://hub.windmill.dev/flows/{id}/json"), + email, + username, + host, + false, + ) + .await? + .json() + .await + .map_err(to_anyhow)?; + Ok(Json(value)) +} + async fn create_flow( authed: Authed, Extension(user_db): Extension, diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index 69cd19edad..d4b08db748 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -5,6 +5,7 @@ * LICENSE-AGPL for a copy of the license. */ +use axum::extract::Host; use chrono::Duration; use sql_builder::prelude::*; @@ -12,6 +13,7 @@ use sqlx::{query_scalar, Postgres, Transaction}; use std::collections::HashMap; use tracing::instrument; +use crate::error::to_anyhow; use crate::scripts::{get_hub_script_by_path, ScriptLang}; use crate::worker_flow::init_flow_status; use crate::{ @@ -19,7 +21,7 @@ use crate::{ db::{UserDB, DB}, error, error::Error, - flow::FlowValue, + flows::FlowValue, schedule::get_schedule_opt, scripts::ScriptHash, users::{owner_to_token_owner, Authed}, @@ -1127,6 +1129,13 @@ pub async fn push<'c>( groups: vec![], }, Path(StripPath(path)), + Extension( + reqwest::ClientBuilder::new() + .user_agent("windmill/beta") + .build() + .map_err(to_anyhow)?, + ), + Host(std::env::var("BASE_URL").unwrap_or_else(|_| "".to_string())), ) .await?, ), @@ -1337,7 +1346,12 @@ pub async fn schedule_again_if_scheduled( let mut tx = db.begin().await?; let schedule = get_schedule_opt(&mut tx, &w_id, &schedule_path) .await? - .unwrap(); + .ok_or_else(|| { + Error::InternalErr(format!( + "Could not find schedule {:?} for workspace {}", + schedule_path, w_id + )) + })?; if schedule.enabled && script_path.is_some() && script_path.unwrap() == schedule.script_path { tx = crate::schedule::push_scheduled_job(tx, schedule).await?; diff --git a/backend/src/lib.rs b/backend/src/lib.rs index 0ed7cd57ec..cb5275f4a9 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -25,7 +25,7 @@ mod client; mod db; mod email; mod error; -mod flow; +mod flows; mod granular_acls; mod groups; mod jobs; @@ -146,7 +146,7 @@ pub async fn run_server( .nest("/audit", audit::workspaced_service()) .nest("/acls", granular_acls::workspaced_service()) .nest("/workspaces", workspaces::workspaced_service()) - .nest("/flows", flow::workspaced_service()), + .nest("/flows", flows::workspaced_service()), ) .nest("/workspaces", workspaces::global_service()) .nest( @@ -155,6 +155,7 @@ pub async fn run_server( ) .nest("/workers", worker_ping::global_service()) .nest("/scripts", scripts::global_service()) + .nest("/flows", flows::global_service()) .nest("/schedules", schedule::global_service()) .route_layer(from_extractor::()) .route_layer(from_extractor::()) diff --git a/backend/src/scripts.rs b/backend/src/scripts.rs index 6c4d51afab..f6bbf27b2e 100644 --- a/backend/src/scripts.rs +++ b/backend/src/scripts.rs @@ -5,6 +5,7 @@ * LICENSE-AGPL for a copy of the license. */ +use reqwest::Client; use serde::Deserializer; use sql_builder::prelude::*; @@ -14,7 +15,7 @@ use crate::{ error::{to_anyhow, Error, JsonResult, Result}, jobs, parser, users::{owner_to_token_owner, truncate_token, Authed, Tokened}, - utils::{require_admin, Pagination, StripPath}, + utils::{http_get_from_hub, list_elems_from_hub, require_admin, Pagination, StripPath}, }; use axum::{ extract::{Extension, Host, Path, Query}, @@ -251,88 +252,22 @@ async fn list_scripts( Ok(Json(rows)) } -#[derive(Deserialize, Serialize)] -struct SearchScriptData { - asks: Vec, -} - -#[derive(Deserialize, Serialize)] -struct ScriptSearch { - id: i32, - ask_id: i32, - summary: String, - app: String, - approved: bool, - is_trigger: bool, - views: i32, - votes: i32, -} - async fn list_hub_scripts( Authed { email, username, .. }: Authed, + Extension(http_client): Extension, Host(host): Host, -) -> JsonResult> { - let http_client = reqwest::ClientBuilder::new() - .user_agent("windmill/beta") - .build() - .map_err(to_anyhow)?; - let rows = http_client - .get("https://hub.windmill.dev/searchData?approved=true") - .header("X-email", email.unwrap_or_else(|| "".to_string())) - .header("X-username", username) - .header("X-hostname", host) - .send() - .await - .map_err(to_anyhow)? - .json::() - .await - .map_err(to_anyhow)? - .asks; - Ok(Json(rows)) -} - -#[derive(Deserialize, Serialize)] -struct SearchFlowData { - asks: Vec, -} - -#[derive(Deserialize, Serialize)] -struct FlowSearch { - id: i32, - ask_id: i32, - summary: String, - app: String, - approved: bool, - is_trigger: bool, - views: i32, - votes: i32, -} - -async fn list_hub_flows( - Authed { - email, username, .. - }: Authed, - Host(host): Host, -) -> JsonResult> { - let http_client = reqwest::ClientBuilder::new() - .user_agent("windmill/beta") - .build() - .map_err(to_anyhow)?; - let rows = http_client - .get("https://hub.windmill.dev/searchFlowData?approved=true") - .header("X-email", email.unwrap_or_else(|| "".to_string())) - .header("X-username", username) - .header("X-hostname", host) - .send() - .await - .map_err(to_anyhow)? - .json::() - .await - .map_err(to_anyhow)? - .asks; - Ok(Json(rows)) +) -> JsonResult { + let asks = list_elems_from_hub( + http_client, + "https://hub.windmill.dev/searchData?approved=true", + email, + username, + host, + ) + .await?; + Ok(Json(asks)) } fn hash_script(ns: &NewScript) -> i64 { @@ -548,26 +483,26 @@ pub async fn get_hub_script_by_path( email, username, .. }: Authed, Path(path): Path, + Extension(http_client): Extension, + Host(host): Host, ) -> Result { let path = path .to_path() .strip_prefix("hub/") .ok_or_else(|| Error::BadRequest("Impossible to remove prefix hex".to_string()))?; - let http_client = reqwest::ClientBuilder::new() - .user_agent("windmill/beta") - .build() - .map_err(to_anyhow)?; - let content = http_client - .get(format!("https://hub.windmill.dev/raw/{path}.ts")) - .header("X-email", email.unwrap_or_else(|| "".to_string())) - .header("X-username", username) - .send() - .await - .map_err(to_anyhow)? - .text() - .await - .map_err(to_anyhow)?; + let content = http_get_from_hub( + http_client, + &format!("https://hub.windmill.dev/raw/{path}.ts"), + email, + username, + host, + true, + ) + .await? + .text() + .await + .map_err(to_anyhow)?; Ok(content) } diff --git a/backend/src/utils.rs b/backend/src/utils.rs index d3b6d15f4c..64d059ac9e 100644 --- a/backend/src/utils.rs +++ b/backend/src/utils.rs @@ -6,10 +6,11 @@ */ use rand::{distributions::Alphanumeric, thread_rng, Rng}; +use reqwest::Response; use serde::Deserialize; use sqlx::{Postgres, Transaction}; -use crate::error::{Error, Result}; +use crate::error::{to_anyhow, Error, Result}; pub const MAX_PER_PAGE: usize = 1000; pub const DEFAULT_PER_PAGE: usize = 100; @@ -95,3 +96,46 @@ pub fn not_found_if_none>(opt: Option, kind: &str, name: U) pub fn get_owner_from_path(path: &str) -> String { path.split('/').take(2).collect::>().join("/") } + +pub async fn list_elems_from_hub( + http_client: reqwest::Client, + url: &str, + email: Option, + username: String, + host: String, +) -> Result { + let rows = http_get_from_hub(http_client, url, email, username, host, false) + .await? + .json::() + .await + .map_err(to_anyhow)?; + Ok(rows) +} + +pub async fn http_get_from_hub( + http_client: reqwest::Client, + url: &str, + email: Option, + username: String, + host: String, + plain: bool, +) -> Result { + let response = http_client + .get(url) + .header( + "Accept", + if plain { + "text/plain" + } else { + "application/json" + }, + ) + .header("X-email", email.unwrap_or_else(|| "".to_string())) + .header("X-username", username) + .header("X-hostname", host) + .send() + .await + .map_err(to_anyhow)?; + + Ok(response) +} diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs index 3ce1e3bfbc..7d449e3243 100644 --- a/backend/src/worker_flow.rs +++ b/backend/src/worker_flow.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; -use crate::flow::{FlowModuleValue, FlowValue, InputTransform}; +use crate::flows::{FlowModuleValue, FlowValue, InputTransform}; use crate::jobs::{ add_completed_job, add_completed_job_error, get_queued_job, postprocess_queued_job, push, script_path_to_payload, JobPayload, @@ -394,9 +394,9 @@ async fn push_next_flow_job( let module = &flow.modules[i]; let mut tx = db.begin().await?; let job_payload = match &module.value { - FlowModuleValue::Script { - path: script_path - } => script_path_to_payload(script_path, &mut tx, &flow_job.workspace_id).await?, + FlowModuleValue::Script { path: script_path } => { + script_path_to_payload(script_path, &mut tx, &flow_job.workspace_id).await? + } FlowModuleValue::RawScript(raw_code) => { let mut raw_code = raw_code.clone(); if raw_code.path.is_none() { diff --git a/backend/src/workspaces.rs b/backend/src/workspaces.rs index d4afb1db87..5f39a07fe6 100644 --- a/backend/src/workspaces.rs +++ b/backend/src/workspaces.rs @@ -8,7 +8,7 @@ use crate::{ db::{UserDB, DB}, error::{Error, JsonResult, Result}, - users::{Authed, WorkspaceInvite}, utils::{require_admin, require_super_admin, Pagination}, audit::{audit_log, ActionKind}, scripts::{Script, Schema}, resources::{Resource, ResourceType}, flow::Flow, variables::ListableVariable, + users::{Authed, WorkspaceInvite}, utils::{require_admin, require_super_admin, Pagination}, audit::{audit_log, ActionKind}, scripts::{Script, Schema}, resources::{Resource, ResourceType}, flows::Flow, variables::ListableVariable, }; use axum::{extract::{Extension, Path, Query}, routing::{get, post, delete}, Json, Router, response::{IntoResponse}, body::StreamBody}; diff --git a/frontend/package.json b/frontend/package.json index 355dd332ca..f90de39b99 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "windmill", - "version": "1.22.0", + "version": "1.22.42", "scripts": { "dev": "vite dev", "build": "vite build", diff --git a/frontend/src/lib/components/Editor.svelte b/frontend/src/lib/components/Editor.svelte index 9d74db801f..08bed3112b 100644 --- a/frontend/src/lib/components/Editor.svelte +++ b/frontend/src/lib/components/Editor.svelte @@ -25,6 +25,8 @@ let websockets: WebSocket[] = [] let websocketInterval: NodeJS.Timer | undefined + let lastWsAttempt: Date | undefined + let nbWsAttempt = 0 let uri: string = '' let disposeMethod: () => void | undefined const dispatch = createEventDispatcher() @@ -154,6 +156,8 @@ const writer = new WebSocketMessageWriter(socket) const languageClient = createLanguageClient({ reader, writer }, name, options) languageClient.start() + lastWsAttempt = undefined + nbWsAttempt = 0 reader.onClose(() => { try { languageClient.stop() @@ -239,14 +243,27 @@ } }) } + websocketInterval && clearInterval(websocketInterval) websocketInterval = setInterval(() => { if (document.visibilityState == 'visible') { - if (!websocketAlive.black && !websocketAlive.deno && !websocketAlive.pyright) { - sendUserToast( - 'Smart assistant got disconnected. Reconnecting to windmill language server for smart assistance' - ) - reloadWebsocket() + if ( + !lastWsAttempt || + (lastWsAttempt.getTime() - new Date().getTime() > 60000 && nbWsAttempt < 2) + ) { + if (!websocketAlive.black && !websocketAlive.deno && !websocketAlive.pyright) { + sendUserToast( + 'Smart assistant got disconnected. Reconnecting to windmill language server for smart assistance' + ) + lastWsAttempt = new Date() + nbWsAttempt++ + reloadWebsocket() + } else { + if (nbWsAttempt >= 2) { + sendUserToast('Giving up on establishing smart assistant connection', true) + clearInterval(websocketInterval) + } + } } } }, 5000) @@ -380,9 +397,8 @@ }) onDestroy(() => { - if (disposeMethod) { - disposeMethod() - } + disposeMethod && disposeMethod() + websocketInterval && clearInterval(websocketInterval) }) diff --git a/frontend/src/lib/components/FlowBuilder.svelte b/frontend/src/lib/components/FlowBuilder.svelte index 87fe7c8966..fe4e201a24 100644 --- a/frontend/src/lib/components/FlowBuilder.svelte +++ b/frontend/src/lib/components/FlowBuilder.svelte @@ -1,12 +1,10 @@ @@ -39,14 +41,40 @@ {/if} {#if tab == 'ui'}
-

+

+ Inputs +

+ {#if flow.schema && flow.schema.properties && Object.keys(flow.schema.properties).length > 0 && flow.schema} +
    + {#each Object.entries(flow.schema.properties) as [inp, v]} +
  • + {v.default != undefined ? 'default: ' + JSON.stringify(v.default) : ''} +
    • + {/each} +
+ {:else} +
+ This script has no argument or is ill-defined +
+ {/if} +

{flow?.value?.modules?.length} Steps

@@ -107,6 +135,17 @@ {/each}
-{:else} - +{:else if tab == 'json'} +
+ + +
+{:else if tab == 'schema'} + {/if} diff --git a/frontend/src/lib/components/Modal.svelte b/frontend/src/lib/components/Modal.svelte index 10c040d73c..453f5b1235 100644 --- a/frontend/src/lib/components/Modal.svelte +++ b/frontend/src/lib/components/Modal.svelte @@ -32,45 +32,18 @@ -
+{#if open} +
-
-
- {#if open} -
- -

-
-
-
- -
-
-
+
+
+ {#if open} +
-   +

+
-
- {/if} +
+ +
+
+
+ +   +
+
+ {/if} +
-
+{/if}