From caecbfd0d9eaadc38372ce7238ed6d3baf9ba6e3 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 27 Jul 2022 12:18:02 +0200 Subject: [PATCH] feat: add schedule settable from pull flows --- backend/openapi.yaml | 3 + backend/src/schedule.rs | 13 ++- frontend/src/lib/components/ArgInput.svelte | 2 +- frontend/src/lib/components/CronInput.svelte | 11 ++- frontend/src/lib/components/Editor.svelte | 4 +- .../src/lib/components/FlowBuilder.svelte | 97 ++++++++++++++---- frontend/src/lib/components/FlowEditor.svelte | 96 ++++++++++++------ .../src/lib/components/RadioButtonV2.svelte | 2 +- .../src/lib/components/SchemaEditor.svelte | 5 + .../src/lib/components/SharedBadge.svelte | 16 +-- .../src/lib/components/flows/flowStore.ts | 14 +++ frontend/src/lib/components/flows/utils.ts | 9 -- .../propertyPicker/ObjectViewer.svelte | 6 +- frontend/src/lib/utils.ts | 4 + frontend/src/routes/flows/add.svelte | 3 +- .../src/routes/flows/edit/[...path].svelte | 4 +- .../src/routes/flows/get/[...path].svelte | 99 +++++++++++++------ frontend/src/routes/schedules.svelte | 31 +++--- .../src/routes/scripts/get/[...hash].svelte | 85 ++++++++-------- 19 files changed, 343 insertions(+), 161 deletions(-) diff --git a/backend/openapi.yaml b/backend/openapi.yaml index 39843f3fba..2c939952b9 100644 --- a/backend/openapi.yaml +++ b/backend/openapi.yaml @@ -3793,6 +3793,7 @@ components: - offset_ - extra_perms - is_flow + - enabled NewSchedule: type: object @@ -3809,6 +3810,8 @@ components: type: boolean args: $ref: "#/components/schemas/ScriptArgs" + enabled: + type: boolean required: - path - schedule diff --git a/backend/src/schedule.rs b/backend/src/schedule.rs index 45f7bb1e26..a68131b7e2 100644 --- a/backend/src/schedule.rs +++ b/backend/src/schedule.rs @@ -64,6 +64,7 @@ pub struct NewSchedule { pub script_path: String, pub is_flow: bool, pub args: Option, + pub enabled: Option, } pub async fn push_scheduled_job<'c>( @@ -133,7 +134,7 @@ async fn create_schedule( check_flow_conflict(&mut tx, &w_id, &ns.path, ns.is_flow, &ns.script_path).await?; let schedule = sqlx::query_as!(Schedule, - "INSERT INTO schedule (workspace_id, path, schedule, offset_, edited_by, script_path, is_flow, args) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING *", + "INSERT INTO schedule (workspace_id, path, schedule, offset_, edited_by, script_path, is_flow, args, enabled) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING *", w_id, ns.path, ns.schedule, @@ -141,7 +142,8 @@ async fn create_schedule( &authed.username, ns.script_path, ns.is_flow, - ns.args + ns.args, + ns.enabled ) .fetch_one(&mut tx) .await?; @@ -165,8 +167,13 @@ async fn create_schedule( ) .await?; - let tx = push_scheduled_job(tx, schedule).await?; + let tx = if ns.enabled.unwrap_or(true) { + push_scheduled_job(tx, schedule).await? + } else { + tx + }; tx.commit().await?; + Ok(ns.path.to_string()) } diff --git a/frontend/src/lib/components/ArgInput.svelte b/frontend/src/lib/components/ArgInput.svelte index 46e3185c36..e86d64aa7a 100644 --- a/frontend/src/lib/components/ArgInput.svelte +++ b/frontend/src/lib/components/ArgInput.svelte @@ -296,7 +296,7 @@ {disabled} class="default-button-secondary items-center leading-4 py-0 my-px px-1 float-right" on:click={() => (value = undefined)} - >ResetReset to default valueReset Reset to default value {/if} diff --git a/frontend/src/lib/components/CronInput.svelte b/frontend/src/lib/components/CronInput.svelte index a75387e841..6ce1e8976d 100644 --- a/frontend/src/lib/components/CronInput.svelte +++ b/frontend/src/lib/components/CronInput.svelte @@ -12,6 +12,7 @@ let preview: string[] = [] let cronError = '' export let schedule: string = '0 0 12 * *' + let limit = 3 $: handleScheduleInput(schedule) @@ -69,11 +70,17 @@ {#if preview && preview.length > 0}
-
The next 10 runs will be scheduled at:
+
The next runs will be scheduled at:
    - {#each preview as p} + {#each preview.slice(0, limit) as p}
  • {displayDate(p)}
    • {/each} +
  • ...
    • + {#if limit != 10} + + {:else} + + {/if}
{/if} diff --git a/frontend/src/lib/components/Editor.svelte b/frontend/src/lib/components/Editor.svelte index 08bed3112b..6721187eb0 100644 --- a/frontend/src/lib/components/Editor.svelte +++ b/frontend/src/lib/components/Editor.svelte @@ -252,9 +252,7 @@ (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' - ) + console.log('reconnecting to language servers') lastWsAttempt = new Date() nbWsAttempt++ reloadWebsocket() diff --git a/frontend/src/lib/components/FlowBuilder.svelte b/frontend/src/lib/components/FlowBuilder.svelte index fe4e201a24..233a3b2f1b 100644 --- a/frontend/src/lib/components/FlowBuilder.svelte +++ b/frontend/src/lib/components/FlowBuilder.svelte @@ -1,12 +1,13 @@ @@ -63,7 +83,7 @@
See JSON
- +
@@ -106,7 +126,7 @@ } url.searchParams.append( 'flow', - btoa(JSON.stringify(flowToMode(openFlow, mode))) + btoa(JSON.stringify(flowToMode(openFlow, $mode))) ) window.open(url, '_blank')?.focus() } @@ -154,35 +174,48 @@ options={[ [ { - title: 'Push', + title: 'UI or webhook triggered', desc: 'Trigger this flow through the generated UI, a manual schedule or by calling the associated webhook' }, 'push' ], [ { - title: 'Pull', + title: 'Watching changes regularly', desc: 'The first module of this flow is a trigger script whose purpose is to pull data from an external source and return all new items since last run. This flow is meant to be scheduled very regularly to reduce latency to react to new events. It will trigger the rest of the flow once per item. If no new items, the flow will be skipped.' }, 'pull' ] ]} - bind:value={mode} + bind:value={$mode} /> - + {#if $mode == 'pull'} +
+ + The primary schedule of a flow is simply a schedule that has the same name as a + flow. It can be set and enabled directly within the flow editor. "Watching for + new changes" flows are meant to be watching regularly for new items in an + external systems. The primary schedule purpose is there to set the periodicity + at which you want this watcher to operate. +   + +
+ {#if !scheduleEnabled} + No next scheduled run when disabled + {/if} + +
+ + +
+ {/if}
  • @@ -194,10 +227,15 @@
    - + { + $flowStore = $flowStore + }} + schema={$flowStore.schema} + />
    • - + {/each}
  • diff --git a/frontend/src/lib/components/RadioButtonV2.svelte b/frontend/src/lib/components/RadioButtonV2.svelte index 5af7a3f839..60aac5f449 100644 --- a/frontend/src/lib/components/RadioButtonV2.svelte +++ b/frontend/src/lib/components/RadioButtonV2.svelte @@ -20,7 +20,7 @@ class="flex flex-col default-secondary-button-v2 mb-2 grow" class:selected={value == val} > -

    {label.title} {label.desc}

    +

    {label.title} {label.desc}

    {/each}
    diff --git a/frontend/src/lib/components/SchemaEditor.svelte b/frontend/src/lib/components/SchemaEditor.svelte index c92ada4f97..4559ba7ba2 100644 --- a/frontend/src/lib/components/SchemaEditor.svelte +++ b/frontend/src/lib/components/SchemaEditor.svelte @@ -6,6 +6,9 @@ import { emptySchema, sendUserToast } from '$lib/utils' import Tooltip from './Tooltip.svelte' import TableCustom from './TableCustom.svelte' + import { createEventDispatcher } from 'svelte' + + const dispatch = createEventDispatcher() export let schema: Schema = emptySchema() @@ -63,6 +66,7 @@ } schema = schema schemaString = JSON.stringify(schema, null, '\t') + dispatch('change', schema) } function startEditArgument(argName: string): void { @@ -87,6 +91,7 @@ delete schema.properties[argName] schema = schema schemaString = JSON.stringify(schema, null, '\t') + dispatch('change', schema) } else { throw Error('Argument not found!') } diff --git a/frontend/src/lib/components/SharedBadge.svelte b/frontend/src/lib/components/SharedBadge.svelte index 5a980c7ddc..e5e6199c94 100644 --- a/frontend/src/lib/components/SharedBadge.svelte +++ b/frontend/src/lib/components/SharedBadge.svelte @@ -46,10 +46,12 @@ } - - {#if kind == 'read'} - shared to you (read-only) - {:else if kind == 'write'} - shared to you - {/if} - +{#if kind == 'read' || kind == 'write'} + + {#if kind == 'read'} + shared to you (read-only) + {:else if kind == 'write'} + shared to you + {/if} + +{/if} diff --git a/frontend/src/lib/components/flows/flowStore.ts b/frontend/src/lib/components/flows/flowStore.ts index ae1720f9d0..8eaa0f65cf 100644 --- a/frontend/src/lib/components/flows/flowStore.ts +++ b/frontend/src/lib/components/flows/flowStore.ts @@ -7,10 +7,14 @@ import { createInlineScriptModuleFromPath, getFirstStepSchema, loadSchemaFromMod export type FlowMode = 'push' | 'pull' +export const mode = writable('push') export const flowStore = writable(undefined) export const schemasStore = writable([]) export function initFlow(flow: Flow) { + const newMode = flow.value.modules[1]?.value.type === FlowModuleValue.type.FORLOOPFLOW ? 'pull' : 'push' + mode.set(newMode) + flow = flattenForloopFlows(flow, newMode) schemasStore.set([]) flowStore.set(flow) // For each module in flow, we should load the corresponding schema @@ -19,6 +23,16 @@ export function initFlow(flow: Flow) { }) } +export function flattenForloopFlows(flow: Flow, mode: FlowMode): Flow { + let newFlow: Flow = JSON.parse(JSON.stringify(flow)) + if (mode == 'pull') { + const oldModules = newFlow.value.modules[1].value.value?.modules ?? [] + newFlow.value.modules = newFlow.value.modules.slice(0, 1) + newFlow.value.modules.push(...oldModules) + } + return newFlow +} + export const isCopyFirstStepSchemaDisabled = derived(flowStore, (flow: Flow | undefined) => { if (flow) { const modules = flow.value.modules diff --git a/frontend/src/lib/components/flows/utils.ts b/frontend/src/lib/components/flows/utils.ts index b969f3843a..9a54ab761c 100644 --- a/frontend/src/lib/components/flows/utils.ts +++ b/frontend/src/lib/components/flows/utils.ts @@ -35,15 +35,6 @@ export function flowToMode(flow: Flow | any, mode: FlowMode): Flow { return flow } -export function flattenForloopFlows(flow: Flow): Flow { - let newFlow: Flow = JSON.parse(JSON.stringify(flow)) - if (newFlow.value.modules[1]?.value.type == FlowModuleValue.type.FORLOOPFLOW) { - const oldModules = newFlow.value.modules[1].value.value?.modules ?? [] - newFlow.value.modules = newFlow.value.modules.slice(0, 1) - newFlow.value.modules.push(...oldModules) - } - return newFlow -} export function getTypeAsString(arg: any): string { if (arg === null) { diff --git a/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte b/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte index dea4632097..3e3c385773 100644 --- a/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte +++ b/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte @@ -7,6 +7,7 @@ export let level = 0 export let isLast = true export let currentPath: string = '' + export let pureViewer = false const collapsedSymbol = '...' let keys: string | any[] @@ -67,6 +68,7 @@ level={level + 1} isLast={index === keys.length - 1} currentPath={computeKey(key)} + {pureViewer} on:select /> {:else} @@ -75,7 +77,9 @@ {:else} {JSON.stringify(json[key])} - + {#if !pureViewer} + + {/if} {/if} {/if} diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts index e5c958111a..b8569f93d2 100644 --- a/frontend/src/lib/utils.ts +++ b/frontend/src/lib/utils.ts @@ -176,6 +176,10 @@ export function canWrite( return false } +export function defaultIfEmptyString(str: string | undefined, dflt: string): string { + return str == undefined || str == '' ? dflt : str +} + export function removeKeysWithEmptyValues(obj: any): any { Object.keys(obj).forEach((key) => (obj[key] === undefined ? delete obj[key] : {})) } diff --git a/frontend/src/routes/flows/add.svelte b/frontend/src/routes/flows/add.svelte index d0d7f1a4d0..6877da4140 100644 --- a/frontend/src/routes/flows/add.svelte +++ b/frontend/src/routes/flows/add.svelte @@ -10,7 +10,7 @@ import { page } from '$app/stores' import FlowBuilder from '$lib/components/FlowBuilder.svelte' - import { initFlow } from '$lib/components/flows/flowStore' + import { initFlow, mode } from '$lib/components/flows/flowStore' import { FlowService, type Flow } from '$lib/gen' import { emptySchema, sendUserToast } from '$lib/utils' @@ -42,6 +42,7 @@ $page.url.searchParams.delete('hub') sendUserToast(`Flow has been loaded from hub flow id ${hubId}.`) } + $mode = 'push' initFlow(flow) } diff --git a/frontend/src/routes/flows/edit/[...path].svelte b/frontend/src/routes/flows/edit/[...path].svelte index 2ec8797835..e71704f95f 100644 --- a/frontend/src/routes/flows/edit/[...path].svelte +++ b/frontend/src/routes/flows/edit/[...path].svelte @@ -7,13 +7,12 @@