starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-05 16:03:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(backend): implement new OpenFlow module Branches (#692)

Browse Source 
* branches

* iterate

* branches

* progress

* progress

* progress

* done

* adapt frontend

* adapt frontend

* sqlx
This commit is contained in:
Ruben Fiszel
2022-10-09 23:12:05 +02:00
committed by GitHub
parent 430e22a4b4
commit b87b03a673
18 changed files with 562 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/sqlx-data.json
+14
View File
@@ -2651,6 +2651,20 @@
},
"query": "DELETE FROM resource WHERE path = $1 AND workspace_id = $2"
},
"c110717810a30b3b5b1b9179401f82ae30d96105335dcd2bc7b20e45019d2325": {
"describe": {
"columns": [],
"nullable": [],
"parameters": {
"Left": [
"Jsonb",
"Jsonb",
"Uuid"
]
}
},
"query": "\n UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules'], $1),\n raw_flow = JSONB_SET(raw_flow, ARRAY['modules'], $2)\n WHERE id = $3\n "
},
"c213427a32e8903fff649a3e7aa8392d2215665ed04f5a8f2178861e9dba298a": {
"describe": {
"columns": [
backend/src/flows.rs
+13 -4
View File
@@ -178,6 +178,14 @@ pub enum InputTransform {
Javascript { expr: String },
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct BranchModules {
#[serde(skip_serializing_if = "Option::is_none")]
pub summary: Option<String>,
pub expr: String,
pub modules: Vec<FlowModule>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(
tag = "type",
@@ -193,8 +201,9 @@ pub enum FlowModuleValue {
#[serde(default = "default_true")]
skip_failures: bool,
},
Flow {
path: String,
Branches {
branches: Vec<BranchModules>,
default: Vec<FlowModule>,
},
RawScript(RawCode),
}
@@ -545,7 +554,7 @@ mod tests {
],
failure_module: Some(FlowModule {
input_transforms: HashMap::new(),
value: FlowModuleValue::Flow { path: "test".to_string() },
value: FlowModuleValue::Script { path: "test".to_string() },
stop_after_if: Some(StopAfterIf {
expr: "previous.isEmpty()".to_string(),
skip_if_stopped: false,
@@ -621,7 +630,7 @@ mod tests {
"failure_module": {
"input_transforms": {},
"value": {
"type": "flow",
"type": "script",
"path": "test"
},
"stop_after_if": {
backend/src/parser_ts.rs
+1 -1
View File
@@ -119,7 +119,7 @@ pub fn parse_deno_signature(code: &str) -> error::Result<MainArgSignature> {
})
} else {
Err(error::Error::ExecutionErr(
"main function was not findable (expected to find 'export main function(...)'"
"main function was not findable (expected to find 'export function main(...)'"
.to_string(),
))
}
backend/src/worker.rs
+101
View File
@@ -2440,6 +2440,107 @@ def main():
assert_eq!(result, serde_json::json!(9));
}
fn module_add_item_to_list(i: i32) -> serde_json::Value {
json!({
"input_transform": {
"array": {
"type": "javascript",
"expr": "previous_result",
},
"i": {
"type": "static",
"value": json!(i),
}
},
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(array, i){ array.push(i); return array}",
}
})
}
#[sqlx::test(fixtures("base"))]
async fn test_branches_simple(db: DB) {
initialize_tracing().await;
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [1] }",
}
},
{
"value": {
"branches": [],
"default": [module_add_item_to_list(2)],
"type": "branches",
}
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow).await;
assert_eq!(result, serde_json::json!([1, 2]));
}
#[sqlx::test(fixtures("base"))]
async fn test_branches_nested(db: DB) {
initialize_tracing().await;
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [] }",
}
},
module_add_item_to_list(1),
{
"value": {
"branches": [
{
"expr": "false",
"modules": []
},
{
"expr": "true",
"modules": [ {
"value": {
"branches": [
{
"expr": "false",
"modules": []
}],
"default": [module_add_item_to_list(2)],
"type": "branches",
}
}]
},
],
"default": [module_add_item_to_list(-4)],
"type": "branches",
}
},
module_add_item_to_list(3),
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow).await;
assert_eq!(result, serde_json::json!([1, 2, 3]));
}
#[sqlx::test(fixtures("base"))]
async fn test_failure_module(db: DB) {
initialize_tracing().await;
backend/src/worker_flow.rs
+385 -189
View File
@@ -53,15 +53,46 @@ pub struct Iterator {
pub args: Map<String, Value>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub enum BranchChosen {
Default,
Branch(usize),
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(tag = "type")]
pub enum FlowStatusModule {
WaitingForPriorSteps,
WaitingForEvents { count: u16, job: Uuid },
WaitingForExecutor { job: Uuid },
InProgress { job: Uuid, iterator: Option<Iterator>, forloop_jobs: Option<Vec<Uuid>> },
Success { job: Uuid, forloop_jobs: Option<Vec<Uuid>> },
Failure { job: Uuid, forloop_jobs: Option<Vec<Uuid>> },
WaitingForEvents {
count: u16,
job: Uuid,
},
WaitingForExecutor {
job: Uuid,
},
InProgress {
job: Uuid,
#[serde(skip_serializing_if = "Option::is_none")]
iterator: Option<Iterator>,
#[serde(skip_serializing_if = "Option::is_none")]
forloop_jobs: Option<Vec<Uuid>>,
#[serde(skip_serializing_if = "Option::is_none")]
branch_chosen: Option<BranchChosen>,
},
Success {
job: Uuid,
#[serde(skip_serializing_if = "Option::is_none")]
forloop_jobs: Option<Vec<Uuid>>,
#[serde(skip_serializing_if = "Option::is_none")]
branch_chosen: Option<BranchChosen>,
},
Failure {
job: Uuid,
#[serde(skip_serializing_if = "Option::is_none")]
forloop_jobs: Option<Vec<Uuid>>,
#[serde(skip_serializing_if = "Option::is_none")]
branch_chosen: Option<BranchChosen>,
},
}
impl FlowStatus {
@@ -94,7 +125,7 @@ pub async fn update_flow_status_after_job_completion(
worker_dir: &str,
keep_job_dir: bool,
) -> error::Result<()> {
tracing::debug!("HANDLE FLOW: {job:?} {success} {result:?}");
tracing::debug!("UPDATE FLOW STATUS: {job:?} {success} {result:?}");
let mut tx = db.begin().await?;
@@ -147,12 +178,12 @@ pub async fn update_flow_status_after_job_completion(
if success || (forloop_jobs.is_some() && skip_loop_failures) {
(
old_status.step + 1,
FlowStatusModule::Success { job: job.id, forloop_jobs },
FlowStatusModule::Success { job: job.id, forloop_jobs, branch_chosen: None },
)
} else {
(
old_status.step,
FlowStatusModule::Failure { job: job.id, forloop_jobs },
FlowStatusModule::Failure { job: job.id, forloop_jobs, branch_chosen: None },
)
}
}
@@ -163,12 +194,6 @@ pub async fn update_flow_status_after_job_completion(
.map(|i| !(..old_status.modules.len()).contains(&i))
.unwrap_or(true);
tracing::debug!(
"old status: {:#?}\n{:#?}\n{is_last_step}",
old_status,
new_status
);
let (stop_early_expr, skip_if_stop_early) =
sqlx::query_as::<_, (Option<String>, Option<bool>)>(
"
@@ -190,11 +215,9 @@ pub async fn update_flow_status_after_job_completion(
.await
.map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e}")))?;
tracing::debug!("UPDATE: {:?}", new_status);
let stop_early = success
&& if let Some(expr) = stop_early_expr.clone() {
compute_stop_early(expr, result.clone()).await?
compute_bool_from_expr(expr, result.clone()).await?
} else {
false
};
@@ -378,8 +401,19 @@ fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u16, Duration)> {
.map(|d| (status.fail_count + 1, std::cmp::min(d, MAX_RETRY_INTERVAL)))
}
async fn compute_stop_early(expr: String, result: serde_json::Value) -> error::Result<bool> {
match eval_timeout(expr, [("result".to_string(), result)].into(), None, vec![]).await? {
async fn compute_bool_from_expr(expr: String, result: serde_json::Value) -> error::Result<bool> {
match eval_timeout(
expr,
[
("result".to_string(), result.clone()),
("previous_result".to_string(), result),
]
.into(),
None,
vec![],
)
.await?
{
serde_json::Value::Bool(true) => Ok(true),
serde_json::Value::Bool(false) => Ok(false),
a @ _ => Err(Error::ExecutionErr(format!(
@@ -497,7 +531,12 @@ pub async fn handle_flow(
.to_owned();
let flow = serde_json::from_value::<FlowValue>(value.to_owned())?;
push_next_flow_job(flow_job, flow, db, last_result, same_worker_tx).await?;
let status: FlowStatus =
serde_json::from_value::<FlowStatus>(flow_job.flow_status.clone().unwrap_or_default())
.with_context(|| format!("parse flow status {}", flow_job.id))?;
tracing::debug!("handle_flow: {:#?} {:#?}", status, flow);
push_next_flow_job(flow_job, status, flow, db, last_result, same_worker_tx).await?;
Ok(())
}
@@ -505,17 +544,13 @@ pub async fn handle_flow(
#[instrument(level = "trace", skip_all)]
async fn push_next_flow_job(
flow_job: &QueuedJob,
flow: FlowValue,
mut status: FlowStatus,
mut flow: FlowValue,
db: &sqlx::Pool<sqlx::Postgres>,
mut last_result: serde_json::Value,
same_worker_tx: Sender<Uuid>,
) -> anyhow::Result<()> {
let status: FlowStatus =
serde_json::from_value::<FlowStatus>(flow_job.flow_status.clone().unwrap_or_default())
.with_context(|| format!("parse flow status {}", flow_job.id))?;
/* `mut` because reassigned on FlowStatusModule::Failure when failure_module is Some */
let mut i = usize::try_from(status.step)
.with_context(|| format!("invalid module index {}", status.step))?;
@@ -568,7 +603,7 @@ async fn push_next_flow_job(
.unwrap_or_else(|| status.failure_module.clone());
tracing::debug!(
"PUSH: module: {:#?}, status: {:#?}",
"push_next_flow_job: module: {:#?}, status: {:#?}",
module.value,
status_module
);
@@ -801,168 +836,88 @@ async fn push_next_flow_job(
Map::new()
};
/* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */
let next_flow_transform = compute_next_flow_transform(
flow_job,
&flow,
transform_context,
&db,
&module,
&status,
&status_module,
last_result.clone(),
&mut args,
)
.await?;
let next_loop_status: Option<NextLoopStatus> = if let FlowModuleValue::ForloopFlow {
iterator,
..
} = &module.value
{
match status_module {
FlowStatusModule::WaitingForPriorSteps => {
let (token, steps) = if let Some(x) = transform_context {
x
} else {
get_transform_context(&db, &flow_job, &status).await?
};
/* Iterator is an InputTransform, evaluate it into an array. */
let itered = iterator
.clone()
.evaluate_with(
|| {
vec![
("result".to_string(), last_result.clone()),
("previous_result".to_string(), last_result.clone()),
]
},
token,
flow_job.workspace_id.clone(),
steps,
)
.await?
.into_array()
.map_err(|not_array| {
Error::ExecutionErr(format!("Expected an array value, found: {not_array}"))
})?;
let (job_payload, next_status) = match next_flow_transform {
NextFlowTransform::Continue(job_payload, next_state) => (job_payload, next_state),
NextFlowTransform::BranchChosen(branch, modules) => {
let added_flow_statuses = vec![FlowStatusModule::WaitingForPriorSteps; modules.len()];
let first = if let Some(first) = itered.first() {
first
} else {
/* Nothing to iterate, complete immediately and bail. */
let next_step = i
.checked_add(1)
.filter(|i| (..flow.modules.len()).contains(i));
let index = (status.step + 1) as usize;
status.modules.splice(index..index, added_flow_statuses);
flow.modules.splice(index..index, modules);
let new_job = sqlx::query_as::<_, QueuedJob>(
r#"
UPDATE queue
SET flow_status = JSONB_SET(
JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2),
ARRAY['step'], $3)
WHERE id = $4
RETURNING *
"#,
)
.bind(status.step)
.bind(json!(FlowStatusModule::Success {
job: flow_job.id,
forloop_jobs: Some(vec![])
}))
.bind(json!(next_step.unwrap_or(i)))
.bind(flow_job.id)
.fetch_one(db)
.await?;
let mut tx = db.begin().await?;
sqlx::query!(
"
UPDATE queue
SET flow_status = JSONB_SET(flow_status, ARRAY['modules'], $1),
raw_flow = JSONB_SET(raw_flow, ARRAY['modules'], $2)
WHERE id = $3
",
json!(status.modules),
json!(flow.modules),
flow_job.id
)
.execute(&mut tx)
.await?;
return if next_step.is_some() {
push_next_flow_job(&new_job, flow, db, json!([]), same_worker_tx).await
} else {
let success = true;
let skipped = false;
let logs = "Forloop completed without iteration".to_string();
let _uuid =
add_completed_job(db, &new_job, success, skipped, json!([]), logs)
.await?;
postprocess_queued_job(
false,
new_job.schedule_path,
new_job.script_path,
&new_job.workspace_id,
new_job.id,
db,
)
.await?;
Ok(())
};
};
args.insert("iter".to_string(), json!({ "index": 0, "value": first }));
Some(NextLoopStatus { index: 0, itered, forloop_jobs: vec![] })
}
FlowStatusModule::InProgress {
iterator: Some(Iterator { itered, index, args: iterator_args }),
forloop_jobs: Some(forloop_jobs),
..
} => {
let (index, next) = index
.checked_add(1)
.and_then(|i| itered.get(i).map(|next| (i, next)))
/* we shouldn't get here because update_flow_status_after_job_completion
* should leave this state if there iteration is complete, but also it should
* be reasonable to just enter a completed state instead of failing, similar to
* iterating an empty list above */
.with_context(|| format!("could not iterate index {index} of {itered:?}"))?;
args.extend(iterator_args);
args.insert("iter".to_string(), json!({ "index": index, "value": next }));
Some(NextLoopStatus { index, itered, forloop_jobs })
}
_ => Err(Error::BadRequest(format!(
"Unrecognized module status for ForloopFlow {status_module:?}"
)))?,
return jump_to_next_step(
status.step,
i,
&flow_job.id,
flow,
tx,
&db,
FlowStatusModule::Success {
job: flow_job.id,
forloop_jobs: None,
branch_chosen: Some(branch),
},
last_result,
same_worker_tx,
)
.await;
}
} else {
None
};
NextFlowTransform::EmptyIterator => {
let tx = db.begin().await?;
if matches!(&module.value, FlowModuleValue::ForloopFlow { .. }) {
if let Some(value) = &flow_job.args {
value
.as_object()
.ok_or_else(|| {
Error::BadRequest(format!("Expected an object value, found: {value:?}"))
})
.map(|map| args.extend(map.clone()))?;
}
}
/* Finally, push the job into the queue */
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::RawScript(raw_code) => {
let mut raw_code = raw_code.clone();
if raw_code.path.is_none() {
raw_code.path = Some(format!("{}/{}", flow_job.script_path(), status.step));
}
JobPayload::Code(raw_code)
}
FlowModuleValue::ForloopFlow { modules, .. } => JobPayload::RawFlow {
value: FlowValue {
modules: (*modules).clone(),
failure_module: flow.failure_module.clone(),
same_worker: flow.same_worker,
},
path: Some(format!("{}/{}", flow_job.script_path(), status.step)),
},
a @ FlowModuleValue::Flow { .. } => {
tracing::info!("Unrecognized module values {:?}", a);
Err(Error::BadRequest(format!(
"Unrecognized module values {:?}",
a
)))?
return jump_to_next_step(
status.step,
i,
&flow_job.id,
flow.clone(),
tx,
&db,
FlowStatusModule::Success {
job: flow_job.id,
forloop_jobs: Some(vec![]),
branch_chosen: None,
},
json!([]),
same_worker_tx,
)
.await;
}
};
let continue_on_same_worker =
flow.same_worker && !matches!(job_payload, JobPayload::RawFlow { .. });
/* Finally, push the job into the queue */
let tx = db.begin().await?;
let (uuid, mut tx) = push(
tx,
&flow_job.workspace_id,
@@ -978,18 +933,19 @@ async fn push_next_flow_job(
)
.await?;
let new_status =
if let Some(NextLoopStatus { index, itered, mut forloop_jobs }) = next_loop_status {
let new_status = match next_status {
NextStatus::NextLoopIteration(NextIteration { index, itered, mut forloop_jobs }) => {
forloop_jobs.push(uuid);
FlowStatusModule::InProgress {
job: uuid,
iterator: Some(Iterator { index, itered, args }),
forloop_jobs: Some(forloop_jobs),
branch_chosen: None,
}
} else {
FlowStatusModule::WaitingForExecutor { job: uuid }
};
}
_ => FlowStatusModule::WaitingForExecutor { job: uuid },
};
sqlx::query(
"
@@ -1013,13 +969,253 @@ async fn push_next_flow_job(
same_worker_tx.send(uuid).await?;
}
return Ok(());
}
/// Some state about the current/last forloop FlowStatusModule used to initialized the next
/// iteration's FlowStatusModule after pushing a job
struct NextLoopStatus {
index: usize,
itered: Vec<Value>,
forloop_jobs: Vec<Uuid>,
async fn jump_to_next_step<'c>(
status_step: i32,
i: usize,
job_id: &Uuid,
flow: FlowValue,
mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
db: &DB,
status_module: FlowStatusModule,
last_result: serde_json::Value,
same_worker_tx: Sender<Uuid>,
) -> anyhow::Result<()> {
let next_step = i
.checked_add(1)
.filter(|i| (..flow.modules.len()).contains(i));
let new_job = sqlx::query_as::<_, QueuedJob>(
r#"
UPDATE queue
SET flow_status = JSONB_SET(
JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2),
ARRAY['step'], $3)
WHERE id = $4
RETURNING *
"#,
)
.bind(status_step)
.bind(json!(status_module))
.bind(json!(next_step.unwrap_or(i)))
.bind(job_id)
.fetch_one(&mut tx)
.await?;
tx.commit().await?;
let new_status = new_job.parse_flow_status().ok_or_else(|| {
Error::ExecutionErr("Impossible to parse new status after jump".to_string())
})?;
if next_step.is_some() {
tracing::debug!("Jumping to next step with flow {flow:#?}");
return push_next_flow_job(&new_job, new_status, flow, db, last_result, same_worker_tx)
.await;
} else {
let success = true;
let skipped = false;
let logs = "Forloop completed without iteration".to_string();
let _uuid = add_completed_job(db, &new_job, success, skipped, json!([]), logs).await?;
postprocess_queued_job(
false,
new_job.schedule_path,
new_job.script_path,
&new_job.workspace_id,
new_job.id,
db,
)
.await?;
return Ok(());
}
}
/// Some state about the current/last forloop FlowStatusModule used to initialized the next
/// iteration's FlowStatusModule after pushing a job
struct NextIteration {
index: usize,
itered: Vec<Value>,
forloop_jobs: Vec<Uuid>,
}
enum LoopStatus {
NextIteration(NextIteration),
EmptyIterator,
}
enum NextStatus {
NextStep,
NextLoopIteration(NextIteration),
}
enum NextFlowTransform {
EmptyIterator,
Continue(JobPayload, NextStatus),
BranchChosen(BranchChosen, Vec<FlowModule>),
}
async fn compute_next_flow_transform(
flow_job: &QueuedJob,
flow: &FlowValue,
transform_context: Option<(String, Vec<String>)>,
db: &DB,
module: &FlowModule,
status: &FlowStatus,
status_module: &FlowStatusModule,
last_result: serde_json::Value,
args: &mut Map<String, serde_json::Value>,
) -> error::Result<NextFlowTransform> {
match &module.value {
FlowModuleValue::Script { path: script_path } => Ok(NextFlowTransform::Continue(
script_path_to_payload(script_path, &mut db.begin().await?, &flow_job.workspace_id)
.await?,
NextStatus::NextStep,
)),
FlowModuleValue::RawScript(raw_code) => {
let mut raw_code = raw_code.clone();
if raw_code.path.is_none() {
raw_code.path = Some(format!("{}/{}", flow_job.script_path(), status.step));
}
Ok(NextFlowTransform::Continue(
JobPayload::Code(raw_code),
NextStatus::NextStep,
))
}
/* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */
FlowModuleValue::ForloopFlow { modules, iterator, .. } => {
let next_loop_status = match status_module {
FlowStatusModule::WaitingForPriorSteps => {
let (token, steps) = if let Some(x) = transform_context {
x
} else {
get_transform_context(&db, &flow_job, &status).await?
};
/* Iterator is an InputTransform, evaluate it into an array. */
let itered = iterator
.clone()
.evaluate_with(
|| {
vec![
("result".to_string(), last_result.clone()),
("previous_result".to_string(), last_result.clone()),
]
},
token,
flow_job.workspace_id.clone(),
steps,
)
.await?
.into_array()
.map_err(|not_array| {
Error::ExecutionErr(format!(
"Expected an array value, found: {not_array}"
))
})?;
if let Some(first) = itered.first() {
args.insert("iter".to_string(), json!({ "index": 0, "value": first }));
LoopStatus::NextIteration(NextIteration {
index: 0,
itered,
forloop_jobs: vec![],
})
} else {
LoopStatus::EmptyIterator
}
}
FlowStatusModule::InProgress {
iterator: Some(Iterator { itered, index, args: iterator_args }),
forloop_jobs: Some(forloop_jobs),
..
} => {
let (index, next) = index
.checked_add(1)
.and_then(|i| itered.get(i).map(|next| (i, next)))
/* we shouldn't get here because update_flow_status_after_job_completion
* should leave this state if there iteration is complete, but also it should
* be reasonable to just enter a completed state instead of failing, similar to
* iterating an empty list above */
.with_context(|| {
format!("could not iterate index {index} of {itered:?}")
})?;
args.extend(iterator_args.clone());
args.insert("iter".to_string(), json!({ "index": index, "value": next }));
LoopStatus::NextIteration(NextIteration {
index,
itered: itered.clone(),
forloop_jobs: forloop_jobs.clone(),
})
}
_ => Err(Error::BadRequest(format!(
"Unrecognized module status for ForloopFlow {status_module:?}"
)))?,
};
match next_loop_status {
LoopStatus::EmptyIterator => Ok(NextFlowTransform::EmptyIterator),
LoopStatus::NextIteration(ns) => {
/* embedded flow input is augmented with embedding flow input */
if let Some(value) = &flow_job.args {
value
.as_object()
.ok_or_else(|| {
Error::BadRequest(format!(
"Expected an object value, found: {value:?}"
))
})
.map(|map| args.extend(map.clone()))?;
}
Ok(NextFlowTransform::Continue(
JobPayload::RawFlow {
value: FlowValue {
modules: (*modules).clone(),
failure_module: flow.failure_module.clone(),
same_worker: flow.same_worker,
},
path: Some(format!("{}/loop-{}", flow_job.script_path(), status.step)),
},
NextStatus::NextLoopIteration(ns),
))
}
}
}
FlowModuleValue::Branches { branches, default, .. } => {
let branch = match status_module {
FlowStatusModule::WaitingForPriorSteps => {
let mut branch_chosen = BranchChosen::Default;
for (i, b) in branches.iter().enumerate() {
let pred =
compute_bool_from_expr(b.expr.to_string(), last_result.clone()).await?;
if pred {
branch_chosen = BranchChosen::Branch(i);
break;
}
}
branch_chosen
}
_ => Err(Error::BadRequest(format!(
"Unrecognized module status for Branches {status_module:?}"
)))?,
};
let modules = if let BranchChosen::Branch(index) = branch {
&branches[index].modules
} else {
&default
};
// match inner_flow_transform {}
Ok(NextFlowTransform::BranchChosen(branch, modules.clone()))
}
}
}
frontend/src/lib/components/FlowModulesViewer.svelte
-2
View File
@@ -89,8 +89,6 @@
/>
</div>
{/if}
{:else if mod?.value?.type == 'flow'}
Flow at path {mod?.value?.path}
{:else if mod?.value?.type == 'forloopflow'}
For loop over all the elements of the list returned as a result of step {i}:
<svelte:self modules={mod.value.modules} />
frontend/src/lib/components/RunForm.svelte
-2
View File
@@ -1,11 +1,9 @@
<script lang="ts">
import { page } from '$app/stores'
import type { Flow, Script } from '$lib/gen'
import { decodeState, getToday } from '$lib/utils'
import { slide } from 'svelte/transition'
import { faChevronDown, faChevronUp } from '@fortawesome/free-solid-svg-icons'
import Icon from 'svelte-awesome'
import SvelteMarkdown from 'svelte-markdown'
import SchemaForm from './SchemaForm.svelte'
import Tooltip from './Tooltip.svelte'
frontend/src/lib/components/ScriptEditor.svelte
+1 -7
View File
@@ -3,13 +3,7 @@
import { CompletedJob, Job, JobService } from '$lib/gen'
import { userStore, workspaceStore } from '$lib/stores'
import { emptySchema, scriptLangToEditorLang } from '$lib/utils'
import {
faCheck,
faExclamationTriangle,
faPlay,
faRotateRight
} from '@fortawesome/free-solid-svg-icons'
import Icon from 'svelte-awesome'
import { faPlay, faRotateRight } from '@fortawesome/free-solid-svg-icons'
import Editor from './Editor.svelte'
import { inferArgs } from '$lib/infer'
frontend/src/lib/components/SharedBadge.svelte
+2
View File
@@ -2,6 +2,7 @@
import { userStore } from '$lib/stores'
import { faPeopleGroup } from '@fortawesome/free-solid-svg-icons'
import Badge from './common/badge/Badge.svelte'
import Tooltip from './Tooltip.svelte'
export let extraPerms: Record<string, boolean> = {}
export let canWrite: boolean
@@ -50,5 +51,6 @@
{#if kind === 'read' || kind === 'write'}
<Badge icon={{ data: faPeopleGroup }} capitalize color="blue">
{kind}
<Tooltip>{reason}</Tooltip>
</Badge>
{/if}
frontend/src/lib/components/flows/content/FlowInput.svelte
-1
View File
@@ -2,7 +2,6 @@
import { flowStore } from '$lib/components/flows/flowStore'
import SchemaEditor from '$lib/components/SchemaEditor.svelte'
import SchemaForm from '$lib/components/SchemaForm.svelte'
import { emptySchema } from '$lib/utils'
import FlowCard from '../common/FlowCard.svelte'
import CopyFirstStepSchema from './CopyFirstStepSchema.svelte'
frontend/src/lib/components/flows/content/RemoveStepConfirmationModal.svelte
-1
View File
@@ -1,6 +1,5 @@
<script lang="ts">
import Alert from '$lib/components/common/alert/Alert.svelte'
import Badge from '$lib/components/common/badge/Badge.svelte'
import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
import { createEventDispatcher } from 'svelte'
frontend/src/lib/components/flows/flowStateUtils.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import type { Schema } from '$lib/common'
import { CompletedJob, Job, Script, ScriptService, type FlowModule, type RawScript } from '$lib/gen'
import { DENO_FAILURE_MODULE_CODE, initialCode } from '$lib/script_helpers'
import { Job, Script, ScriptService, type FlowModule, type RawScript } from '$lib/gen'
import { initialCode } from '$lib/script_helpers'
import { userStore, workspaceStore } from '$lib/stores'
import {
buildExtraLib,
frontend/src/lib/components/flows/header/FlowImportExportMenu.svelte
+1 -1
View File
@@ -7,7 +7,7 @@
import FlowViewer from '$lib/components/FlowViewer.svelte'
import SimpleEditor from '$lib/components/SimpleEditor.svelte'
import { sendUserToast } from '$lib/utils'
import { faFileExport, faFileImport, faGlobe } from '@fortawesome/free-solid-svg-icons'
import { faFileExport, faFileImport } from '@fortawesome/free-solid-svg-icons'
import Icon from 'svelte-awesome'
import { Button } from '../../common'
import { flowStore, initFlow } from '../flowStore'
frontend/src/lib/components/flows/map/FlowModuleSchemaMap.svelte
+1 -1
View File
@@ -147,7 +147,7 @@
<div slot="content" class="w-full truncate block">
<span
>{mod.summary ||
mod.value.path ||
(`path` in mod.value ? mod.value.path : undefined) ||
(mod.value.type === 'rawscript'
? `Inline ${mod.value.language}`
: 'Select a script')}</span
frontend/src/routes/flows/get/[...path].svelte
-1
View File
@@ -16,7 +16,6 @@
defaultIfEmptyString,
flowToHubUrl
} from '$lib/utils'
import Icon from 'svelte-awesome'
import {
faPlay,
faEdit,
frontend/src/routes/scripts.svelte
+3 -48
View File
@@ -55,9 +55,6 @@
let groupedScripts: Section[] = []
let communityScripts: Section[] = []
let templateScripts: Script[] = []
let templateFilter = ''
let filteredTemplates: Script[] | undefined
let tab: Tab = 'all'
let shareModal: ShareModal
@@ -68,8 +65,6 @@
}
const fuse: Fuse<ScriptW> = new Fuse(scripts, fuseOptions)
const templateFuse: Fuse<Script> = new Fuse(templateScripts, fuseOptions)
const hubScriptsFuse: Fuse<any> = new Fuse($hubScripts ?? [], {
includeScore: false,
keys: ['app', 'path', 'summary']
@@ -87,11 +82,6 @@
? hubScriptsFuse.search(hubFilter).map((value) => value.item)
: $hubScripts ?? []
$: filteredTemplates =
templateFilter.length > 0
? templateFuse.search(templateFilter).map((value) => value.item)
: templateScripts
$: {
let defaults: string[] = []
@@ -159,7 +149,6 @@
loadScripts()
}
}
</script>
<Modal bind:this={codeViewer}>
@@ -182,7 +171,7 @@
<CreateActions />
</PageHeader>
<Tabs bind:selected={tab} >
<Tabs bind:selected={tab}>
<Tab value="all">All</Tab>
<Tab value="hub">Hub</Tab>
<Tab value="personal">{`Personal space (${$userStore?.username})`}</Tab>
@@ -190,11 +179,11 @@
<Tab value="shared">Shared</Tab>
<Tab value="examples">Examples</Tab>
</Tabs>
{#if tab != 'hub'}
<input placeholder="Search scripts" bind:value={scriptFilter} class="search-bar mt-2" />
{/if}
<div class="grid grid-cols-1 divide-y">
{#each tab == 'all' ? ['personal', 'groups', 'shared', 'examples', 'hub'] : [tab] as sectionTab}
<div class="shadow p-4 my-2">
@@ -430,40 +419,6 @@
}}
/>
<!-- <Modal
bind:this={templateModal}
on:open={() => {
loadTemplateScripts()
}}
>
<div slot="title">Pick a template</div>
<div slot="content">
<div class="w-12/12 pb-4">
<input placeholder="Search templates" bind:value={templateFilter} class="search-bar" />
</div>
<div class="flex flex-col mb-2 md:mb-6">
{#if filteredTemplates && filteredTemplates.length > 0}
{#each filteredTemplates as { summary, path, hash }}
<a
class="p-1 flex flex-row items-baseline gap-2 selected text-gray-700"
href="/scripts/add?template={path}"
>
{#if summary}
<p class="text-sm font-semibold">{summary}</p>
{/if}
<p class="text-sm">{path}</p>
<p class="text-gray-400 text-xs text-right grow">
Last version: {hash}
</p>
</a>
{/each}
{:else}
<p class="text-sm text-gray-700">No templates</p>
{/if}
</div>
</div>
</Modal> -->
<style>
.selected:hover {
@apply border border-gray-500 rounded-md border-opacity-50;
frontend/src/routes/workspace_settings.svelte
-1
View File
@@ -20,7 +20,6 @@
import PageHeader from '$lib/components/PageHeader.svelte'
import { userStore, usersWorkspaceStore, workspaceStore, oauthStore } from '$lib/stores'
import CenteredPage from '$lib/components/CenteredPage.svelte'
import Icon from 'svelte-awesome'
import { faSlack } from '@fortawesome/free-brands-svg-icons'
import TableCustom from '$lib/components/TableCustom.svelte'
import { goto } from '$app/navigation'
openflow.openapi.yaml
+38 -14
View File
@@ -138,14 +138,14 @@ components:
- $ref: "#/components/schemas/RawScript"
- $ref: "#/components/schemas/PathScript"
- $ref: "#/components/schemas/ForloopFlow"
- $ref: "#/components/schemas/PathFlow"
- $ref: "#/components/schemas/Branches"
discriminator:
propertyName: type
mapping:
rawscript: "#/components/schemas/RawScript"
script: "#/components/schemas/PathScript"
forloopflow: "#/components/schemas/ForloopFlow"
flow: "#/components/schemas/PathFlow"
branches: "#/components/schemas/Branches"
RawScript:
type: object
@@ -180,18 +180,6 @@ components:
- type
- path
PathFlow:
type: object
properties:
path:
type: string
type:
type: string
enum:
- flow
required:
- type
ForloopFlow:
type: object
properties:
@@ -213,6 +201,42 @@ components:
- skip_failures
- type
Branches:
type: object
properties:
branches:
type: array
items:
type: object
properties:
summary:
type: string
expr:
type: string
modules:
type: array
items:
$ref: "#/components/schemas/FlowModule"
required:
- modules
- expr
default:
type: object
properties:
modules:
type: array
items:
$ref: "#/components/schemas/FlowModule"
required: [modules]
type:
type: string
enum:
- forloopflow
required:
- branches
- default
- type
FlowStatus:
type: object
properties: