diff --git a/backend/add-pipeline-menu.png b/backend/add-pipeline-menu.png new file mode 100644 index 0000000000..706fed1ce2 Binary files /dev/null and b/backend/add-pipeline-menu.png differ diff --git a/backend/after-create.png b/backend/after-create.png new file mode 100644 index 0000000000..082c8634b8 Binary files /dev/null and b/backend/after-create.png differ diff --git a/backend/downstream-menu.png b/backend/downstream-menu.png new file mode 100644 index 0000000000..090a92bf6b Binary files /dev/null and b/backend/downstream-menu.png differ diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 39dbfad1cf..43a43d343e 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -123,17 +123,14 @@ pub enum TriggerSpec { #[serde(skip_serializing_if = "Option::is_none", default)] debounce: Option, }, - // Refresh on cron. The raw expression is passed through as-is so the - // existing schedule subsystem can validate it. - Schedule { - cron: String, - }, // `// on ` — marker-only declaration that this script wants to be // triggered by a native trigger of the given kind. No path: the binding // is the trigger row's own `script_path` field (set when the user creates - // the kafka/mqtt/… trigger in its dedicated UI). The graph endpoint - // discovers attached triggers by `WHERE script_path = ` and - // surfaces a "missing" placeholder when an annotation has no matching row. + // the kafka/mqtt/schedule/… trigger in its dedicated UI). The graph + // endpoint discovers attached triggers by `WHERE script_path = ` and surfaces a "missing" placeholder when an annotation has no + // matching row. + Schedule, Webhook, Email, Kafka, @@ -509,22 +506,6 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } - if let Some(after_kw) = rest.strip_prefix("schedule") { - if !after_kw.starts_with(|c: char| c.is_whitespace()) { - continue; - } - let after = after_kw.trim(); - if let Some(cron) = unquote(after) { - if !cron.trim().is_empty() { - let trig = TriggerSpec::Schedule { cron: cron.to_string() }; - if !out.triggers.contains(&trig) { - out.triggers.push(trig); - } - } - } - continue; - } - if let Some(after_kw) = rest.strip_prefix("partitioned") { if !after_kw.starts_with(|c: char| c.is_whitespace()) { continue; @@ -683,14 +664,13 @@ fn parse_partitioned_spec(s: &str) -> Option { // Native trigger keywords are *marker-only* — no trailing path. The actual // binding lives on the native trigger row (`script_path` column). Anything // trailing the keyword is rejected so the form stays unambiguous. -// -// Note: `on schedule "..."` is no longer accepted — schedule moved to a -// top-level `// schedule "..."` annotation. The `Schedule` TriggerSpec -// variant is still produced, just from a different keyword. fn parse_trigger_spec(s: &str) -> Option { // Marker-only native trigger keywords. The match table keeps the - // annotation set in lockstep with `TriggerSpec`. + // annotation set in lockstep with `TriggerSpec`. `schedule` is in here + // too — the cron lives on the schedule row the user creates separately; + // the annotation is just the binding declaration. const NATIVE_KINDS: &[(&str, TriggerSpec)] = &[ + ("schedule", TriggerSpec::Schedule), ("webhook", TriggerSpec::Webhook), ("email", TriggerSpec::Email), ("kafka", TriggerSpec::Kafka), @@ -751,28 +731,22 @@ mod pipeline_annotation_tests { } #[test] - fn top_level_schedule() { - let out = parse_pipeline_annotations("// schedule \"0 */6 * * *\""); + fn on_schedule_marker() { + // `// on schedule` is marker-only — the binding is the schedule row's + // own `script_path` field, just like kafka/mqtt/etc. + let out = parse_pipeline_annotations("// on schedule"); assert_eq!(out.triggers.len(), 1); - assert_eq!( - out.triggers[0], - TriggerSpec::Schedule { cron: "0 */6 * * *".to_string() } - ); + assert_eq!(out.triggers[0], TriggerSpec::Schedule); } #[test] - fn top_level_schedule_single_quotes_and_other_prefixes() { - let out = parse_pipeline_annotations("# schedule '0 0 * * *'\n-- schedule \"@daily\""); - assert_eq!(out.triggers.len(), 2); - } - - #[test] - fn rejects_on_schedule_form() { - // The old `on schedule "..."` form is no longer recognized — the - // `on` branch tries `schedule` as a native trigger kind and falls - // through. + fn rejects_schedule_with_trailing_content() { + // Marker-only — the old `// schedule ""` form is gone, and a + // trailing path/cron on the `on schedule` form is malformed. let out = parse_pipeline_annotations("// on schedule \"0 0 * * *\""); assert!(out.triggers.is_empty()); + let out = parse_pipeline_annotations("// schedule \"0 0 * * *\""); + assert!(out.triggers.is_empty()); } #[test] diff --git a/backend/pipeline-test.png b/backend/pipeline-test.png new file mode 100644 index 0000000000..49f36f7946 Binary files /dev/null and b/backend/pipeline-test.png differ diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 417ef4a969..be685b2fe6 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -407,10 +407,10 @@ struct GraphEdge { } // Declared `// on ` trigger edge — the actual execution DAG. -// Asset / Schedule come from `script_trigger`; the seven native variants -// (Email/Kafka/…/Gcp) come from the per-kind trigger tables joined on -// `script_path`. Each native variant carries just the trigger row's path; -// the config (broker, topic, auth, …) lives in its own UI. +// Asset edges come from `script_trigger`; the eight native variants +// (Schedule/Email/Kafka/…/Gcp) come from the per-kind trigger tables joined +// on `script_path`. Each native variant carries just the trigger row's path; +// the config (cron, broker, topic, auth, …) lives in its own UI. // // `webhook` is parsed as an annotation marker but has no dedicated trigger // table — every script gets an implicit webhook endpoint — so no variant @@ -425,7 +425,7 @@ enum TriggerEdge { runnable_path: String, }, Schedule { - cron: String, + path: String, runnable_kind: AssetUsageKind, runnable_path: String, }, @@ -518,10 +518,10 @@ async fn asset_graph( .fetch_all(&mut *tx) .await?; - // Pipeline asset / schedule trigger edges, fetched separately so we can - // widen the runnable_set for trigger-only endpoints (e.g. an asset - // trigger whose asset has no usage in the pipeline yet). Native trigger - // kinds (kafka, mqtt, …) are *not* in `script_trigger` anymore — they're + // Pipeline asset trigger edges, fetched separately so we can widen the + // runnable_set for trigger-only endpoints (e.g. an asset trigger whose + // asset has no usage in the pipeline yet). Native trigger kinds + // (schedule, kafka, mqtt, …) are *not* in `script_trigger` — they're // discovered below by querying each native trigger table directly. let trigger_rows = sqlx::query!( r#" @@ -532,7 +532,7 @@ async fn asset_graph( trigger_ref AS "trigger_ref!" FROM script_trigger WHERE workspace_id = $1 - AND trigger_kind IN ('asset', 'schedule') + AND trigger_kind = 'asset' AND ($2::text IS NULL OR runnable_path LIKE $2) "#, &w_id, @@ -545,11 +545,17 @@ async fn asset_graph( // single-destination `script_path` directly, so we resolve attachment by // joining on that field rather than via `script_trigger`. UNION ALL keeps // it a single round trip; the `kind` column drives the TriggerEdge ctor - // below. + // below. `schedule` lives in the `schedule` table, which has its own + // shape (no workspace_id-only filter — it shares `is_flow` like the + // others), but the columns we need line up. let native_trigger_rows = sqlx::query!( r#" SELECT kind, path, script_path, is_flow FROM ( - SELECT 'email' AS kind, path, script_path, is_flow FROM email_trigger + SELECT 'schedule' AS kind, path, script_path, is_flow FROM schedule + WHERE workspace_id = $1 + AND script_path IS NOT NULL + UNION ALL + SELECT 'email', path, script_path, is_flow FROM email_trigger WHERE workspace_id = $1 UNION ALL SELECT 'kafka', path, script_path, is_flow FROM kafka_trigger @@ -595,10 +601,43 @@ async fn asset_graph( .fetch_all(&mut *tx) .await?; + // Existing scripts / flows in the workspace. Used to filter out + // orphan trigger rows whose `script_path` no longer resolves — those + // would otherwise be added to `runnable_set` below and surface as + // phantom "deployed" runnables on the canvas (matching what the user + // can deploy a new trigger against: nothing). + let existing_script_paths = sqlx::query_scalar!( + r#"SELECT path AS "path!" FROM script + WHERE workspace_id = $1 + AND archived = false + AND deleted = false"#, + &w_id, + ) + .fetch_all(&mut *tx) + .await?; + let existing_flow_paths = sqlx::query_scalar!( + r#"SELECT path AS "path!" FROM flow WHERE workspace_id = $1 AND archived = false"#, + &w_id, + ) + .fetch_all(&mut *tx) + .await?; + tx.commit().await?; let pipeline_member_script_paths: std::collections::HashSet = pipeline_member_paths.into_iter().map(|r| r.path).collect(); + let existing_script_paths: std::collections::HashSet = + existing_script_paths.into_iter().collect(); + let existing_flow_paths: std::collections::HashSet = + existing_flow_paths.into_iter().collect(); + let runnable_exists = |kind: AssetUsageKind, path: &str| match kind { + AssetUsageKind::Script => existing_script_paths.contains(path), + AssetUsageKind::Flow => existing_flow_paths.contains(path), + // `Job` is a runtime-detected ephemeral runnable (asset usage rows + // only), never a target of a stored trigger row. Treat as existing + // so we don't accidentally drop ephemeral lineage edges. + AssetUsageKind::Job => true, + }; let mut edges = Vec::with_capacity(rows.len()); let mut asset_set: std::collections::HashSet<(AssetKind, String)> = Default::default(); @@ -613,6 +652,14 @@ async fn asset_graph( } for r in rows { + // Drop asset usage rows whose runnable target was archived/deleted + // but whose row in `asset` is still around — those would otherwise + // surface as a phantom "deployed" runnable on the canvas with no + // way to interact with it, since the underlying script/flow no + // longer exists. + if !runnable_exists(r.usage_kind, &r.usage_path) { + continue; + } asset_set.insert((r.asset_kind, r.asset_path.clone())); runnable_set.insert((r.usage_kind, r.usage_path.clone())); edges.push(GraphEdge { @@ -627,37 +674,38 @@ async fn asset_graph( let mut triggers: Vec = Vec::with_capacity(trigger_rows.len() + native_trigger_rows.len()); for t in trigger_rows { + // Drop orphan asset-trigger rows — their target runnable no longer + // exists (script/flow archived or deleted, or was never deployed). + // Without this, an orphan row would surface as a phantom "deployed" + // runnable on the canvas (no `unsaved` flag, can't actually be + // run / re-targeted by a new trigger). + if !runnable_exists(t.runnable_kind, &t.runnable_path) { + continue; + } runnable_set.insert((t.runnable_kind, t.runnable_path.clone())); - match t.trigger_kind.as_str() { - "asset" => { - // trigger_ref is `` — parse back out so both - // endpoints match what the frontend uses for node ids. - if let Some((asset_kind, asset_path)) = parse_asset_trigger_ref(&t.trigger_ref) { - // Make sure the source asset has a node even if nothing - // reads/writes it in this folder. - asset_set.insert((asset_kind, asset_path.clone())); - triggers.push(TriggerEdge::Asset { - asset_kind, - asset_path, - runnable_kind: t.runnable_kind, - runnable_path: t.runnable_path, - }); - } - } - "schedule" => { - triggers.push(TriggerEdge::Schedule { - cron: t.trigger_ref, + if t.trigger_kind.as_str() == "asset" { + // trigger_ref is `` — parse back out so both + // endpoints match what the frontend uses for node ids. + if let Some((asset_kind, asset_path)) = parse_asset_trigger_ref(&t.trigger_ref) { + // Make sure the source asset has a node even if nothing + // reads/writes it in this folder. + asset_set.insert((asset_kind, asset_path.clone())); + triggers.push(TriggerEdge::Asset { + asset_kind, + asset_path, runnable_kind: t.runnable_kind, runnable_path: t.runnable_path, }); } - _ => {} // Native kinds come from per-kind trigger tables below. } + // Native kinds (schedule, kafka, mqtt, …) come from per-kind trigger + // tables below. } // Native trigger attachments — one TriggerEdge per row, the kind chosen // from the discriminator. Add the runnable to the set so a script with - // no asset edges but a kafka attachment still renders on the canvas. + // no asset edges but a kafka/schedule attachment still renders on the + // canvas. for t in native_trigger_rows { let kind = t.kind.unwrap_or_default(); let path = t.path.unwrap_or_default(); @@ -667,8 +715,15 @@ async fn asset_graph( } else { AssetUsageKind::Script }; + // Same orphan filter as the asset-trigger loop above — drop trigger + // rows whose target script/flow no longer exists so the graph + // doesn't synthesize a phantom deployed runnable. + if !runnable_exists(runnable_kind, &script_path) { + continue; + } runnable_set.insert((runnable_kind, script_path.clone())); let edge = match kind.as_str() { + "schedule" => TriggerEdge::Schedule { path, runnable_kind, runnable_path: script_path }, "email" => TriggerEdge::Email { path, runnable_kind, runnable_path: script_path }, "kafka" => TriggerEdge::Kafka { path, runnable_kind, runnable_path: script_path }, "mqtt" => TriggerEdge::Mqtt { path, runnable_kind, runnable_path: script_path }, diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index bd74e49b25..77f70d4762 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -45,7 +45,7 @@ use windmill_common::{ assets::{ clear_script_triggers, clear_static_asset_usage, clear_static_asset_usage_by_script_hash, delete_managed_pipeline_schedule, insert_script_trigger, insert_static_asset_usage, - parse_duration_secs, parse_pipeline_annotations, reconcile_pipeline_schedule, + parse_duration_secs, parse_pipeline_annotations, trigger_spec_to_row, AssetUsageKind, AssetWithAltAccessType, TriggerSpec, }, error::{self, to_anyhow}, @@ -1622,28 +1622,11 @@ async fn create_script_internal<'c>( .await?; } - // Phase 4 reconciliation: a `// schedule ""` annotation creates or - // updates a managed schedule row that fires this script on the given - // cron. Removing the annotation deletes the managed row. Manual - // schedules at the same path are untouched (see - // `reconcile_pipeline_schedule` for the conflict policy). First-write - // wins if multiple `// schedule` lines are declared. - let pipeline_cron: Option<&str> = pipeline_triggers.iter().find_map(|t| match t { - TriggerSpec::Schedule { cron } => Some(cron.as_str()), - _ => None, - }); - let permissioned_as_for_schedule = username_to_permissioned_as(&authed.username); - reconcile_pipeline_schedule( - &mut *tx, - &w_id, - &ns.path, - false, // is_flow — scripts only for now - &authed.email, - &authed.username, - &permissioned_as_for_schedule, - pipeline_cron, - ) - .await?; + // Schedule annotations (`// on schedule`) are marker-only — the binding + // lives on the schedule row's own `script_path` field, which the user + // creates separately via the schedule editor. No script-create-time + // reconciliation is needed (and there are no "managed" schedules to + // upsert/delete anymore). let permissioned_as = username_to_permissioned_as(&authed.username); if let Some(parent_hash) = ns.parent_hash { diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index e512776e03..8484c3226d 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -103,82 +103,13 @@ pub async fn clear_script_triggers<'e>( Ok(()) } -// Reconcile a managed schedule for a pipeline script based on its parsed -// `// schedule ""` annotation. Idempotent: each call brings the -// `schedule` row in line with the annotation as of *this* deploy. -// -// The schedule lives at the same path as the runnable. The `managed` flag -// disambiguates auto-created rows from user-managed ones — only managed -// rows are updated or removed by reconciliation; manually-created schedules -// at the same path are left alone (the annotation is silently ignored). -// -// Pass `cron = None` when the annotation has been removed → drops the -// managed row. Pass `cron = Some(...)` to upsert. -pub async fn reconcile_pipeline_schedule<'e>( - executor: impl PgExecutor<'e>, - workspace_id: &str, - runnable_path: &str, - is_flow: bool, - email: &str, - edited_by: &str, - permissioned_as: &str, - cron: Option<&str>, -) -> error::Result<()> { - match cron { - Some(cron) => { - // Upsert. ON CONFLICT only updates rows that are already managed - // (the WHERE clause guards against trampling user-created - // schedules that happen to live at the runnable's path). - sqlx::query!( - r#" - INSERT INTO schedule ( - workspace_id, path, schedule, timezone, edited_by, script_path, - is_flow, enabled, email, permissioned_as, - ws_error_handler_muted, no_flow_overlap, cron_version, - managed - ) - VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', true) - ON CONFLICT (workspace_id, path) DO UPDATE - SET schedule = EXCLUDED.schedule, - edited_at = now(), - edited_by = EXCLUDED.edited_by, - managed = true - WHERE schedule.managed - OR schedule.script_path = EXCLUDED.script_path - "#, - workspace_id, - runnable_path, - cron, - edited_by, - is_flow, - email, - permissioned_as, - ) - .execute(executor) - .await?; - } - None => { - // Drop any prior managed schedule for this runnable. Manual - // schedules at the same path keep `managed = false` and are - // unaffected. - sqlx::query!( - r#"DELETE FROM schedule - WHERE workspace_id = $1 - AND script_path = $2 - AND managed"#, - workspace_id, - runnable_path, - ) - .execute(executor) - .await?; - } - } - Ok(()) -} - -// Drop the managed schedule (if any) for a runnable that's been deleted. -// Equivalent to `reconcile_pipeline_schedule(..., None)` but exposed -// separately so script-delete sites can call it without parsing annotations. +// Drop any managed schedule rows left over from an earlier release of the +// pipeline editor (when `// schedule ""` annotations auto-created a +// `managed = true` row on every deploy). Schedule annotations are now +// marker-only — the binding lives on the schedule row's own `script_path` +// field, same as kafka/mqtt/etc. — but this cleanup hook stays in place so +// script archive/delete still nukes the orphaned managed rows. Manual +// schedules at the same path keep `managed = false` and are untouched. pub async fn delete_managed_pipeline_schedule<'e>( executor: impl PgExecutor<'e>, workspace_id: &str, @@ -314,8 +245,11 @@ pub fn trigger_spec_to_row(spec: &TriggerSpec) -> Option<(ScriptTriggerKind, Str }; Some((ScriptTriggerKind::Asset, format!("{}{}", prefix, path))) } - TriggerSpec::Schedule { cron } => Some((ScriptTriggerKind::Schedule, cron.clone())), - TriggerSpec::Webhook + // Schedule joins the native-trigger family — no script_trigger row + // is inserted for the annotation. The binding lives on the schedule + // row's own `script_path` field, same as kafka/mqtt/etc. + TriggerSpec::Schedule + | TriggerSpec::Webhook | TriggerSpec::Email | TriggerSpec::Kafka | TriggerSpec::Mqtt diff --git a/docs/pipelines-vs-dbt.md b/docs/pipelines-vs-dbt.md new file mode 100644 index 0000000000..9ec6dd11d3 --- /dev/null +++ b/docs/pipelines-vs-dbt.md @@ -0,0 +1,374 @@ +# Pipelines vs. dbt + +Positioning analysis and architectural notes for the data-pipeline abstraction +currently landing on `feat/asset-graph-view`. Covers what we're building, how +it differs from dbt, which dbt features are real gaps vs. TODO, and a focused +deep-dive on incremental materialization — including a recommendation to +collapse it into partitioning rather than ship it as a separate concept. + +## What we're building + +Asset-centric, polyglot, annotation-driven, event-aware: + +- Assets (`datatable`, `ducklake`, `s3object`, `volume`) are graph nodes; + scripts are edges that produce/consume them. See + `backend/parsers/windmill-parser/src/asset_parser.rs:25`. +- Lineage comes from two sources: parsed annotations (`// pipeline`, + `// on datatable://...`, `// partitioned daily`, `// freshness 1h`, + `// trigger any`, `// debounce`, `// tag`, `// retry`) and body-inferred + reads/writes via the asset parser. +- Triggers are first-class: schedule, webhook, email, kafka, mqtt, nats, + postgres, sqs, gcp — all wired into the same DAG view + (`frontend/src/lib/components/assets/AssetGraph/types.ts:56`). +- Per-language scaffolds (DuckDB ATTACH, Postgres, Python, TS, Bash) generate + starter code per `PipelineOutputKind` (`datatable | ducklake | s3_parquet | + s3_object | none`). See + `frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts:11`. + +## Differentiators vs. dbt + +1. **Event-driven + batch in one DAG.** dbt is batch-on-warehouse. Kafka → + Python normalize → DuckDB aggregate → Postgres view → Slack notify is + native here; in dbt land it's "use Airflow/Prefect for the non-SQL parts." +2. **Polyglot, not SQL+Jinja.** Python/TS/Bash/Duck/PG transformations live + in the same graph. No Jinja templating language; annotations are real + comments parsed strictly. +3. **Multi-substrate by design.** `datatable` (Postgres), `ducklake` + (lakehouse), `s3_parquet`, `s3_object` are peers. dbt's universe is + "tables in your warehouse." +4. **One platform.** Same runtime as workflows, internal apps, background + jobs, RBAC, secrets, schedules. dbt is single-purpose. +5. **Inferred lineage from code.** Body parser picks up `CREATE TABLE` / S3 + writes — annotations are not strictly required to get edges. dbt requires + explicit `{{ ref() }}` everywhere. + +## Where dbt wins today + +| Gap | Architectural blocker? | Verdict | +|---|---|---| +| Data tests | No | Pure TODO | +| Incremental materializations | No, but pick a philosophy | TODO with design decision | +| Column lineage + docs site | No | Pure TODO | +| Snapshots / SCD2 | No | New output kind | +| Selective execution grammar | No | UI/CLI surface | +| Schema contracts | No, but design metadata model | TODO with design work | +| Packages / community | Closed annotation parser starts to bind | Decide extensibility model | +| Semantic layer / metrics | No | Large additive scope | + +The three items where the current abstraction needs deliberate decisions +before more weight lands on it: **incremental philosophy, schema metadata, +and annotation extensibility**. The rest is execution. + +### 1. Data tests + +dbt: `unique`, `not_null`, `accepted_values`, custom generic tests, plus +singular tests. Run as `SELECT` statements that pass when they return 0 rows. + +Today: nothing. Annotation parser is the natural hook — +`// test unique col_name`, `// test not_null col_name`, +`// test ` for custom. Pipeline runtime already handles +failure propagation. Lowest-risk, highest-payoff item. + +### 2. Incremental materializations + +See [Incremental deep-dive](#incremental-deep-dive) below. + +### 3. Column lineage + docs + +dbt: SQL-AST parsing for column-level deps; `dbt docs serve` produces a +static lineage site with descriptions. + +Today: graph is asset-level. `SqlQueryDetails` in the parser +(`backend/parsers/windmill-parser/src/asset_parser.rs:44`) already has a +column map — the scaffolding exists. No `// column` annotation, no docs +surface. Pure TODO; no abstraction stands in the way. + +### 4. Snapshots / SCD2 + +dbt: `{% snapshot %}` blocks with `strategy='timestamp'` or `'check'`. +Today: nothing. Add as a new `PipelineOutputKind` + `// snapshot strategy= +timestamp updated_at=updated_at unique_key=id` annotation. Same shape as +other output kinds. + +### 5. Selective execution grammar + +dbt: `--select tag:nightly+ state:modified+ +my_model+`. +Today: `requestRunCascadeSignal` in the canvas, `// tag` annotation parsed. +Graph + tags + last-run state has all the inputs. UI/CLI surface, not +abstraction work. + +### 6. Schema contracts + +dbt: `contract: enforced` + `columns: [{name, data_type}]`. Compile-time +check that model output matches the declaration. + +Today: `// on datatable://users/active` is a string. Rename a column +upstream → downstream breaks at runtime, silently. + +This is the item where the current asset abstraction is thinnest. +To do contracts well: capture output schemas after a run (substrate-specific +DESCRIBE), persist them as asset metadata, validate consumer references at +save time. The asset-as-typed-node model accommodates it — but **where** +schemas live (asset row, sidecar?), **when** they're captured (post-run? +edit-time?), and **how** versioning works are non-trivial design choices. +Worth doing intentionally now while the asset surface is still young. + +### 7. Packages / community + +dbt: `dbt deps`, `dbt-utils`, `dbt-expectations`. Whole ecosystem on Jinja +macros. + +Today: closed-vocabulary annotation parser — `parsePipelineAnnotations` +hardcodes `pipeline`, `partitioned`, `freshness`, `trigger`, `debounce`, +`tag`, `retry`, `on`. No way for a package to register +`// test rows_between 100 1000000` or `// hook on_failure my_alert`. + +This is the one place the current abstraction starts to bind. Macros are +also dbt's biggest pain source — we don't have to replicate them. Possible +shapes: + +- **Hooks-as-scripts**: `// on_failure f/lib/alert`, `// pre_run f/lib/setup`. + Value is a script path. Stays inside the closed annotation set; new + hook *types* still require parser changes but third-party *behavior* + ships as scripts. +- **Test types as scripts**: a test is a script that returns 0/1, packaged + via the hub like anything else. +- **Materialization plugins**: harder; template-generator would need to be + extensible. + +Doing this *after* you've shipped 30 hardcoded annotations is much harder +than doing it now. + +### 8. Semantic layer / metrics + +dbt: `metrics:` blocks, MetricFlow, BI-tool query API. Large scope, +additive. Lowest priority of the eight. + +## Incremental deep-dive + +### How dbt incremental works + +```sql +-- models/marts/orders_daily.sql +{{ config( + materialized='incremental', + unique_key='order_id', + incremental_strategy='merge', + on_schema_change='append_new_columns' +) }} + +SELECT order_id, user_id, amount, created_at +FROM {{ ref('orders_raw') }} +{% if is_incremental() %} +WHERE created_at > (SELECT MAX(created_at) FROM {{ this }}) +{% endif %} +``` + +- **First run** (target doesn't exist): `CREATE TABLE orders_daily AS + SELECT ...` — full build, no WHERE. +- **Subsequent runs**: stage to temp table, then MERGE on `unique_key`. + +Knobs: `incremental_strategy` ∈ {`merge`, `append`, `delete+insert`, +`insert_overwrite`, `microbatch`}. `on_schema_change` ∈ {`fail`, `ignore`, +`append_new_columns`, `sync_all_columns`}. `--full-refresh` forces rebuild. + +Pain points: watermark + unique_key interaction is subtle (late-arriving +rows past the watermark are silently dropped); `on_schema_change` defaults +to `ignore` (silent column drop); Jinja `is_incremental()` runs at compile, +not runtime — debugging requires `dbt compile`; cross-warehouse MERGE +dialect is dbt's biggest internal complexity. + +### Where Windmill stands today + +- `// partitioned daily|hourly|weekly|monthly|dynamic key=...` parsed into + `PartitionSpec` at + `backend/parsers/windmill-parser/src/asset_parser.rs:172`. +- `// freshness 1h` parsed. +- Templates emit `CREATE TABLE IF NOT EXISTS ... AS SELECT *` — full + refresh, every run, no partition substitution. +- No `WM_PARTITION_*` context flowing into scripts. +- No materialized-partition state per asset. + +Annotations are present but metadata-only. Nothing actually executes +incrementally yet. + +### Path A — Literal templates ("script is the truth") + +Philosophy: WYSIWYG. Windmill never wraps. Templates scaffold boilerplate, +partition context is injected as bind / env vars, the user owns the SQL. + +```sql +-- pipeline +-- on datatable://prod/orders_raw +-- partitioned daily +-- unique_key order_id + +ATTACH 'datatable://prod' AS pg; + +CREATE TABLE IF NOT EXISTS pg.orders_daily ( + order_id BIGINT PRIMARY KEY, + user_id BIGINT, + amount NUMERIC, + created_at TIMESTAMPTZ +); + +CREATE OR REPLACE TEMP TABLE _stage AS +SELECT order_id, user_id, amount, created_at +FROM pg.orders_raw +WHERE created_at >= $WM_PARTITION_START + AND created_at < $WM_PARTITION_END; + +BEGIN; +DELETE FROM pg.orders_daily + WHERE created_at >= $WM_PARTITION_START + AND created_at < $WM_PARTITION_END; +INSERT INTO pg.orders_daily SELECT * FROM _stage; +COMMIT; +``` + +Runtime: resolve `(value, start, end)` from scheduler tick / trigger event +/ backfill range → bind as SQL params → execute script as-is → record +`(asset_path, partition_value)` on success. + +**Pros**: no compile step; backfill is trivial (idempotent DELETE+INSERT); +late-arriving data → just re-run the affected partition; no dialect +rewriting in core; Python/TS/Bash/SQL all fit the same model. + +**Cons**: boilerplate per script; materialization changes require script +edits; user owns dialect specifics. + +### Path B — dbt-style wrapping + +Philosophy: separate intent (SELECT) from execution (DDL). User declares +what; Windmill compiles to per-substrate DDL. + +User writes: + +```sql +SELECT order_id, user_id, amount, created_at +FROM pg.orders_raw +WHERE created_at >= $WM_PARTITION_START + AND created_at < $WM_PARTITION_END +``` + +Runtime parses, looks up target schema, wraps per output kind + +strategy + first-run/subsequent-run state. + +**Pros**: concise; materialization is a config flip; automatic schema-drift +handling; cross-substrate consistency. + +**Cons**: two-layer execution ("what ran?" needs a compile-output view); +SELECT-only restricts pre/post-statement work (dbt's answer: `pre_hook` / +`post_hook` — more surface); doesn't generalize to Python/TS (you end up +with two execution models); cross-substrate MERGE dialect is where dbt has +burned the most engineering — we'd inherit that tax forever; schema +introspection per substrate is its own project. + +### Path C — Hybrid (recommended) + +- **Literal-by-default**: scaffolds emit full DDL with `WM_PARTITION_*` + substitution. WYSIWYG for all languages. +- **Helper library** (e.g. `wmll.partition`, `wmll.datatable.upsert_partition`): + lifts boilerplate into library calls without hiding semantics — readable + source. +- **Opt-in wrapping** for single-SELECT SQL scripts via + `// materialized incremental wrap=true`. Limit to DuckDB first; add + others as needed. Always log the compiled SQL. +- **State + backfill UI**: persist materialized partitions per asset; UI + to backfill a range with concurrency cap. + +Ships A's 80% case first without committing to B's dialect-rewriting tax. +Wrapping becomes opt-in convenience for users who want dbt-style ergonomics. + +### Decisions either path forces + +1. **Partition window provenance.** Scheduler tick? Trigger event time + (Kafka `event_time` header)? Explicit backfill? Default = "now's bucket"? +2. **Surface.** Bind params (`$WM_PARTITION_START`), env vars + (`WM_PARTITION_START`), helper library — probably all three for + different languages, but pick canonical names. +3. **First-run bootstrap.** Template scaffolds `CREATE TABLE IF NOT + EXISTS` (A), or runtime detects "table missing → full refresh" (B). +4. **State tracking.** `materialized_partitions` keyed by `(workspace, + asset_kind, asset_path)`. Drives "run stale," backfill gap detection, + downstream waiting. +5. **Backfill execution.** N partitions → serial? Parallel with + concurrency cap per asset? +6. **Idempotency contract.** `// partitioned` should imply "re-running the + same partition is safe." Templates and helpers must enforce. + +## Partitioning vs. incremental: the reframing + +Partitioning covers ~80% of what dbt's incremental does. What it gives +for free: + +- Unit of work (one partition per run) +- Idempotency (DELETE-by-partition + INSERT is safe to rerun) +- State (track which partitions are materialized) +- Backfill (re-run a range) +- First-run vs. subsequent-run (every run is "process partition P" — no + special case) +- "Process only new data" (the partition window IS the filter) + +dbt itself has been migrating toward partition-first thinking via +`microbatch` strategy — essentially `incremental` with mandatory partition +key. + +### What partitioning alone doesn't address + +**Dedup within a partition by a separate key.** Example: partition by +`created_at` daily, but `orders_raw` is mutable — the same `order_id` can +appear multiple times in one partition (initial create, then amendments). +You want `orders_daily` to hold the latest version per `order_id`. + +DELETE-by-partition + INSERT works only if you reprocess from a +source-of-truth source. If you're consuming amendments and need dedup +*within* the slice, you need MERGE on `order_id`, not DELETE on partition. + +This is what dbt's `unique_key` does. Orthogonal to partitioning: +`partitioned` answers "which slice?"; `unique_key` answers "how do I dedup +inside the slice?" + +**Pure watermark-based incremental.** Mostly subsumed by +`// partitioned dynamic key=updated_at` — a partition becomes "everything +since the last seen value of `key`." + +### Recommended annotation shape + +Don't build "incremental" as a concept. Build: + +- `// partitioned ` — unit of work + state + backfill (already exists). +- `// unique_key ` — opt-in dedup-within-partition. Drives MERGE + template vs. DELETE+INSERT template. +- `// append` — opt-out of dedup entirely (INSERT-only, no DELETE). + +This collapses dbt's `materialized=incremental` + `incremental_strategy` + +`unique_key` into orthogonal annotations that compose. Partition-first is +the better mental model. + +Schema drift handling (`on_schema_change`) is genuinely separate — applies +to full-refresh too — and belongs with the schema-contracts work (gap #6). + +## First implementation slice + +Sequencing if we go with the hybrid + partition-first reframing: + +1. **Partition runtime context** — resolve `(value, start, end)` from + scheduler / trigger / backfill, surface as bind vars + env vars. No + materialization change yet. +2. **Helper library** — `wmll.partition.window()`, + `wmll.datatable.upsert_partition()` for Python/TS, SQL macros for + DuckDB/PG. +3. **Template updates** — when `// partitioned X` is present, scaffold + DELETE+INSERT (or MERGE when `// unique_key` also present, or INSERT + when `// append`). +4. **Materialized-partition state** — new table keyed by + `(workspace, asset_kind, asset_path, partition_value)`. Asset metadata + read API exposes it. +5. **Backfill UI** — date range picker on the pipeline folder page; fans + out runs with concurrency cap. +6. *Later, behind a flag:* opt-in wrap mode for single-SELECT DuckDB. + +Delivers dbt's pragmatic value (incremental, backfill, idempotent reruns) +without buying the compile-layer maintenance, and keeps Windmill +recognizably Windmill-shaped. diff --git a/frontend/src/lib/components/MoveDrawer.svelte b/frontend/src/lib/components/MoveDrawer.svelte index f661ae4817..212e45d58d 100644 --- a/frontend/src/lib/components/MoveDrawer.svelte +++ b/frontend/src/lib/components/MoveDrawer.svelte @@ -8,6 +8,7 @@ import { updateItemPathAndSummary, checkFlowOnBehalfOf } from './moveRenameManager' import Label from './Label.svelte' import TextInput from './text_input/TextInput.svelte' + import { FlowService, ScriptService, type TriggersCount } from '$lib/gen' const dispatch = createEventDispatcher() @@ -24,8 +25,46 @@ let own = $state(false) let onBehalfOfEmail = $state(undefined) + // Counts of triggers/schedules/etc. that reference this script or flow. + // The backend cascades `script_path` on rename across all trigger tables + // (see `windmill_common::triggers::update_triggers_script_path` invoked + // from script/flow create), so the user just needs to know what will be + // moved along — not opt in per-row. + let attachedTriggers = $state(undefined) let hasChanges = $derived((summary ?? '') !== initialSummary || dirtyPath) + // Flatten the count buckets into a uniform list for rendering. Order + // reflects user-facing prominence: schedules first (most common), then + // the seven native trigger kinds, then HTTP / webhook / websocket / + // email-default / cloud-service installations. Buckets with count 0 + // are dropped so the panel only mentions triggers that actually exist. + let attachedSummary = $derived.by>(() => { + const c = attachedTriggers + if (!c) return [] + const out: Array<{ label: string; count: number }> = [] + const push = (label: string, n: number | undefined) => { + if (typeof n === 'number' && n > 0) out.push({ label, count: n }) + } + push('schedule', c.schedule_count) + push('kafka', c.kafka_count) + push('mqtt', c.mqtt_count) + push('nats', c.nats_count) + push('postgres', c.postgres_count) + push('sqs', c.sqs_count) + push('gcp', c.gcp_count) + push('email', c.email_count) + push('http route', c.http_routes_count) + push('websocket', c.websocket_count) + push('webhook token', c.webhook_count) + push('default-email token', c.default_email_count) + push('nextcloud', c.nextcloud_count) + push('google', c.google_count) + push('github', c.github_count) + push('azure', c.azure_count) + return out + }) + let attachedTotal = $derived(attachedSummary.reduce((s, { count }) => s + count, 0)) + export async function openDrawer( initialPath_l: string, summary_l: string | undefined, @@ -35,6 +74,7 @@ path = undefined dirtyPath = false onBehalfOfEmail = undefined + attachedTriggers = undefined initialPath = initialPath_l initialSummary = summary_l ?? '' summary = summary_l @@ -43,6 +83,22 @@ if (kind === 'flow') { onBehalfOfEmail = await checkFlowOnBehalfOf($workspaceStore!, initialPath_l) } + if (kind === 'script' || kind === 'flow') { + void loadAttachedTriggers() + } + } + + async function loadAttachedTriggers() { + try { + const workspace = $workspaceStore! + attachedTriggers = + kind === 'flow' + ? await FlowService.getTriggersCountOfFlow({ workspace, path: initialPath }) + : await ScriptService.getTriggersCountOfScript({ workspace, path: initialPath }) + } catch { + // Non-fatal: the rename still works without the summary panel. + attachedTriggers = undefined + } } function loadOwner() { @@ -76,6 +132,23 @@ This flow will be redeployed on behalf of you ({$userStore?.email}) instead of {onBehalfOfEmail} {/if} + {#if (kind === 'script' || kind === 'flow') && attachedTotal > 0} + +
+ {#each attachedSummary as { label, count } (label)} + {count} {label}{count === 1 + ? '' + : 's'} + {/each} +
+ + {/if} {#if !hideTarget}
-

- Pick a script or flow to be triggered -

-
- + {#if fixedScriptPath != ''} + + {:else} +

+ Pick a script or flow to be triggered +

+
+ - {#if emptyString(script_path) && is_flow === false} -
- -
- {/if} -
+ {#if emptyString(script_path) && is_flow === false} +
+ +
+ {/if} +
+ {/if}
{/if} diff --git a/frontend/src/lib/components/triggers/schedules/ScheduleEditor.svelte b/frontend/src/lib/components/triggers/schedules/ScheduleEditor.svelte index c75e215e79..a0010829a5 100644 --- a/frontend/src/lib/components/triggers/schedules/ScheduleEditor.svelte +++ b/frontend/src/lib/components/triggers/schedules/ScheduleEditor.svelte @@ -5,20 +5,21 @@ let { onUpdate }: { onUpdate?: (path?: string) => void } = $props() let open = $state(false) - export async function openEdit(ePath: string, isFlow: boolean) { + export async function openEdit(ePath: string, isFlow: boolean, fixedScriptPath?: string) { open = true await tick() - drawer?.openEdit(ePath, isFlow) + drawer?.openEdit(ePath, isFlow, undefined, fixedScriptPath) } export async function openNew( is_flow: boolean, initial_script_path?: string, - schedule_path?: string + schedule_path?: string, + fixedScriptPath?: string ) { open = true await tick() - drawer?.openNew(is_flow, initial_script_path, undefined, schedule_path) + drawer?.openNew(is_flow, initial_script_path, undefined, schedule_path, fixedScriptPath) } let drawer: ScheduleEditorInner | undefined = $state() diff --git a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte index b7652da98f..657f2147c8 100644 --- a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte +++ b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte @@ -8,6 +8,7 @@ import LabelsInput from '$lib/components/LabelsInput.svelte' import Required from '$lib/components/Required.svelte' import ScriptPicker from '$lib/components/ScriptPicker.svelte' + import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte' import ErrorOrRecoveryHandler from '$lib/components/ErrorOrRecoveryHandler.svelte' import Toggle from '$lib/components/Toggle.svelte' import Tooltip from '$lib/components/Tooltip.svelte' @@ -95,6 +96,10 @@ let dynamicSkipPath: string | undefined = $state(undefined) let script_path = $state('') let initialScriptPath = $state('') + // When non-empty, the drawer was opened from the pipeline editor for an + // already-bound script. We swap the runnable ScriptPicker for a read-only + // viewer so the trigger can't be silently reassigned off the pipeline. + let fixedScriptPath = $state('') let runnable: Script | Flow | undefined = $state() let args: Record = $state({}) let loading = $state(false) @@ -141,7 +146,12 @@ deployed: () => initialConfig }) - export async function openEdit(ePath: string, isFlow: boolean, defaultCfg?: Record) { + export async function openEdit( + ePath: string, + isFlow: boolean, + defaultCfg?: Record, + fixedScriptPath_?: string + ) { let loadingTimeout = setTimeout(() => { showLoading = true }, 100) // Do not show loading spinner for the first 100ms @@ -151,6 +161,7 @@ initialPath = ePath itemKind = isFlow ? 'flow' : 'script' path = defaultCfg?.path ?? ePath + fixedScriptPath = fixedScriptPath_ ?? '' await loadSchedule(defaultCfg) edit = true if (!defaultCfg) { @@ -276,7 +287,8 @@ nis_flow: boolean, initial_script_path?: string, defaultValues?: Schedule, - schedule_path?: string + schedule_path?: string, + fixedScriptPath_?: string ) { let loadingTimeout = setTimeout(() => { showLoading = true @@ -298,6 +310,7 @@ edit = false itemKind = (s?.is_flow ?? nis_flow) ? 'flow' : 'script' initialScriptPath = initial_script_path ?? '' + fixedScriptPath = fixedScriptPath_ ?? '' path = initNewPath ? '' : (defaultValues?.path ?? (trigger?.isPrimary ? initialScriptPath : '')) @@ -874,7 +887,9 @@
{#if !hideTarget} - {#if !edit} + {#if fixedScriptPath != ''} + + {:else if !edit}

Pick a script or flow to be triggered by the schedule

diff --git a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditor.svelte b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditor.svelte index f60dfc83a5..e27f78e792 100644 --- a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditor.svelte +++ b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditor.svelte @@ -5,10 +5,10 @@ let { onUpdate }: { onUpdate?: (path?: string) => void } = $props() let open = $state(false) - export async function openEdit(ePath: string, isFlow: boolean) { + export async function openEdit(ePath: string, isFlow: boolean, fixedScriptPath?: string) { open = true await tick() - drawer?.openEdit(ePath, isFlow) + drawer?.openEdit(ePath, isFlow, undefined, fixedScriptPath) } export async function openNew( diff --git a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte index 205db6f925..76f61f0429 100644 --- a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte @@ -18,6 +18,7 @@ import SqsTriggerEditorConfigSection from './SqsTriggerEditorConfigSection.svelte' import Section from '$lib/components/Section.svelte' import ScriptPicker from '$lib/components/ScriptPicker.svelte' + import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte' import Required from '$lib/components/Required.svelte' import { untrack, type Snippet } from 'svelte' import TriggerEditorToolbar from '../TriggerEditorToolbar.svelte' @@ -125,7 +126,8 @@ export async function openEdit( ePath: string, isFlow: boolean, - defaultConfig?: Record + defaultConfig?: Record, + fixedScriptPath_?: string ) { let loadingTimeout = setTimeout(() => { showLoading = true @@ -137,6 +139,7 @@ itemKind = isFlow ? 'flow' : 'script' edit = true dirtyPath = false + fixedScriptPath = fixedScriptPath_ ?? '' await loadTrigger(defaultConfig) // Snapshot the *backend* config as the baseline before overlaying // any local autosave, so hasChanged / onConfigChange correctly @@ -449,34 +452,38 @@ {#if !hideTarget}
-

- Pick a script or flow to be triggered -

-
- - {#if emptyString(script_path)} - - {/if} -
+ initialPath={initialScriptPath} + kinds={['script']} + allowFlow={true} + bind:itemKind + bind:scriptPath={script_path} + allowRefresh={can_write} + allowEdit={!$userStore?.operator} + clearable + /> + {#if emptyString(script_path)} + + {/if} + + {/if}
{/if} diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index 513234bf7a..d3de4e09fd 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -9,6 +9,7 @@ import PipelineEventLog from '$lib/components/assets/AssetGraph/PipelineEventLog.svelte' import AssetGraphDetailsPane from '$lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte' import PipelinePickerModal from '$lib/components/assets/AssetGraph/PipelinePickerModal.svelte' + import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte' import { extractWrites, extractReads, @@ -45,9 +46,17 @@ Save } from 'lucide-svelte' import { + EmailTriggerService, + GcpTriggerService, JobService, + KafkaTriggerService, + MqttTriggerService, + NatsTriggerService, OpenAPI, + PostgresTriggerService, + ScheduleService, ScriptService, + SqsTriggerService, type AssetKind, type Script, type ScriptLang @@ -67,6 +76,7 @@ import SqsTriggerEditor from '$lib/components/triggers/sqs/SqsTriggerEditor.svelte' import GcpTriggerEditor from '$lib/components/triggers/gcp/GcpTriggerEditor.svelte' import EmailTriggerEditor from '$lib/components/triggers/email/EmailTriggerEditor.svelte' + import ScheduleEditor from '$lib/components/triggers/schedules/ScheduleEditor.svelte' // Variables and resources are declarative config, not pipeline assets — // they're hub-shaped (referenced by most runnables) and would swamp the @@ -91,19 +101,27 @@ // with a placeholder name. let pathPrefix = $derived(`f/${folder}/`) const DEFAULT_PATH_SUFFIX = 'new_pipeline_script' - // Default cron for top + pipeline scripts (pipeline roots). Every hour is - // a sane middle ground between batch and real-time; users edit the - // `// schedule "..."` line in the editor before saving if they want - // something different. Asset-triggered scripts don't get a schedule by - // default — they inherit their trigger from the upstream asset. - const DEFAULT_SCHEDULE_CRON = '0 * * * *' - // In-flight drafts keyed by script path. Multiple can coexist — clicking // + repeatedly creates additional drafts, each with its own random // output asset, and they all render on the graph simultaneously. // Saving removes a draft from the map; closing the pane keeps it so the // user can come back to it. + // Counter-based id source — sufficient for "stable across renames in + // this session"; doesn't need to survive a reload. (We have crypto. + // randomUUID() too but a short numeric id keeps localStorage tidy.) + let nextDraftLocalIdCounter = 0 + function newDraftLocalId(): string { + nextDraftLocalIdCounter += 1 + return `d${nextDraftLocalIdCounter}-${Date.now()}` + } + type Draft = { + // Stable per-draft identifier, generated on first create and + // preserved across renames. Used to track concurrent deploys (a + // fast double-rename otherwise fires two saves that each leave a + // persisted script behind — the latest deploy archives the prior + // one keyed on this id). + localId: string script: Script // Undefined when the user picked `outputKind === 'none'` — the draft // has no auto-generated output asset, so the graph overlay skips @@ -180,7 +198,12 @@ const loaded = new Map() for (const entry of state.drafts) { if (entry && typeof entry[0] === 'string' && entry[1]?.script) { - loaded.set(entry[0], entry[1] as Draft) + const d = entry[1] as Draft + // Backfill localId for state persisted by older builds. + if (typeof d.localId !== 'string' || d.localId === '') { + d.localId = newDraftLocalId() + } + loaded.set(entry[0], d) } } if (loaded.size > 0) drafts = loaded @@ -253,7 +276,6 @@ annotations: { inPipeline: false, triggerAssets: [], - schedules: [], nativeTriggers: [] } }) @@ -269,27 +291,79 @@ assets: AssetWithAltAccessType[] }>({ scriptPath: undefined, assets: [] }) - // Sticky cache of inferred body assets per script path — accumulates as - // the user opens scripts in this session, so once we've seen a write - // for `f/foo/bar`, the edge stays on the canvas even after the user - // selects a different node. Without this, switching selection would - // drop the previous script's edges back to whatever's in `base.edges` - // (often nothing for scripts whose deploy didn't extract body assets, - // e.g. when an older WASM was used at save time). - let inferredWritesByPath = $state>>( - new Map() - ) - // Same sticky cache for read usages (e.g. duckdb `read_parquet('s3://…')`, - // loadS3File). Keeps the asset → reader lineage edge live as the body is - // edited / across selection, instead of only after Save re-derives the - // persisted asset rows. - let inferredReadsByPath = $state>>(new Map()) + // Only-add cache of (script_path → body content) populated lazily by + // `bodyFetchEffect` for every script in the current folder. We never + // remove entries: stale keys (renamed-away, deleted) are simply ignored + // at read time because the derived maps below only iterate paths that + // 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()) + // Sibling cache: the parsed asset usages from `inferAssets` (wasm), one + // pass per body. Same only-add semantics as `bodiesByPath`. + let inferredAssetsByPath = $state>(new Map()) + // Bumped on folder change so an in-flight prefetch sweep stops before + // writing into the new folder's state. + let bodyFetchGen = 0 - // Sticky cache of native trigger kinds declared via `// on kafka` etc. - // in each deployed script's source. Filled by the prefetch sweep below. - // resolveGraph uses this to flag scripts whose annotation has no - // matching trigger row — red placeholder on the canvas. - let annotatedNativeKindsByPath = $state>>(new Map()) + // Inferred write/read edges per script-in-graph. Derived from + // (a) the open pane's live overlay (`liveBodyAssets`) — current + // keystrokes for the script the user is editing right now, and + // (b) the prefetched assets cache for everyone else. + // Iteration is gated on `graphRes.current.runnables`, so a path that + // gets renamed / deleted disappears from the derived map as soon as + // the refetch lands — no manual rekey, no phantom edges. + let inferredWritesByPath = $derived.by(() => { + const out = new Map>() + const g = graphRes.current + if (!g) return out + const liveAssetsForPath = (path: string) => + liveBodyAssets.scriptPath === path ? liveBodyAssets.assets : inferredAssetsByPath.get(path) + for (const r of g.runnables) { + if (r.usage_kind !== 'script') continue + const assets = liveAssetsForPath(r.path) + if (!assets) continue + const w = extractWrites(assets) + if (w.length > 0) out.set(r.path, w) + } + return out + }) + let inferredReadsByPath = $derived.by(() => { + const out = new Map>() + const g = graphRes.current + if (!g) return out + const liveAssetsForPath = (path: string) => + liveBodyAssets.scriptPath === path ? liveBodyAssets.assets : inferredAssetsByPath.get(path) + for (const r of g.runnables) { + if (r.usage_kind !== 'script') continue + const assets = liveAssetsForPath(r.path) + if (!assets) continue + const reads = extractReads(assets) + if (reads.length > 0) out.set(r.path, reads) + } + return out + }) + // Same derived shape for `// on kafka` etc. annotations. Live buffer + // wins for the open script; everyone else is parsed from the + // prefetched body content. + let annotatedNativeKindsByPath = $derived.by(() => { + const out = new Map>() + const g = graphRes.current + if (!g) return out + const livePath = liveAnnotations.scriptPath + for (const r of g.runnables) { + if (r.usage_kind !== 'script') continue + let kinds: Set + if (r.path === livePath) { + kinds = new Set(liveAnnotations.annotations.nativeTriggers.map((n) => n.kind)) + } else { + const body = bodiesByPath.get(r.path) + if (!body) continue + kinds = new Set(parsePipelineAnnotations(body).nativeTriggers.map((n) => n.kind)) + } + if (kinds.size > 0) out.set(r.path, kinds) + } + return out + }) // Build a runnable Script from picked language / triggers / output. // Delegates to the shared template generator (pipelineTemplates.ts) so @@ -346,7 +420,7 @@ // the user picked `none`, `outputAsset` is undefined and the graph // overlay skips synthesizing a write edge. const next = new Map(drafts) - next.set(scriptPath, { script, outputAsset: out }) + next.set(scriptPath, { localId: newDraftLocalId(), script, outputAsset: out }) drafts = next activeDraftPath = scriptPath selection = undefined @@ -542,7 +616,17 @@ if (!savedPaths.includes(k)) next.set(k, v) } drafts = next + // If the open draft just got deployed, transfer the focus to + // its now-persisted runnable so the pane stays on the same + // script the user was editing — otherwise the pane closes, + // the canvas re-fits, and the user has to re-find their + // script after every save. if (activeDraftPath && savedPaths.includes(activeDraftPath)) { + selection = { + kind: 'runnable', + runnable_kind: 'script', + path: activeDraftPath + } activeDraftPath = undefined } await graphRes.refetch() @@ -563,8 +647,7 @@ const next = new Map(drafts) next.delete(path) drafts = next - if (activeDraftPath === path) activeDraftPath = undefined - clearSaveError(path) + forgetPath(path) } function clearSaveError(path: string) { @@ -574,6 +657,30 @@ saveErrors = next } + // "This path is gone" cleanup. The three big inferred-* / annotated-* + // maps used to live here too, but they're now derived from + // `bodiesByPath` × `g.runnables` — entries for missing paths drop out + // implicitly when `g.runnables` no longer mentions them, so the only + // things left to flush are the live overlays for the open pane + the + // selection + per-path save errors. `bodiesByPath` keeps its entry + // (only-add cache, harmless if stale). + function forgetPath(path: string) { + if (activeDraftPath === path) activeDraftPath = undefined + if (selection?.kind === 'runnable' && selection.path === path) { + selection = undefined + } + if (liveAnnotations.scriptPath === path) { + liveAnnotations = { + scriptPath: undefined, + annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] } + } + } + if (liveBodyAssets.scriptPath === path) { + liveBodyAssets = { scriptPath: undefined, assets: [] } + } + clearSaveError(path) + } + // Rename a draft in place — re-key its entry in the drafts map and // repoint activeDraftPath. Returns false (or an error string) if the // new path collides with another draft so the dialog can keep itself @@ -596,6 +703,25 @@ } drafts = next if (activeDraftPath === oldPath) activeDraftPath = newPath + // Path-keyed live overlays: re-key for the renamed draft so the + // graph stays consistent between the moment we mutate `drafts` + // here and the next editor event that re-emits annotations / + // asset usages with the new path. Without this re-key, + // `resolveGraph` strips seeded triggers for the OLD path while + // re-applying live overlays against the same OLD path — leaving + // phantom edges that displace the + node off the top of the + // graph and shuffle the layout. + if (liveAnnotations.scriptPath === oldPath) { + liveAnnotations = { ...liveAnnotations, scriptPath: newPath } + } + if (liveBodyAssets.scriptPath === oldPath) { + liveBodyAssets = { ...liveBodyAssets, scriptPath: newPath } + } + // `inferredWritesByPath` / `inferredReadsByPath` / + // `annotatedNativeKindsByPath` are derived from `g.runnables` × + // the body caches, so a rename naturally re-derives them when the + // drafts loop in `resolveGraph` swaps to the new path — no + // per-mutation rekey needed. // Errors are keyed by path — re-key the entry so a previously // failed draft keeps its error visible against the new path // after rename. @@ -606,9 +732,122 @@ nextErrors.set(newPath, msg) saveErrors = nextErrors } + // Auto-deploy the rename. Without this, native triggers can't be + // attached to the renamed script (the trigger row's script_path + // needs to point at a real script) and the user has to remember + // a separate "save" step. Backend `update_triggers_script_path` + // already cascades the path change across every trigger table on + // script create, so any triggers previously attached to the old + // path follow the rename automatically. + void deployRenamedDraft(newPath) return true } + // Per-draft deploy queue. Each draft (keyed by stable `localId`) has at + // most one deploy in flight; concurrent renames append to a single + // `queuedPath` slot so we only ever fire one extra deploy at the + // latest target. `lastDeployedPath` lets the runner archive any + // intermediate path it deployed before the user's final rename + // settled — otherwise a double-rename leaves a phantom script at the + // intermediate path. + const deployQueue = new Map< + string, + { inflight: boolean; queuedPath: string | undefined; lastDeployedPath: string | undefined } + >() + + function deployRenamedDraft(path: string) { + const draft = drafts.get(path) + if (!draft) return + const localId = draft.localId + let state = deployQueue.get(localId) + if (state?.inflight) { + // Coalesce: only the latest target matters. The runner picks + // it up when the current deploy completes. + state.queuedPath = path + return + } + state = state ?? { inflight: false, queuedPath: undefined, lastDeployedPath: undefined } + state.inflight = true + state.queuedPath = undefined + deployQueue.set(localId, state) + void runDeployQueue(localId, path) + } + + async function runDeployQueue(localId: string, initialPath: string) { + let path = initialPath + const state = deployQueue.get(localId)! + try { + while (true) { + if (!$workspaceStore) break + const draft = drafts.get(path) + if (!draft) break + try { + await saveDraft(path, draft, $workspaceStore) + // Archive the previously-deployed intermediate path (if + // any) — a double-rename otherwise leaves it as an + // orphan script visible on the canvas as a "deployed" + // runnable that the user never intended to keep. + const prev = state.lastDeployedPath + if (prev && prev !== path) { + try { + await ScriptService.archiveScriptByPath({ + workspace: $workspaceStore, + path: prev + }) + } catch (archiveErr) { + // Non-fatal: surface a warning so the user + // knows to clean up manually. The fresh deploy + // at the new path still landed. + const msg = (archiveErr as any)?.body ?? (archiveErr as any)?.message ?? String(archiveErr) + sendUserToast( + `Renamed to "${path}" but couldn't archive the old "${prev}": ${msg}`, + true + ) + } + } + state.lastDeployedPath = path + // Drop the now-deployed draft from the local map only + // if no newer rename was queued during the save; if a + // queued path is waiting, the next loop iteration will + // pick it up and we keep the draft live. + if (!state.queuedPath) { + const nextDrafts = new Map(drafts) + nextDrafts.delete(path) + drafts = nextDrafts + if (activeDraftPath === path) { + selection = { kind: 'runnable', runnable_kind: 'script', path } + activeDraftPath = undefined + } + if (saveErrors.has(path)) { + const nextErrors = new Map(saveErrors) + nextErrors.delete(path) + saveErrors = nextErrors + } + await graphRes.refetch() + } + } catch (e: any) { + const msg = e?.body ?? e?.message ?? String(e) + sendUserToast(`Could not deploy rename to "${path}": ${msg}`, true) + const nextErrors = new Map(saveErrors) + nextErrors.set(path, msg) + saveErrors = nextErrors + } + // Pick up the next queued rename, if any. + const next = state.queuedPath + if (!next || next === path) break + path = next + state.queuedPath = undefined + } + } finally { + state.inflight = false + // On the rare path where the draft is also gone (deployed + + // no queued path), drop the slot to keep the map bounded. + if (!drafts.has(path) && !state.queuedPath) { + deployQueue.delete(localId) + } + } + } + // Currently-open draft shape (if any) — fed into the details pane. let activeDraft = $derived(activeDraftPath ? drafts.get(activeDraftPath) : undefined) @@ -626,31 +865,13 @@ liveAnnotations = { scriptPath, annotations } } function handleAssetsChange(scriptPath: string | undefined, assets: AssetWithAltAccessType[]) { + // Single update site for the live overlay. `inferredWritesByPath` + // / `inferredReadsByPath` are now derived from `liveBodyAssets` + // (for the open script) + `inferredAssetsByPath` (prefetched + // snapshot for every other script), so we don't have to write + // into those caches here — the derive picks up our update on the + // next reactive tick. liveBodyAssets = { scriptPath, assets } - // Seed the sticky cache so the script's writes survive selection - // changes. Only the active script's entry is updated (the - // scriptPath the WASM just parsed); other entries are untouched - // so previously-seen scripts retain their last-known writes. - // `untrack` around the read+write of `inferredWritesByPath` is - // crucial: this fn is called from AssetGraphDetailsPane's $effect - // for onAssetsChange, and Svelte tracks every state read inside - // that effect's closure. Without `untrack`, cloning the Map - // registers it as a dep, and writing the new Map immediately - // re-fires the effect → infinite loop. - if (scriptPath) { - const writes = extractWrites(assets) - const reads = extractReads(assets) - untrack(() => { - const nextW = new Map(inferredWritesByPath) - if (writes.length > 0) nextW.set(scriptPath, writes) - else nextW.delete(scriptPath) - inferredWritesByPath = nextW - const nextR = new Map(inferredReadsByPath) - if (reads.length > 0) nextR.set(scriptPath, reads) - else nextR.delete(scriptPath) - inferredReadsByPath = nextR - }) - } } function handleDraftPersist( p: string, @@ -829,9 +1050,24 @@ let sqsEditor: SqsTriggerEditor | undefined = $state() let gcpEditor: GcpTriggerEditor | undefined = $state() let emailEditor: EmailTriggerEditor | undefined = $state() + let scheduleEditor: ScheduleEditor | undefined = $state() function openMissingTriggerDrawer(kind: NativeTriggerKind, scriptPath: string) { + // A native trigger row stores `script_path` as a hard reference — + // pointing it at a never-saved draft would either fail at create + // 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). + if (drafts.has(scriptPath)) { + sendUserToast( + `Save the script "${scriptPath}" first — triggers can only be attached to deployed scripts.`, + true + ) + return + } switch (kind) { + case 'schedule': + return scheduleEditor?.openNew(false, scriptPath, undefined, scriptPath) case 'kafka': return kafkaEditor?.openNew(false, scriptPath) case 'mqtt': @@ -846,7 +1082,98 @@ return gcpEditor?.openNew(false, scriptPath) case 'email': return emailEditor?.openNew(false, scriptPath) - // webhook has no dedicated editor; schedule is inline-managed. + // webhook has no dedicated editor. + default: + return + } + } + + // Lock the script-picker to the related script so the user can't + // reassign the trigger off this pipeline from the canvas. The trigger + // can still be edited everywhere else (TriggersPanel, etc.) where the + // picker stays editable. + // Delete-trigger confirmation state. Kebab → Delete opens the standard + // ConfirmationModal; the actual delete is dispatched from onConfirmed + // so the dialog stays consistent with the rest of the app. + let triggerDeleteTarget = $state< + { kind: NativeTriggerKind; path: string } | undefined + >(undefined) + let triggerDeleteLoading = $state(false) + let triggerDeleteOpen = $derived(triggerDeleteTarget != undefined) + + function deleteAttachedTrigger(kind: NativeTriggerKind, triggerPath: string) { + triggerDeleteTarget = { kind, path: triggerPath } + } + + async function confirmDeleteAttachedTrigger() { + if (!triggerDeleteTarget || !$workspaceStore) return + const { kind, path: triggerPath } = triggerDeleteTarget + const workspace = $workspaceStore + triggerDeleteLoading = true + try { + switch (kind) { + case 'schedule': + await ScheduleService.deleteSchedule({ workspace, path: triggerPath }) + break + case 'kafka': + await KafkaTriggerService.deleteKafkaTrigger({ workspace, path: triggerPath }) + break + case 'mqtt': + await MqttTriggerService.deleteMqttTrigger({ workspace, path: triggerPath }) + break + case 'nats': + await NatsTriggerService.deleteNatsTrigger({ workspace, path: triggerPath }) + break + case 'postgres': + await PostgresTriggerService.deletePostgresTrigger({ workspace, path: triggerPath }) + break + case 'sqs': + await SqsTriggerService.deleteSqsTrigger({ workspace, path: triggerPath }) + break + case 'gcp': + await GcpTriggerService.deleteGcpTrigger({ workspace, path: triggerPath }) + break + case 'email': + await EmailTriggerService.deleteEmailTrigger({ workspace, path: triggerPath }) + break + default: + return + } + sendUserToast(`Deleted ${kind} trigger "${triggerPath}"`) + triggerDeleteTarget = undefined + await graphRes.refetch() + } catch (e: any) { + sendUserToast( + `Could not delete ${kind} trigger "${triggerPath}": ${e?.body ?? e?.message ?? String(e)}`, + true + ) + } finally { + triggerDeleteLoading = false + } + } + + function openEditTriggerDrawer( + kind: NativeTriggerKind, + triggerPath: string, + scriptPath: string + ) { + switch (kind) { + case 'schedule': + return scheduleEditor?.openEdit(triggerPath, false, scriptPath) + case 'kafka': + return kafkaEditor?.openEdit(triggerPath, false, scriptPath) + case 'mqtt': + return mqttEditor?.openEdit(triggerPath, false, scriptPath) + case 'nats': + return natsEditor?.openEdit(triggerPath, false, scriptPath) + case 'postgres': + return postgresEditor?.openEdit(triggerPath, false, scriptPath) + case 'sqs': + return sqsEditor?.openEdit(triggerPath, false, scriptPath) + case 'gcp': + return gcpEditor?.openEdit(triggerPath, false, scriptPath) + case 'email': + return emailEditor?.openEdit(triggerPath, false, scriptPath) default: return } @@ -920,74 +1247,52 @@ } ) - // Eagerly infer every folder script's writes on load and seed the same - // `inferredWritesByPath` overlay the open-script path uses — otherwise a - // script whose persisted asset rows are missing only gets edges when - // clicked, which re-layouts the graph. WHY untrack: deps must stay - // (workspace, base-graph) only so a keystroke/new draft doesn't re-sweep; - // the generation token cancels an in-flight sweep on folder change. - let assetPrefetchGen = 0 - // True while the load-time sweep above is still parsing scripts — drives - // a small "parsing assets…" hint so the user knows the graph is still - // settling (edges may still appear) rather than already complete. + // Body / inferred-assets prefetch sweep. Watches `g.runnables`; for any + // non-draft path we haven't fetched yet, fetches `getScriptByPath` and + // `inferAssets`, and stores both in their respective only-add caches. + // All three previously-sticky maps (`inferredWritesByPath`, + // `inferredReadsByPath`, `annotatedNativeKindsByPath`) are now derived + // from these caches × the current graph, so rename / delete cleanup + // happens implicitly when `g.runnables` changes — no per-mutation + // cache surgery needed. A generation counter cancels in-flight work on + // folder change so the previous folder's results never leak into the + // new one. let prefetchingAssets = $state(false) $effect(() => { const ws = $workspaceStore const g = graphRes.current if (!ws || !g) return - const gen = ++assetPrefetchGen + const gen = ++bodyFetchGen const targets = untrack(() => g.runnables .filter((r) => r.usage_kind === 'script') .map((r) => r.path) - .filter( - (p) => - !drafts.has(p) && - !inferredWritesByPath.has(p) && - !inferredReadsByPath.has(p) && - !annotatedNativeKindsByPath.has(p) - ) + .filter((p) => !drafts.has(p) && !bodiesByPath.has(p)) ) if (targets.length === 0) return let i = 0 const POOL = 6 const worker = async () => { - while (i < targets.length && gen === assetPrefetchGen) { + while (i < targets.length && gen === bodyFetchGen) { const path = targets[i++] try { const s = await ScriptService.getScriptByPath({ workspace: ws, path }) - if (gen !== assetPrefetchGen) return + if (gen !== bodyFetchGen) return const content = s.content ?? '' const res = await inferAssets(s.language, content) - if (gen !== assetPrefetchGen) return + if (gen !== bodyFetchGen) return const inferred = (res?.assets ?? []) as AssetWithAltAccessType[] - const writes = extractWrites(inferred) - const reads = extractReads(inferred) - // Parse `// on kafka` markers in parallel with the asset - // inference. Cheap pure-TS pass — runs on the same content - // we already loaded for inferAssets. - const annotated = new Set( - parsePipelineAnnotations(content).nativeTriggers.map((n) => n.kind) - ) untrack(() => { - // A live edit / prior sweep may have filled either meanwhile. - if (writes.length > 0 && !inferredWritesByPath.has(path)) { - const next = new Map(inferredWritesByPath) - next.set(path, writes) - inferredWritesByPath = next + if (!bodiesByPath.has(path)) { + const nextBodies = new Map(bodiesByPath) + nextBodies.set(path, content) + bodiesByPath = nextBodies } - if (reads.length > 0 && !inferredReadsByPath.has(path)) { - const next = new Map(inferredReadsByPath) - next.set(path, reads) - inferredReadsByPath = next + if (!inferredAssetsByPath.has(path)) { + const nextAssets = new Map(inferredAssetsByPath) + nextAssets.set(path, inferred) + inferredAssetsByPath = nextAssets } - // Always seed the annotation map (even if empty) so a - // later removal of all `// on kafka` lines retires the - // placeholder on the next deploy + refetch. - const nextAnnot = new Map(annotatedNativeKindsByPath) - if (annotated.size > 0) nextAnnot.set(path, annotated) - else nextAnnot.delete(path) - annotatedNativeKindsByPath = nextAnnot }) } catch { // Skip — that node just falls back to base-graph edges. @@ -997,9 +1302,7 @@ prefetchingAssets = true const pool = Array.from({ length: Math.min(POOL, targets.length) }, () => worker()) void Promise.all(pool).then(() => { - // Only clear for the sweep still current — a folder change starts - // a new gen (and its own true) that this stale resolve mustn't undo. - if (gen === assetPrefetchGen) prefetchingAssets = false + if (gen === bodyFetchGen) prefetchingAssets = false }) }) @@ -1155,8 +1458,9 @@ runStates={activeRunnables.states} {pathPrefix} defaultPathSuffix={DEFAULT_PATH_SUFFIX} - defaultScheduleCron={DEFAULT_SCHEDULE_CRON} onCreateMissingTrigger={openMissingTriggerDrawer} + onEditTrigger={openEditTriggerDrawer} + onDeleteTrigger={deleteAttachedTrigger} onselect={(s) => { // Clicking a draft runnable node re-opens it in the pane; // clicking anything else selects it normally and detaches @@ -1365,7 +1669,6 @@ annotations: { inPipeline: false, triggerAssets: [], - schedules: [], nativeTriggers: [] } } @@ -1375,7 +1678,24 @@ if (activeDraftPath) discardDraft(activeDraftPath) }} onDraftSaved={async (savedPath) => { - discardDraft(savedPath) + // Drop the now-deployed draft and hand focus to its + // persisted runnable so the pane stays open on the + // same script. `discardDraft` would clear + // activeDraftPath without setting selection — the + // canvas would deselect and the view reset on the + // next refetch. + const nextDrafts = new Map(drafts) + nextDrafts.delete(savedPath) + drafts = nextDrafts + if (activeDraftPath === savedPath) { + selection = { + kind: 'runnable', + runnable_kind: 'script', + path: savedPath + } + activeDraftPath = undefined + } + clearSaveError(savedPath) await graphRes.refetch() }} onPersistedSaved={async () => { @@ -1400,12 +1720,15 @@ await graphRes.refetch() }} onScriptRemoved={async (removedPath) => { - // Drop the selection (the script is gone) and - // refetch so the runnable node disappears from - // the canvas. - if (selection?.kind === 'runnable' && selection.path === removedPath) { - selection = undefined - } + // Drop every path-keyed overlay / cache entry + // pointing at the now-archived runnable so + // resolveGraph doesn't keep emitting lineage + // edges or missing-trigger placeholders against + // a script that no longer exists. Without this + // the inferred writes / annotation maps would + // keep dragging phantom nodes onto the canvas + // until the next folder change. + forgetPath(removedPath) await graphRes.refetch() }} /> @@ -1418,6 +1741,26 @@ + { + if (!triggerDeleteLoading) triggerDeleteTarget = undefined + }} + > + {#if triggerDeleteTarget} +

+ Delete {triggerDeleteTarget.path}? The + // on {triggerDeleteTarget.kind} annotation on the script + stays — the trigger will read as missing on the canvas until you recreate it or remove + the annotation. +

+ {/if} + +