diff --git a/frontend/src/lib/components/CronBuilder.svelte b/frontend/src/lib/components/CronBuilder.svelte index bcf8508c12..3bceabb417 100644 --- a/frontend/src/lib/components/CronBuilder.svelte +++ b/frontend/src/lib/components/CronBuilder.svelte @@ -15,7 +15,7 @@ contentClasses="p-4" > {#snippet trigger()} - {/snippet} diff --git a/frontend/src/lib/components/DatatablePicker.svelte b/frontend/src/lib/components/DatatablePicker.svelte index 53617acf36..17ad6f3361 100644 --- a/frontend/src/lib/components/DatatablePicker.svelte +++ b/frontend/src/lib/components/DatatablePicker.svelte @@ -17,6 +17,9 @@ selectInputClass?: string class?: string onClear?: () => void + /** Size the input to its content, like Path's folder picker. */ + useContentEditable?: boolean + RightIcon?: any } let { @@ -27,7 +30,9 @@ placeholder = undefined, selectInputClass = '', class: className = '', - onClear = undefined + onClear = undefined, + useContentEditable = false, + RightIcon = undefined }: Props = $props() let datatables = usePromise(() => @@ -47,6 +52,8 @@ {placeholder} inputClass={selectInputClass} {onClear} + {useContentEditable} + {RightIcon} /> {#if showSchemaExplorer && value && assetCanBeExplored({ kind: 'datatable', path: value })} diff --git a/frontend/src/lib/components/DucklakePicker.svelte b/frontend/src/lib/components/DucklakePicker.svelte index 3a6f2bf739..d6ed4d0cdf 100644 --- a/frontend/src/lib/components/DucklakePicker.svelte +++ b/frontend/src/lib/components/DucklakePicker.svelte @@ -17,6 +17,9 @@ selectInputClass?: string class?: string onClear?: () => void + /** Size the input to its content, like Path's folder picker. */ + useContentEditable?: boolean + RightIcon?: any } let { @@ -27,7 +30,9 @@ placeholder = undefined, selectInputClass = '', class: className = '', - onClear = undefined + onClear = undefined, + useContentEditable = false, + RightIcon = undefined }: Props = $props() let ducklakes = usePromise(() => @@ -45,6 +50,8 @@ {placeholder} inputClass={selectInputClass} {onClear} + {useContentEditable} + {RightIcon} /> {#if showSchemaExplorer && value && assetCanBeExplored({ kind: 'ducklake', path: value })} diff --git a/frontend/src/lib/components/SummaryPathDisplay.svelte b/frontend/src/lib/components/SummaryPathDisplay.svelte index 22925e98e7..ea534bd917 100644 --- a/frontend/src/lib/components/SummaryPathDisplay.svelte +++ b/frontend/src/lib/components/SummaryPathDisplay.svelte @@ -28,6 +28,13 @@ editable?: boolean onSaved?: (newPath: string) => void kind?: 'flow' | 'script' + /** Saves instead of the API, for an item that exists only locally (a draft). + * Returns an error message to keep the popover open. */ + saveOverride?: (next: { + path: string + summary: string + labels: string[] | undefined + }) => string | undefined } let { @@ -37,7 +44,8 @@ inheritedLabels = undefined, editable = false, onSaved, - kind = 'flow' + kind = 'flow', + saveOverride = undefined }: Props = $props() let editSummary = $state('') @@ -47,7 +55,8 @@ let ownPath = $state(undefined) // Derived: ownership answers about the operating workspace, whose user resolves asynchronously. const own = $derived( - ownPath === undefined ? false : isOwner(ownPath, actingUser, $operatingWorkspace) + saveOverride !== undefined || + (ownPath === undefined ? false : isOwner(ownPath, actingUser, $operatingWorkspace)) ) let onBehalfOfEmail = $state(undefined) let summaryInput: ReturnType | undefined = $state() @@ -72,6 +81,17 @@ async function save(close: () => void) { const initialPath = path ?? '' const newPath = own ? editPath : initialPath + if (saveOverride) { + const error = saveOverride({ path: newPath, summary: editSummary, labels }) + if (error) { + sendUserToast(error, true) + return + } + labelsDirty = false + close() + onSaved?.(newPath) + return + } try { await updateItemPathAndSummary({ diff --git a/frontend/src/lib/components/assets/AssetGraph/AddNode.svelte b/frontend/src/lib/components/assets/AssetGraph/AddNode.svelte index 880abb685a..c9fd8111db 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AddNode.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AddNode.svelte @@ -31,7 +31,8 @@ path: string, source: { kind: NativeTriggerKind; path: string | undefined }, outputKind: PipelineOutputKind, - aiPrompt?: string + aiPrompt?: string, + options?: import('./PipelineInsertMenu.svelte').PipelineInsertOptions ) => void pathPrefix: string defaultPathSuffix: string @@ -51,7 +52,8 @@ pick.path, { kind: kindId, path: undefined }, outputKind, - pick.aiPrompt + pick.aiPrompt, + { schedule: pick.schedule, outputAsset: pick.outputAsset } ) } @@ -78,48 +80,56 @@ }, { id: 'email', + configuredAfterCreate: true, label: 'On email', description: 'Triggered by incoming email', icon: Mail }, { id: 'kafka', + configuredAfterCreate: true, label: 'On Kafka', description: 'Triggered by a Kafka message', icon: Zap }, { id: 'mqtt', + configuredAfterCreate: true, label: 'On MQTT', description: 'Triggered by an MQTT message', icon: Radio }, { id: 'amqp', + configuredAfterCreate: true, label: 'On AMQP', description: 'Triggered by an AMQP (RabbitMQ) message', icon: Radio }, { id: 'nats', + configuredAfterCreate: true, label: 'On NATS', description: 'Triggered by a NATS message', icon: MessageSquare }, { id: 'postgres', + configuredAfterCreate: true, label: 'On Postgres', description: 'Triggered by a Postgres event', icon: Database }, { id: 'sqs', + configuredAfterCreate: true, label: 'On SQS', description: 'Triggered by an SQS message', icon: Send }, { id: 'gcp', + configuredAfterCreate: true, label: 'On GCP Pub/Sub', description: 'Triggered by a Pub/Sub message', icon: CloudCog diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte index 27f6ce6be7..e4b15d4e01 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte @@ -49,7 +49,8 @@ language: import('$lib/gen').ScriptLang, scriptPath: string, outputKind: import('./pipelineTemplates').PipelineOutputKind, - aiPrompt?: string + aiPrompt?: string, + options?: import('./PipelineInsertMenu.svelte').PipelineInsertOptions ) => void // Pipeline-wide + node shown at the top of the graph. Picking any // kind from the menu invokes this one callback with the chosen @@ -65,7 +66,8 @@ path: string | undefined }, outputKind: import('./pipelineTemplates').PipelineOutputKind, - aiPrompt?: string + aiPrompt?: string, + options?: import('./PipelineInsertMenu.svelte').PipelineInsertOptions ) => void // Folder-scoped prefix shown as a read-only chip in the insert menu // path input (e.g. `f/{folder}/`). Shared across top + and per-asset +. @@ -526,6 +528,7 @@ data: { runnable_kind: r.usage_kind, path: r.path, + summary: r.summary, in_pipeline: r.in_pipeline ?? false, partition_kind: r.partition_kind, freshness: r.freshness, @@ -724,6 +727,7 @@ kind: TriggerNodeKind ref: string missing: boolean + draft: boolean // First target script (drives the per-script create/edit flows). runnable_path?: string // Every target script: a single (kind, ref) — e.g. one schedule — @@ -739,6 +743,7 @@ ref: string, unsaved: boolean, missing: boolean, + draft: boolean, runnable_path?: string ) { const prev = triggerSourceNodes.get(id) @@ -748,6 +753,7 @@ kind, ref, missing, + draft, runnable_path, runnable_paths: runnable_path ? [runnable_path] : [] }) @@ -796,6 +802,7 @@ ref, !!t.unsaved, isMissing, + t.draft === true, // Always thread the target script so the trigger node can // reach back to it — drives both the missing-trigger "create" // flow and the attached-trigger "edit" flow's script-path @@ -830,6 +837,7 @@ ref: info.ref, unsaved: info.allUnsaved, missing: info.missing, + draft: info.draft, runnable_path: info.runnable_path, runnable_unsaved: info.runnable_path ? unsavedRunnablePaths.has(info.runnable_path) diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index 6050c2d1cd..28c15ff864 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -22,8 +22,6 @@ X, Pencil } from 'lucide-svelte' - import Popover from '$lib/components/meltComponents/Popover.svelte' - import { tick } from 'svelte' import { inferArgs } from '$lib/infer' import { emptySchema, sendUserToast } from '$lib/utils' import type { Schema } from '$lib/common' @@ -41,6 +39,7 @@ import { extractDraftMacros } from './resolveGraph' import { assetColumnNodes, type ColumnLineageGraph } from './columnLineageGraph' import SummaryPathDisplay from '$lib/components/SummaryPathDisplay.svelte' + import Badge from '$lib/components/common/badge/Badge.svelte' import S3FilePreview from '$lib/components/S3FilePreview.svelte' import DataTablePreview from './DataTablePreview.svelte' import DucklakeAssetPanel from './DucklakeAssetPanel.svelte' @@ -215,15 +214,12 @@ // the edges. Counter rather than boolean so back-to-back runs // re-fire the effect even if no other prop changed. requestRunSignal?: number - // Folder-scoped non-editable prefix shown next to the suffix - // editor when the user renames a draft (e.g. `f//`). The - // new path = pathPrefix + suffix. - pathPrefix?: string // Called when the user renames a draft — the parent reseats the // path key in its drafts map and updates activeDraftPath. Returns // true on success; false on collision/validation failure so the // popover can keep itself open and surface the error inline. onDraftPathChange?: (oldPath: string, newPath: string) => boolean | string + onDraftMetaChange?: (path: string, meta: { summary: string; labels: string[] | undefined }) => void // Bumped by the parent (e.g. from the runnable-node action menu) to // auto-open the archive/delete confirmation modal for the currently // loaded persisted script. No-ops while the pane is showing a draft @@ -315,8 +311,8 @@ activeRunnable, onTestStateChange, requestRunSignal, - pathPrefix = '', onDraftPathChange, + onDraftMetaChange, requestRemoveSignal, downstreamSubscribers = 0, onStartBoundedRun, @@ -910,49 +906,6 @@ JSON.stringify(script.schema ?? null) === JSON.stringify(orig.schema ?? null) ) }) - - // Suffix editor for the draft-path popover. Seeded from the current - // path each time the popover opens so the user starts with what they - // see, not stale state from an earlier rename. - let draftPathSuffix = $state('') - let draftPathError = $state(undefined) - let draftPathInput: HTMLInputElement | undefined = $state(undefined) - - function suffixOf(fullPath: string): string { - return fullPath.startsWith(pathPrefix) ? fullPath.slice(pathPrefix.length) : fullPath - } - - async function openDraftPathEditor() { - draftPathSuffix = script ? suffixOf(script.path) : '' - draftPathError = undefined - await tick() - draftPathInput?.focus() - draftPathInput?.select() - } - - function confirmDraftPath(close: () => void) { - if (!script) return - const suffix = draftPathSuffix.trim() - if (!suffix) { - draftPathError = 'Path cannot be empty' - return - } - const newPath = pathPrefix + suffix - if (newPath === script.path) { - close() - return - } - const result = onDraftPathChange?.(script.path, newPath) - if (result === true || result === undefined) { - script.path = newPath - draftPathError = undefined - close() - } else if (typeof result === 'string') { - draftPathError = result - } else { - draftPathError = 'Path already in use' - } - }
@@ -964,80 +917,31 @@ {@const draftScriptPath = script.path} {#if onDraftPathChange && !readOnly} - - { - if (e.detail) openDraftPathEditor() + + {}} + saveOverride={({ path, summary, labels }) => { + // Before the rename, which carries the draft entry over as it is. + onDraftMetaChange?.(draftScriptPath, { summary, labels }) + if (script) { + script.summary = summary + script.labels = labels + } + if (path === draftScriptPath) return undefined + const result = onDraftPathChange(draftScriptPath, path) + if (result !== true && result !== undefined) { + return typeof result === 'string' ? result : 'Path already in use' + } + if (script) script.path = path + return undefined }} - > - {#snippet trigger()} - - {/snippet} - {#snippet content({ close })} -
- Path -
- {#if pathPrefix} - - {pathPrefix} - - {/if} - { - if (e.key === 'Enter') { - e.preventDefault() - confirmDraftPath(close) - } else if (e.key === 'Escape') { - e.preventDefault() - close() - } - }} - class="flex-1 min-w-0 px-2 py-1.5 text-sm font-mono bg-transparent focus:outline-none" - placeholder="my_script" - /> -
- {#if draftPathError} - {draftPathError} - {/if} -
- -
-
- {/snippet} - + /> + Draft {:else}
diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphEdge.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphEdge.svelte index f39c2d8d18..d7be5b15e6 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphEdge.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphEdge.svelte @@ -1,6 +1,7 @@
-
-

- Existing pipelines -

- {#if pipelines.loading && !pipelines.current} -
- - Loading… -
- {:else if pipelines.error} -
Failed: {pipelines.error.message}
- {:else if visiblePipelines.length === 0} -
- {currentFolder - ? 'No other pipelines in this workspace.' - : 'No pipelines yet. A pipeline is any folder whose scripts carry pipeline annotations.'} -
- {:else} -
- {#each visiblePipelines as p (p.folder)} - - {/each} -
- {/if} -
+ {#if !hideExisting} +
+

Existing pipelines

+ {#if pipelines.loading && !pipelines.current} +
+ + Loading… +
+ {:else if pipelines.error} +
Failed: {pipelines.error.message}
+ {:else if visiblePipelines.length === 0} +
+ {currentFolder + ? 'No other pipelines in this workspace.' + : 'No pipelines yet. A pipeline is any folder whose scripts carry pipeline annotations.'} +
+ {:else} +
+ {#each visiblePipelines as p (p.folder)} + + {/each} +
+ {/if} +
+ {/if} {#if !$userStore?.operator} -
-

- Pick or create a folder -

-
+
+

Pick or create a folder

+

+ Each folder holds one pipeline, made of its pipeline scripts. Pick a folder to open its + pipeline, or create one to start a new pipeline. +

+
diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte index b71a51ea0a..048fc44f86 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte @@ -18,8 +18,10 @@ AssetGraphSelection, NativeTriggerKind, PipelineMode, + PipelineTriggerDraft, DbtAssetProvenance } from './types' + import { isDraftableTriggerKind, triggerDraftKey } from './pipelineTriggerDrafts' import type { AssetKind, Script, ScriptLang } from '$lib/gen' import type { RunnableRunState, PipelineEvent } from './activeRunnables.svelte' import type { PipelineOutputKind } from './pipelineTemplates' @@ -153,14 +155,16 @@ language: ScriptLang, scriptPath: string, outputKind: PipelineOutputKind, - aiPrompt?: string + aiPrompt?: string, + options?: import('./PipelineInsertMenu.svelte').PipelineInsertOptions ) => void onAddPipelineScript?: ( language: ScriptLang, path: string, source: { kind: NativeTriggerKind; path: string | undefined }, outputKind: PipelineOutputKind, - aiPrompt?: string + aiPrompt?: string, + options?: import('./PipelineInsertMenu.svelte').PipelineInsertOptions ) => void onRunnableMenuRemove?: (...args: any[]) => void onRunProducer?: (producer: RunProducer) => Promise @@ -280,7 +284,11 @@ // FlowBuilder's autosave analogue — gated by `persistDrafts`. let pipelineDraftPath = $derived(pipelineBundlePath(folder)) let storageKey = $derived(`pipeline-${folder}`) - type PipelineDraftBundle = { drafts: Array<[string, PipelineDraft]>; activeDraftPath?: string } + type PipelineDraftBundle = { + drafts: Array<[string, PipelineDraft]> + activeDraftPath?: string + triggerDrafts?: PipelineTriggerDraft[] + } // Hydration is tracked on the editor instance (`editor.hydratedFromDb`), not a // component flag: the in-session preview reuses one instance across editor // hide/show, and it must hydrate ONCE per instance, not on every remount. @@ -300,6 +308,18 @@ if (loaded.size > 0) editor.drafts = loaded } if (typeof bundle.activeDraftPath === 'string') editor.activeDraftPath = bundle.activeDraftPath + if (Array.isArray(bundle.triggerDrafts)) { + editor.triggerDrafts = new Map( + bundle.triggerDrafts + .filter( + (d) => + isDraftableTriggerKind(d?.kind) && + typeof d.config?.path === 'string' && + typeof d.config?.script_path === 'string' + ) + .map((d) => [triggerDraftKey(d.kind, d.config.path), d]) + ) + } } function readLocalBundle(): PipelineDraftBundle | undefined { @@ -311,7 +331,8 @@ if (s && (Array.isArray(s.drafts) || typeof s.activeDraftPath === 'string')) { return { drafts: Array.isArray(s.drafts) ? s.drafts : [], - activeDraftPath: typeof s.activeDraftPath === 'string' ? s.activeDraftPath : undefined + activeDraftPath: typeof s.activeDraftPath === 'string' ? s.activeDraftPath : undefined, + triggerDrafts: Array.isArray(s.triggerDrafts) ? s.triggerDrafts : undefined } } } catch (e) { @@ -413,24 +434,26 @@ return [p, { ...d, script, outputAssets, inputAssets }] as [string, PipelineDraft] }) const activePath = editor.activeDraftPath + const triggerDrafts = [...editor.triggerDrafts.values()] const key = storageKey const ws = workspace const path = pipelineDraftPath const hydrated = editor.hydratedFromDb untrack(() => { if (!hydrated) return - const isEmpty = serialized.length === 0 && !activePath + const isEmpty = serialized.length === 0 && !activePath && triggerDrafts.length === 0 const bundle: PipelineDraftBundle | undefined = isEmpty ? undefined - : { drafts: serialized, activeDraftPath: activePath } + : { + drafts: serialized, + activeDraftPath: activePath, + ...(triggerDrafts.length > 0 ? { triggerDrafts } : {}) + } try { if (typeof localStorage !== 'undefined') { if (isEmpty) localStorage.removeItem(key) else - localStorage.setItem( - key, - encodeState({ drafts: serialized, activeDraftPath: activePath }) - ) + localStorage.setItem(key, encodeState(bundle)) } } catch (e) { console.warn('failed to mirror pipeline state', e) @@ -552,8 +575,8 @@ draftScript={activeDraft?.script} draftOutputAssets={activeDraft?.outputAssets} draftInputAssets={activeDraft?.inputAssets} - {pathPrefix} {onDraftPathChange} + onDraftMetaChange={editor.setDraftMeta} {workspace} onAnnotationsChange={editor.handleAnnotationsChange} onAssetsChange={editor.handleAssetsChange} diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte index e341707250..f9e920a678 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte @@ -1,6 +1,7 @@ {#if show} -
-
- -
-

Finish setting up pipelines

-

- Pipelines materialize data into DuckLake tables backed by object storage. Configure the - following before your first pipeline can run. -

-
+ +
+

+ Pipelines materialize data into DuckLake tables backed by object storage. Configure the + following before your first pipeline can run. +

+
    + {#each steps as step (step.title)} + {@const Icon = step.icon} +
  • + {#if step.done === true} + + {:else if step.done === false} + + {:else} + + {/if} +
    + {step.title} + {step.description} +
    + {#if step.done !== true} + + {step.cta} + + + {:else} + Configured + {/if} +
    • + {/each} +
- -
    - {#each steps as step (step.title)} - {@const Icon = step.icon} -
  • - {#if step.done === true} - - {:else if step.done === false} - - {:else} - - {/if} -
    - {step.title} - {step.description} -
    - {#if step.done !== true} - - {step.cta} - - - {:else} - Configured - {/if} -
    • - {/each} -
-
+ {/if} diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte index 4f97039a1c..fba438ed70 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte @@ -1,7 +1,7 @@ diff --git a/frontend/src/lib/components/triggers/amqp/AmqpTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/amqp/AmqpTriggerEditorInner.svelte index f2f3f81c0c..8e359061a3 100644 --- a/frontend/src/lib/components/triggers/amqp/AmqpTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/amqp/AmqpTriggerEditorInner.svelte @@ -172,12 +172,17 @@ is_flow = itemKind === 'flow' }) + // Set by `openNew(…, { onSaveDraft })`: the caller keeps the trigger as its own + // draft, so Save hands it the config instead of writing it. + let saveDraftHandler: ((cfg: Record) => void) | undefined = $state(undefined) + export async function openEdit( ePath: string, isFlow: boolean, defaultConfig?: Record, fixedScriptPath_?: string ) { + saveDraftHandler = undefined // A `whoami` that failed earlier would otherwise pin this workspace to "unknown user". operatingUser.forgetFailures() let loadingTimeout = setTimeout(() => { @@ -212,8 +217,14 @@ export async function openNew( nis_flow: boolean, fixedScriptPath_?: string, - defaultValues?: Record + defaultValues?: Record, + opts: { + onSaveDraft?: (cfg: Record) => void + /** A config this editor saved before, re-applied over the defaults. */ + draftConfig?: Record + } = {} ) { + saveDraftHandler = opts.onSaveDraft let loadingTimeout = setTimeout(() => { showLoading = true }, 100) @@ -243,6 +254,7 @@ selectedPermissionedAs = undefined preservePermissionedAs = false originalConfig = undefined + if (opts.draftConfig) await loadTriggerConfig(opts.draftConfig) } finally { clearTimeout(loadingTimeout) drawerLoading = false @@ -335,6 +347,11 @@ } async function updateTrigger(): Promise { + if (saveDraftHandler) { + saveDraftHandler($state.snapshot(getSaveCfg())) + drawer?.closeDrawer() + return + } deploymentLoading = true const previousPath = initialPath const cfg = getSaveCfg() @@ -431,7 +448,9 @@ ? can_write ? `Edit AMQP trigger ${initialPath}` : `AMQP trigger ${initialPath}` - : 'New AMQP trigger'} + : saveDraftHandler + ? 'Draft AMQP trigger' + : 'New AMQP trigger'} on:close={() => (inline ? onClose?.() : drawer?.closeDrawer())} > {#snippet actions()} @@ -489,6 +508,7 @@ isLoading={deploymentLoading} {isDeployed} onUpdate={updateTrigger} + saveLabel={saveDraftHandler ? 'Save draft' : undefined} {onReset} {onDelete} onToggleMode={handleToggleMode} diff --git a/frontend/src/lib/components/triggers/email/EmailTriggerEditor.svelte b/frontend/src/lib/components/triggers/email/EmailTriggerEditor.svelte index 101b84f770..4b9bfc7803 100644 --- a/frontend/src/lib/components/triggers/email/EmailTriggerEditor.svelte +++ b/frontend/src/lib/components/triggers/email/EmailTriggerEditor.svelte @@ -27,6 +27,21 @@ drawer?.openNew(is_flow, initial_script_path, defaultValues) } + /** Edit a trigger that exists only as the caller's draft; Save hands the config back. */ + export async function openDraft( + scriptPath: string, + defaults: Record, + saved: Record | undefined, + onSaveDraft: (cfg: Record) => void + ) { + open = true + await tickPainted() + drawer?.openNew(false, scriptPath, defaults as any, { + onSaveDraft, + draftConfig: saved + }) + } + let drawer: EmailTriggerEditorInner | undefined = $state() diff --git a/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte index fd3b20864a..5fff3067f4 100644 --- a/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte @@ -139,12 +139,17 @@ is_flow = itemKind === 'flow' }) + // Set by `openNew(…, { onSaveDraft })`: the caller keeps the trigger as its own + // draft, so Save hands it the config instead of writing it. + let saveDraftHandler: ((cfg: Record) => void) | undefined = $state(undefined) + export async function openEdit( ePath: string, isFlow: boolean, defaultConfig?: Partial, fixedScriptPath_?: string ) { + saveDraftHandler = undefined // A `whoami` that failed earlier would otherwise pin this workspace to "unknown user". operatingUser.forgetFailures() drawerLoading = true @@ -188,8 +193,14 @@ export async function openNew( nis_flow: boolean, fixedScriptPath_?: string, - defaultValues?: Partial + defaultValues?: Partial, + opts: { + onSaveDraft?: (cfg: Record) => void + /** A config this editor saved before, re-applied over the defaults. */ + draftConfig?: Record + } = {} ) { + saveDraftHandler = opts.onSaveDraft drawerLoading = true let loader = setTimeout(() => { showLoader = true @@ -218,6 +229,7 @@ selectedPermissionedAs = undefined preservePermissionedAs = false originalConfig = undefined + if (opts.draftConfig) await loadTriggerConfig(opts.draftConfig as Partial) } finally { clearTimeout(loader) drawerLoading = false @@ -276,6 +288,11 @@ } async function triggerScript(): Promise { + if (saveDraftHandler) { + saveDraftHandler($state.snapshot(getEmailTriggerConfig())) + drawer?.closeDrawer() + return + } if (customSaveBehavior) { customSaveBehavior(emailConfig) drawer?.closeDrawer() @@ -501,6 +518,7 @@ {edit} isLoading={deploymentLoading} onUpdate={triggerScript} + saveLabel={saveDraftHandler ? 'Save draft' : undefined} {onReset} {onDelete} {isDeployed} @@ -520,7 +538,9 @@ ? can_write ? `Edit email trigger ${initialPath}` : `Email trigger ${initialPath}` - : 'New email trigger'} + : saveDraftHandler + ? 'Draft email trigger' + : 'New email trigger'} on:close={() => (inline ? onClose?.() : drawer?.closeDrawer())} > {#snippet actions()} diff --git a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditor.svelte b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditor.svelte index f86706e8ab..ee3af7941a 100644 --- a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditor.svelte +++ b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditor.svelte @@ -22,6 +22,21 @@ drawer?.openNew(is_flow, initial_script_path, defaultValues) } + /** Edit a trigger that exists only as the caller's draft; Save hands the config back. */ + export async function openDraft( + scriptPath: string, + defaults: Record, + saved: Record | undefined, + onSaveDraft: (cfg: Record) => void + ) { + open = true + await tickPainted() + drawer?.openNew(false, scriptPath, defaults as any, { + onSaveDraft, + draftConfig: saved + }) + } + let drawer: GcpTriggerEditorInner | undefined = $state() diff --git a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte index 43d5f8e28c..41d494b8ed 100644 --- a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte @@ -163,12 +163,17 @@ ) const captureConfig = $derived.by(untrack(() => isEditor) ? getGcpCaptureConfig : () => ({})) + // Set by `openNew(…, { onSaveDraft })`: the caller keeps the trigger as its own + // draft, so Save hands it the config instead of writing it. + let saveDraftHandler: ((cfg: Record) => void) | undefined = $state(undefined) + export async function openEdit( ePath: string, isFlow: boolean, defaultValues?: Record, fixedScriptPath_?: string ) { + saveDraftHandler = undefined // A `whoami` that failed earlier would otherwise pin this workspace to "unknown user". operatingUser.forgetFailures() drawerLoading = true @@ -201,8 +206,14 @@ export async function openNew( nis_flow: boolean, fixedScriptPath_?: string, - defaultValues?: Record + defaultValues?: Record, + opts: { + onSaveDraft?: (cfg: Record) => void + /** A config this editor saved before, re-applied over the defaults. */ + draftConfig?: Record + } = {} ) { + saveDraftHandler = opts.onSaveDraft drawerLoading = true try { drawer?.openDrawer() @@ -244,6 +255,7 @@ selectedPermissionedAs = undefined preservePermissionedAs = false originalConfig = undefined + if (opts.draftConfig) await loadTriggerConfig(opts.draftConfig) } finally { drawerLoading = false } @@ -307,6 +319,11 @@ } async function updateTrigger(): Promise { + if (saveDraftHandler) { + saveDraftHandler($state.snapshot(getGcpConfig())) + drawer?.closeDrawer() + return + } deploymentLoading = true const previousPath = initialPath const cfg = gcpConfig @@ -441,7 +458,9 @@ ? can_write ? `Edit GCP Pub/Sub trigger ${initialPath}` : `GCP Pub/Sub trigger ${initialPath}` - : 'New GCP Pub/Sub trigger'} + : saveDraftHandler + ? 'Draft GCP Pub/Sub trigger' + : 'New GCP Pub/Sub trigger'} on:close={() => (inline ? onClose?.() : drawer?.closeDrawer())} > {#snippet actions()} @@ -498,6 +517,7 @@ {allowDraft} {isDeployed} onUpdate={updateTrigger} + saveLabel={saveDraftHandler ? 'Save draft' : undefined} {onReset} {onDelete} onToggleMode={handleToggleMode} diff --git a/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditor.svelte b/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditor.svelte index 73d32c0d52..bbce2315e6 100644 --- a/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditor.svelte +++ b/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditor.svelte @@ -21,6 +21,21 @@ drawer?.openNew(is_flow, initial_script_path, defaultValues) } + /** Edit a trigger that exists only as the caller's draft; Save hands the config back. */ + export async function openDraft( + scriptPath: string, + defaults: Record, + saved: Record | undefined, + onSaveDraft: (cfg: Record) => void + ) { + open = true + await tickPainted() + drawer?.openNew(false, scriptPath, defaults as any, { + onSaveDraft, + draftConfig: saved + }) + } + let drawer: KafkaTriggerEditorInner | undefined = $state() diff --git a/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte index cd79c98015..3e21cc2124 100644 --- a/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte @@ -174,12 +174,17 @@ is_flow = itemKind === 'flow' }) + // Set by `openNew(…, { onSaveDraft })`: the caller keeps the trigger as its own + // draft, so Save hands it the config instead of writing it. + let saveDraftHandler: ((cfg: Record) => void) | undefined = $state(undefined) + export async function openEdit( ePath: string, isFlow: boolean, defaultConfig?: Record, fixedScriptPath_?: string ) { + saveDraftHandler = undefined // A `whoami` that failed earlier would otherwise pin this workspace to "unknown user". operatingUser.forgetFailures() let loadingTimeout = setTimeout(() => { @@ -217,8 +222,14 @@ export async function openNew( nis_flow: boolean, fixedScriptPath_?: string, - nDefaultValues?: Record + nDefaultValues?: Record, + opts: { + onSaveDraft?: (cfg: Record) => void + /** A config this editor saved before, re-applied over the defaults. */ + draftConfig?: Record + } = {} ) { + saveDraftHandler = opts.onSaveDraft let loadingTimeout = setTimeout(() => { showLoading = true }, 100) // Do not show loading spinner for the first 100ms @@ -252,6 +263,7 @@ selectedPermissionedAs = undefined preservePermissionedAs = false originalConfig = undefined + if (opts.draftConfig) await loadTriggerConfig(opts.draftConfig) } finally { clearTimeout(loadingTimeout) drawerLoading = false @@ -334,6 +346,11 @@ } async function updateTrigger(): Promise { + if (saveDraftHandler) { + saveDraftHandler($state.snapshot(getSaveCfg())) + drawer?.closeDrawer() + return + } deploymentLoading = true const previousPath = initialPath const cfg = getSaveCfg() @@ -453,7 +470,9 @@ ? can_write ? `Edit Kafka trigger ${initialPath}` : `Kafka trigger ${initialPath}` - : 'New Kafka trigger'} + : saveDraftHandler + ? 'Draft Kafka trigger' + : 'New Kafka trigger'} on:close={() => (inline ? onClose?.() : drawer?.closeDrawer())} > {#snippet actions()} @@ -511,6 +530,7 @@ {isDeployed} {saveDisabled} onUpdate={updateTrigger} + saveLabel={saveDraftHandler ? 'Save draft' : undefined} {onReset} {onDelete} onToggleMode={handleToggleMode} diff --git a/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditor.svelte b/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditor.svelte index fea56a3750..b5af1243d6 100644 --- a/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditor.svelte +++ b/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditor.svelte @@ -21,6 +21,21 @@ drawer?.openNew(is_flow, initial_script_path, defaultValues) } + /** Edit a trigger that exists only as the caller's draft; Save hands the config back. */ + export async function openDraft( + scriptPath: string, + defaults: Record, + saved: Record | undefined, + onSaveDraft: (cfg: Record) => void + ) { + open = true + await tickPainted() + drawer?.openNew(false, scriptPath, defaults as any, { + onSaveDraft, + draftConfig: saved + }) + } + let drawer: MqttTriggerEditorInner | undefined = $state() diff --git a/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte index 09c944ed7c..16bd38d8fd 100644 --- a/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte @@ -169,12 +169,17 @@ is_flow = itemKind === 'flow' }) + // Set by `openNew(…, { onSaveDraft })`: the caller keeps the trigger as its own + // draft, so Save hands it the config instead of writing it. + let saveDraftHandler: ((cfg: Record) => void) | undefined = $state(undefined) + export async function openEdit( ePath: string, isFlow: boolean, defaultConfig?: Record, fixedScriptPath_?: string ) { + saveDraftHandler = undefined // A `whoami` that failed earlier would otherwise pin this workspace to "unknown user". operatingUser.forgetFailures() let loadingTimeout = setTimeout(() => { @@ -212,8 +217,14 @@ export async function openNew( nis_flow: boolean, fixedScriptPath_?: string, - defaultValues?: Record + defaultValues?: Record, + opts: { + onSaveDraft?: (cfg: Record) => void + /** A config this editor saved before, re-applied over the defaults. */ + draftConfig?: Record + } = {} ) { + saveDraftHandler = opts.onSaveDraft let loadingTimeout = setTimeout(() => { showLoading = true }, 100) @@ -248,6 +259,7 @@ selectedPermissionedAs = undefined preservePermissionedAs = false originalConfig = undefined + if (opts.draftConfig) await loadTriggerConfig(opts.draftConfig) } finally { clearTimeout(loadingTimeout) drawerLoading = false @@ -347,6 +359,11 @@ } async function updateTrigger(): Promise { + if (saveDraftHandler) { + saveDraftHandler($state.snapshot(getSaveCfg())) + drawer?.closeDrawer() + return + } deploymentLoading = true const previousPath = initialPath const cfg = getSaveCfg() @@ -427,7 +444,9 @@ ? can_write ? `Edit MQTT trigger ${initialPath}` : `MQTT trigger ${initialPath}` - : 'New MQTT trigger'} + : saveDraftHandler + ? 'Draft MQTT trigger' + : 'New MQTT trigger'} on:close={() => (inline ? onClose?.() : drawer?.closeDrawer())} > {#snippet actions()} @@ -485,6 +504,7 @@ isLoading={deploymentLoading} {isDeployed} onUpdate={updateTrigger} + saveLabel={saveDraftHandler ? 'Save draft' : undefined} {onReset} {onDelete} onToggleMode={handleToggleMode} diff --git a/frontend/src/lib/components/triggers/nats/NatsTriggerEditor.svelte b/frontend/src/lib/components/triggers/nats/NatsTriggerEditor.svelte index aa5585a7b6..0e38be90f4 100644 --- a/frontend/src/lib/components/triggers/nats/NatsTriggerEditor.svelte +++ b/frontend/src/lib/components/triggers/nats/NatsTriggerEditor.svelte @@ -21,6 +21,21 @@ drawer?.openNew(is_flow, initial_script_path, defaultValues) } + /** Edit a trigger that exists only as the caller's draft; Save hands the config back. */ + export async function openDraft( + scriptPath: string, + defaults: Record, + saved: Record | undefined, + onSaveDraft: (cfg: Record) => void + ) { + open = true + await tickPainted() + drawer?.openNew(false, scriptPath, defaults as any, { + onSaveDraft, + draftConfig: saved + }) + } + let drawer: NatsTriggerEditorInner | undefined = $state() diff --git a/frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte index 33af8a3d40..96b8fbb493 100644 --- a/frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte @@ -157,12 +157,17 @@ is_flow = itemKind === 'flow' }) + // Set by `openNew(…, { onSaveDraft })`: the caller keeps the trigger as its own + // draft, so Save hands it the config instead of writing it. + let saveDraftHandler: ((cfg: Record) => void) | undefined = $state(undefined) + export async function openEdit( ePath: string, isFlow: boolean, defaultConfig?: Record, fixedScriptPath_?: string ) { + saveDraftHandler = undefined // A `whoami` that failed earlier would otherwise pin this workspace to "unknown user". operatingUser.forgetFailures() let loadingTimeout = setTimeout(() => { @@ -206,8 +211,14 @@ export async function openNew( nis_flow: boolean, fixedScriptPath_?: string, - nDefaultValues?: Record + nDefaultValues?: Record, + opts: { + onSaveDraft?: (cfg: Record) => void + /** A config this editor saved before, re-applied over the defaults. */ + draftConfig?: Record + } = {} ) { + saveDraftHandler = opts.onSaveDraft let loadingTimeout = setTimeout(() => { showLoading = true }, 100) @@ -241,6 +252,7 @@ selectedPermissionedAs = undefined preservePermissionedAs = false originalConfig = undefined + if (opts.draftConfig) await loadTriggerConfig(opts.draftConfig) } finally { clearTimeout(loadingTimeout) drawerLoading = false @@ -326,6 +338,11 @@ } async function updateTrigger(): Promise { + if (saveDraftHandler) { + saveDraftHandler($state.snapshot(getSaveCfg())) + drawer?.closeDrawer() + return + } deploymentLoading = true const previousPath = initialPath const cfg = natsConfig @@ -424,7 +441,9 @@ ? can_write ? `Edit NATS trigger ${initialPath}` : `NATS trigger ${initialPath}` - : 'New NATS trigger'} + : saveDraftHandler + ? 'Draft NATS trigger' + : 'New NATS trigger'} on:close={() => (inline ? onClose?.() : drawer?.closeDrawer())} > {#snippet actions()} @@ -481,6 +500,7 @@ isLoading={deploymentLoading} {isDeployed} onUpdate={updateTrigger} + saveLabel={saveDraftHandler ? 'Save draft' : undefined} {onReset} {onDelete} {cloudDisabled} diff --git a/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditor.svelte b/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditor.svelte index 66702a54ab..195829dda0 100644 --- a/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditor.svelte +++ b/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditor.svelte @@ -21,6 +21,21 @@ drawer?.openNew(is_flow, initial_script_path, defaultValues) } + /** Edit a trigger that exists only as the caller's draft; Save hands the config back. */ + export async function openDraft( + scriptPath: string, + defaults: Record, + saved: Record | undefined, + onSaveDraft: (cfg: Record) => void + ) { + open = true + await tickPainted() + drawer?.openNew(false, scriptPath, defaults as any, undefined, { + onSaveDraft, + draftConfig: saved + }) + } + let drawer: PostgresTriggerEditorInner | undefined = $state() diff --git a/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte index 2072041c28..762f13735b 100644 --- a/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte @@ -258,12 +258,17 @@ is_flow = itemKind === 'flow' }) + // Set by `openNew(…, { onSaveDraft })`: the caller keeps the trigger as its own + // draft, so Save hands it the config instead of writing it. + let saveDraftHandler: ((cfg: Record) => void) | undefined = $state(undefined) + export async function openEdit( ePath: string, isFlow: boolean, defaultConfig?: Record, fixedScriptPath_?: string ) { + saveDraftHandler = undefined // A `whoami` that failed earlier would otherwise pin this workspace to "unknown user". operatingUser.forgetFailures() let loadingTimeout = setTimeout(() => { @@ -307,8 +312,14 @@ nis_flow: boolean, fixedScriptPath_?: string, defaultValues?: Record, - newDraft?: boolean + newDraft?: boolean, + opts: { + onSaveDraft?: (cfg: Record) => void + /** A config this editor saved before, re-applied over the defaults. */ + draftConfig?: Record + } = {} ) { + saveDraftHandler = opts.onSaveDraft let loadingTimeout = setTimeout(() => { showLoading = true }, 100) // Do not show loading spinner for the first 100ms @@ -352,6 +363,7 @@ selectedPermissionedAs = undefined preservePermissionedAs = false originalConfig = undefined + if (opts.draftConfig) await loadTriggerConfig(opts.draftConfig) } finally { clearTimeout(loadingTimeout) drawerLoading = false @@ -477,6 +489,11 @@ } async function updateTrigger(): Promise { + if (saveDraftHandler) { + saveDraftHandler($state.snapshot(getSaveCfg())) + drawer?.closeDrawer() + return + } const cfg = postgresConfig if (!cfg) { return @@ -608,7 +625,9 @@ ? can_write ? `Edit Postgres trigger ${initialPath}` : `Postgres trigger ${initialPath}` - : 'New Postgres trigger'} + : saveDraftHandler + ? 'Draft Postgres trigger' + : 'New Postgres trigger'} on:close={() => (inline ? onClose?.() : drawer?.closeDrawer())} > {#snippet actions()}{@render actionsSnippet()}{/snippet} @@ -663,6 +682,7 @@ isLoading={deploymentLoading} {isDeployed} onUpdate={updateTrigger} + saveLabel={saveDraftHandler ? 'Save draft' : undefined} {onReset} {onDelete} {mode} diff --git a/frontend/src/lib/components/triggers/schedules/ScheduleAdvancedOptions.svelte b/frontend/src/lib/components/triggers/schedules/ScheduleAdvancedOptions.svelte new file mode 100644 index 0000000000..116828a3de --- /dev/null +++ b/frontend/src/lib/components/triggers/schedules/ScheduleAdvancedOptions.svelte @@ -0,0 +1,597 @@ + + + + +
+ {#if !loading} + + + + + + + {#if itemKind === 'script'} + + {/if} + + {#if optionTabSelected === 'error_handler'} +
+ {#snippet header()} +
+ {#if !$enterpriseLicense}(ee only){/if} +
+ {/snippet} + {#snippet action()} +
+ saveAsDefaultErrorHandler(false) + }, + { + displayName: 'Override all existing', + type: 'delete', + action: () => saveAsDefaultErrorHandler(true) + } + ]} + > + {#snippet buttonReplacement()} + + Set as default + {/snippet} + +
+ {/snippet} +
+ +
+ + + {#snippet customTabTooltip()} + +
+
The following args will be passed to the error handler: +
    +
  • workspace_id: The ID of the workspace that the schedule belongs to.
    • +
  • job_id: The UUID of the job that errored.
    • +
  • 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.
    • +
+
+
+ + {/snippet} + +
+
+

+ Triggered when schedule failed

+ + +

time{failedTimes > 1 ? 's in a row' : ''}

+
+
+
+ {:else if optionTabSelected === 'recovery_handler'} + {@const disabled = !canWrite || 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 buttonReplacement()} + + Set as default + {/snippet} + +
+ {/snippet} + + {#snippet customTabTooltip()} + +
+
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 errored
    • +
  • error_started_at: The start datetime of the last job that errored
    • +
  • success_times: The number of times the schedule succeeded before + calling the recovery handler.
    • +
  • success_result: The result of the latest successful job
    • +
  • success_started_at: The start datetime of the latest successful job
    • +
+
+
+ + {/snippet} + +
+
+

Triggered when schedule recovered

+ +

time{recoveredTimes > 1 ? 's in a row' : ''}

+
+
+
+ {:else if optionTabSelected === 'success_handler'} + {@const disabled = !canWrite || 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 buttonReplacement()} + + Set as default + {/snippet} + +
+ {/snippet} + + {#snippet customTabTooltip()} + +
+
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 job
    • +
  • success_started_at: The start datetime of the successful job
    • +
+
+
+ + {/snippet} + +
+ {:else if optionTabSelected === 'retries'} + {@const disabled = !canWrite || 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. +
+ This is only available for individual script. For flows, retries can be set on each flow + step in the flow editor. + + {/snippet} + {#if itemKind !== 'script'} + + Error Handler and Retries are only available for scripts. For flows, use the built-in error handler + and retries. + + {:else} + + {/if} +
+ {:else if optionTabSelected === 'dynamic_skip'} +
+ {#snippet header()} + + Optional script to filter scheduled dates. Receives the proposed datetime and returns + boolean. True = run on this date, False = skip to next occurrence. + + {/snippet} +
+ + + Handler must return a boolean value. Return true to execute the scheduled job, false + to skip. + +
+
+ {:else if optionTabSelected === 'tag'} +
+ +
+ {/if} + {:else} + + {/if} +
diff --git a/frontend/src/lib/components/triggers/schedules/ScheduleEditor.svelte b/frontend/src/lib/components/triggers/schedules/ScheduleEditor.svelte index 78e569d3f3..648250e52b 100644 --- a/frontend/src/lib/components/triggers/schedules/ScheduleEditor.svelte +++ b/frontend/src/lib/components/triggers/schedules/ScheduleEditor.svelte @@ -1,6 +1,7 @@ diff --git a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte index 12cd4c143e..1317da425b 100644 --- a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte +++ b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte @@ -1,11 +1,12 @@ diff --git a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte index 0995a2e1b6..d9612e64a5 100644 --- a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte @@ -152,12 +152,17 @@ is_flow = itemKind === 'flow' }) + // Set by `openNew(…, { onSaveDraft })`: the caller keeps the trigger as its own + // draft, so Save hands it the config instead of writing it. + let saveDraftHandler: ((cfg: Record) => void) | undefined = $state(undefined) + export async function openEdit( ePath: string, isFlow: boolean, defaultConfig?: Record, fixedScriptPath_?: string ) { + saveDraftHandler = undefined // A `whoami` that failed earlier would otherwise pin this workspace to "unknown user". operatingUser.forgetFailures() let loadingTimeout = setTimeout(() => { @@ -199,8 +204,14 @@ export async function openNew( nis_flow: boolean, fixedScriptPath_?: string, - defaultValues?: Record + defaultValues?: Record, + opts: { + onSaveDraft?: (cfg: Record) => void + /** A config this editor saved before, re-applied over the defaults. */ + draftConfig?: Record + } = {} ) { + saveDraftHandler = opts.onSaveDraft let loadingTimeout = setTimeout(() => { showLoading = true }, 100) @@ -229,6 +240,7 @@ selectedPermissionedAs = undefined preservePermissionedAs = false originalConfig = undefined + if (opts.draftConfig) await loadTriggerConfig(opts.draftConfig) } finally { initialConfig = structuredClone($state.snapshot(getSaveCfg())) clearTimeout(loadingTimeout) @@ -336,6 +348,11 @@ } async function updateTrigger(): Promise { + if (saveDraftHandler) { + saveDraftHandler($state.snapshot(getSaveCfg())) + drawer?.closeDrawer() + return + } deploymentLoading = true const previousPath = initialPath const cfg = getSaveCfg() @@ -406,7 +423,9 @@ ? can_write ? `Edit SQS trigger ${initialPath}` : `SQS trigger ${initialPath}` - : 'New SQS trigger'} + : saveDraftHandler + ? 'Draft SQS trigger' + : 'New SQS trigger'} on:close={() => (inline ? onClose?.() : drawer?.closeDrawer())} > {#snippet actions()} @@ -463,6 +482,7 @@ isLoading={deploymentLoading} {isDeployed} onUpdate={updateTrigger} + saveLabel={saveDraftHandler ? 'Save draft' : undefined} {onReset} {onDelete} {mode} diff --git a/frontend/src/routes/(root)/(logged)/pipeline/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/+page.svelte index 1515040f18..07ff0a6bd4 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/+page.svelte @@ -3,8 +3,8 @@ import PipelineFolderList from '$lib/components/assets/AssetGraph/PipelineFolderList.svelte' import PipelineSetupSignpost from '$lib/components/assets/AssetGraph/PipelineSetupSignpost.svelte' import PipelineAlphaAckModal from '$lib/components/assets/AssetGraph/PipelineAlphaAckModal.svelte' - import PipelineDbtSignpost from '$lib/components/assets/AssetGraph/PipelineDbtSignpost.svelte' import { BookOpen, NetworkIcon } from 'lucide-svelte' + import Badge from '$lib/components/common/badge/Badge.svelte' import { onMount } from 'svelte' const ACK_STORAGE_KEY = 'pipeline-alpha-ack' @@ -41,34 +41,24 @@ class="border-b flex flex-row justify-between gap-2 px-2 py-1 items-center min-h-10 shrink-0 whitespace-nowrap" >
- -

Pipelines

- - Alpha - + +

Pipelines

+ Alpha
-
+
-

Data pipelines

-

+

Data pipelines

+

Chain ingestion, transformation and materialization steps into an asset-aware graph. - {#if !$userStore?.operator} - Each pipeline lives in a folder: open one below, or pick a folder to start a new - pipeline. - {:else} - Open a pipeline below to view its graph and runs. - {/if}

Pipelines documentation @@ -79,14 +69,7 @@ {/if} - - - - {#if !$userStore?.operator} - - {/if} +
diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index b4ddb8350c..b63d4171bf 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -27,7 +27,9 @@ AssetGraphResponse, AssetGraphSelection, NativeTriggerKind, - PipelineMode + PipelineMode, + PipelineTriggerDraft, + PipelineTriggerDraftKind } from '$lib/components/assets/AssetGraph/types' import PipelineModeToggle from '$lib/components/assets/AssetGraph/PipelineModeToggle.svelte' import MacroExplorerDrawer from '$lib/components/assets/AssetGraph/MacroExplorerDrawer.svelte' @@ -79,7 +81,17 @@ createPipelineAiHelpers, type PipelineDraft } from '$lib/components/assets/AssetGraph/pipelineAiHelpers' - import { PipelineEditorState } from '$lib/components/assets/AssetGraph/pipelineEditorState.svelte' + import { + PipelineEditorState + } from '$lib/components/assets/AssetGraph/pipelineEditorState.svelte' + import { + defaultTriggerPath, + isDraftableTriggerKind, + triggerDraftKey + } from '$lib/components/assets/AssetGraph/pipelineTriggerDrafts' + import { deployTriggerDraft } from '$lib/components/assets/AssetGraph/pipelineTriggerDraftDeploy' + import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte' + import type { PipelineInsertOptions } from '$lib/components/assets/AssetGraph/PipelineInsertMenu.svelte' import { createPipelineRecording, finalizePipelineRecording @@ -288,6 +300,8 @@ // appear in the current `g.runnables`. That self-cleaning property is // the whole reason for the refactor — no rename/delete cleanup needed. let bodiesByPath = $state>(new Map()) + // Deployed summaries from the same fetch, for node titles. Same only-add cache. + let summariesByPath = $state>(new Map()) // Sibling cache: the parsed asset usages from `inferAssets` (wasm), one // pass per body. Same only-add semantics as `bodiesByPath`. let inferredAssetsByPath = $state>(new Map()) @@ -412,9 +426,10 @@ triggers: DraftTriggerSource[], outputKind: PipelineOutputKind, input?: { kind: AssetKind; path: string }, - aiPrompt?: string + aiPrompt?: string, + options?: PipelineInsertOptions ) { - const out = autoOutputAsset(outputKind, folder, language) + const out = options?.outputAsset ?? autoOutputAsset(outputKind, folder, language) const script = buildDraft(language, scriptPath, triggers, outputKind, out, input) // Write the new draft into the map (structural update so Svelte // re-derives graphWithDraft) and focus it in the details pane. When @@ -429,6 +444,16 @@ pe.drafts = next pe.activeDraftPath = scriptPath pe.selection = undefined + if (options?.schedule) { + pe.setTriggerDraft({ + kind: 'schedule', + config: { + ...options.schedule, + path: options.schedule.path || defaultTriggerPath(scriptPath, 'schedule'), + script_path: scriptPath + } + }) + } // Follow the new node with a smooth pan. The id matches the runnable // node the canvas builds for a draft script (`script:`). @@ -528,7 +553,7 @@ bypassNavigationGuard = false return } - if (pe.drafts.size === 0) return + if (pendingCount === 0) return // `leave` covers tab close / hard reload / cross-origin nav. SvelteKit // turns a cancelled leave into a browser-native "Leave site?" prompt, // which we explicitly don't want — match the rest of the editors and @@ -563,6 +588,7 @@ // stale active path don't bleed into the next page. localStorage // is overwritten by the persist effect on the next tick. pe.drafts = new Map() + pe.triggerDrafts = new Map() pe.activeDraftPath = undefined saveErrors = new Map() const target = pendingNavigationUrl @@ -582,7 +608,7 @@ // page so they can deal with the failures via the bar's error // popover. Otherwise resume the navigation that triggered the // guard. - if (pe.drafts.size === 0) { + if (pendingCount === 0) { const target = pendingNavigationUrl leaveModalOpen = false pendingNavigationUrl = undefined @@ -670,8 +696,39 @@ }) } + // Scripts plus trigger drafts: what "Save all" deploys. + let pendingCount = $derived(pe.drafts.size + pe.triggerDrafts.size) + let confirmTriggerDeployOpen = $state(false) + + function requestSaveAll() { + if (pe.triggerDrafts.size > 0) confirmTriggerDeployOpen = true + else void saveAllDrafts() + } + + // Runs after the script deploys: a trigger's `script_path` must already + // exist. Triggers whose script is still an undeployed draft stay drafts. + async function deployTriggerDrafts( + ws: string, + savedScriptPaths: string[], + errors: Map + ): Promise { + const ready = [...pe.triggerDrafts].filter( + ([, d]) => + !pe.drafts.has(d.config.script_path) || savedScriptPaths.includes(d.config.script_path) + ) + const results = await Promise.all(ready.map(([, d]) => deployTriggerDraft(d, ws))) + const saved: string[] = [] + results.forEach((ok, i) => { + const [key, d] = ready[i] + if (ok) saved.push(key) + else errors.set(d.config.path, `Could not create the ${d.kind} trigger, see the notification`) + }) + for (const key of saved) pe.discardTriggerDraft(key) + return saved + } + async function saveAllDrafts() { - if (!$workspaceStore || pe.drafts.size === 0 || savingAll) return + if (!$workspaceStore || pendingCount === 0 || savingAll) return savingAll = true const ws = $workspaceStore // The open pane's keystrokes live in `pe.liveContent` until the pane is @@ -715,6 +772,7 @@ errors.set(path, e?.body ?? e?.message ?? String(e)) } } + const savedTriggers = await deployTriggerDrafts(ws, savedPaths, errors) // Drop the saved drafts from the map; failed ones stay so the user // can fix them and retry. Build the new map from the still-failing // entries to keep insertion order stable. @@ -737,6 +795,8 @@ } pe.activeDraftPath = undefined } + } + if (savedPaths.length > 0 || savedTriggers.length > 0) { await graphRes.refetch() // Verify only what actually deployed — failed drafts would // otherwise false-positive as "missing" rows. @@ -744,10 +804,11 @@ } saveErrors = errors savingAll = false - if (savedPaths.length > 0 && errors.size === 0) { - sendUserToast(`Saved ${savedPaths.length} draft${savedPaths.length === 1 ? '' : 's'}`) - } else if (savedPaths.length > 0 && errors.size > 0) { - sendUserToast(`Saved ${savedPaths.length}, ${errors.size} failed — see details`, true) + const savedCount = savedPaths.length + savedTriggers.length + if (savedCount > 0 && errors.size === 0) { + sendUserToast(`Saved ${savedCount} draft${savedCount === 1 ? '' : 's'}`) + } else if (savedCount > 0 && errors.size > 0) { + sendUserToast(`Saved ${savedCount}, ${errors.size} failed — see details`, true) } else if (errors.size > 0) { sendUserToast(`${errors.size} draft${errors.size === 1 ? '' : 's'} failed to save`, true) } @@ -759,6 +820,7 @@ next.delete(path) pe.drafts = next forgetPath(path) + pe.discardTriggerDraftsFor(path) } function clearSaveError(path: string) { @@ -819,6 +881,7 @@ } } pe.drafts = next + pe.retargetTriggerDrafts(oldPath, newPath) if (pe.activeDraftPath === oldPath) pe.activeDraftPath = newPath // Path-keyed live overlays: re-key for the renamed draft so the // graph stays consistent between the moment we mutate `drafts` @@ -1050,7 +1113,8 @@ language: ScriptLang, scriptPath: string, outputKind: PipelineOutputKind, - aiPrompt?: string + aiPrompt?: string, + options?: PipelineInsertOptions ) { const ref = `${ASSET_PREFIX[asset.kind]}${asset.path}` openMaterializerDraft( @@ -1062,7 +1126,8 @@ kind: asset.kind, path: asset.path }, - aiPrompt + aiPrompt, + options ) } function handleAddPipelineScript( @@ -1070,9 +1135,15 @@ scriptPath: string, source: { kind: NativeTriggerKind; path: string | undefined }, outputKind: PipelineOutputKind, - aiPrompt?: string + aiPrompt?: string, + options?: PipelineInsertOptions ) { - openMaterializerDraft(language, scriptPath, [source], outputKind, undefined, aiPrompt) + openMaterializerDraft(language, scriptPath, [source], outputKind, undefined, aiPrompt, options) + // A schedule was configured by the wizard itself; other row-backed kinds + // are configured in their own editor, as a draft of the new script. + if (source.kind !== 'schedule' && isDraftableTriggerKind(source.kind)) { + openTriggerDraft(source.kind, scriptPath) + } } function handleRunnableMenuRemove(info: { runnable_kind: 'script' | 'flow' @@ -1227,7 +1298,8 @@ liveAnnotations: pe.liveAnnotations, inferredWritesByPath, inferredReadsByPath, - annotatedNativeKindsByPath + annotatedNativeKindsByPath, + triggerDrafts: pe.triggerDrafts.values() }) ) @@ -1250,7 +1322,21 @@ // What the canvas renders: drafts merged in edit, deployed-only in view — // unless the "show drafts" chip is on, which overlays the drafts onto the // view (what View will show once they're deployed). - let displayGraph = $derived(mode === 'edit' || includeDrafts ? graphWithDraft : deployedGraph) + let displayGraph = $derived( + withSummaries(mode === 'edit' || includeDrafts ? graphWithDraft : deployedGraph) + ) + function withSummaries(g: AssetGraphResponse): AssetGraphResponse { + const showDrafts = mode === 'edit' || includeDrafts + return { + ...g, + runnables: g.runnables.map((r) => { + const summary = + (showDrafts ? pe.drafts.get(r.path)?.script.summary : undefined) || + summariesByPath.get(r.path) + return summary ? { ...r, summary } : r + }) + } + } // Leaving edit mode: drop the live editor overlays — they substitute the // last keystroke buffer for one script inside inferredWritesByPath / @@ -2168,6 +2254,12 @@ // time or silently bind to nothing. Surface that as a toast and // keep the drawer closed; the user needs to save the script first // (which also creates it under the new path if they renamed it). + // Kinds with a trigger row can instead wait as a pipeline draft and + // deploy with the pipeline, after the script. + if (isDraftableTriggerKind(kind) && pe.drafts.has(scriptPath)) { + openTriggerDraft(kind, scriptPath) + return + } if (pe.drafts.has(scriptPath)) { sendUserToast( `Save the script "${scriptPath}" first — triggers can only be attached to deployed scripts.`, @@ -2183,13 +2275,40 @@ // can still be edited everywhere else (TriggersPanel, etc.) where the // picker stays editable. function deleteAttachedTrigger(kind: NativeTriggerKind, triggerPath: string) { + const key = triggerDraftKey(kind, triggerPath) + if (pe.triggerDrafts.has(key)) { + pe.discardTriggerDraft(key) + return + } triggerEditors?.requestDelete(kind, triggerPath) } function openEditTriggerDrawer(kind: NativeTriggerKind, triggerPath: string, scriptPath: string) { + const draft = pe.triggerDrafts.get(triggerDraftKey(kind, triggerPath)) + if (draft) { + openTriggerDraft(draft.kind, scriptPath, draft) + return + } triggerEditors?.openEdit(kind, triggerPath, scriptPath) } + /** The kind's editor over a new draft, or over `draft` when re-editing one. */ + function openTriggerDraft( + kind: PipelineTriggerDraftKind, + scriptPath: string, + draft?: PipelineTriggerDraft + ) { + const defaults = { path: defaultTriggerPath(scriptPath, kind) } + triggerEditors?.openTriggerDraft(kind, scriptPath, defaults, draft?.config, (cfg) => { + const { extra_perms: _, ...config } = cfg + pe.setTriggerDraft( + { kind, config: { ...config, path: cfg.path, script_path: scriptPath, is_flow: false } }, + // The editor's Path field can rename it; drafts are keyed by path. + draft ? triggerDraftKey(kind, draft.config.path) : undefined + ) + }) + } + // Reuse the empty AssetGraphResponse so we can still render the canvas // (layout, controls, mini-map) on a fresh pipeline. const EMPTY_GRAPH: AssetGraphResponse = { @@ -2328,6 +2447,9 @@ nextBodies.set(path, content) bodiesByPath = nextBodies } + if (s.summary && !summariesByPath.has(path)) { + summariesByPath = new Map(summariesByPath).set(path, s.summary) + } if (!inferredAssetsByPath.has(path)) { const nextAssets = new Map(inferredAssetsByPath) nextAssets.set(path, inferred) @@ -2436,8 +2558,8 @@ anchored between the two flex-1 side groups so it stays centered, with breathing room on both sides. -->
- setMode(m)} /> - {#if mode === 'view' && pe.drafts.size > 0} + setMode(m)} /> + {#if mode === 'view' && pendingCount > 0} {/if}
@@ -2533,7 +2655,7 @@ {/snippet} {/if} - {#if mode === 'edit' && pe.drafts.size > 0} + {#if mode === 'edit' && pendingCount > 0} @@ -2550,11 +2672,11 @@ variant="accent" unifiedSize="sm" startIcon={{ icon: savingAll ? Loader2 : Save }} - onclick={saveAllDrafts} + onclick={requestSaveAll} disabled={savingAll} - title={savingAll ? 'Saving drafts…' : `Deploy all ${pe.drafts.size} drafts`} + title={savingAll ? 'Saving drafts…' : `Deploy all ${pendingCount} drafts`} > - {savingAll ? 'Saving…' : `Save all (${pe.drafts.size})`} + {savingAll ? 'Saving…' : `Save all (${pendingCount})`} {/if} {#if mode === 'view'} @@ -2759,6 +2881,38 @@ + { + confirmTriggerDeployOpen = false + void saveAllDrafts() + }} + onCanceled={() => (confirmTriggerDeployOpen = false)} +> +
+

+ {pe.triggerDrafts.size === 1 + ? 'Saving the pipeline also creates this trigger. It starts running its script once deployed.' + : `Saving the pipeline also creates these ${pe.triggerDrafts.size} triggers. They start running their scripts once deployed.`} +

+
    + {#each [...pe.triggerDrafts] as [key, d] (key)} +
  • + {d.config.path} + + {d.kind}{#if d.kind === 'schedule'} + · {d.config.schedule} ({d.config.timezone}){/if} + → {d.config.script_path} + +
    • + {/each} +
+
+ +

- {pe.drafts.size === 1 ? 'Unsaved draft' : `${pe.drafts.size} unsaved drafts`} + {pendingCount === 1 ? 'Unsaved draft' : `${pendingCount} unsaved drafts`}

- You have {pe.drafts.size === 1 - ? 'a draft pipeline script' - : `${pe.drafts.size} draft pipeline scripts`} that {pe.drafts.size === 1 + You have {pendingCount === 1 + ? 'a pipeline draft' + : `${pendingCount} pipeline drafts`} that {pendingCount === 1 ? 'has' : 'have'} not been deployed yet. What would you like to do?

    - {#each [...pe.drafts.keys()] as p} + {#each [...pe.drafts.keys(), ...[...pe.triggerDrafts.values()].map((d) => d.config.path)] as p}
  • {p}
    • {/each}
@@ -2817,7 +2971,7 @@ startIcon={{ icon: leaveSaving ? Loader2 : Save }} > {leaveSaving ? 'Saving…' : `Save all (${pe.drafts.size})`}{leaveSaving ? 'Saving…' : `Save all (${pendingCount})`}