+
+
+
Metadata
+
+
+
-
-
-
- {#if cronVersion === 'v1'}
- Schedules use CRON syntax. Seconds are mandatory.
- {:else}
- Schedules use extended CRON syntax .
+ {#if !edit && !primary}
+
+
- {/if}
-
- {#if initialCronVersion !== 'v2'}
-
{/if}
-
+
+
+
-
+
+ {#snippet header()}
+ {#if cronVersion === 'v1'}
+ Schedules use CRON syntax. Seconds are mandatory.
+ {:else}
+ Schedules use extended CRON syntax .
{/if}
-
-
+ {/snippet}
+ {#if initialCronVersion !== 'v2'}
+
+ {/if}
+
+
+
+ {#if !hideTarget}
{#if !edit}
Pick a script or flow to be triggered by the schedule
{/if}
-
- {#if !loading}
- {#if runnable}
- {#if runnable?.schema && runnable.schema.properties && Object.keys(runnable.schema.properties).length > 0}
- {#await import('$lib/components/SchemaForm.svelte')}
-
- {:then Module}
-
- {/await}
- {:else}
-
- This {is_flow ? 'flow' : 'script'} takes no argument
-
- {/if}
- {:else if script_path != ''}
-
- You cannot see the the {is_flow ? 'flow' : 'script'} input form as you do not have
- access to it.
-
- {:else}
-
- Pick a {is_flow ? 'flow' : 'script'} and fill its argument here
-
- {/if}
- {:else}
-
- {/if}
-
-
-
-
+ {/if}
+
{#if !loading}
-
- Error Handler
- Recovery Handler
- Success Handler
- {#if itemKind === 'script'}
- Retries
- Custom tag
- {/if}
-
-
- {#if optionTabSelected === 'error_handler'}
-
-
-
- {#if !$enterpriseLicense}(ee only) {/if}
-
-
-
-
- defaults
- saveAsDefaultErrorHandler(false)
- },
- {
- displayName: 'Override all existing',
- type: 'delete',
- action: () => saveAsDefaultErrorHandler(true)
- }
- ]}
- >
-
-
-
-
-
-
-
0}
+ {#await import('$lib/components/SchemaForm.svelte')}
+
+ This {is_flow ? 'flow' : 'script'} takes no argument
-
-
-
-
-
-
The following args will be passed to the error handler:
-
- path : The path of the script or flow that failed.is_flow : Whether the runnable is a flow.schedule_path : The path of the schedule.error : The error details.failed_times : Minimum number of times the schedule failed
- before calling the error handler.started_at : The start datetime of the latest job that failed.
-
-
-
-
-
-
-
- Triggered when schedule failed
-
- >=
- ==
-
-
-
time{failedTimes > 1 ? 's in a row' : ''}
-
-
-
-
- {#if !$enterpriseLicense}(ee only) {/if}
-
-
-
-
- defaults
- saveAsDefaultRecoveryHandler(false)
- },
- {
- displayName: 'Override all existing',
- type: 'delete',
- action: () => saveAsDefaultRecoveryHandler(true)
- }
- ]}
- >
-
-
-
-
-
-
-
-
-
-
The following args will be passed to the recovery handler:
-
- path : The path of the script or flow that recovered.is_flow : Whether the runnable is a flow.schedule_path : The path of the schedule.error : The error of the last job that errorederror_started_at : The start datetime of the last job that
- erroredsuccess_times : The number of times the schedule succeeded
- before calling the recovery handler.success_result : The result of the latest successful jobsuccess_started_at : The start datetime of the latest
- successful job
-
-
-
-
-
-
-
Triggered when schedule recovered
-
-
time{recoveredTimes > 1 ? 's in a row' : ''}
-
-
- {:else if optionTabSelected === 'success_handler'}
- {@const disabled = !can_write || emptyString($enterpriseLicense)}
-
-
-
- {#if !$enterpriseLicense}(ee only) {/if}
-
-
-
-
- defaults
- saveAsDefaultSuccessHandler(false)
- },
- {
- displayName: 'Override all existing',
- type: 'delete',
- action: () => saveAsDefaultSuccessHandler(true)
- }
- ]}
- >
-
-
-
-
-
-
-
-
-
-
The following args will be passed to the success handler:
-
- path : The path of the script or flow that succeeded.is_flow : Whether the runnable is a flow.schedule_path : The path of the schedule.success_result : The result of the successful jobsuccess_started_at : The start datetime of the successful job
-
-
-
-
-
- {:else if optionTabSelected === 'retries'}
- {@const disabled = !can_write || emptyString($enterpriseLicense)}
-
-
-
- {#if !$enterpriseLicense}(ee only) {/if}
-
-
- If defined, upon error this schedule will be retried with a delay and a maximum
- number of attempts as defined below.
-
-
-
- {:else if optionTabSelected === 'tag'}
-
+ {/if}
+ {:else if script_path != ''}
+
+ You cannot see the the {is_flow ? 'flow' : 'script'} input form as you do not have access
+ to it.
+
+ {:else}
+
+ Pick a {is_flow ? 'flow' : 'script'} and fill its argument here
+
{/if}
{:else}
+
+
+
+ {@render errorHandler()}
+
+
+ {/if}
+{/snippet}
+
+{#snippet errorHandler()}
+
+ {#if !loading}
+
+ Error Handler
+ Recovery Handler
+ Success Handler
+ {#if itemKind === 'script'}
+ Retries
+ Custom tag
+ {/if}
+
+
+ {#if optionTabSelected === 'error_handler'}
+
+ {#snippet header()}
+
+ {#if !$enterpriseLicense}(ee only) {/if}
+
+ {/snippet}
+ {#snippet action()}
+
+ defaults
+ saveAsDefaultErrorHandler(false)
+ },
+ {
+ displayName: 'Override all existing',
+ type: 'delete',
+ action: () => saveAsDefaultErrorHandler(true)
+ }
+ ]}
+ >
+ {#snippet children()}
+
+
+ {/snippet}
+
+
+
+
+
+
+
+
+
The following args will be passed to the error handler:
+
+ path : The path of the script or flow that failed.is_flow : Whether the runnable is a flow.schedule_path : The path of the schedule.error : The error details.failed_times : Minimum number of times the schedule failed before
+ calling the error handler.started_at : The start datetime of the latest job that failed.
+
+
+
+
+
+
+
+ Triggered when schedule failed
+
+ >=
+ ==
+
+
+
time{failedTimes > 1 ? 's in a row' : ''}
+
+
+ {:else if optionTabSelected === 'recovery_handler'}
+ {@const disabled = !can_write || emptyString($enterpriseLicense)}
+
+ {#snippet header()}
+
+ {#if !$enterpriseLicense}(ee only) {/if}
+
+ {/snippet}
+ {#snippet action()}
+
+ defaults
+ saveAsDefaultRecoveryHandler(false)
+ },
+ {
+ displayName: 'Override all existing',
+ type: 'delete',
+ action: () => saveAsDefaultRecoveryHandler(true)
+ }
+ ]}
+ >
+ {#snippet children()}
+
+
+ {/snippet}
+
+
+
+
+
+
The following args will be passed to the recovery handler:
+
+ path : The path of the script or flow that recovered.is_flow : Whether the runnable is a flow.schedule_path : The path of the schedule.error : The error of the last job that errorederror_started_at : The start datetime of the last job that erroredsuccess_times : The number of times the schedule succeeded before
+ calling the recovery handler.success_result : The result of the latest successful jobsuccess_started_at : The start datetime of the latest successful job
+
+
+
+
+
+
+
Triggered when schedule recovered
+
+
time{recoveredTimes > 1 ? 's in a row' : ''}
+
+
+ {:else if optionTabSelected === 'success_handler'}
+ {@const disabled = !can_write || emptyString($enterpriseLicense)}
+
+ {#snippet header()}
+
+ {#if !$enterpriseLicense}(ee only) {/if}
+
+ {/snippet}
+ {#snippet action()}
+
+ defaults
+ saveAsDefaultSuccessHandler(false)
+ },
+ {
+ displayName: 'Override all existing',
+ type: 'delete',
+ action: () => saveAsDefaultSuccessHandler(true)
+ }
+ ]}
+ >
+ {#snippet children()}
+
+
+ {/snippet}
+
+
+
+
+
+
The following args will be passed to the success handler:
+
+ path : The path of the script or flow that succeeded.is_flow : Whether the runnable is a flow.schedule_path : The path of the schedule.success_result : The result of the successful jobsuccess_started_at : The start datetime of the successful job
+
+
+
+
+
+ {:else if optionTabSelected === 'retries'}
+ {@const disabled = !can_write || emptyString($enterpriseLicense)}
+
+ {#snippet header()}
+
+ {#if !$enterpriseLicense}(ee only) {/if}
+
+
+ If defined, upon error this schedule will be retried with a delay and a maximum number
+ of attempts as defined below.
+
+ {/snippet}
+
+ {:else if optionTabSelected === 'tag'}
+
+ {/if}
+ {:else}
+
{/if}
-
-
+
+{/snippet}
+
+{#if useDrawer}
+
+
+
+
+ {@render saveButton()}
+
+
+ {@render content()}
+
+
+{:else}
+
+
+ {#if customLabel}
+ {@render customLabel()}
+ {/if}
+
+
+
+ {@render saveButton()}
+
+
+ {#if docDescription}
+ {@render docDescription()}
+ {/if}
+ {@render content()}
+
+{/if}
diff --git a/frontend/src/lib/components/triggers/sqs/SqsCapture.svelte b/frontend/src/lib/components/triggers/sqs/SqsCapture.svelte
new file mode 100644
index 0000000000..d61e663d0c
--- /dev/null
+++ b/frontend/src/lib/components/triggers/sqs/SqsCapture.svelte
@@ -0,0 +1,52 @@
+
+
+{#if captureInfo}
+
+ {#snippet description()}
+ {#if captureInfo.active}
+
+ Listening to SQS messages...
+
+ {:else}
+
+ Start capturing to listen to SQS messages.
+
+ {/if}
+ {/snippet}
+
+{/if}
diff --git a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditor.svelte b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditor.svelte
index 2de3e17567..2e1cc47727 100644
--- a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditor.svelte
+++ b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditor.svelte
@@ -2,7 +2,9 @@
import { tick } from 'svelte'
import SqsTriggerEditorInner from './SqsTriggerEditorInner.svelte'
- let open = false
+ let { onUpdate } = $props()
+
+ let open = $state(false)
export async function openEdit(ePath: string, isFlow: boolean) {
open = true
await tick()
@@ -19,9 +21,9 @@
drawer?.openNew(is_flow, initial_script_path, defaultValues)
}
- let drawer: SqsTriggerEditorInner
+ let drawer: SqsTriggerEditorInner | undefined = $state()
{#if open}
-
+
{/if}
diff --git a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorConfigSection.svelte b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorConfigSection.svelte
index 3de41919fb..55282e8762 100644
--- a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorConfigSection.svelte
+++ b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorConfigSection.svelte
@@ -1,7 +1,5 @@
- {#if showCapture && captureInfo}
-
- {/if}
+
+ {#if showTestingBadge}
+
@@ -64,9 +53,13 @@
Select an AWS resource to authenticate your account.
-
{
- aws_resource_path = ''
- }} let:item>
+ {
+ aws_resource_path = ''
+ }}
+ let:item
+ >
diff --git a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte
index eee25de840..c23a683b2c 100644
--- a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte
+++ b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte
@@ -5,40 +5,96 @@
import Path from '$lib/components/Path.svelte'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, emptyString, sendUserToast } from '$lib/utils'
- import { createEventDispatcher } from 'svelte'
- import { Loader2, Save } from 'lucide-svelte'
+ import { Loader2 } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
- import Toggle from '$lib/components/Toggle.svelte'
import { SqsTriggerService, type AwsAuthResourceType } from '$lib/gen'
import SqsTriggerEditorConfigSection from './SqsTriggerEditorConfigSection.svelte'
import Section from '$lib/components/Section.svelte'
import ScriptPicker from '$lib/components/ScriptPicker.svelte'
import Required from '$lib/components/Required.svelte'
+ import type { Snippet } from 'svelte'
+ import TriggerEditorToolbar from '../TriggerEditorToolbar.svelte'
+ import { saveSqsTriggerFromCfg } from './utils'
+ import { handleConfigChange } from '../utils'
- let drawer: Drawer
- let is_flow: boolean = false
- let initialPath = ''
- let edit = true
- let itemKind: 'flow' | 'script' = 'script'
- let script_path = ''
- let initialScriptPath = ''
- let fixedScriptPath = ''
- let path: string = ''
- let pathError = ''
- let enabled = false
- let dirtyPath = false
- let can_write = true
- let drawerLoading = true
- let aws_resource_path: string = ''
- let queue_url = ''
- let message_attributes: string[] = []
- let aws_auth_resource_type: AwsAuthResourceType = 'credentials'
- let isValid = false
- const dispatch = createEventDispatcher()
+ interface Props {
+ useDrawer?: boolean
+ description?: Snippet | undefined
+ hideTarget?: boolean
+ hideTooltips?: boolean
+ allowDraft?: boolean
+ hasDraft?: boolean
+ isDraftOnly?: boolean
+ isEditor?: boolean
+ customLabel?: Snippet
+ isDeployed?: boolean
+ cloudDisabled?: boolean
+ onConfigChange?: (cfg: Record, saveDisabled: boolean, updated: boolean) => void
+ onCaptureConfigChange?: (cfg: Record, isValid: boolean) => void
+ onUpdate?: (path?: string) => void
+ onDelete?: () => void
+ onReset?: () => void
+ }
- $: is_flow = itemKind === 'flow'
+ let {
+ useDrawer = true,
+ description = undefined,
+ hideTarget = false,
+ hideTooltips = false,
+ allowDraft = false,
+ hasDraft = false,
+ isDraftOnly = false,
+ isEditor = false,
+ customLabel = undefined,
+ isDeployed = false,
+ cloudDisabled = false,
+ onConfigChange = undefined,
+ onCaptureConfigChange = undefined,
+ onUpdate = undefined,
+ onDelete = undefined,
+ onReset = undefined
+ }: Props = $props()
- export async function openEdit(ePath: string, isFlow: boolean) {
+ let drawer: Drawer | undefined = $state(undefined)
+ let is_flow: boolean = $state(false)
+ let initialPath = $state('')
+ let edit = $state(true)
+ let itemKind: 'flow' | 'script' = $state('script')
+ let script_path = $state('')
+ let initialScriptPath = $state('')
+ let fixedScriptPath = $state('')
+ let path: string = $state('')
+ let pathError = $state('')
+ let enabled = $state(false)
+ let dirtyPath = $state(false)
+ let can_write = $state(true)
+ let drawerLoading = $state(true)
+ let showLoading = $state(false)
+ let aws_resource_path: string = $state('')
+ let queue_url = $state('')
+ let message_attributes: string[] = $state([])
+ let aws_auth_resource_type: AwsAuthResourceType = $state('credentials')
+ let isValid = $state(false)
+ let initialConfig: Record | undefined = undefined
+ let deploymentLoading = $state(false)
+
+ const sqsConfig = $derived.by(getSaveCfg)
+ const captureConfig = $derived.by(getCaptureConfig)
+ const saveDisabled = $derived(
+ pathError != '' || emptyString(script_path) || !isValid || !can_write
+ )
+ $effect(() => {
+ is_flow = itemKind === 'flow'
+ })
+
+ export async function openEdit(
+ ePath: string,
+ isFlow: boolean,
+ defaultConfig?: Record
+ ) {
+ let loadingTimeout = setTimeout(() => {
+ showLoading = true
+ }, 100) // Do not show loading spinner for the first 100ms
drawerLoading = true
try {
drawer?.openDrawer()
@@ -46,11 +102,13 @@
itemKind = isFlow ? 'flow' : 'script'
edit = true
dirtyPath = false
- await loadTrigger()
+ await loadTrigger(defaultConfig)
} catch (err) {
sendUserToast(`Could not load sqs trigger: ${err.body}`, true)
} finally {
+ clearTimeout(loadingTimeout)
drawerLoading = false
+ showLoading = false
}
}
@@ -59,6 +117,9 @@
fixedScriptPath_?: string,
defaultValues?: Record
) {
+ let loadingTimeout = setTimeout(() => {
+ showLoading = true
+ }, 100)
drawerLoading = true
try {
drawer?.openDrawer()
@@ -69,143 +130,207 @@
script_path = fixedScriptPath
aws_resource_path = defaultValues?.aws_resource_path ?? ''
queue_url = defaultValues?.queue_url ?? ''
- path = ''
+ path = defaultValues?.path ?? ''
message_attributes = defaultValues?.message_attributes ?? []
aws_auth_resource_type = defaultValues?.aws_auth_resource_type ?? 'credentials'
initialPath = ''
edit = false
dirtyPath = false
+ enabled = defaultValues?.enabled ?? false
} finally {
+ initialConfig = structuredClone($state.snapshot(getSaveCfg()))
+ clearTimeout(loadingTimeout)
drawerLoading = false
+ showLoading = false
}
}
- async function loadTrigger(): Promise {
+ async function loadTriggerConfig(cfg?: Record): Promise {
try {
- const s = await SqsTriggerService.getSqsTrigger({
- workspace: $workspaceStore!,
- path: initialPath
- })
- script_path = s.script_path
- initialScriptPath = s.script_path
- aws_resource_path = s.aws_resource_path
- queue_url = s.queue_url
- is_flow = s.is_flow
- message_attributes = s.message_attributes ?? []
- path = s.path
- enabled = s.enabled
- aws_auth_resource_type = s.aws_auth_resource_type
- can_write = canWrite(s.path, s.extra_perms, $userStore)
+ script_path = cfg?.script_path
+ initialScriptPath = cfg?.script_path
+ aws_resource_path = cfg?.aws_resource_path
+ queue_url = cfg?.queue_url
+ is_flow = cfg?.is_flow
+ message_attributes = cfg?.message_attributes ?? []
+ path = cfg?.path
+ enabled = cfg?.enabled
+ aws_auth_resource_type = cfg?.aws_auth_resource_type
+ can_write = canWrite(cfg?.path, cfg?.extra_perms, $userStore)
+ } catch (error) {
+ sendUserToast(`Could not load SQS trigger config: ${error.body}`, true)
+ }
+ }
+
+ async function loadTrigger(defaultConfig?: Record): Promise {
+ try {
+ if (defaultConfig) {
+ loadTriggerConfig(defaultConfig)
+ return
+ } else {
+ const s = await SqsTriggerService.getSqsTrigger({
+ workspace: $workspaceStore!,
+ path: initialPath
+ })
+ loadTriggerConfig(s)
+ }
} catch (error) {
sendUserToast(`Could not load SQS trigger: ${error.body}`, true)
}
}
- async function updateTrigger(): Promise {
- if (edit) {
- await SqsTriggerService.updateSqsTrigger({
- workspace: $workspaceStore!,
- path: initialPath,
- requestBody: {
- path,
- script_path,
- aws_auth_resource_type,
- enabled,
- is_flow,
- queue_url,
- aws_resource_path,
- message_attributes
- }
- })
- sendUserToast(`SQS trigger ${path} updated`)
- } else {
- await SqsTriggerService.createSqsTrigger({
- workspace: $workspaceStore!,
- requestBody: {
- enabled: true,
- aws_resource_path,
- queue_url,
- aws_auth_resource_type,
- path,
- script_path,
- is_flow,
- message_attributes
- }
- })
- sendUserToast(`SQS trigger ${path} created`)
+ function getSaveCfg(): Record {
+ return {
+ script_path,
+ is_flow,
+ path,
+ aws_resource_path,
+ queue_url,
+ message_attributes,
+ aws_auth_resource_type,
+ enabled
}
-
- if (!$usedTriggerKinds.includes('sqs')) {
- $usedTriggerKinds = [...$usedTriggerKinds, 'sqs']
- }
- dispatch('update')
- drawer.closeDrawer()
}
+
+ async function handleToggleEnabled(nEnabled: boolean) {
+ enabled = nEnabled
+ if (!isDraftOnly && !hasDraft) {
+ await SqsTriggerService.setSqsTriggerEnabled({
+ path: initialPath,
+ workspace: $workspaceStore ?? '',
+ requestBody: { enabled: nEnabled }
+ })
+ sendUserToast(`${nEnabled ? 'enabled' : 'disabled'} SQS trigger ${initialPath}`)
+ }
+ }
+
+ async function updateTrigger(): Promise {
+ deploymentLoading = true
+ const cfg = getSaveCfg()
+ const isSaved = await saveSqsTriggerFromCfg(
+ initialPath,
+ cfg,
+ edit,
+ $workspaceStore!,
+ usedTriggerKinds
+ )
+ if (isSaved) {
+ onUpdate?.(cfg.path)
+ drawer?.closeDrawer()
+ }
+ deploymentLoading = false
+ }
+
+ function getCaptureConfig(): Record {
+ return {
+ aws_resource_path,
+ queue_url,
+ message_attributes,
+ aws_auth_resource_type,
+ path
+ }
+ }
+
+ $effect(() => {
+ onCaptureConfigChange?.(captureConfig, isValid)
+ })
+
+ $effect(() => {
+ if (!drawerLoading) {
+ handleConfigChange(sqsConfig, initialConfig, saveDisabled, edit, onConfigChange)
+ }
+ })
-
-
-
- {#if !drawerLoading && can_write}
- {#if edit}
-
- {
- sendUserToast(`${e.detail ? 'enabled' : 'disabled'} sqs trigger ${initialPath}`)
- }}
- />
-
- {/if}
-
- Save
-
+{#if useDrawer}
+
+
+
+ {@render actions()}
+
+ {@render config()}
+
+
+{:else}
+
+
+ {#if customLabel}
+ {@render customLabel()}
{/if}
- {#if drawerLoading}
-
- {:else}
-
-
+
+ {@render actions()}
+
+ {@render config()}
+
+{/if}
+
+{#snippet actions()}
+ {#if !drawerLoading}
+
+ {#if description}
+ {@render description()}
+ {/if}
+ {#if !hideTooltips}
+
{#if edit}
Changes can take up to 30 seconds to take effect.
{:else}
- New postgres triggers can take up to 30 seconds to start listening.
+ New SQS triggers can take up to 30 seconds to start listening.
{/if}
+ {/if}
+
+
-
-
+ {#if !hideTarget}
Pick a script or flow to be triggered Create from template
+ Create from template
+
{/if}
+ {/if}
-
-
{
- loadTriggers()
- }}
- bind:this={sqsTriggerEditor}
-/>
+ onMount(() => {
+ sqsTriggerEditor && openSqsTriggerEditor(isFlow, selectedTrigger.isDraft ?? false)
+ })
+
+ const cloudDisabled = $derived(isCloudHosted())
+
{#if !$enterpriseLicense}
SQS triggers are an enterprise only feature.
-{:else if isCloudHosted()}
-
- SQS triggers are disabled in the multi-tenant cloud.
-
{:else}
-
- SQS triggers allow your scripts/flows to process messages from Amazon Simple Queue Service
- (SQS) in real time. Each trigger listens to an SQS queue and executes a script or a flow when
- new messages arrive.
-
-
- {#if !newItem && sqsTriggers && sqsTriggers.length > 0}
-
-
-
- {#each sqsTriggers as sqsTriggers (sqsTriggers.path)}
-
-
{sqsTriggers.path}
-
- {sqsTriggers.queue_url}
-
-
- sqsTriggerEditor?.openEdit(sqsTriggers.path, isFlow)}
- class="px-2"
- >
- {#if sqsTriggers.canWrite}
- Edit
- {:else}
- View
- {/if}
-
-
-
- {/each}
-
-
- {/if}
-
-
{
- sqsTriggerEditor?.openNew(isFlow, path, e.detail.config)
- }}
- on:addPreprocessor
- on:updateSchema
- on:testWithArgs
- cloudDisabled={false}
- triggerType="sqs"
- {isFlow}
- {path}
- {isEditor}
- {canHavePreprocessor}
- {hasPreprocessor}
- {newItem}
- {openForm}
- bind:showCapture={dontCloseOnLoad}
- />
+ {#snippet description()}
+ {#if cloudDisabled}
+
+ SQS triggers are disabled in the multi-tenant cloud.
+
+ {:else}
+
+ SQS triggers allow you to execute scripts and flows in response to messages in an AWS
+ SQS queue. They can be configured to filter messages based on message attributes.
+
+ {/if}
+ {/snippet}
+
,
+ edit: boolean,
+ workspace: string,
+ usedTriggerKinds: Writable
+): Promise {
+ const requestBody = {
+ path: cfg.path,
+ script_path: cfg.script_path,
+ is_flow: cfg.is_flow,
+ aws_resource_path: cfg.aws_resource_path,
+ queue_url: cfg.queue_url,
+ message_attributes: cfg.message_attributes,
+ aws_auth_resource_type: cfg.aws_auth_resource_type,
+ enabled: cfg.enabled
+ }
+ try {
+ if (edit) {
+ await SqsTriggerService.updateSqsTrigger({
+ workspace,
+ path: initialPath,
+ requestBody
+ })
+ sendUserToast(`SQS trigger ${cfg.path} updated`)
+ } else {
+ await SqsTriggerService.createSqsTrigger({
+ workspace,
+ requestBody: { ...requestBody, enabled: true }
+ })
+ sendUserToast(`SQS trigger ${cfg.path} created`)
+ }
+
+ if (!get(usedTriggerKinds).includes('sqs')) {
+ usedTriggerKinds.update((t) => [...t, 'sqs'])
+ }
+ return true
+ } catch (error) {
+ sendUserToast(error.body || error.message, true)
+ return false
+ }
+}
diff --git a/frontend/src/lib/components/triggers/testingBadge.svelte b/frontend/src/lib/components/triggers/testingBadge.svelte
new file mode 100644
index 0000000000..eb8d288fbf
--- /dev/null
+++ b/frontend/src/lib/components/triggers/testingBadge.svelte
@@ -0,0 +1,9 @@
+
+
+
+ Config used for creating a testing endpoint
+
diff --git a/frontend/src/lib/components/triggers/triggers.svelte.ts b/frontend/src/lib/components/triggers/triggers.svelte.ts
new file mode 100644
index 0000000000..5f40fb6126
--- /dev/null
+++ b/frontend/src/lib/components/triggers/triggers.svelte.ts
@@ -0,0 +1,456 @@
+import {
+ KafkaTriggerService,
+ MqttTriggerService,
+ NatsTriggerService,
+ PostgresTriggerService,
+ ScheduleService,
+ SqsTriggerService,
+ WebsocketTriggerService,
+ type GcpTrigger,
+ type KafkaTrigger,
+ type PostgresTrigger,
+ type Schedule,
+ type TriggersCount,
+ type HttpTrigger,
+ HttpTriggerService,
+ GcpTriggerService
+} from '$lib/gen'
+import { getLightConfig, sortTriggers, updateTriggersCount, type Trigger } from './utils'
+import type { Writable } from 'svelte/store'
+import type { TriggerType } from './utils'
+import type { UserExt } from '$lib/stores'
+import type { ScheduleTrigger } from '../triggers'
+import { canWrite, formatCron } from '$lib/utils'
+
+export class Triggers {
+ #triggers = $state([])
+ #selectedTriggerIndex = $state(undefined)
+ #selectedTrigger = $derived(
+ this.#selectedTriggerIndex !== undefined
+ ? this.#triggers[this.#selectedTriggerIndex]
+ : undefined
+ )
+ #updateDraftCallback: (() => void) | undefined = undefined
+
+ constructor(
+ triggers: Trigger[] = [],
+ selectedIndex?: number,
+ updateDraftCallback?: (() => void) | undefined
+ ) {
+ this.#triggers = triggers
+ this.#selectedTriggerIndex = selectedIndex
+ this.#updateDraftCallback = updateDraftCallback
+ }
+
+ get selectedTrigger(): Trigger | undefined {
+ return this.#selectedTrigger
+ }
+
+ get selectedTriggerIndex(): number | undefined {
+ return this.#selectedTriggerIndex
+ }
+
+ set selectedTriggerIndex(index: number | undefined) {
+ if (index === undefined || index < 0 || index >= this.#triggers.length) {
+ this.#selectedTriggerIndex = undefined
+ } else {
+ this.#selectedTriggerIndex = index
+ }
+ this.#updateDraftCallback?.()
+ }
+
+ get triggers(): Trigger[] {
+ return this.#triggers
+ }
+
+ setTriggers(triggers: Trigger[]) {
+ this.#triggers = triggers
+ this.#updateDraftCallback?.()
+ }
+
+ setDraftConfig(triggerIndex: number, draftConfig: Record | undefined) {
+ if (triggerIndex === undefined || triggerIndex < 0 || triggerIndex >= this.#triggers.length) {
+ return
+ }
+ this.#triggers[triggerIndex].draftConfig = draftConfig
+ this.#updateDraftCallback?.()
+ }
+
+ getDraftTriggersSnapshot(): Trigger[] | undefined {
+ const draftTriggers = this.#triggers.filter((t) => t.draftConfig)
+ return draftTriggers.length > 0 ? $state.snapshot(draftTriggers) : undefined
+ }
+
+ getSelectedTriggerSnapshot(): number | undefined {
+ return $state.snapshot(this.#selectedTriggerIndex)
+ }
+
+ addDraftTrigger(
+ triggersCountStore: Writable,
+ type: TriggerType,
+ path?: string,
+ draftCfg?: Record
+ ): number {
+ const primaryScheduleExists = this.#triggers.some((t) => t.type === 'schedule' && t.isPrimary)
+
+ // Create the new draft trigger
+ const draftId = crypto.randomUUID()
+ const isPrimary = type === 'schedule' && !primaryScheduleExists
+ const newTrigger = {
+ id: draftId,
+ type,
+ path,
+ isPrimary,
+ isDraft: true,
+ draftConfig: draftCfg
+ }
+
+ this.#triggers.push(newTrigger)
+ this.#updateDraftCallback?.()
+
+ updateTriggersCount(triggersCountStore, type, 'add', newTrigger.draftConfig)
+
+ return this.#triggers.length - 1
+ }
+
+ deleteTrigger(
+ triggersCountStore: Writable,
+ triggerIndex: number
+ ): void {
+ if (triggerIndex === undefined || triggerIndex < 0 || triggerIndex >= this.#triggers.length) {
+ return
+ }
+ const { type } = this.#triggers[triggerIndex]
+
+ this.#triggers = this.#triggers.filter((_, index) => index !== triggerIndex)
+
+ updateTriggersCount(triggersCountStore, type, 'remove')
+ this.#updateDraftCallback?.()
+ }
+
+ updateTriggers(
+ remoteTriggers: any[],
+ type: TriggerType,
+ user: UserExt | undefined = undefined
+ ): number {
+ const currentTriggers = this.#triggers
+ // Identify triggers with draftConfig to preserve
+ const configuredTriggers = currentTriggers.filter(
+ (t) => t.type === type && !t.isDraft && t.draftConfig
+ )
+
+ const configMap = new Map }>()
+
+ configuredTriggers.forEach((t) => {
+ configMap.set(t.path ?? '', { draftConfig: t.draftConfig! })
+ })
+
+ const backendTriggers = remoteTriggers.map((trigger) => {
+ const { draftConfig } = configMap.get(trigger.path) ?? {}
+ return {
+ type: type as TriggerType,
+ path: trigger.path,
+ isPrimary: type === 'schedule' && trigger.path === trigger.script_path,
+ isDraft: false,
+ canWrite: canWrite(trigger.path, trigger.extra_perms, user),
+ draftConfig: draftConfig,
+ lightConfig: getLightConfig(type, trigger)
+ }
+ })
+
+ const filteredTriggers = currentTriggers.filter((t) => t.type !== type || t.isDraft)
+ const newTriggers = sortTriggers([...filteredTriggers, ...backendTriggers])
+ this.#triggers = newTriggers
+
+ this.#updateDraftCallback?.()
+ return newTriggers.filter((t) => t.type === type).length
+ }
+
+ async fetchSchedules(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ primarySchedule?: ScheduleTrigger | undefined | false,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+ try {
+ //First update the store with legacy primary schedule
+ if (primarySchedule && !this.#triggers.some((s) => s.isPrimary)) {
+ const primary = {
+ type: 'schedule' as TriggerType,
+ path,
+ isPrimary: true,
+ isDraft: false,
+ draftConfig: {
+ schedule: primarySchedule.cron ? formatCron(primarySchedule.cron) : undefined,
+ args: primarySchedule.args,
+ timezone: primarySchedule.timezone,
+ summary: primarySchedule.summary,
+ description: primarySchedule.description,
+ enabled: primarySchedule.enabled
+ }
+ }
+ this.#triggers = [...this.#triggers, primary]
+ }
+
+ const allDeployedSchedules: Schedule[] = await ScheduleService.listSchedules({
+ workspace: workspaceId,
+ path,
+ isFlow
+ })
+
+ const scheduleCount = this.updateTriggers(allDeployedSchedules, 'schedule', user)
+ const updatedPrimarySchedule = this.#triggers.find((s) => s.isPrimary)
+ triggersCountStore.update((triggersCount) => {
+ return {
+ ...(triggersCount ?? {}),
+ schedule_count: scheduleCount,
+ primary_schedule: {
+ schedule:
+ updatedPrimarySchedule?.draftConfig?.schedule ??
+ updatedPrimarySchedule?.lightConfig?.schedule
+ }
+ }
+ })
+
+ return
+ } catch (error) {
+ console.error('Failed to fetch schedules:', error)
+ return
+ }
+ }
+
+ async fetchWebsocketTriggers(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+ try {
+ const wsTriggers = await WebsocketTriggerService.listWebsocketTriggers({
+ workspace: workspaceId,
+ path,
+ isFlow
+ })
+ const wsCount = this.updateTriggers(wsTriggers, 'websocket', user)
+ triggersCountStore.update((triggersCount) => {
+ return {
+ ...(triggersCount ?? {}),
+ websocket_count: wsCount
+ }
+ })
+ } catch (error) {
+ console.error('Failed to fetch Websocket triggers:', error)
+ }
+ }
+
+ async fetchPostgresTriggers(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+ try {
+ const pgTriggers: PostgresTrigger[] = await PostgresTriggerService.listPostgresTriggers({
+ workspace: workspaceId,
+ path,
+ isFlow
+ })
+ const pgCount = this.updateTriggers(pgTriggers, 'postgres', user)
+ triggersCountStore.update((triggersCount) => {
+ return {
+ ...(triggersCount ?? {}),
+ postgres_count: pgCount
+ }
+ })
+ } catch (error) {
+ console.error('Failed to fetch Postgres triggers:', error)
+ }
+ }
+
+ async fetchKafkaTriggers(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+ try {
+ const kafkaTriggers: KafkaTrigger[] = await KafkaTriggerService.listKafkaTriggers({
+ workspace: workspaceId,
+ path,
+ isFlow
+ })
+ const kafkaCount = this.updateTriggers(kafkaTriggers, 'kafka', user)
+ triggersCountStore.update((triggersCount) => {
+ return {
+ ...(triggersCount ?? {}),
+ kafka_count: kafkaCount
+ }
+ })
+ } catch (error) {
+ console.error('Failed to fetch Kafka triggers:', error)
+ }
+ }
+
+ async fetchNatsTriggers(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+ try {
+ const natsTriggers = await NatsTriggerService.listNatsTriggers({
+ workspace: workspaceId,
+ path,
+ isFlow
+ })
+ const natsCount = this.updateTriggers(natsTriggers, 'nats', user)
+ triggersCountStore.update((triggersCount) => {
+ return {
+ ...(triggersCount ?? {}),
+ nats_count: natsCount
+ }
+ })
+ } catch (error) {
+ console.error('Failed to fetch NATS triggers:', error)
+ }
+ }
+
+ async fetchMqttTriggers(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+ try {
+ const mqttTriggers = await MqttTriggerService.listMqttTriggers({
+ workspace: workspaceId,
+ path,
+ isFlow
+ })
+ const mqttCount = this.updateTriggers(mqttTriggers, 'mqtt', user)
+ triggersCountStore.update((triggersCount) => {
+ return {
+ ...(triggersCount ?? {}),
+ mqtt_count: mqttCount
+ }
+ })
+ } catch (error) {
+ console.error('Failed to fetch MQTT triggers:', error)
+ }
+ }
+
+ async fetchSqsTriggers(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+ try {
+ const sqsTriggers = await SqsTriggerService.listSqsTriggers({
+ workspace: workspaceId,
+ path,
+ isFlow
+ })
+ const sqsCount = this.updateTriggers(sqsTriggers, 'sqs', user)
+ triggersCountStore.update((triggersCount) => {
+ return {
+ ...(triggersCount ?? {}),
+ sqs_count: sqsCount
+ }
+ })
+ } catch (error) {
+ console.error('Failed to fetch SQS triggers:', error)
+ }
+ }
+
+ async fetchGcpTriggers(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+ try {
+ const gcpTriggers: GcpTrigger[] = await GcpTriggerService.listGcpTriggers({
+ workspace: workspaceId,
+ path,
+ isFlow
+ })
+ const gcpCount = this.updateTriggers(gcpTriggers, 'gcp', user)
+ triggersCountStore.update((triggersCount) => {
+ return {
+ ...(triggersCount ?? {}),
+ gcp_count: gcpCount
+ }
+ })
+ } catch (error) {
+ console.error('Failed to fetch GCP Pub/Sub triggers:', error)
+ }
+ }
+
+ async fetchHttpTriggers(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+ try {
+ const httpTriggers: HttpTrigger[] = await HttpTriggerService.listHttpTriggers({
+ workspace: workspaceId,
+ path,
+ isFlow
+ })
+ const httpCount = this.updateTriggers(httpTriggers, 'http', user)
+ triggersCountStore.update((triggersCount) => {
+ return {
+ ...(triggersCount ?? {}),
+ http_routes_count: httpCount
+ }
+ })
+ } catch (error) {
+ console.error('Failed to fetch HTTP triggers:', error)
+ }
+ }
+
+ async fetchTriggers(
+ triggersCountStore: Writable,
+ workspaceId: string | undefined,
+ path: string,
+ isFlow: boolean,
+ primarySchedule: ScheduleTrigger | undefined | false = undefined,
+ user: UserExt | undefined = undefined
+ ): Promise {
+ if (!workspaceId) return
+
+ // Fetch each type of trigger
+ await Promise.all([
+ this.fetchSchedules(triggersCountStore, workspaceId, path, isFlow, primarySchedule, user),
+ this.fetchHttpTriggers(triggersCountStore, workspaceId, path, isFlow, user),
+ this.fetchWebsocketTriggers(triggersCountStore, workspaceId, path, isFlow, user),
+ this.fetchPostgresTriggers(triggersCountStore, workspaceId, path, isFlow, user),
+ this.fetchKafkaTriggers(triggersCountStore, workspaceId, path, isFlow, user),
+ this.fetchNatsTriggers(triggersCountStore, workspaceId, path, isFlow, user),
+ this.fetchMqttTriggers(triggersCountStore, workspaceId, path, isFlow, user),
+ this.fetchSqsTriggers(triggersCountStore, workspaceId, path, isFlow, user),
+ this.fetchGcpTriggers(triggersCountStore, workspaceId, path, isFlow, user)
+ ])
+ }
+}
diff --git a/frontend/src/lib/components/triggers/utils.ts b/frontend/src/lib/components/triggers/utils.ts
new file mode 100644
index 0000000000..7daf829366
--- /dev/null
+++ b/frontend/src/lib/components/triggers/utils.ts
@@ -0,0 +1,529 @@
+import { Webhook, Mail, Calendar, Route, Unplug, Database, Terminal } from 'lucide-svelte'
+import KafkaIcon from '$lib/components/icons/KafkaIcon.svelte'
+import NatsIcon from '$lib/components/icons/NatsIcon.svelte'
+import MqttIcon from '$lib/components/icons/MqttIcon.svelte'
+import AwsIcon from '$lib/components/icons/AwsIcon.svelte'
+import GoogleCloudIcon from '$lib/components/icons/GoogleCloudIcon.svelte'
+import type { CaptureTriggerKind, Flow, NewScript, TriggersCount } from '$lib/gen/types.gen'
+import type { Writable } from 'svelte/store'
+import SchedulePollIcon from '../icons/SchedulePollIcon.svelte'
+import { type TriggerKind } from '$lib/components/triggers'
+import { saveScheduleFromCfg } from '$lib/components/flows/scheduleUtils'
+import { saveHttpRouteFromCfg } from './http/utils'
+import { saveWebsocketTriggerFromCfg } from './websocket/utils'
+import { savePostgresTriggerFromCfg } from './postgres/utils'
+import { saveKafkaTriggerFromCfg } from './kafka/utils'
+import { saveSqsTriggerFromCfg } from './sqs/utils'
+import { saveNatsTriggerFromCfg } from './nats/utils'
+import { saveMqttTriggerFromCfg } from './mqtt/utils'
+import { saveGcpTriggerFromCfg } from './gcp/utils'
+import type { Triggers } from './triggers.svelte'
+import { emptyString } from '$lib/utils'
+
+export const CLOUD_DISABLED_TRIGGER_TYPES = [
+ 'nats',
+ 'kafka',
+ 'sqs',
+ 'mqtt',
+ 'gcp',
+ 'websocket',
+ 'postgres'
+]
+
+export type TriggerType =
+ | 'webhook'
+ | 'email'
+ | 'schedule'
+ | 'http'
+ | 'websocket'
+ | 'postgres'
+ | 'kafka'
+ | 'nats'
+ | 'mqtt'
+ | 'sqs'
+ | 'gcp'
+ | 'poll'
+ | 'cli'
+
+export type Trigger = {
+ type: TriggerType
+ path?: string
+ isDraft?: boolean
+ isPrimary?: boolean
+ canWrite?: boolean
+ id?: string
+ draftConfig?: Record
+ captureConfig?: Record
+ extra?: Record
+ lightConfig?: Record
+}
+
+// Map of trigger kinds to icons
+export const triggerIconMap = {
+ webhook: Webhook,
+ email: Mail,
+ schedule: Calendar,
+ http: Route,
+ websocket: Unplug,
+ postgres: Database,
+ kafka: KafkaIcon,
+ nats: NatsIcon,
+ mqtt: MqttIcon,
+ sqs: AwsIcon,
+ gcp: GoogleCloudIcon,
+ primary_schedule: Calendar,
+ poll: SchedulePollIcon,
+ cli: Terminal
+}
+
+/**
+ * Converts a TriggerType to a CaptureTriggerKind when a mapping exists
+ * @param triggerType The trigger type to convert
+ * @returns The corresponding CaptureTriggerKind or undefined if no mapping exists
+ */
+export function triggerTypeToCaptureKind(triggerType: TriggerType): CaptureTriggerKind | undefined {
+ // Define types that can be mapped to CaptureTriggerKind
+ const capturableTriggerTypes: TriggerType[] = [
+ 'webhook',
+ 'email',
+ 'http',
+ 'websocket',
+ 'postgres',
+ 'kafka',
+ 'nats',
+ 'mqtt',
+ 'sqs',
+ 'gcp',
+ 'cli'
+ ]
+
+ if (capturableTriggerTypes.includes(triggerType)) {
+ return triggerType as CaptureTriggerKind
+ }
+
+ return undefined
+}
+
+export function updateTriggersCount(
+ triggersCountStore: Writable,
+ type: TriggerType,
+ action: 'add' | 'remove',
+ primaryCfg?: Record,
+ isPrimary?: boolean
+) {
+ // Map trigger types to their corresponding count property names
+ const countPropertyMap: Record = {
+ webhook: undefined,
+ email: undefined,
+ schedule: 'schedule_count',
+ http: 'http_routes_count',
+ websocket: 'websocket_count',
+ postgres: 'postgres_count',
+ kafka: 'kafka_count',
+ nats: 'nats_count',
+ mqtt: 'mqtt_count',
+ sqs: 'sqs_count',
+ gcp: 'gcp_count',
+ poll: undefined,
+ cli: undefined
+ }
+
+ const countProperty = countPropertyMap[type]
+
+ triggersCountStore.update((triggersCount) => {
+ // Handle special case for schedule
+ if (type === 'schedule') {
+ if (action === 'add' && primaryCfg) {
+ return {
+ ...(triggersCount ?? {}),
+ schedule_count: (triggersCount?.schedule_count ?? 0) + 1,
+ primary_schedule: primaryCfg?.schedule
+ }
+ } else if (action === 'remove') {
+ return {
+ ...(triggersCount ?? {}),
+ schedule_count: (triggersCount?.schedule_count ?? 1) - 1,
+ primary_schedule: isPrimary ? undefined : triggersCount?.primary_schedule
+ }
+ }
+ }
+
+ // Handle standard count updates
+ if (countProperty && action === 'add') {
+ return {
+ ...(triggersCount ?? {}),
+ [countProperty]: (triggersCount?.[countProperty] ?? 0) + 1
+ }
+ } else if (countProperty && action === 'remove') {
+ return {
+ ...(triggersCount ?? {}),
+ [countProperty]: (triggersCount?.[countProperty] ?? 1) - 1
+ }
+ }
+
+ return triggersCount
+ })
+}
+
+// TODO: Remove this once we've migrated all the trigger kinds to the new TriggerType enum
+export function triggerKindToTriggerType(kind: TriggerKind): TriggerType | undefined {
+ switch (kind) {
+ case 'webhooks':
+ return 'webhook'
+ case 'emails':
+ return 'email'
+ case 'schedules':
+ return 'schedule'
+ case 'routes':
+ return 'http'
+ case 'websockets':
+ return 'websocket'
+ case 'postgres':
+ return 'postgres'
+ case 'kafka':
+ return 'kafka'
+ case 'nats':
+ return 'nats'
+ case 'mqtt':
+ return 'mqtt'
+ case 'sqs':
+ return 'sqs'
+ case 'gcp':
+ return 'gcp'
+ case 'scheduledPoll':
+ return 'poll'
+ default:
+ throw new Error(`Unknown TriggerKind: ${kind}`)
+ }
+}
+
+export async function deployTriggers(
+ triggersToDeploy: Trigger[],
+ workspaceId: string | undefined,
+ isAdmin: boolean,
+ usedTriggerKinds: Writable,
+ initialPath?: string,
+ isNew?: boolean
+) {
+ if (!workspaceId) return
+
+ if (isNew && initialPath) {
+ triggersToDeploy.forEach((trigger) => {
+ trigger.draftConfig = {
+ ...trigger.draftConfig,
+ script_path: initialPath
+ }
+ })
+ }
+
+ // Map of trigger types to their save functions
+ const triggerSaveFunctions: Record = {
+ webhook: undefined,
+ email: undefined,
+ schedule: (trigger: Trigger) => {
+ if (trigger.isPrimary && initialPath) {
+ trigger.draftConfig = {
+ ...trigger.draftConfig,
+ path: initialPath,
+ script_path: initialPath
+ }
+ }
+ return saveScheduleFromCfg(trigger.draftConfig ?? {}, !trigger.isDraft, workspaceId)
+ },
+ http: (trigger: Trigger) =>
+ saveHttpRouteFromCfg(
+ trigger.path ?? trigger.draftConfig?.path ?? '',
+ trigger.draftConfig ?? {},
+ !trigger.isDraft,
+ workspaceId,
+ isAdmin,
+ usedTriggerKinds
+ ),
+ websocket: (trigger: Trigger) =>
+ saveWebsocketTriggerFromCfg(
+ trigger.path ?? trigger.draftConfig?.path ?? '',
+ trigger.draftConfig ?? {},
+ !trigger.isDraft,
+ workspaceId,
+ usedTriggerKinds
+ ),
+ postgres: (trigger: Trigger) =>
+ savePostgresTriggerFromCfg(
+ trigger.path ?? trigger.draftConfig?.path ?? '',
+ trigger.draftConfig ?? {},
+ !trigger.isDraft,
+ workspaceId,
+ usedTriggerKinds
+ ),
+ kafka: (trigger: Trigger) =>
+ saveKafkaTriggerFromCfg(
+ trigger.path ?? trigger.draftConfig?.path ?? '',
+ trigger.draftConfig ?? {},
+ !trigger.isDraft,
+ workspaceId,
+ usedTriggerKinds
+ ),
+ nats: (trigger: Trigger) =>
+ saveNatsTriggerFromCfg(
+ trigger.path ?? trigger.draftConfig?.path ?? '',
+ trigger.draftConfig ?? {},
+ !trigger.isDraft,
+ workspaceId,
+ usedTriggerKinds
+ ),
+ mqtt: (trigger: Trigger) =>
+ saveMqttTriggerFromCfg(
+ trigger.path ?? trigger.draftConfig?.path ?? '',
+ trigger.draftConfig ?? {},
+ !trigger.isDraft,
+ workspaceId,
+ usedTriggerKinds
+ ),
+ sqs: (trigger: Trigger) =>
+ saveSqsTriggerFromCfg(
+ trigger.path ?? trigger.draftConfig?.path ?? '',
+ trigger.draftConfig ?? {},
+ !trigger.isDraft,
+ workspaceId,
+ usedTriggerKinds
+ ),
+ gcp: (trigger: Trigger) =>
+ saveGcpTriggerFromCfg(
+ trigger.path ?? trigger.draftConfig?.path ?? '',
+ trigger.draftConfig ?? {},
+ !trigger.isDraft,
+ workspaceId,
+ usedTriggerKinds
+ ),
+ poll: undefined,
+ cli: undefined
+ }
+
+ await Promise.all(
+ triggersToDeploy.map(async (trigger) => {
+ const saveFunction = triggerSaveFunctions[trigger.type]
+ if (saveFunction) {
+ await saveFunction(trigger)
+ } else {
+ console.warn(`No save function defined for trigger type: ${trigger.type}`)
+ }
+ })
+ )
+}
+
+export function handleSelectTriggerFromKind(
+ triggersState: Triggers,
+ triggersCountStore: Writable,
+ initialPath: string | undefined,
+ triggerKind: TriggerKind
+) {
+ const triggerType = triggerKindToTriggerType(triggerKind)
+
+ if (!triggerType) {
+ return
+ }
+
+ const existingTriggerIndex = triggersState.triggers.findIndex(
+ (trigger) => trigger.type === triggerType
+ )
+
+ if (existingTriggerIndex !== -1) {
+ triggersState.selectedTriggerIndex = existingTriggerIndex
+ } else {
+ const newTrigger = triggersState.addDraftTrigger(
+ triggersCountStore,
+ triggerType,
+ triggerType === 'schedule' ? initialPath : undefined
+ )
+ triggersState.selectedTriggerIndex = newTrigger
+ }
+}
+
+export function handleConfigChange(
+ nCfg: Record,
+ initialConfig: Record | undefined,
+ saveDisabled: boolean,
+ edit: boolean,
+ onConfigChange?: (cfg: Record, saveDisabled: boolean, updated: boolean) => void
+) {
+ let updated = false
+ if (!edit || !initialConfig) {
+ updated = true
+ } else {
+ // We ignore changes to enabled
+ let newCfg = { ...nCfg }
+ if ('enabled' in newCfg) {
+ delete newCfg.enabled
+ }
+ let initialCfg = { ...initialConfig }
+ if ('enabled' in initialCfg) {
+ delete initialCfg.enabled
+ }
+ if (JSON.stringify(newCfg) !== JSON.stringify(initialCfg)) {
+ updated = true
+ }
+ }
+
+ onConfigChange?.(nCfg, saveDisabled, updated)
+}
+
+export function getLightConfig(
+ triggerType: TriggerType,
+ trigger: Record
+): Record | undefined {
+ if (triggerType === 'schedule') {
+ return { schedule: trigger.schedule, enable: trigger.enable, summary: trigger.summary }
+ } else if (triggerType === 'http') {
+ return { route_path: trigger.route_path, http_method: trigger.http_method }
+ } else if (triggerType === 'websocket') {
+ return { url: trigger.url }
+ } else if (triggerType === 'postgres') {
+ return { postgres_resource_path: trigger.postgres_resource_path }
+ } else if (triggerType === 'kafka') {
+ return { kafka_resource_path: trigger.kafka_resource_path, topics: trigger.topics }
+ } else if (triggerType === 'nats') {
+ return { nats_resource_path: trigger.nats_resource_path, subjects: trigger.subjects }
+ } else if (triggerType === 'mqtt') {
+ return {
+ mqtt_resource_path: trigger.mqtt_resource_path,
+ subscribe_topics: trigger.subscribe_topics
+ }
+ } else if (triggerType === 'sqs') {
+ return { queue_url: trigger.queue_url }
+ } else if (triggerType === 'gcp') {
+ return { gcp_resource_path: trigger.gcp_resource_path, topic: trigger.topic }
+ } else {
+ return undefined
+ }
+}
+
+export function getTriggerLabel(trigger: Trigger): string {
+ const { type, isDraft, draftConfig, lightConfig, path } = trigger
+ const config = draftConfig ?? lightConfig
+
+ if (type === 'webhook') {
+ return 'Webhook'
+ } else if (type === 'email') {
+ return 'Email'
+ } else if (type === 'cli') {
+ return 'CLI'
+ } else if (type === 'http' && !emptyString(config?.route_path)) {
+ return `${(draftConfig?.http_method ?? lightConfig?.http_method ?? 'post').toUpperCase()} ${draftConfig?.route_path ?? lightConfig?.route_path}`
+ } else if (type === 'schedule' && config?.summary) {
+ return `${config?.summary}`
+ } else if (type === 'kafka' && config?.topics && config?.kafka_resource_path) {
+ return `${config?.kafka_resource_path} - ${config?.topics.join(', ')}`
+ } else if (type === 'nats' && config?.subjects && config?.nats_resource_path) {
+ return `${config?.nats_resource_path} - ${config?.subjects.join(', ')}`
+ } else if (type === 'mqtt' && config?.subscribe_topics && config?.mqtt_resource_path) {
+ const topics = config?.subscribe_topics.map((topic: any) => topic.topic).join(', ')
+ return `${config?.mqtt_resource_path} - ${topics}`
+ } else if (type === 'sqs' && config?.queue_url) {
+ return `${config?.queue_url}`
+ } else if (type === 'gcp' && config?.gcp_resource_path && config?.topic) {
+ return `${config?.gcp_resource_path} - ${config?.topic}`
+ } else if (type === 'websocket' && config?.url) {
+ return `${config?.url}`
+ } else if (isDraft && draftConfig?.path) {
+ return `${draftConfig?.path}`
+ } else if (isDraft) {
+ return `New ${type.replace(/s$/, '')} trigger`
+ } else {
+ return path ?? ''
+ }
+}
+
+export function sortTriggers(triggers: Trigger[]): Trigger[] {
+ const triggerTypeOrder = [
+ 'webhook',
+ 'cli',
+ 'email',
+ 'poll',
+ 'schedule',
+ 'http',
+ 'websocket',
+ 'postgres',
+ 'kafka',
+ 'nats',
+ 'mqtt',
+ 'sqs',
+ 'gcp'
+ ]
+
+ return triggers.sort((a, b) => {
+ // Draft triggers always come last
+ if (a.isDraft && !b.isDraft) return 1
+ if (!a.isDraft && b.isDraft) return -1
+
+ // If both are drafts or both are not drafts, sort by type order
+ if (a.isDraft === b.isDraft) {
+ const aIndex = triggerTypeOrder.indexOf(a.type)
+ const bIndex = triggerTypeOrder.indexOf(b.type)
+
+ // If both types are in the order array, sort by their position
+ if (aIndex >= 0 && bIndex >= 0) {
+ return aIndex - bIndex
+ }
+
+ // If only one type is in the order array, it comes first
+ if (aIndex >= 0) return -1
+ if (bIndex >= 0) return 1
+
+ // If neither type is in the order array, maintain original order
+ return 0
+ }
+
+ return 0
+ })
+}
+
+export type FlowWithDraftAndDraftTriggers = Flow & {
+ draft?: Flow & {
+ draft_triggers?: Trigger[]
+ }
+}
+
+export type NewScriptWithDraftAndDraftTriggers = NewScript & {
+ draft?: NewScript & { draft_triggers?: Trigger[] }
+ hash: string
+}
+
+// Get rid of deployed triggers from the saved flow in the case there is a match with a deployed trigger
+export function filterDraftTriggers(
+ savedValue: FlowWithDraftAndDraftTriggers | NewScriptWithDraftAndDraftTriggers,
+ triggersState: Triggers
+): FlowWithDraftAndDraftTriggers | NewScriptWithDraftAndDraftTriggers {
+ const deployedTriggers = triggersState.triggers.filter((t) => !t.draftConfig && !t.isDraft)
+ let newSavedValue = savedValue
+
+ const filterMatchingTriggers = (savedTriggers: Trigger[], deployedTriggers: Trigger[]) => {
+ return savedTriggers.filter(
+ (savedTrigger) =>
+ !deployedTriggers.some(
+ (deployedTrigger) =>
+ deployedTrigger.path === savedTrigger.draftConfig?.path &&
+ deployedTrigger.type === savedTrigger.type
+ )
+ )
+ }
+
+ const savedDraftTriggersFiltered = filterMatchingTriggers(
+ newSavedValue?.draft?.draft_triggers ?? [],
+ deployedTriggers
+ )
+ if (newSavedValue?.draft?.draft_triggers) {
+ newSavedValue = {
+ ...newSavedValue,
+ draft: {
+ ...newSavedValue.draft,
+ draft_triggers:
+ savedDraftTriggersFiltered.length > 0 ? savedDraftTriggersFiltered : undefined
+ }
+ } as typeof newSavedValue
+ }
+ triggersState.setTriggers([
+ ...triggersState.triggers.filter((t) => !t.draftConfig),
+ ...savedDraftTriggersFiltered
+ ])
+ return newSavedValue
+}
diff --git a/frontend/src/lib/components/triggers/webhook/WebhooksCapture.svelte b/frontend/src/lib/components/triggers/webhook/WebhooksCapture.svelte
new file mode 100644
index 0000000000..b6392bf41b
--- /dev/null
+++ b/frontend/src/lib/components/triggers/webhook/WebhooksCapture.svelte
@@ -0,0 +1,80 @@
+
+
+{#if captureInfo}
+
+ {#snippet description()}
+ {#if captureInfo.active}
+
+ Send a POST request to the URL below to simulate a webhook event.
+
+ {:else}
+
+ Start capturing to listen to webhook events on this test URL.
+
+ {/if}
+ {/snippet}
+
+
+
+{/if}
diff --git a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte
index bd8bb7d044..1ee21df607 100644
--- a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte
+++ b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte
@@ -21,9 +21,6 @@
import { workspaceStore, userStore } from '$lib/stores'
import UserSettings from '../../UserSettings.svelte'
import { generateRandomString } from '$lib/utils'
- import CopyableCodeBlock from '../../details/CopyableCodeBlock.svelte'
- import CaptureSection, { type CaptureInfo } from '../CaptureSection.svelte'
- import CaptureTable from '../CaptureTable.svelte'
export let isFlow: boolean = false
export let path: string = ''
@@ -32,9 +29,6 @@
export let runnableArgs: any
export let triggerTokens: TriggerTokens | undefined = undefined
export let scopes: string[] = []
- export let showCapture: boolean = false
- export let captureTable: CaptureTable | undefined = undefined
- export let captureInfo: CaptureInfo | undefined = undefined
let webhooks: {
async: {
@@ -198,17 +192,6 @@ function waitForJobCompletion(UUID) {
return `${mainFunction}\n\n${triggerJobFunction}\n\n${waitForJobCompletionFunction}`
}
- let captureUrl = `${location.origin}/api/w/${$workspaceStore}/capture_u/webhook/${
- isFlow ? 'flow' : 'script'
- }/${path}`
-
- function captureCurlCode() {
- return `curl \\
--X POST ${captureUrl} \\
--H 'Content-Type: application/json' \\
--d '${JSON.stringify(cleanedRunnableArgs ?? {}, null, 2)}'`
- }
-
function curlCode() {
return `TOKEN='${token}'
${requestType !== 'get_path' ? `BODY='${JSON.stringify(cleanedRunnableArgs ?? {})}'` : ''}
@@ -263,195 +246,162 @@ done`
{scopes}
/>
-
- {#if showCapture && captureInfo}
-
-
-
-
-
-
+ {#if SCRIPT_VIEW_SHOW_CREATE_TOKEN_BUTTON}
+
+
+ userSettings.openDrawer()}>
+ Create a Webhook-specific Token
+
+ The token will have a scope such that it can only be used to trigger this script. It is
+ safe to share as it cannot be used to impersonate you.
+
+
+
+
{/if}
-
- {#if SCRIPT_VIEW_SHOW_CREATE_TOKEN_BUTTON}
-
-
- userSettings.openDrawer()}
- >
- Create a Webhook-specific Token
-
- The token will have a scope such that it can only be used to trigger this script. It
- is safe to share as it cannot be used to impersonate you.
-
-
-
-
- {/if}
-
-
-
-
-
Call method
-
-
-
-
-
Token configuration
-
-
-
+
+
-
-
-
-
- REST
- {#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL}
- Curl
+
+
Call method
+
+ Fetch
-
- {#key token}
-
-
-
-
+
+
Token configuration
+
+
+
+
+ REST
+ {#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL}
+ Curl
+ {/if}
+ Fetch
+
+
+ {#key token}
+
+
+
+
+
+ {#if requestType !== 'get_path'}
+
+
-
- {#if requestType !== 'get_path'}
-
-
+
- {/if}
- {#key requestType}
- {#key tokenType}
-
-
- {/key}
{/key}
-
-
-
-
- {#key runnableArgs}
- {#key requestType}
- {#key webhookType}
- {#key tokenType}
-
{
- e.preventDefault()
- copyToClipboard(curlCode())
- }}
- >
-
- {/key}
- {/key}
- {/key}
- {/key}
-
-
+ {/key}
+
+
{#key runnableArgs}
{#key requestType}
{#key webhookType}
{#key tokenType}
- {#key token}
-
{
- e.preventDefault()
- copyToClipboard(fetchCode())
- }}
- >
-
- {/key}{/key}{/key}{/key}
+
{
+ e.preventDefault()
+ copyToClipboard(curlCode())
+ }}
+ >
+
+ {/key}
+ {/key}
+ {/key}
{/key}
-
- {/key}
-
-
-
+
+
+ {#key runnableArgs}
+ {#key requestType}
+ {#key webhookType}
+ {#key tokenType}
+ {#key token}
+ {
+ e.preventDefault()
+ copyToClipboard(fetchCode())
+ }}
+ >
+
+ {/key}{/key}{/key}{/key}
+ {/key}
+
+ {/key}
+
+
+
diff --git a/frontend/src/lib/components/triggers/webhook/WebhooksPanel.svelte b/frontend/src/lib/components/triggers/webhook/WebhooksPanel.svelte
index 485a433e97..5893ee2a17 100644
--- a/frontend/src/lib/components/triggers/webhook/WebhooksPanel.svelte
+++ b/frontend/src/lib/components/triggers/webhook/WebhooksPanel.svelte
@@ -2,30 +2,40 @@
import { Alert } from '$lib/components/common'
import Description from '$lib/components/Description.svelte'
import HighlightTheme from '$lib/components/HighlightTheme.svelte'
- import TriggersEditorSection from '../TriggersEditorSection.svelte'
+ import WebhooksConfigSection from './WebhooksConfigSection.svelte'
+ import { Section } from '$lib/components/common'
- export let token: string
- export let args: Record
= {}
- export let scopes: string[] = []
- export let isFlow: boolean = false
- export let hash: string | undefined = undefined
- export let path: string
- export let newItem: boolean = false
- export let isEditor: boolean = false
- export let canHavePreprocessor: boolean = false
- export let hasPreprocessor: boolean = false
+ interface Props {
+ token: string
+ args?: Record
+ scopes?: string[]
+ isFlow?: boolean
+ hash?: string | undefined
+ path: string
+ newItem?: boolean
+ }
- $: data = {
+ let {
+ token,
+ args = {},
+ scopes = [],
+ isFlow = false,
+ hash = undefined,
+ path,
+ newItem = false
+ }: Props = $props()
+
+ let data = $derived({
hash,
token,
scopes,
args
- }
+ })
+
Webhooks trigger scripts or flows via HTTP requests. Each webhook can be configured to run
synchronously or asynchronously. You can secure webhooks using tokens with specific permissions.
@@ -37,22 +47,12 @@
{/if}
-
+
diff --git a/frontend/src/lib/components/triggers/websocket/WebsocketCapture.svelte b/frontend/src/lib/components/triggers/websocket/WebsocketCapture.svelte
new file mode 100644
index 0000000000..ba64404e12
--- /dev/null
+++ b/frontend/src/lib/components/triggers/websocket/WebsocketCapture.svelte
@@ -0,0 +1,55 @@
+
+
+{#if captureInfo}
+
+
+ {#if captureInfo.active}
+ {#if captureInfo.connectionInfo?.connected}
+
+ Listenning to websocket events...
+
+ {:else}
+
+ Connecting to websocket...
+
+ {/if}
+ {:else}
+
+ Start capturing to listen to websocket events.
+
+ {/if}
+
+
+{/if}
diff --git a/frontend/src/lib/components/triggers/websocket/WebsocketEditorConfigSection.svelte b/frontend/src/lib/components/triggers/websocket/WebsocketEditorConfigSection.svelte
index 4f773de7ed..a229b57215 100644
--- a/frontend/src/lib/components/triggers/websocket/WebsocketEditorConfigSection.svelte
+++ b/frontend/src/lib/components/triggers/websocket/WebsocketEditorConfigSection.svelte
@@ -5,23 +5,20 @@
import ScriptPicker from '$lib/components/ScriptPicker.svelte'
import Required from '$lib/components/Required.svelte'
import { Loader2 } from 'lucide-svelte'
- import CaptureSection, { type CaptureInfo } from '../CaptureSection.svelte'
- import CaptureTable from '../CaptureTable.svelte'
import { sendUserToast } from '$lib/utils'
import type { Schema } from '$lib/common'
import { FlowService, ScriptService, type Flow, type Script } from '$lib/gen'
import { workspaceStore } from '$lib/stores'
import TestTriggerConnection from '../TestTriggerConnection.svelte'
+ import TestingBadge from '$lib/components/triggers/testingBadge.svelte'
export let url: string | undefined
export let url_runnable_args: Record | undefined
export let dirtyUrl: boolean = false
export let can_write: boolean = false
export let headless: boolean = false
- export let showCapture: boolean = false
- export let captureTable: CaptureTable | undefined = undefined
- export let captureInfo: CaptureInfo | undefined = undefined
export let isValid: boolean = false
+ export let showTestingBadge: boolean = false
let areRunnableArgsValid: boolean = true
@@ -35,11 +32,11 @@
? await FlowService.getFlowByPath({
workspace: $workspaceStore!,
path: url.split(':')[1]
- })
+ })
: await ScriptService.getScriptByPath({
workspace: $workspaceStore!,
path: url.split(':')[1]
- })
+ })
urlRunnableSchema = scriptOrFlow.schema as Schema
} catch (err) {
sendUserToast(
@@ -81,20 +78,12 @@
- {#if showCapture && captureInfo}
-
- {/if}
+
+ {#if showTestingBadge}
+
-
{#if url?.startsWith('$')}
@@ -126,6 +117,7 @@
const { path, itemKind } = ev.detail
url = `$${itemKind}:${path ?? ''}`
}}
+ disabled={!can_write}
/>
{dirtyUrl ? urlError : ''}
@@ -146,6 +138,7 @@
bind:isValid={areRunnableArgsValid}
shouldHideNoInputs
class="text-xs"
+ disabled={!can_write}
/>
{/key}
{/await}
diff --git a/frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditor.svelte b/frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditor.svelte
index 65c0f7f1b3..d1112ccd20 100644
--- a/frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditor.svelte
+++ b/frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditor.svelte
@@ -1,8 +1,23 @@
{#if open}
-
+
{/if}
diff --git a/frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte
index fea948ee30..fda17690e8 100644
--- a/frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte
+++ b/frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte
@@ -16,45 +16,118 @@
} from '$lib/gen'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, emptySchema, emptyString, sendUserToast } from '$lib/utils'
- import { createEventDispatcher } from 'svelte'
import Section from '$lib/components/Section.svelte'
- import { Loader2, Save, X, Plus } from 'lucide-svelte'
+ import { Loader2, X, Plus } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
import { fade } from 'svelte/transition'
import type { Schema } from '$lib/common'
import JsonEditor from '$lib/components/JsonEditor.svelte'
import Toggle from '$lib/components/Toggle.svelte'
import WebsocketEditorConfigSection from './WebsocketEditorConfigSection.svelte'
+ import type { Snippet } from 'svelte'
- let drawer: Drawer
- let is_flow: boolean = false
- let initialPath = ''
- let edit = true
- let itemKind: 'flow' | 'script' = 'script'
- let script_path = ''
- let initialScriptPath = ''
- let fixedScriptPath = ''
- let path: string = ''
- let pathError = ''
- let url = ''
- let dirtyUrl = false
- let enabled = false
+ import TriggerEditorToolbar from '../TriggerEditorToolbar.svelte'
+ import { saveWebsocketTriggerFromCfg } from './utils'
+ import { handleConfigChange } from '../utils'
+
+ interface Props {
+ useDrawer?: boolean
+ description?: Snippet | undefined
+ hideTarget?: boolean
+ hideTooltips?: boolean
+ useEditButton?: boolean
+ isEditor?: boolean
+ allowDraft?: boolean
+ hasDraft?: boolean
+ isDraftOnly?: boolean
+ isDeployed?: boolean
+ cloudDisabled?: boolean
+ customLabel?: Snippet
+ onConfigChange?: (cfg: Record
, saveDisabled: boolean, updated: boolean) => void
+ onCaptureConfigChange?: (cfg: Record, isValid: boolean) => void
+ onUpdate?: (path?: string) => void
+ onDelete?: () => void
+ onReset?: () => void
+ }
+
+ let {
+ useDrawer = true,
+ description = undefined,
+ hideTarget = false,
+ hideTooltips = false,
+ isEditor = false,
+ allowDraft = false,
+ hasDraft = false,
+ isDraftOnly = false,
+ isDeployed = false,
+ customLabel = undefined,
+ onConfigChange = undefined,
+ onCaptureConfigChange = undefined,
+ onUpdate = undefined,
+ onDelete = undefined,
+ onReset = undefined,
+ cloudDisabled = false
+ }: Props = $props()
+
+ let drawer: Drawer | undefined = $state()
+ let is_flow: boolean = $state(false)
+ let initialPath = $state('')
+ let edit = $state(true)
+ let itemKind: 'flow' | 'script' = $state('script')
+ let script_path = $state('')
+ let initialScriptPath = $state('')
+ let fixedScriptPath = $state('')
+ let path: string = $state('')
+ let pathError = $state('')
+ let url = $state('')
+ let dirtyUrl = $state(false)
+ let enabled = $state(false)
let filters: {
key: string
value: any
- }[] = []
- let initial_messages: WebsocketTriggerInitialMessage[] = []
- let url_runnable_args: Record | undefined = {}
- let can_return_message = false
- let dirtyPath = false
- let can_write = true
- let drawerLoading = true
+ }[] = $state([])
+ let initial_messages: WebsocketTriggerInitialMessage[] = $state([])
+ let url_runnable_args: Record | undefined = $state({})
+ let can_return_message = $state(false)
+ let dirtyPath = $state(false)
+ let can_write = $state(true)
+ let drawerLoading = $state(true)
+ let showLoading = $state(false)
+ let initialConfig: Record | undefined = undefined
+ let deploymentLoading = $state(false)
+ let isValid = $state(false)
- const dispatch = createEventDispatcher()
+ const websocketCfg = $derived.by(getSaveCfg)
+ const captureConfig = $derived.by(isEditor ? getCaptureConfig : () => ({}))
+ const saveDisabled = $derived.by(() => {
+ const invalidInitialMessages = initial_messages.some((v) => {
+ if ('runnable_result' in v) {
+ return !v.runnable_result.path
+ }
+ return false
+ })
+ return (
+ pathError !== '' ||
+ !isValid ||
+ invalidInitialMessages ||
+ drawerLoading ||
+ !can_write ||
+ emptyString(script_path)
+ )
+ })
- $: is_flow = itemKind === 'flow'
+ $effect(() => {
+ is_flow = itemKind === 'flow'
+ })
- export async function openEdit(ePath: string, isFlow: boolean) {
+ export async function openEdit(
+ ePath: string,
+ isFlow: boolean,
+ defaultConfig?: Record
+ ) {
+ let loadingTimeout = setTimeout(() => {
+ showLoading = true
+ }, 100) // Do not show loading spinner for the first 100ms
drawerLoading = true
try {
drawer?.openDrawer()
@@ -63,11 +136,16 @@
edit = true
dirtyPath = false
dirtyUrl = false
- await loadTrigger()
+ await loadTrigger(defaultConfig)
} catch (err) {
sendUserToast(`Could not load websocket trigger: ${err}`, true)
} finally {
+ if (!defaultConfig) {
+ initialConfig = structuredClone($state.snapshot(getSaveCfg()))
+ }
+ clearTimeout(loadingTimeout)
drawerLoading = false
+ showLoading = false
}
}
@@ -76,6 +154,9 @@
fixedScriptPath_?: string,
defaultValues?: Record
) {
+ let loadingTimeout = setTimeout(() => {
+ showLoading = true
+ }, 100) // Do not show loading spinner for the first 100ms
drawerLoading = true
try {
drawer?.openDrawer()
@@ -87,7 +168,7 @@
initialScriptPath = ''
fixedScriptPath = fixedScriptPath_ ?? ''
script_path = fixedScriptPath
- path = ''
+ path = defaultValues?.path ?? ''
initialPath = ''
filters = []
initial_messages = []
@@ -95,31 +176,55 @@
dirtyPath = false
can_return_message = false
} finally {
+ clearTimeout(loadingTimeout)
drawerLoading = false
+ showLoading = false
}
}
- async function loadTrigger(): Promise {
- const s = await WebsocketTriggerService.getWebsocketTrigger({
- workspace: $workspaceStore!,
- path: initialPath
- })
- script_path = s.script_path
- initialScriptPath = s.script_path
-
- is_flow = s.is_flow
- path = s.path
- url = s.url
- enabled = s.enabled
- filters = s.filters
- initial_messages = s.initial_messages ?? []
- url_runnable_args = s.url_runnable_args
- can_return_message = s.can_return_message
-
- can_write = canWrite(s.path, s.extra_perms, $userStore)
+ function loadTriggerConfig(cfg?: Record): void {
+ script_path = cfg?.script_path
+ initialScriptPath = cfg?.script_path
+ is_flow = cfg?.is_flow
+ path = cfg?.path
+ url = cfg?.url
+ enabled = cfg?.enabled
+ filters = cfg?.filters
+ initial_messages = cfg?.initial_messages ?? []
+ url_runnable_args = cfg?.url_runnable_args
+ can_return_message = cfg?.can_return_message
+ can_write = canWrite(path, cfg?.extra_perms, $userStore)
}
- let initialMessageRunnableSchemas: Record = {}
+ function getSaveCfg() {
+ return {
+ script_path,
+ initialScriptPath,
+ is_flow,
+ path,
+ url,
+ filters,
+ initial_messages,
+ url_runnable_args,
+ can_return_message,
+ enabled
+ }
+ }
+
+ async function loadTrigger(defaultConfig?: Record): Promise {
+ if (defaultConfig) {
+ loadTriggerConfig(defaultConfig)
+ return
+ } else {
+ const s = await WebsocketTriggerService.getWebsocketTrigger({
+ workspace: $workspaceStore!,
+ path: initialPath
+ })
+ loadTriggerConfig(s)
+ }
+ }
+
+ let initialMessageRunnableSchemas: Record = $state({})
async function loadInitialMessageRunnableSchemas(
initialMessageRunnables: {
path: string
@@ -147,133 +252,152 @@
}
}
}
- $: initialMessageRunnables = initial_messages
- .map((v) => ('runnable_result' in v ? v.runnable_result : undefined))
- .filter((v): v is { path: string; is_flow: boolean; args: ScriptArgs } => !!v)
- $: loadInitialMessageRunnableSchemas(initialMessageRunnables)
-
- $: invalidInitialMessages = initial_messages.some((v) => {
- if ('runnable_result' in v) {
- return !v.runnable_result.path
- }
- return false
+ let initialMessageRunnables = $derived(
+ initial_messages
+ .map((v) => ('runnable_result' in v ? v.runnable_result : undefined))
+ .filter((v): v is { path: string; is_flow: boolean; args: ScriptArgs } => !!v)
+ )
+ $effect(() => {
+ loadInitialMessageRunnableSchemas(initialMessageRunnables)
})
async function updateTrigger(): Promise {
- if (edit) {
- await WebsocketTriggerService.updateWebsocketTrigger({
- workspace: $workspaceStore!,
- path: initialPath,
- requestBody: {
- path,
- script_path,
- is_flow,
- url,
- filters,
- initial_messages,
- url_runnable_args,
- can_return_message
- }
- })
- sendUserToast(`Websocket trigger ${path} updated`)
- } else {
- await WebsocketTriggerService.createWebsocketTrigger({
- workspace: $workspaceStore!,
- requestBody: {
- path,
- script_path,
- is_flow,
- url,
- enabled: true,
- filters,
- initial_messages,
- url_runnable_args,
- can_return_message
- }
- })
- sendUserToast(`Websocket trigger ${path} created`)
+ deploymentLoading = true
+ const saveCfg = getSaveCfg()
+ const isSaved = await saveWebsocketTriggerFromCfg(
+ initialPath,
+ saveCfg,
+ edit,
+ $workspaceStore!,
+ usedTriggerKinds
+ )
+ if (isSaved) {
+ onUpdate?.(saveCfg.path)
+ drawer?.closeDrawer()
}
- if (!$usedTriggerKinds.includes('ws')) {
- $usedTriggerKinds = [...$usedTriggerKinds, 'ws']
- }
- dispatch('update')
- drawer.closeDrawer()
+ deploymentLoading = false
}
- let isValid = false
+ function getCaptureConfig() {
+ return {
+ url,
+ url_runnable_args,
+ path
+ }
+ }
+
+ async function handleToggleEnabled(newEnabled: boolean) {
+ enabled = newEnabled
+ if (!isDraftOnly && !hasDraft) {
+ await WebsocketTriggerService.setWebsocketTriggerEnabled({
+ path: initialPath,
+ workspace: $workspaceStore!,
+ requestBody: { enabled: newEnabled }
+ })
+ sendUserToast(`${newEnabled ? 'enabled' : 'disabled'} websocket trigger ${initialPath}`)
+ }
+ }
+
+ $effect(() => {
+ onCaptureConfigChange?.(captureConfig, isValid)
+ })
+
+ $effect(() => {
+ if (!drawerLoading) {
+ handleConfigChange(websocketCfg, initialConfig, saveDisabled, edit, onConfigChange)
+ }
+ })
-
-
-
- {#if !drawerLoading}
- {#if edit}
-
- {
- await WebsocketTriggerService.setWebsocketTriggerEnabled({
- path: initialPath,
- workspace: $workspaceStore ?? '',
- requestBody: { enabled: e.detail }
- })
- sendUserToast(
- `${e.detail ? 'enabled' : 'disabled'} websocket trigger ${initialPath}`
- )
- }}
- />
-
- {/if}
- {#if can_write}
-
- Save
-
- {/if}
+{#if useDrawer}
+
+
+
+ {@render actionsButtons()}
+
+ {@render config()}
+
+
+{:else}
+
+
+ {#if customLabel}
+ {@render customLabel()}
{/if}
- {#if drawerLoading}
-
- {#if edit}
- Changes can take up to 30 seconds to take effect.
- {:else}
- New WebSocket triggers can take up to 30 seconds to start listening.
- {/if}
-
-
-
+
+ {@render actionsButtons()}
+
+ {@render config()}
+
+{/if}
+{#snippet actionsButtons()}
+ {#if !drawerLoading}
+
+ {/if}
+{/snippet}
+
+{#snippet config()}
+ {#if drawerLoading}
+ {#if showLoading}
+
+ {/if}
+ {:else}
+
+ {#if description}
+ {@render description()}
+ {/if}
+ {#if !hideTooltips}
+
+ {#if edit}
+ Changes can take up to 30 seconds to take effect.
+ {:else}
+ New WebSocket triggers can take up to 30 seconds to start listening.
+ {/if}
+
+ {/if}
+
+
+
+
+ {#if !hideTarget}
@@ -295,9 +419,12 @@
btnClasses="ml-4 mt-2"
color="dark"
size="xs"
+ disabled={!can_write}
href={itemKind === 'flow' ? '/flows/add?hub=64' : '/scripts/add?hub=hub%2F19660'}
- target="_blank">Create from template
+ Create from template
+
{/if}
@@ -312,230 +439,244 @@
rightTooltip:
'Whether the runnable result should be sent as a message to the websocket server when not null.'
}}
+ disabled={!can_write}
/>
+ {/if}
-
+
-
-
- Initial messages are sent at the beginning of the connection. They are sent in order.
-
- {#each initial_messages as v, i}
-
-
-
-
- Type
- {
- if (e.target?.['value'] === 'raw_message') {
- initial_messages[i] = {
- raw_message: '""'
- }
- } else {
- initial_messages[i] = {
- runnable_result: {
- path: '',
- args: {},
- is_flow: false
- }
- }
+
+
+ Initial messages are sent at the beginning of the connection. They are sent in order.
+
+ {#each initial_messages as v, i}
+
+
+
+
+ Type
+ {
+ if (e.target?.['value'] === 'raw_message') {
+ initial_messages[i] = {
+ raw_message: '""'
}
- }}
- value={'runnable_result' in v ? 'runnable_result' : 'raw_message'}
- >
- Raw message
- Runnable result
-
-
-
- {#if 'raw_message' in v}
-
-
- Raw JSON message (if a string, wrapping quotes will be discarded)
-
-
{
- const { code } = ev.detail
- v = {
- raw_message: code
- }
- }}
- code={v.raw_message}
- />
-
- {:else if 'runnable_result' in v}
-
-
Runnable
-
{
- const { path, itemKind } = ev.detail
- v = {
+ } else {
+ initial_messages[i] = {
runnable_result: {
- path: path ?? '',
+ path: '',
args: {},
- is_flow: itemKind === 'flow'
+ is_flow: false
}
}
- }}
- />
-
- {#if v.runnable_result?.path}
- {@const schema =
- initialMessageRunnableSchemas[
- v.runnable_result.is_flow
- ? 'flow/' + v.runnable_result.path
- : v.runnable_result.path
- ]}
- {#if schema}
- Arguments
- {#await import('$lib/components/SchemaForm.svelte')}
- This runnable takes no arguments
- {/if}
- {:else}
- Raw message
+ Runnable result
+
+
+
+ {#if 'raw_message' in v}
+
+
+ Raw JSON message (if a string, wrapping quotes will be discarded)
- {:else}
- Unknown type
- {/if}
-
-
{
- initial_messages = initial_messages.filter((_, index) => index !== i)
- }}
- >
-
-
- {/each}
-
-
- {
- if (initial_messages == undefined || !Array.isArray(initial_messages)) {
- initial_messages = []
- }
- initial_messages = initial_messages.concat({
- raw_message: '""'
- })
- }}
- startIcon={{ icon: Plus }}
- >
- Add item
-
-
-
-
-
-
-
- Filters will limit the execution of the trigger to only messages that match all
- criteria.
-
- {#each filters as v, i}
-
-
-
-
- Type
- {
- if (e.target?.['value']) {
- filters[i] = {
- key: '',
- value: ''
- }
- }
- }}
- value={'json'}
- >
- JSON
-
-
+
{
+ const { code } = ev.detail
+ initial_messages[i] = {
+ raw_message: code
+ }
+ }}
+ code={v.raw_message}
+ disabled={!can_write}
+ />
+ {:else if 'runnable_result' in v}
+
+
Runnable
+
{
+ const { path, itemKind } = ev.detail
+ initial_messages[i] = {
+ runnable_result: {
+ path: path ?? '',
+ args: {},
+ is_flow: itemKind === 'flow'
+ }
+ }
+ }}
+ disabled={!can_write}
+ />
+
+ {#if v.runnable_result?.path}
+ {@const schema =
+ initialMessageRunnableSchemas[
+ v.runnable_result.is_flow
+ ? 'flow/' + v.runnable_result.path
+ : v.runnable_result.path
+ ]}
+ {#if schema}
+ Arguments
+ {#await import('$lib/components/SchemaForm.svelte')}
+ This runnable takes no arguments
+ {/if}
+ {:else}
+
+ {:else}
+ Unknown type
+ {/if}
+
+
{
+ initial_messages = initial_messages.filter((_, index) => index !== i)
+ }}
+ disabled={!can_write}
+ >
+
+
+ {/each}
+
+
+ {
+ if (initial_messages == undefined || !Array.isArray(initial_messages)) {
+ initial_messages = []
+ }
+ initial_messages = initial_messages.concat({
+ raw_message: '""'
+ })
+ }}
+ disabled={!can_write}
+ startIcon={{ icon: Plus }}
+ >
+ Add item
+
+
+
+
+
+
+ Filters will limit the execution of the trigger to only messages that match all criteria.
+
+ {#each filters as v, i}
+
+
+
- Key
-
-
-
- Value
- Type
+ {
+ if (e.target?.['value']) {
+ filters[i] = {
+ key: '',
+ value: ''
+ }
+ }
+ }}
+ value={'json'}
+ disabled={!can_write}
+ >
+ JSON
+
-
{
- filters = filters.filter((_, index) => index !== i)
- }}
- >
-
+
+ Key
+
+
+
+ Value
+
- {/each}
-
-
- {
- if (filters == undefined || !Array.isArray(filters)) {
- filters = []
- }
- filters = filters.concat({
- key: '',
- value: ''
- })
+ {
+ filters = filters.filter((_, index) => index !== i)
}}
- startIcon={{ icon: Plus }}
+ disabled={!can_write}
>
- Add item
-
+
+ {/each}
+
+
+ {
+ if (filters == undefined || !Array.isArray(filters)) {
+ filters = []
+ }
+ filters = filters.concat({
+ key: '',
+ value: ''
+ })
+ }}
+ disabled={!can_write}
+ startIcon={{ icon: Plus }}
+ >
+ Add item
+
-
-
- {/if}
-
-
+
+
{
- loadTriggers()
- }}
- bind:this={wsTriggerEditor}
-/>
-
-{#if isCloudHosted()}
-
- WebSocket triggers are disabled in the multi-tenant cloud.
-
-{:else}
-
-
- WebSocket triggers allow real-time bidirectional communication between your scripts/flows and
- external systems. Each trigger creates a unique WebSocket endpoint.
-
-
- {#if !newItem && wsTriggers && wsTriggers.length > 0}
-
-
-
- {#each wsTriggers as wsTriggers (wsTriggers.path)}
-
-
{wsTriggers.path}
-
- {wsTriggers.url}
-
-
- wsTriggerEditor?.openEdit(wsTriggers.path, isFlow)}
- class="px-2"
- >
- {#if wsTriggers.canWrite}
- Edit
- {:else}
- View
- {/if}
-
-
-
- {/each}
-
-
- {/if}
-
-
{
- wsTriggerEditor?.openNew(isFlow, path, e.detail.config)
- }}
- on:addPreprocessor
- on:updateSchema
- on:testWithArgs
- cloudDisabled={false}
- triggerType="websocket"
- {isFlow}
- {path}
- {isEditor}
- {canHavePreprocessor}
- {hasPreprocessor}
- {newItem}
- {openForm}
- />
-
+
+ {#snippet description()}
+ {#if cloudDisabled}
+
+ WebSocket triggers are disabled in the multi-tenant cloud.
+
+ {:else}
+
+ WebSocket triggers allow real-time bidirectional communication between your scripts/flows
+ and external systems. Each trigger creates a unique WebSocket endpoint.
+
+ {/if}
+ {/snippet}
+
+
,
+ edit: boolean,
+ workspace: string,
+ usedTriggerKinds: Writable
+): Promise {
+ const requestBody = {
+ path: triggerCfg.path,
+ script_path: triggerCfg.script_path,
+ is_flow: triggerCfg.is_flow,
+ url: triggerCfg.url,
+ filters: triggerCfg.filters,
+ initial_messages: triggerCfg.initial_messages,
+ url_runnable_args: triggerCfg.url_runnable_args,
+ can_return_message: triggerCfg.can_return_message
+ }
+ try {
+ if (edit) {
+ await WebsocketTriggerService.updateWebsocketTrigger({
+ workspace: workspace,
+ path: initialPath,
+ requestBody: requestBody
+ })
+ sendUserToast(`Websocket trigger ${triggerCfg.path} updated`)
+ } else {
+ await WebsocketTriggerService.createWebsocketTrigger({
+ workspace: workspace,
+ requestBody: { ...requestBody, enabled: true }
+ })
+ sendUserToast(`Websocket trigger ${triggerCfg.path} created`)
+ }
+ if (!get(usedTriggerKinds).includes('ws')) {
+ usedTriggerKinds.update((t) => [...t, 'ws'])
+ }
+ return true
+ } catch (error) {
+ sendUserToast(error.body || error.message, true)
+ return false
+ }
+}
diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts
index 2609eba53d..6c2bde6431 100644
--- a/frontend/src/lib/utils.ts
+++ b/frontend/src/lib/utils.ts
@@ -1161,6 +1161,7 @@ export type Item = {
disabled?: boolean
type?: 'action' | 'delete'
hide?: boolean | undefined
+ extra?: Snippet
}
export function isObjectTooBig(obj: any): boolean {
@@ -1245,6 +1246,15 @@ export function formatDateShort(dateString: string | undefined): string {
}).format(date)
}
+export function toJsonStr(result: any) {
+ try {
+ // console.log(result)
+ return JSON.stringify(result ?? null, null, 4) ?? 'null'
+ } catch (e) {
+ return 'error stringifying object: ' + e.toString()
+ }
+}
+
export function getOS() {
const userAgent = window.navigator.userAgent
const platform = window.navigator.platform
@@ -1268,6 +1278,7 @@ export function getOS() {
import { type ClassValue, clsx } from 'clsx'
import { twMerge } from 'tailwind-merge'
+import type { Snippet } from 'svelte'
export function cn(...inputs: ClassValue[]) {
return twMerge(clsx(inputs))
diff --git a/frontend/src/routes/(root)/(logged)/flows/add/+page.svelte b/frontend/src/routes/(root)/(logged)/flows/add/+page.svelte
index e291917235..056424922b 100644
--- a/frontend/src/routes/(root)/(logged)/flows/add/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/flows/add/+page.svelte
@@ -13,9 +13,9 @@
import { decodeState, emptySchema } from '$lib/utils'
import { tick } from 'svelte'
import { writable } from 'svelte/store'
- import type { ScheduleTrigger } from '$lib/components/triggers'
import type { GetInitialAndModifiedValues } from '$lib/components/common/confirmationModal/unsavedTypes'
import { replaceScriptPlaceholderWithItsValues } from '$lib/hub'
+ import type { Trigger } from '$lib/components/triggers/utils'
let nodraft = $page.url.searchParams.get('nodraft')
@@ -43,6 +43,7 @@
initialArgs = $initialArgsStore
$initialArgsStore = undefined
}
+ let flowBuilder: FlowBuilder | undefined = undefined
export const flowStore = writable({
summary: '',
@@ -56,7 +57,8 @@
})
const flowStateStore = writable({})
- let savedPrimarySchedule: ScheduleTrigger | undefined = undefined
+ let draftTriggersFromUrl: Trigger[] | undefined = undefined
+ let selectedTriggerIndexFromUrl: number | undefined = undefined
async function loadFlow() {
loading = true
let flow: Flow = {
@@ -101,7 +103,10 @@
flow = state.flow
pathStoreInit = state.path
- savedPrimarySchedule = state.primarySchedule
+ draftTriggersFromUrl = state.draft_triggers
+ selectedTriggerIndexFromUrl = state.selected_trigger
+ flowBuilder?.setDraftTriggers(draftTriggersFromUrl)
+ flowBuilder?.setSelectedTriggerIndex(selectedTriggerIndexFromUrl)
state?.selectedId && (selectedId = state?.selectedId)
} else {
if (templatePath) {
@@ -153,7 +158,6 @@
loadFlow()
let getSelectedId: (() => string) | undefined = undefined
- let flowBuilder: FlowBuilder | undefined = undefined
let getInitialAndModifiedValues: GetInitialAndModifiedValues | undefined = undefined
@@ -181,7 +185,8 @@
{flowStateStore}
{selectedId}
{loading}
- {savedPrimarySchedule}
+ {draftTriggersFromUrl}
+ {selectedTriggerIndexFromUrl}
>
{
@@ -82,12 +86,19 @@
})
const draftOrDeployed = cleanValueProperties(savedFlow?.draft || savedFlow)
- const urlScript = cleanValueProperties(stateLoadedFromUrl.flow)
+ const urlScript = cleanValueProperties({
+ ...stateLoadedFromUrl.flow,
+ draft_triggers: stateLoadedFromUrl.draft_triggers
+ })
flow = stateLoadedFromUrl.flow
- savedPrimarySchedule = stateLoadedFromUrl.primarySchedule
+ draftTriggersFromUrl = stateLoadedFromUrl.draft_triggers
+ selectedTriggerIndexFromUrl = stateLoadedFromUrl.selected_trigger
+ flowBuilder?.setDraftTriggers(draftTriggersFromUrl)
+ flowBuilder?.setSelectedTriggerIndex(selectedTriggerIndexFromUrl)
+ const selectedId = stateLoadedFromUrl?.selectedId ?? 'settings-metadata'
const reloadAction = () => {
stateLoadedFromUrl = undefined
- goto(`/flows/edit/${statePath}`)
+ goto(`/flows/edit/${statePath}?selected=${selectedId}`)
loadFlow()
}
if (orderedJsonStringify(draftOrDeployed) === orderedJsonStringify(urlScript)) {
@@ -133,15 +144,18 @@
? {
...structuredClone(flowWithDraft.draft),
path: flowWithDraft.draft.path ?? flowWithDraft.path // backward compatibility for old drafts missing path
- }
+ }
: undefined
} as Flow & {
- draft?: Flow
+ draft?: Flow & {
+ draft_triggers?: Trigger[]
+ }
}
if (flowWithDraft.draft != undefined && !nobackenddraft) {
flow = flowWithDraft.draft
savedPrimarySchedule = flowWithDraft?.draft?.['primary_schedule']
flowBuilder?.setPrimarySchedule(savedPrimarySchedule)
+ flowBuilder?.setDraftTriggers(flowWithDraft?.draft?.['draft_triggers'])
if (!flowWithDraft.draft_only) {
const deployed = cleanValueProperties(flowWithDraft)
@@ -178,6 +192,7 @@
}
} else {
flow = flowWithDraft
+ flowBuilder?.setDraftTriggers(undefined)
}
}
@@ -200,6 +215,7 @@
return
}
diffDrawer.closeDrawer()
+ stateLoadedFromUrl = undefined
goto(`/flows/edit/${savedFlow.draft.path}`)
loadFlow()
}
@@ -217,6 +233,7 @@
path: savedFlow.path
})
}
+ stateLoadedFromUrl = undefined
goto(`/flows/edit/${savedFlow.path}`)
loadFlow()
}
@@ -251,6 +268,8 @@
bind:savedFlow
{diffDrawer}
{savedPrimarySchedule}
+ {draftTriggersFromUrl}
+ {selectedTriggerIndexFromUrl}
bind:version
bind:getInitialAndModifiedValues
>
diff --git a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte b/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte
index a20cfd427a..55bf1d5d1c 100644
--- a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte
@@ -41,35 +41,26 @@
} from 'lucide-svelte'
import DetailPageHeader from '$lib/components/details/DetailPageHeader.svelte'
- import WebhooksPanel from '$lib/components/triggers/webhook/WebhooksPanel.svelte'
- import CliHelpBox from '$lib/components/CliHelpBox.svelte'
import FlowGraphViewer from '$lib/components/FlowGraphViewer.svelte'
- import RunPageSchedules from '$lib/components/RunPageSchedules.svelte'
import { createAppFromFlow } from '$lib/components/details/createAppFromScript'
import { importStore } from '$lib/components/apps/store'
import TimeAgo from '$lib/components/TimeAgo.svelte'
- import ClipboardPanel from '$lib/components/details/ClipboardPanel.svelte'
import FlowGraphViewerStep from '$lib/components/FlowGraphViewerStep.svelte'
import GfmMarkdown from '$lib/components/GfmMarkdown.svelte'
import FlowHistory from '$lib/components/flows/FlowHistory.svelte'
- import EmailTriggerPanel from '$lib/components/details/EmailTriggerPanel.svelte'
import Star from '$lib/components/Star.svelte'
- import RoutesPanel from '$lib/components/triggers/http/RoutesPanel.svelte'
import { Highlight } from 'svelte-highlight'
import json from 'svelte-highlight/languages/json'
import { writable } from 'svelte/store'
- import TriggersBadge from '$lib/components/graph/renderers/triggers/TriggersBadge.svelte'
import InputSelectedBadge from '$lib/components/schema/InputSelectedBadge.svelte'
- import WebsocketTriggersPanel from '$lib/components/triggers/websocket/WebsocketTriggersPanel.svelte'
- import KafkaTriggersPanel from '$lib/components/triggers/kafka/KafkaTriggersPanel.svelte'
- import NatsTriggersPanel from '$lib/components/triggers/nats/NatsTriggersPanel.svelte'
- import PostgresTriggersPanel from '$lib/components/triggers/postgres/PostgresTriggersPanel.svelte'
import Toggle from '$lib/components/Toggle.svelte'
- import MqttTriggersPanel from '$lib/components/triggers/mqtt/MqttTriggersPanel.svelte'
- import SqsTriggerPanel from '$lib/components/triggers/sqs/SqsTriggerPanel.svelte'
- import { onDestroy } from 'svelte'
+ import { onDestroy, tick } from 'svelte'
import LogViewer from '$lib/components/LogViewer.svelte'
- import GcpTriggerPanel from '$lib/components/triggers/gcp/GcpTriggerPanel.svelte'
+ import TriggersEditor from '$lib/components/triggers/TriggersEditor.svelte'
+ import type { TriggerContext } from '$lib/components/triggers'
+ import { setContext } from 'svelte'
+ import TriggersBadge from '$lib/components/graph/renderers/triggers/TriggersBadge.svelte'
+ import { Triggers } from '$lib/components/triggers/triggers.svelte'
let flow: Flow | undefined
let can_write = false
@@ -85,9 +76,29 @@
let intervalId: NodeJS.Timeout | undefined = undefined
+ $: {
+ const cliTrigger = triggersState.triggers.find((t) => t.type === 'cli')
+ if (cliTrigger) {
+ cliTrigger.extra = {
+ cliCommand: `wmill flow run ${flow?.path} -d '${JSON.stringify(args)}'`
+ }
+ }
+ }
+
const triggersCount = writable(undefined)
- $: cliCommand = `wmill flow run ${flow?.path} -d '${JSON.stringify(args)}'`
+ // Add triggers context store
+ const triggersState = new Triggers([
+ { type: 'webhook', path: '', isDraft: false },
+ { type: 'email', path: '', isDraft: false },
+ { type: 'cli', path: '', isDraft: false }
+ ])
+ setContext('TriggerContext', {
+ triggersCount,
+ simplifiedPoll: writable(false),
+ showCaptureHint: writable(undefined),
+ triggersState
+ })
let previousPath: string | undefined = undefined
$: {
@@ -96,6 +107,7 @@
previousPath = path
loadFlow()
loadTriggersCount()
+ loadTriggers()
}
}
}
@@ -124,6 +136,17 @@
})
}
+ async function loadTriggers(): Promise {
+ await triggersState.fetchTriggers(
+ triggersCount,
+ $workspaceStore,
+ path,
+ true,
+ undefined,
+ $userStore
+ )
+ }
+
async function loadFlow(): Promise {
flow = await FlowService.getFlowByPath({
workspace: $workspaceStore!,
@@ -371,7 +394,7 @@
}
}
let stepDetail: FlowModule | string | undefined = undefined
- let token = 'TOKEN_TO_CREATE'
+
let rightPaneSelected = 'saved_inputs'
let savedInputsV2: SavedInputsV2 | undefined = undefined
let flowHistory: FlowHistory | undefined = undefined
@@ -398,7 +421,6 @@
{#if flow}
{
+ onSelect={async (triggerIndex: number) => {
rightPaneSelected = 'triggers'
+ await tick()
+ triggersState.selectedTriggerIndex = triggerIndex
}}
+ small={false}
/>
{#if $workspaceStore}
@@ -590,89 +616,23 @@
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
{#if stepDetail}
+
+
{/if}
diff --git a/frontend/src/routes/(root)/(logged)/gcp_triggers/+page.svelte b/frontend/src/routes/(root)/(logged)/gcp_triggers/+page.svelte
index 9fcaa0f202..2fd36a1cc1 100644
--- a/frontend/src/routes/(root)/(logged)/gcp_triggers/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/gcp_triggers/+page.svelte
@@ -263,7 +263,7 @@
-
+
{
@@ -122,7 +120,6 @@
searchParams={$page.url.searchParams}
{script}
{showMeta}
- {savedPrimarySchedule}
replaceStateFn={(path) => replaceState(path, $page.state)}
>
{
fullyLoaded = false
@@ -98,10 +100,12 @@
savedScript = structuredClone(scriptWithDraft)
if (scriptWithDraft.draft != undefined) {
script = scriptWithDraft.draft
+ scriptBuilder?.setDraftTriggers(script.draft_triggers)
if (script['primary_schedule']) {
savedPrimarySchedule = script['primary_schedule']
scriptBuilder?.setPrimarySchedule(savedPrimarySchedule)
}
+
if (!scriptWithDraft.draft_only) {
reloadAction = async () => {
scriptLoadedFromUrl = undefined
@@ -153,6 +157,7 @@
if (script) {
initialPath = script.path
+ scriptBuilder?.setDraftTriggers(script.draft_triggers)
scriptBuilder?.setCode(script.content)
if (topHash) {
script.parent_hash = topHash
@@ -176,6 +181,7 @@
}
diffDrawer.closeDrawer()
goto(`/scripts/edit/${savedScript.draft.path}`)
+ scriptLoadedFromUrl = undefined
loadScript()
}
@@ -193,6 +199,7 @@
})
}
goto(`/scripts/edit/${savedScript.path}`)
+ scriptLoadedFromUrl = undefined
loadScript()
}
diff --git a/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte b/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte
index 0a7d41a4dc..dd4f7fa15f 100644
--- a/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte
@@ -21,7 +21,7 @@
import { enterpriseLicense, hubBaseUrlStore, userStore, workspaceStore } from '$lib/stores'
import { isDeployable, ALL_DEPLOYABLE } from '$lib/utils_deployable'
- import { onDestroy } from 'svelte'
+ import { onDestroy, setContext, tick } from 'svelte'
import HighlightCode from '$lib/components/HighlightCode.svelte'
import {
Tabs,
@@ -42,10 +42,8 @@
import DeployWorkspaceDrawer from '$lib/components/DeployWorkspaceDrawer.svelte'
import SavedInputsV2 from '$lib/components/SavedInputsV2.svelte'
- import WebhooksPanel from '$lib/components/triggers/webhook/WebhooksPanel.svelte'
import DetailPageLayout from '$lib/components/details/DetailPageLayout.svelte'
import DetailPageHeader from '$lib/components/details/DetailPageHeader.svelte'
- import CliHelpBox from '$lib/components/CliHelpBox.svelte'
import {
Activity,
Archive,
@@ -68,30 +66,22 @@
import { scriptToHubUrl } from '$lib/hub'
import SharedBadge from '$lib/components/SharedBadge.svelte'
import ScriptVersionHistory from '$lib/components/ScriptVersionHistory.svelte'
- import RunPageSchedules from '$lib/components/RunPageSchedules.svelte'
import { createAppFromScript } from '$lib/components/details/createAppFromScript'
import { importStore } from '$lib/components/apps/store'
import TimeAgo from '$lib/components/TimeAgo.svelte'
- import ClipboardPanel from '$lib/components/details/ClipboardPanel.svelte'
import PersistentScriptDrawer from '$lib/components/PersistentScriptDrawer.svelte'
import GfmMarkdown from '$lib/components/GfmMarkdown.svelte'
- import EmailTriggerPanel from '$lib/components/details/EmailTriggerPanel.svelte'
import Star from '$lib/components/Star.svelte'
import LogViewer from '$lib/components/LogViewer.svelte'
- import RoutesPanel from '$lib/components/triggers/http/RoutesPanel.svelte'
import { Highlight } from 'svelte-highlight'
import json from 'svelte-highlight/languages/json'
import { writable } from 'svelte/store'
- import TriggersBadge from '$lib/components/graph/renderers/triggers/TriggersBadge.svelte'
- import WebsocketTriggersPanel from '$lib/components/triggers/websocket/WebsocketTriggersPanel.svelte'
- import KafkaTriggersPanel from '$lib/components/triggers/kafka/KafkaTriggersPanel.svelte'
- import NatsTriggersPanel from '$lib/components/triggers/nats/NatsTriggersPanel.svelte'
- import PostgresTriggersPanel from '$lib/components/triggers/postgres/PostgresTriggersPanel.svelte'
import Toggle from '$lib/components/Toggle.svelte'
import InputSelectedBadge from '$lib/components/schema/InputSelectedBadge.svelte'
- import MqttTriggersPanel from '$lib/components/triggers/mqtt/MqttTriggersPanel.svelte'
- import SqsTriggerPanel from '$lib/components/triggers/sqs/SqsTriggerPanel.svelte'
- import GcpTriggerPanel from '$lib/components/triggers/gcp/GcpTriggerPanel.svelte'
+ import type { TriggerContext } from '$lib/components/triggers'
+ import TriggersBadge from '$lib/components/graph/renderers/triggers/TriggersBadge.svelte'
+ import TriggersEditor from '$lib/components/triggers/TriggersEditor.svelte'
+ import { Triggers } from '$lib/components/triggers/triggers.svelte'
let script: Script | undefined
let topHash: string | undefined
@@ -107,7 +97,14 @@
let inputSelected: 'saved' | 'history' | undefined = undefined
let jsonView = false
- $: cliCommand = `wmill script run ${script?.path} -d '${JSON.stringify(args)}'`
+ $: {
+ const cliTrigger = triggersState.triggers.find((t) => t.type === 'cli')
+ if (cliTrigger) {
+ cliTrigger.extra = {
+ cliCommand: `wmill script run ${script?.path} -d '${JSON.stringify(args)}'`
+ }
+ }
+ }
$: loading = !script
@@ -121,6 +118,19 @@
const triggersCount = writable(undefined)
+ // Add triggers context store
+ const triggersState = new Triggers([
+ { type: 'webhook', path: '', isDraft: false },
+ { type: 'email', path: '', isDraft: false },
+ { type: 'cli', path: '', isDraft: false }
+ ])
+ setContext('TriggerContext', {
+ triggersCount,
+ simplifiedPoll: writable(false),
+ showCaptureHint: writable(undefined),
+ triggersState
+ })
+
async function deleteScript(hash: string): Promise {
try {
await ScriptService.deleteScriptByHash({ workspace: $workspaceStore!, hash })
@@ -167,11 +177,15 @@
}
let starred: boolean | undefined = undefined
- async function loadTriggersCount(path: string) {
- $triggersCount = await ScriptService.getTriggersCountOfScript({
- workspace: $workspaceStore!,
- path: path
- })
+ async function loadTriggers(path: string): Promise {
+ await triggersState.fetchTriggers(
+ triggersCount,
+ $workspaceStore,
+ path,
+ false,
+ undefined,
+ $userStore
+ )
}
async function loadScript(hash: string): Promise {
@@ -194,7 +208,7 @@
can_write =
script.workspace_id == $workspaceStore &&
canWrite(script.path, script.extra_perms!, $userStore)
- loadTriggersCount(script.path)
+ loadTriggers(script.path)
if (script.path && script.archived) {
const script_by_path = await ScriptService.getScriptByPath({
@@ -497,7 +511,6 @@
}
}
- let token = 'TOKEN_TO_CREATE'
let rightPaneSelected = 'saved_inputs'
let savedInputsV2: SavedInputsV2 | undefined = undefined
@@ -533,11 +546,7 @@
{#key script.hash}
-
+
{
+ rightPaneSelected = 'triggers'
+ }}
>
{
- rightPaneSelected = 'triggers'
+ selected={rightPaneSelected === 'triggers'}
+ onSelect={async (triggerIndex: number) => {
+ if (rightPaneSelected !== 'triggers') {
+ rightPaneSelected = 'triggers'
+ }
+ await tick()
+ triggersState.selectedTriggerIndex = triggerIndex
}}
/>
@@ -740,76 +757,18 @@
/>
{/if}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
@@ -878,12 +837,6 @@
{/if}
-
-
-
-
{/key}
{/if}
diff --git a/frontend/src/routes/(root)/(logged)/sqs_triggers/+page.svelte b/frontend/src/routes/(root)/(logged)/sqs_triggers/+page.svelte
index b5a5522652..f117702fba 100644
--- a/frontend/src/routes/(root)/(logged)/sqs_triggers/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/sqs_triggers/+page.svelte
@@ -220,7 +220,7 @@
-
+
>({})
const selectedIdStore = writable('settings-metadata')
- const primaryScheduleStore = writable(undefined)
const triggersCount = writable(undefined)
- const selectedTriggerStore = writable<
- 'webhooks' | 'emails' | 'schedules' | 'cli' | 'routes' | 'websockets' | 'scheduledPoll'
- >('webhooks')
setContext('TriggerContext', {
- primarySchedule: primaryScheduleStore,
- selectedTrigger: selectedTriggerStore,
triggersCount: triggersCount,
simplifiedPoll: writable(false),
- defaultValues: writable(undefined),
- captureOn: writable(undefined),
- showCaptureHint: writable(undefined)
+ showCaptureHint: writable(undefined),
+ triggersState: new Triggers()
})
setContext('FlowEditorContext', {