This commit is contained in:
Ruben Fiszel
2026-05-27 20:25:17 +00:00
parent ef6c39601c
commit b96d68f0eb
40 changed files with 1744 additions and 728 deletions
Binary file not shown.

After

Width:  |  Height:  |  Size: 76 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 91 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 111 KiB

@@ -123,17 +123,14 @@ pub enum TriggerSpec {
#[serde(skip_serializing_if = "Option::is_none", default)]
debounce: Option<String>,
},
// Refresh on cron. The raw expression is passed through as-is so the
// existing schedule subsystem can validate it.
Schedule {
cron: String,
},
// `// on <kind>` — 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 = <this script>` 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 = <this
// script>` 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<PartitionSpec> {
// 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<TriggerSpec> {
// 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 "<cron>"` 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]
Binary file not shown.

After

Width:  |  Height:  |  Size: 32 KiB

+88 -33
View File
@@ -407,10 +407,10 @@ struct GraphEdge {
}
// Declared `// on <trigger>` 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<String> =
pipeline_member_paths.into_iter().map(|r| r.path).collect();
let existing_script_paths: std::collections::HashSet<String> =
existing_script_paths.into_iter().collect();
let existing_flow_paths: std::collections::HashSet<String> =
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<TriggerEdge> =
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 `<prefix><path>` — 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 `<prefix><path>` — 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 },
+6 -23
View File
@@ -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 "<cron>"` 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 {
+12 -78
View File
@@ -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 "<cron>"` 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 "<cron>"` 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
+374
View File
@@ -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 <script_path>` 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 <kind>` — unit of work + state + backfill (already exists).
- `// unique_key <col>` — 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.
@@ -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<string | undefined>(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<TriggersCount | undefined>(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<Array<{ label: string; count: number }>>(() => {
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}
</Alert>
{/if}
{#if (kind === 'script' || kind === 'flow') && attachedTotal > 0}
<Alert
type="info"
title={`Will also update ${attachedTotal} attached trigger${attachedTotal === 1 ? '' : 's'}`}
class="mb-4"
>
<div class="flex flex-wrap gap-x-3 gap-y-1 mt-1">
{#each attachedSummary as { label, count } (label)}
<span class="text-xs"
><span class="font-mono font-semibold">{count}</span> {label}{count === 1
? ''
: 's'}</span
>
{/each}
</div>
</Alert>
{/if}
<Label label="Summary" class="mb-6">
<TextInput
inputProps={{
@@ -21,22 +21,19 @@
// source>". id === the SCRIPT_TRIGGER_KIND value, so the handler can
// dispatch on it uniformly. Asset-triggered scripts are not in this
// menu; those live under the per-asset + inside the graph.
type KindId = 'schedule' | NativeTriggerKind
type KindId = NativeTriggerKind
interface Props {
data: {
onAddPipelineScript: (
language: ScriptLang,
path: string,
source:
| { kind: 'schedule'; cron: string }
| { kind: NativeTriggerKind; path: string | undefined },
source: { kind: NativeTriggerKind; path: string | undefined },
outputKind: PipelineOutputKind,
aiPrompt?: string
) => void
pathPrefix: string
defaultPathSuffix: string
defaultScheduleCron: string
}
}
let { data }: Props = $props()
@@ -45,27 +42,16 @@
if (!pick.language || !pick.path) return
const kindId = pick.kindId as KindId
const outputKind = (pick.outputKind ?? 'none') as PipelineOutputKind
if (kindId === 'schedule') {
data.onAddPipelineScript(
pick.language as ScriptLang,
pick.path,
{ kind: 'schedule', cron: data.defaultScheduleCron },
outputKind,
pick.aiPrompt
)
} else {
// Native trigger reference: user is expected to fill in the
// trigger path themselves in the editor (or configure it in the
// trigger's own UI). We seed the annotation with an empty ref
// the user replaces.
data.onAddPipelineScript(
pick.language as ScriptLang,
pick.path,
{ kind: kindId, path: undefined },
outputKind,
pick.aiPrompt
)
}
// Native trigger annotation is marker-only — the binding lives on
// the trigger row's own `script_path`, which the user creates
// separately. Seed with `path: undefined`.
data.onAddPipelineScript(
pick.language as ScriptLang,
pick.path,
{ kind: kindId, path: undefined },
outputKind,
pick.aiPrompt
)
}
</script>
@@ -74,7 +60,7 @@
{
id: 'schedule',
label: 'On schedule',
description: 'Cron-driven pipeline script',
description: 'Triggered by a schedule you create',
icon: Clock
},
{
@@ -49,12 +49,10 @@
onAddPipelineScript?: (
language: import('$lib/gen').ScriptLang,
path: string,
source:
| { kind: 'schedule'; cron: string }
| {
kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp'
path: string | undefined
},
source: {
kind: NativeTriggerKind
path: string | undefined
},
outputKind: import('./pipelineTemplates').PipelineOutputKind,
aiPrompt?: string
) => void
@@ -63,8 +61,6 @@
pathPrefix?: string
// Seeded editable suffix (e.g. `new_pipeline_script`).
defaultPathSuffix?: string
// Default cron expression seeded into the top + template.
defaultScheduleCron?: string
// Page-supplied dispatch for the per-asset run button. Receives the
// producer (kind/path/unsaved) and returns the new job id. The page
// implements the unsaved branch by calling runScriptPreview with the
@@ -110,6 +106,20 @@
// opens the matching editor with `script_path` pre-filled — no
// navigation, drafts stay intact.
onCreateMissingTrigger?: (kind: NativeTriggerKind, scriptPath: string) => void
// Click handler for an attached (non-missing) native trigger node —
// opens the matching editor in edit mode for the given trigger path.
// `scriptPath` is the related script — the drawer locks its
// script-picker to it so the trigger can't be reassigned off the
// pipeline from this entry point.
onEditTrigger?: (
kind: NativeTriggerKind,
triggerPath: string,
scriptPath: string
) => void
// Click handler for the kebab → Delete entry on an attached trigger.
// The page is expected to confirm + call the matching delete API
// and refetch the graph.
onDeleteTrigger?: (kind: NativeTriggerKind, triggerPath: string) => void
}
let {
graph,
@@ -119,13 +129,14 @@
onAddPipelineScript,
pathPrefix = '',
defaultPathSuffix = '',
defaultScheduleCron = '',
onRunProducer,
onRunnableMenuRemove,
activeRunnable,
activeRunnableIds,
runStates,
onCreateMissingTrigger
onCreateMissingTrigger,
onEditTrigger,
onDeleteTrigger
}: Props = $props()
const ADD_NODE_ID = '__add__'
@@ -142,7 +153,6 @@
| 'lineage-write'
| 'lineage-read'
| 'trigger-asset'
| 'trigger-schedule'
| 'trigger-native'
| 'add-anchor'
unsaved?: boolean
@@ -177,8 +187,7 @@
data: {
onAddPipelineScript: onAddPipelineScript!,
pathPrefix,
defaultPathSuffix,
defaultScheduleCron
defaultPathSuffix
}
})
}
@@ -250,6 +259,14 @@
}
})
}
// Set of `script_path` values for runnables that are still drafts (no
// DB row yet). Trigger nodes use this to swap "Click to create" for
// "Click to create (after draft save)" so the user knows the create
// button is blocked until the script is deployed.
const unsavedRunnablePaths = new Set<string>()
for (const r of g.runnables) {
if (r.unsaved) unsavedRunnablePaths.add(r.path)
}
for (const r of g.runnables) {
const rid = `${r.usage_kind}:${r.path}`
// Optimistic badge: the moment a run is launched from this view
@@ -372,15 +389,12 @@
})
continue
}
const isMissing = t.trigger_kind !== 'schedule' && (t as any).missing === true
// Schedule: cron is the ref. Native (attached): trigger row path.
// Native (missing): synthesize a per-script ref so each placeholder
// is its own node ("missing kafka on f/foo/bar").
const ref = isMissing
? `missing:${t.runnable_path}`
: t.trigger_kind === 'schedule'
? (t as any).cron
: ((t as any).path ?? '')
const isMissing = (t as any).missing === true
// Native (attached): trigger row path. Native (missing):
// synthesize a per-script ref so each placeholder is its own
// node ("missing kafka on f/foo/bar"). Schedule joins the native
// family — its ref is the schedule row's path, same shape.
const ref = isMissing ? `missing:${t.runnable_path}` : ((t as any).path ?? '')
const sourceId = `trigger:${t.trigger_kind}:${ref}`
recordSourceTrigger(
sourceId,
@@ -388,13 +402,19 @@
ref,
!!t.unsaved,
isMissing,
isMissing ? t.runnable_path : undefined
// 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
// lock. Previously only set for missing, which silently
// disabled `canEdit` (and the resulting click affordance)
// on every attached native trigger.
t.runnable_path
)
edges.push({
id: `trig-${t.trigger_kind}:${sourceId}->${runnableId}`,
source: sourceId,
target: runnableId,
kind: t.trigger_kind === 'schedule' ? 'trigger-schedule' : 'trigger-native',
kind: 'trigger-native',
unsaved: t.unsaved,
missing: isMissing
})
@@ -409,7 +429,12 @@
unsaved: info.allUnsaved,
missing: info.missing,
runnable_path: info.runnable_path,
onCreateMissingTrigger
runnable_unsaved: info.runnable_path
? unsavedRunnablePaths.has(info.runnable_path)
: false,
onCreateMissingTrigger,
onEditTrigger,
onDeleteTrigger
}
})
}
@@ -451,25 +476,52 @@
// Bound on the outer wrapper; updates on pane resize via $state.
let paneWidth = $state(800)
// Compute layout positions strictly off the model — selection changing
// must NOT trigger a re-layout. Sugiyama's decross output depends on
// the order in which nodes/edges arrive, so two builds with identical
// topology but different insertion order can produce different layer
// orderings (selecting a node briefly re-emits `liveAnnotations`-
// driven trigger overlays in a different order, etc.). Sorting both
// arrays by their stable identity (node id, edge source→target)
// gives sugiyama a topology+path-deterministic input — same nodes +
// edges + paths → same layout. Renames *do* change the layout for
// the renamed entry, since its id moves in the sort, but that's
// expected: a rename is a path change, which is part of the input.
let layoutInput = $derived({
nodes: model.nodes
.map((n) => ({ id: n.id, data: n.data }))
.sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)),
edges: model.edges
.map((e) => ({ source: e.source, target: e.target }))
.sort((a, b) =>
a.source === b.source
? a.target < b.target
? -1
: a.target > b.target
? 1
: 0
: a.source < b.source
? -1
: 1
)
})
let layoutPositions = $derived(layoutAssetGraph(layoutInput))
let positionedNodes = $derived.by(() => {
const positions = layoutAssetGraph({
nodes: model.nodes.map((n) => ({ id: n.id, data: n.data })),
edges: model.edges.map((e) => ({ source: e.source, target: e.target }))
})
// Compute bbox width from layout; shift every x so the graph is
// horizontally centered inside the pane. y is untouched so layer 0
// sits at the top of the viewport (matches flow editor's Trigger
// placement — no fitView reshuffling).
let minX = Infinity
let maxX = -Infinity
for (const p of positions.values()) {
for (const p of layoutPositions.values()) {
if (p.x < minX) minX = p.x
if (p.x > maxX) maxX = p.x
}
const bboxWidth = isFinite(minX) ? maxX - minX : 0
const xCenter = paneWidth / 2 - bboxWidth / 2
return model.nodes.map<Node>((n) => {
const p = positions.get(n.id) ?? { x: 0, y: 0 }
const p = layoutPositions.get(n.id) ?? { x: 0, y: 0 }
// Compensate for the + node being narrower than its layout slot
// so it visually centers over the node(s) below.
const xShift = n.id === ADD_NODE_ID ? (NODE.width - ADD_NODE_WIDTH) / 2 : 0
@@ -482,7 +534,10 @@
// All nodes non-draggable: the layout is sugiyama-computed,
// dragging would fight the reactive re-layout. Selection is
// still allowed on asset/runnable (not on the + or schedules)
// for the details-pane click-through.
// for the details-pane click-through. Trigger nodes opt out
// of selection but TriggerNode itself sets `pointer-events: auto`
// on its inner box so the edit/create button still receives
// clicks despite svelte-flow's wrapper-level pointer-events: none.
draggable: false,
selectable: n.id !== ADD_NODE_ID && n.type !== 'trigger'
}
@@ -530,13 +585,6 @@
label = 'triggers'
labelStyle = 'fill: rgb(16 185 129); font-size: 10px; font-weight: 600;'
break
case 'trigger-schedule':
style = 'stroke: rgb(245 158 11); stroke-width: 2px;'
strokeDasharray = '6 3'
markerColor = 'rgb(245 158 11)'
label = 'schedule'
labelStyle = 'fill: rgb(245 158 11); font-size: 10px; font-weight: 600;'
break
case 'trigger-native':
// Colour neutral here because the trigger source node
// already carries per-kind colour; edge just needs to
@@ -325,16 +325,29 @@
// React to the parent's remove-signal counter and pop the same modal
// the in-pane trash button uses. Skipped for drafts (the parent calls
// `onDiscard` directly there) and while the script hasn't loaded.
// Counter pattern (not boolean) so successive triggers re-fire even if
// the modal was just closed without acting.
// `onDiscard` directly there). Counter pattern (not boolean) so
// successive triggers re-fire even if the modal was just closed without
// acting. The intent is held in `pendingRemoveSignal` until the script
// is loaded — otherwise a kebab→Delete on a non-selected node consumes
// the signal before the pane has the script, and the modal never opens.
// The first observation of the counter is treated as the baseline (not
// a fresh request) — the parent ships an initial 0 and clicking a
// script must NOT trigger the delete modal.
let lastRemoveSignal = $state<number | undefined>(undefined)
let pendingRemoveSignal = $state<number | undefined>(undefined)
$effect(() => {
if (requestRemoveSignal === undefined) return
if (requestRemoveSignal === lastRemoveSignal) return
const baseline = lastRemoveSignal === undefined
if (!baseline && requestRemoveSignal === lastRemoveSignal) return
lastRemoveSignal = requestRemoveSignal
if (baseline) return
if (isDraft) return
pendingRemoveSignal = requestRemoveSignal
})
$effect(() => {
if (pendingRemoveSignal === undefined) return
if (!script || !script.hash) return
pendingRemoveSignal = undefined
removeOpen = true
})
@@ -62,8 +62,12 @@
onPick
}: Props = $props()
// When there's only one trigger kind, hide the Trigger column entirely
// and pre-select it so the user lands directly on the Language picker.
const singleKind = $derived(kinds.length === 1 ? kinds[0] : undefined)
const buildEmptySelected = () => ({
triggerId: undefined as undefined | string,
triggerId: singleKind?.id,
language: undefined as undefined | ScriptLang,
outputId: undefined as undefined | PipelineOutputKind,
scriptPath: `${defaultPathSuffix ?? 'pipeline_script'}_${shortSlug()}`,
@@ -152,45 +156,47 @@
</Popover>
{#snippet topSection()}
<div
class={twMerge('flex flex-col gap-1 p-2 w-56 shrink-0 overflow-auto')}
{@attach arrowTabNav({ onKeyDown: selectAndAdvanceTo(() => languageEl) })}
>
<div class="text-2xs font-normal text-secondary ml-2 mb-1">Trigger</div>
{#each kinds as k}
{@const isSelected = selected.triggerId == k.id}
<Button
variant="subtle"
btnClasses={'text-left'}
onClick={() => (selected.triggerId = k.id)}
selected={isSelected}
>
{#if k.icon}
{@const Icon = k.icon}
<Icon
size={14}
class={twMerge(
'shrink-0 my-auto mr-1.5',
isSelected ? 'text-accent' : 'text-secondary'
)}
/>
{/if}
<span class="flex flex-col items-start flex-1 min-w-0">
<span class="text-xs font-normal leading-tight">{k.label}</span>
{#if k.description}
<span
{#if !singleKind}
<div
class={twMerge('flex flex-col gap-1 p-2 w-56 shrink-0 overflow-auto')}
{@attach arrowTabNav({ onKeyDown: selectAndAdvanceTo(() => languageEl) })}
>
<div class="text-2xs font-normal text-secondary ml-2 mb-1">Trigger</div>
{#each kinds as k}
{@const isSelected = selected.triggerId == k.id}
<Button
variant="subtle"
btnClasses={'text-left'}
onClick={() => (selected.triggerId = k.id)}
selected={isSelected}
>
{#if k.icon}
{@const Icon = k.icon}
<Icon
size={14}
class={twMerge(
'text-2xs font-normal leading-snug mt-0.5',
isSelected ? 'text-accent/80' : 'text-hint'
'shrink-0 my-auto mr-1.5',
isSelected ? 'text-accent' : 'text-secondary'
)}
>
{k.description}
</span>
/>
{/if}
</span>
</Button>
{/each}
</div>
<span class="flex flex-col items-start flex-1 min-w-0">
<span class="text-xs font-normal leading-tight">{k.label}</span>
{#if k.description}
<span
class={twMerge(
'text-2xs font-normal leading-snug mt-0.5',
isSelected ? 'text-accent/80' : 'text-hint'
)}
>
{k.description}
</span>
{/if}
</span>
</Button>
{/each}
</div>
{/if}
<div
bind:this={languageEl}
@@ -257,7 +263,7 @@
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div bind:this={pathEl} class="flex" onkeydown={selectAndAdvanceTo(() => aiPromptEl)}>
<div
class="border rounded-md rounded-r-none border-r-0 text-xs w-fit flex items-center px-2 text-secondary bg-surface-input whitespace-nowrap"
class="border rounded-md rounded-r-none border-r-0 text-xs w-fit shrink-0 whitespace-nowrap flex items-center px-2 text-secondary bg-surface-input"
>
{pathPrefix}
</div>
@@ -110,7 +110,10 @@
import { Handle, Position } from '@xyflow/svelte'
import { NODE } from '$lib/components/graph/util'
import { twMerge } from 'tailwind-merge'
import { AlertTriangle } from 'lucide-svelte'
import { AlertTriangle, EllipsisVertical, Trash2 } from 'lucide-svelte'
import DropdownV2 from '$lib/components/DropdownV2.svelte'
import { stopPropagation, preventDefault } from 'svelte/legacy'
import type { Item } from '$lib/utils'
interface Props {
// `ref` is the cron expression for schedules, the trigger-path for
@@ -125,15 +128,39 @@
unsaved?: boolean
missing?: boolean
runnable_path?: string
// True iff the target script is still a draft (no DB row yet).
// Drives the "(after draft save)" hint on the missing
// placeholder, since the page-level handler refuses to open the
// create drawer until the script is deployed.
runnable_unsaved?: boolean
// Page-supplied dispatcher that opens the matching native
// trigger drawer with `script_path` pre-filled. When absent
// (e.g. webhook, schedule, or a kind without an editor) the
// placeholder is non-clickable.
onCreateMissingTrigger?: (kind: NativeTriggerKind, scriptPath: string) => void
// Page-supplied dispatcher to open the matching native trigger
// drawer in edit mode for an attached (non-missing) trigger.
// `triggerPath` is the trigger row's path (e.g. the mqtt_trigger
// row); `scriptPath` is the script the trigger targets — the
// drawer locks its script-picker to this so the user can't
// reassign the trigger off the pipeline. Absent for kinds
// without an editor.
onEditTrigger?: (
kind: NativeTriggerKind,
triggerPath: string,
scriptPath: string
) => void
// Page-supplied dispatcher to delete an attached (non-missing)
// trigger. Confirmation is the caller's responsibility — the
// node just exposes the entry point on the kebab menu.
onDeleteTrigger?: (kind: NativeTriggerKind, triggerPath: string) => void
}
}
let { data }: Props = $props()
let hover = $state(false)
let menuOpen = $state(false)
let style = $derived(TRIGGER_NODE_STYLE[data.kind])
let Icon = $derived(data.missing ? AlertTriangle : style.icon)
let missingTitle = $derived(
@@ -141,28 +168,71 @@
? `Missing ${style.label} trigger: ${data.runnable_path ?? ''} declares \`// on ${style.label}\` but no ${style.label} trigger targets it. Click to create one, or remove the annotation.`
: undefined
)
// Schedule and webhook have no dedicated drawer (schedules are
// inline-managed; webhooks are implicit endpoints), so the placeholder
// is not clickable for those kinds.
// Webhook has no dedicated drawer (it's an implicit endpoint), so the
// placeholder is not clickable for that kind. Schedule + the other
// native kinds all have dedicated editors.
let canCreate = $derived(
data.missing &&
data.kind !== 'schedule' &&
data.kind !== 'webhook' &&
!!data.runnable_path &&
!!data.onCreateMissingTrigger
)
// Attached native trigger → clickable to open its drawer in edit mode.
let canEdit = $derived(
!data.missing &&
data.kind !== 'webhook' &&
!!data.ref &&
!!data.runnable_path &&
!!data.onEditTrigger
)
// Same gating as `canEdit`: the trigger row only exists when there's a
// non-missing ref + a backing editor (i.e. excludes webhook). Schedule
// has its own delete endpoint, same shape as the other natives.
let canDelete = $derived(
!data.missing &&
data.kind !== 'webhook' &&
!!data.ref &&
!!data.onDeleteTrigger
)
let menuItems: Item[] = $derived(
canDelete
? [
{
displayName: 'Delete…',
icon: Trash2,
type: 'delete' as const,
action: () => {
if (!data.ref || !data.onDeleteTrigger) return
data.onDeleteTrigger(data.kind as NativeTriggerKind, data.ref)
}
}
]
: []
)
function handleMissingClick() {
if (!canCreate || !data.runnable_path || !data.onCreateMissingTrigger) return
data.onCreateMissingTrigger(data.kind as NativeTriggerKind, data.runnable_path)
}
function handleEditClick() {
if (!canEdit || !data.ref || !data.runnable_path || !data.onEditTrigger) return
data.onEditTrigger(data.kind as NativeTriggerKind, data.ref, data.runnable_path)
}
</script>
<div class="relative">
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div
class="relative"
onmouseenter={() => (hover = true)}
onmouseleave={() => (hover = false)}
>
{#if canCreate}
<!-- Whole missing placeholder is a button so the cursor + a11y
affordance reads "clickable to fix". Sits in the same visual
box as the non-clickable variants. -->
<!-- Same affordance as the asset's downstream + button: render the
whole node as a button so the cursor + hover + click is obvious
without depending on svelte-flow's wrapper-level handlers. -->
<button
type="button"
onclick={handleMissingClick}
@@ -180,7 +250,33 @@
{style.label} · missing
</span>
<span class="text-2xs font-mono truncate text-red-700 dark:text-red-400">
Click to create
{data.runnable_unsaved ? 'Click to create (after draft save)' : 'Click to create'}
</span>
</div>
</button>
{:else if canEdit}
<!-- Mirrors the missing-trigger pattern: render the whole node as a
button so clicks open the editor drawer reliably (don't rely on
svelte-flow's onnodeclick wiring). -->
<button
type="button"
onclick={handleEditClick}
class={twMerge(
'flex items-center rounded-md drop-shadow-sm overflow-hidden outline outline-1 w-full text-left',
style.bg,
data.unsaved ? `opacity-80 ${style.borderUnsaved}` : style.border,
'hover:brightness-95 dark:hover:brightness-110 transition-[filter]'
)}
style="width: {NODE.width}px; min-height: {NODE.height}px;"
title={`Edit ${style.label} trigger: ${data.ref}`}
>
<Icon size={14} class={`shrink-0 ml-2 mr-2 ${style.iconText}`} />
<div class="flex flex-col min-w-0 flex-1 pr-2 py-0.5 leading-tight">
<span class="text-3xs uppercase tracking-wide truncate text-tertiary">
{style.label}{data.unsaved ? ' · unsaved' : ''}
</span>
<span class="text-2xs font-mono truncate text-emphasis">
{data.ref}
</span>
</div>
</button>
@@ -221,6 +317,38 @@
</div>
</div>
{/if}
{#if menuItems.length > 0}
<!-- Hover-revealed kebab menu (Delete only for now). Mirrors the
RunnableNode pattern: positioned just outside the top-right of
the node, rendered only on hover or while the menu is open so
the canvas stays clean at rest. `pointerdown` is stopped so
svelte-flow doesn't kick off node selection / drag when the
user reaches for the menu. -->
<div class="absolute -top-2 -right-2 h-7 p-1 min-w-7" style="will-change: transform;">
<DropdownV2
items={menuItems}
placement="bottom-end"
bind:open={menuOpen}
fixedHeight={false}
usePointerDownOutside
>
{#snippet buttonReplacement()}
<button
class={twMerge(
'center-center p-1 text-secondary shadow-sm bg-surface duration-0 hover:bg-surface-tertiary',
hover || menuOpen ? 'block' : '!hidden',
'shadow-md rounded-md'
)}
onpointerdown={stopPropagation(preventDefault(() => {}))}
title="Actions"
>
<EllipsisVertical size={12} />
</button>
{/snippet}
</DropdownV2>
</div>
{/if}
</div>
<Handle type="source" position={Position.Bottom} isConnectable={false} />
@@ -135,7 +135,7 @@ export function useActiveRunnableIds(
const since = lastPollTs
const pollStartedMs = Date.now()
let next = new Set<string>()
const inFlightThisTick = new Set<string>()
const runningThisTick = new Set<string>()
let anyInFlight = false
try {
const res = await JobService.listExtendedJobs({
@@ -150,13 +150,25 @@ export function useActiveRunnableIds(
if (!id) continue
const jobId: string | undefined = (j as any).id
const startedTs: string | undefined = (j as any).started_at ?? (j as any).created_at
// `type === 'QueuedJob'` covers both queued-and-waiting *and*
// actually-executing jobs. Only the latter (`running === true`)
// should drive the "running" badge / edge animation — queued
// jobs haven't started yet, so the node should keep its prior
// status until a worker picks it up. We still keep the queued
// row alive for the activity log (built further down) and treat
// it as "in flight" for poll-cadence purposes so the loop
// doesn't disarm while runs are pending.
const isQueued = (j as any).type === 'QueuedJob'
if (isQueued) {
// queued or running — currently active
const isRunning = isQueued && (j as any).running === true
if (isRunning) {
next.add(id)
inFlightThisTick.add(id)
runningThisTick.add(id)
anyInFlight = true
if (jobId) seenInFlightJobIds.add(jobId)
} else if (isQueued) {
// Queued but not yet running — keep the poll alive but
// don't surface this as an executing runnable.
anyInFlight = true
} else {
// completed: catch-up — a hop whose whole lifetime fell
// between two polls (never observed in-flight) still gets
@@ -214,13 +226,16 @@ export function useActiveRunnableIds(
anyInFlight = ids.size > 0
}
if (!setEq(ids, next)) ids = next
// Rebuild the badge snapshot: in-flight wins (spinner), otherwise the
// last completed status. completedHistory persists across `stop()`.
// Rebuild the badge snapshot: actually-running wins (spinner),
// otherwise the last completed status sticks. completedHistory
// persists across `stop()`. Queued-but-not-started jobs are
// intentionally NOT surfaced as 'running' — the node keeps its
// previous badge state until a worker picks the job up.
const snap = new Map<string, RunnableRunState>()
for (const [id, h] of completedHistory) {
snap.set(id, { status: inFlightThisTick.has(id) ? 'running' : h.lastStatus, runs: h.runs })
snap.set(id, { status: runningThisTick.has(id) ? 'running' : h.lastStatus, runs: h.runs })
}
for (const id of inFlightThisTick) {
for (const id of runningThisTick) {
if (!snap.has(id)) snap.set(id, { status: 'running', runs: 0 })
}
if (!statesEq(states, snap)) states = snap
@@ -63,7 +63,7 @@ describe('parsePipelineAnnotations: combined', () => {
it('parses all keywords together', () => {
const code = [
'// pipeline',
'// schedule "0 0 * * *"',
'// on schedule',
'// on s3://in.csv',
'// partitioned daily tz="UTC"',
'// freshness 2h',
@@ -72,7 +72,9 @@ describe('parsePipelineAnnotations: combined', () => {
].join('\n')
const out = parsePipelineAnnotations(code)
expect(out.inPipeline).toBe(true)
expect(out.schedules).toEqual(['0 0 * * *'])
// `// on schedule` joins the native-trigger family — marker-only,
// binding lives on the schedule row's `script_path`.
expect(out.nativeTriggers).toEqual([{ kind: 'schedule' }])
expect(out.triggerAssets).toHaveLength(1)
expect(out.partition).toBeDefined()
expect(out.freshness).toEqual({ duration: '2h' })
@@ -21,8 +21,12 @@ const ASSET_PREFIXES: Array<[string, AssetKind]> = [
['volume://', 'volume']
]
// Non-asset, non-schedule trigger keywords — `on <kind> <path>`.
// Native trigger keywords — `// on <kind>`. Each is marker-only: the
// binding lives on the matching trigger row's own `script_path` field.
// Schedule joins the family — the cron is stored on the schedule row the
// user creates separately, not in the annotation.
const NATIVE_TRIGGER_KEYWORDS: NativeTriggerKind[] = [
'schedule',
'webhook',
'email',
'kafka',
@@ -72,7 +76,6 @@ export type RetrySpec = {
export type PipelineAnnotations = {
inPipeline: boolean
triggerAssets: PipelineTriggerAsset[]
schedules: string[] // raw cron expressions, in insertion order
nativeTriggers: PipelineNativeTrigger[]
partition?: PartitionSpec
freshness?: FreshnessSpec
@@ -80,16 +83,6 @@ export type PipelineAnnotations = {
retry?: RetrySpec
}
function unquote(s: string): string | undefined {
if (s.length >= 2) {
const q = s[0]
if ((q === '"' || q === "'") && s.endsWith(q)) {
return s.slice(1, -1)
}
}
return undefined
}
// Tokenize a `key=value [key="quoted value"] ...` option string. Bare
// values run until the next whitespace; quoted values (single or double)
// consume until the matching quote. Malformed pairs are skipped.
@@ -241,7 +234,6 @@ export function parsePipelineAnnotations(code: string): PipelineAnnotations {
const out: PipelineAnnotations = {
inPipeline: false,
triggerAssets: [],
schedules: [],
nativeTriggers: []
}
@@ -257,15 +249,6 @@ export function parsePipelineAnnotations(code: string): PipelineAnnotations {
continue
}
const afterSchedule = consumeKeyword(inner, 'schedule')
if (afterSchedule !== undefined) {
const cron = unquote(afterSchedule.trim())
if (cron && cron.trim() !== '' && !out.schedules.includes(cron)) {
out.schedules.push(cron)
}
continue
}
const afterPart = consumeKeyword(inner, 'partitioned')
if (afterPart !== undefined) {
if (!out.partition) {
@@ -232,10 +232,22 @@ function commentPrefix(lang: ScriptLang): string {
}
export type DraftTriggerSource =
| { kind: 'schedule'; cron: string }
| { kind: 'asset'; ref: string }
| {
kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp'
// All native triggers are marker-only — the binding lives on the
// trigger row's own `script_path` field. `path` is unused for
// drafts (kept as a no-op slot for shape parity with attached
// edges); the user creates the trigger row in its dedicated UI.
kind:
| 'schedule'
| 'webhook'
| 'email'
| 'kafka'
| 'mqtt'
| 'nats'
| 'postgres'
| 'sqs'
| 'gcp'
path: string | undefined
}
@@ -257,12 +269,11 @@ function header(language: ScriptLang, triggers: DraftTriggerSource[]): string {
const p = commentPrefix(language)
const lines = triggers.map((t) => {
switch (t.kind) {
case 'schedule':
return `${p} schedule "${t.cron}"`
case 'asset':
return `${p} on ${t.ref}`
default:
// Native triggers: marker-only — no trailing path.
// Native triggers (incl. schedule): marker-only — the
// binding lives on the trigger row's own `script_path`.
return `${p} on ${t.kind}`
}
})
@@ -7,7 +7,6 @@ import type { AssetWithAltAccessType } from '$lib/components/assets/lib'
const ann = (over: Partial<PipelineAnnotations> = {}): PipelineAnnotations => ({
inPipeline: false,
triggerAssets: [],
schedules: [],
nativeTriggers: [],
...over
})
@@ -209,7 +208,7 @@ describe('resolveGraph', () => {
{ kind: 's3object', path: '/persisted.json' }, // deduped
{ kind: 's3object', path: '/new-trigger.json' } // added unsaved
],
schedules: ['0 * * * *']
nativeTriggers: [{ kind: 'schedule' }]
})
}
const r = resolveGraph(input({ base, liveAnnotations }))
@@ -222,15 +221,22 @@ describe('resolveGraph', () => {
(t) => t.trigger_kind === 'asset' && (t as any).asset_path === '/new-trigger.json'
)?.unsaved
).toBe(true)
// Schedule joins the native-trigger family — marker-only, missing
// until the user creates the schedule row separately.
expect(
r.triggers.some((t) => t.trigger_kind === 'schedule' && (t as any).cron === '0 * * * *')
r.triggers.some(
(t) =>
t.trigger_kind === 'schedule' &&
(t as any).missing === true &&
t.runnable_path === 'f/x/open'
)
).toBe(true)
})
it('live annotations for a draft replace that drafts seeded triggers', () => {
// Draft body seeds a schedule via `// schedule`; the live buffer has
// Draft body seeds a schedule annotation; the live buffer has
// replaced it with an asset trigger — only the live one should remain.
const drafts = new Map([['f/x/d', { script: { content: '// schedule "5 * * * *"' } }]])
const drafts = new Map([['f/x/d', { script: { content: '// on schedule' } }]])
const liveAnnotations = {
scriptPath: 'f/x/d',
annotations: ann({ triggerAssets: [{ kind: 's3object', path: '/live-trig.json' }] })
@@ -71,16 +71,30 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
for (const [path, d] of drafts) {
const parsed = parsePipelineAnnotations(d.script.content)
runnables.push({
path,
usage_kind: 'script',
in_pipeline: true,
partition_kind: parsed.partition?.kind,
freshness: parsed.freshness?.duration,
tag: parsed.tag,
retry: parsed.retry,
unsaved: true
})
// A draft can coexist with a base entry — during save the refetch
// lands before drafts cleanup, and a user re-editing a deployed
// script also produces both. In that case we mutate the existing
// base runnable to carry `unsaved: true` instead of pushing a
// duplicate (which would crash svelte-flow's keyed each), so the
// canvas + trigger-node labels reflect that there's pending body
// editing for this path.
const baseIdx = runnables.findIndex(
(r) => r.usage_kind === 'script' && r.path === path
)
if (baseIdx === -1) {
runnables.push({
path,
usage_kind: 'script',
in_pipeline: true,
partition_kind: parsed.partition?.kind,
freshness: parsed.freshness?.duration,
tag: parsed.tag,
retry: parsed.retry,
unsaved: true
})
} else {
runnables[baseIdx] = { ...runnables[baseIdx], unsaved: true }
}
// Output asset(s): three-tier resolution.
// 1. Active draft (the body the user is editing right now):
// live body inference is authoritative — renaming a
@@ -108,6 +122,15 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
for (const out of writeOuts) {
const hasAsset = assets.some((a) => a.kind === out.kind && a.path === out.path)
if (!hasAsset) assets.push({ kind: out.kind, path: out.path })
const hasWriteEdge = base.edges.some(
(e) =>
e.runnable_kind === 'script' &&
e.runnable_path === path &&
e.asset_kind === out.kind &&
e.asset_path === out.path &&
(e.access_type === 'w' || e.access_type === 'rw')
)
if (hasWriteEdge) continue
edges.push({
runnable_path: path,
runnable_kind: 'script',
@@ -117,19 +140,10 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
unsaved: true
})
}
// Seed trigger edges (schedule + asset) from the draft's template
// so the graph stays stable when the user clicks off this draft.
// Live annotations (below) take over for the currently-open draft
// so keystroke edits still update in real time.
for (const cron of parsed.schedules) {
extraTriggers.push({
trigger_kind: 'schedule',
cron,
runnable_kind: 'script',
runnable_path: path,
unsaved: true
})
}
// Seed trigger edges from the draft's template so the graph stays
// stable when the user clicks off this draft. Live annotations
// (below) take over for the currently-open draft so keystroke
// edits still update in real time.
for (const a of parsed.triggerAssets) {
extraTriggers.push({
trigger_kind: 'asset',
@@ -176,16 +190,6 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
)
.map((t) => (t.trigger_kind === 'asset' ? `${t.asset_kind}:${t.asset_path}` : ''))
)
const persistedScheduleCrons = new Set(
base.triggers
.filter(
(t) =>
t.trigger_kind === 'schedule' &&
t.runnable_kind === 'script' &&
t.runnable_path === livePath
)
.map((t) => (t.trigger_kind === 'schedule' ? t.cron : ''))
)
// Strip seeded triggers we computed above for the active draft;
// live annotations are authoritative for the open buffer.
for (let i = extraTriggers.length - 1; i >= 0; i--) {
@@ -210,16 +214,6 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
assets.push({ kind: a.kind, path: a.path })
}
}
for (const cron of liveAnnotations.annotations.schedules) {
if (persistedScheduleCrons.has(cron)) continue
extraTriggers.push({
trigger_kind: 'schedule',
cron,
runnable_kind: 'script',
runnable_path: livePath,
unsaved: true
})
}
// Native trigger annotations: kinds for which a matching trigger
// row was found in the backend response. If the live buffer
// declares `// on kafka` and at least one kafka_trigger row points
@@ -231,7 +225,6 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
.filter(
(t) =>
t.trigger_kind !== 'asset' &&
t.trigger_kind !== 'schedule' &&
t.runnable_kind === 'script' &&
t.runnable_path === livePath
)
@@ -265,7 +258,6 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
.filter(
(t) =>
t.trigger_kind !== 'asset' &&
t.trigger_kind !== 'schedule' &&
t.runnable_kind === 'script' &&
t.runnable_path === scriptPath
)
@@ -49,7 +49,12 @@ export interface AssetGraphEdge {
// Declared `// on <trigger>` — the actual execution DAG edges.
// `unsaved: true` marks overlays computed live from editor buffer that
// haven't been persisted to script_trigger yet.
//
// `schedule` is in the family — the cron lives on the schedule row the user
// creates separately; the annotation is just the binding declaration, same
// as kafka/mqtt/etc.
export type NativeTriggerKind =
| 'schedule'
| 'webhook'
| 'email'
| 'kafka'
@@ -68,17 +73,10 @@ export type AssetGraphTrigger =
runnable_path: string
unsaved?: boolean
}
| {
trigger_kind: 'schedule'
cron: string
runnable_kind: GraphUsageKind
runnable_path: string
unsaved?: boolean
}
| {
trigger_kind: NativeTriggerKind
// path of the matching trigger row (kafka_trigger.path, etc.).
// Undefined when `missing` is true — the script has the
// path of the matching trigger row (kafka_trigger.path, schedule.path,
// etc.). Undefined when `missing` is true — the script has the
// annotation marker but no trigger row points at it.
path?: string
runnable_kind: GraphUsageKind
@@ -0,0 +1,26 @@
<script lang="ts">
import { Alert } from '$lib/components/common'
// Read-only viewer shown in place of the runnable ScriptPicker when a
// trigger drawer is opened in the pipeline-editor context (the editor
// was called with a fixed script path). Pipeline membership means the
// script's path is part of the pipeline's structure — re-pointing the
// trigger at a different script would silently detach this node from
// the pipeline.
interface Props {
path: string
}
let { path }: Props = $props()
</script>
<div class="flex flex-col gap-2 mb-2">
<div
class="border rounded-md text-xs px-3 py-2 text-secondary bg-surface-input font-mono break-all"
>
{path}
</div>
<Alert type="info" title="Pipeline-bound trigger" size="xs">
Triggers whose runnable is part of a data pipeline are managed from the pipeline editor.
The script path is fixed; edit the pipeline to reassign or remove the trigger.
</Alert>
</div>
@@ -11,10 +11,10 @@
let { onUpdate = undefined, customSaveBehavior }: Props = $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(
@@ -5,6 +5,7 @@
import Path from '$lib/components/Path.svelte'
import Required from '$lib/components/Required.svelte'
import ScriptPicker from '$lib/components/ScriptPicker.svelte'
import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
import {
EmailTriggerService,
type ErrorHandler,
@@ -116,7 +117,8 @@
export async function openEdit(
ePath: string,
isFlow: boolean,
defaultConfig?: Partial<NewEmailTrigger>
defaultConfig?: Partial<NewEmailTrigger>,
fixedScriptPath_?: string
) {
drawerLoading = true
let loader = setTimeout(() => {
@@ -130,6 +132,7 @@
edit = true
dirtyPath = false
dirtyLocalPart = false
fixedScriptPath = fixedScriptPath_ ?? ''
await loadTrigger(defaultConfig)
if (!defaultConfig) {
// If the email trigger is loaded from the backend, we to set the initial config
@@ -364,34 +367,38 @@
{#if !hideTarget}
<Section label="Target">
<p class="text-xs mt-3 mb-1 text-primary">
Pick a script or flow to be triggered<Required required={true} />
</p>
<div class="flex flex-col gap-2">
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={fixedScriptPath != '' || !can_write}
initialPath={fixedScriptPath || initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if fixedScriptPath != ''}
<PipelineLockedRunnableInfo path={fixedScriptPath} />
{:else}
<p class="text-xs mt-3 mb-1 text-primary">
Pick a script or flow to be triggered<Required required={true} />
</p>
<div class="flex flex-col gap-2">
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={!can_write}
initialPath={initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
href={itemKind === 'flow' ? '/flows/add?hub=72' : '/scripts/add?hub=hub%2F19813'}
target="_blank">Create from template</Button
>
{/if}
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
href={itemKind === 'flow' ? '/flows/add?hub=72' : '/scripts/add?hub=hub%2F19813'}
target="_blank">Create from template</Button
>
{/if}
</div>
</div>
</div>
{/if}
</Section>
{/if}
@@ -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(
@@ -19,6 +19,7 @@
} from '$lib/gen'
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 GcpTriggerEditorConfigSection from './GcpTriggerEditorConfigSection.svelte'
import { untrack, type Snippet } from 'svelte'
@@ -127,7 +128,8 @@
export async function openEdit(
ePath: string,
isFlow: boolean,
defaultValues?: Record<string, any>
defaultValues?: Record<string, any>,
fixedScriptPath_?: string
) {
drawerLoading = true
try {
@@ -136,6 +138,7 @@
itemKind = isFlow ? 'flow' : 'script'
edit = true
dirtyPath = false
fixedScriptPath = fixedScriptPath_ ?? ''
await loadTrigger(defaultValues)
if (!defaultValues) {
initialConfig = structuredClone($state.snapshot(getGcpConfig()))
@@ -448,32 +451,36 @@
{#if !hideTarget}
<Section label="Runnable">
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered <Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={fixedScriptPath != '' || !can_write}
initialPath={fixedScriptPath || initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="default"
unifiedSize="md"
{#if fixedScriptPath != ''}
<PipelineLockedRunnableInfo path={fixedScriptPath} />
{:else}
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered <Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={!can_write}
href={itemKind === 'flow' ? '/flows/add?hub=68' : '/scripts/add?hub=hub%2F19796'}
target="_blank">Create from template</Button
>
{/if}
</div>
initialPath={initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="default"
unifiedSize="md"
disabled={!can_write}
href={itemKind === 'flow' ? '/flows/add?hub=68' : '/scripts/add?hub=hub%2F19796'}
target="_blank">Create from template</Button
>
{/if}
</div>
{/if}
</Section>
{/if}
@@ -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(
@@ -6,6 +6,7 @@
import Path from '$lib/components/Path.svelte'
import Required from '$lib/components/Required.svelte'
import ScriptPicker from '$lib/components/ScriptPicker.svelte'
import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
import { KafkaTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
@@ -146,7 +147,8 @@
export async function openEdit(
ePath: string,
isFlow: boolean,
defaultConfig?: Record<string, any>
defaultConfig?: Record<string, any>,
fixedScriptPath_?: string
) {
let loadingTimeout = setTimeout(() => {
showLoading = true
@@ -158,6 +160,7 @@
itemKind = isFlow ? 'flow' : 'script'
edit = true
dirtyPath = false
fixedScriptPath = fixedScriptPath_ ?? ''
await loadTrigger(defaultConfig)
if (!defaultConfig) {
initialConfig = structuredClone($state.snapshot(getSaveCfg()))
@@ -497,32 +500,36 @@
{#if !hideTarget}
<Section label="Runnable">
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered<Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={fixedScriptPath != '' || !can_write}
initialPath={fixedScriptPath || initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
{#if fixedScriptPath != ''}
<PipelineLockedRunnableInfo path={fixedScriptPath} />
{:else}
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered<Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={!can_write}
href={itemKind === 'flow' ? '/flows/add?hub=65' : '/scripts/add?hub=hub%2F19659'}
target="_blank">Create from template</Button
>
{/if}
</div>
initialPath={initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
disabled={!can_write}
href={itemKind === 'flow' ? '/flows/add?hub=65' : '/scripts/add?hub=hub%2F19659'}
target="_blank">Create from template</Button
>
{/if}
</div>
{/if}
</Section>
{/if}
@@ -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(
@@ -6,6 +6,7 @@
import Path from '$lib/components/Path.svelte'
import Required from '$lib/components/Required.svelte'
import ScriptPicker from '$lib/components/ScriptPicker.svelte'
import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
@@ -142,7 +143,8 @@
export async function openEdit(
ePath: string,
isFlow: boolean,
defaultConfig?: Record<string, any>
defaultConfig?: Record<string, any>,
fixedScriptPath_?: string
) {
let loadingTimeout = setTimeout(() => {
showLoading = true
@@ -154,6 +156,7 @@
itemKind = isFlow ? 'flow' : 'script'
edit = true
dirtyPath = false
fixedScriptPath = fixedScriptPath_ ?? ''
await loadTrigger(defaultConfig)
if (!defaultConfig) {
initialConfig = structuredClone($state.snapshot(getSaveCfg()))
@@ -471,34 +474,38 @@
{#if !hideTarget}
<Section label="Runnable">
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered<Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={fixedScriptPath != '' || !can_write}
initialPath={fixedScriptPath || initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
{#if fixedScriptPath != ''}
<PipelineLockedRunnableInfo path={fixedScriptPath} />
{:else}
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered<Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={!can_write}
href={itemKind === 'flow' ? '/flows/add?hub=61' : '/scripts/add?hub=hub%2F19655'}
target="_blank"
>
Create from template
</Button>
{/if}
</div>
initialPath={initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
disabled={!can_write}
href={itemKind === 'flow' ? '/flows/add?hub=61' : '/scripts/add?hub=hub%2F19655'}
target="_blank"
>
Create from template
</Button>
{/if}
</div>
{/if}
</Section>
{/if}
@@ -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(
@@ -5,6 +5,7 @@
import Path from '$lib/components/Path.svelte'
import Required from '$lib/components/Required.svelte'
import ScriptPicker from '$lib/components/ScriptPicker.svelte'
import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
import { NatsTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
@@ -130,7 +131,8 @@
export async function openEdit(
ePath: string,
isFlow: boolean,
defaultConfig?: Record<string, any>
defaultConfig?: Record<string, any>,
fixedScriptPath_?: string
) {
let loadingTimeout = setTimeout(() => {
showLoading = true
@@ -142,6 +144,7 @@
itemKind = isFlow ? 'flow' : 'script'
edit = true
dirtyPath = false
fixedScriptPath = fixedScriptPath_ ?? ''
await loadTrigger(defaultConfig)
if (!defaultConfig) {
initialConfig = structuredClone($state.snapshot(getSaveCfg()))
@@ -453,33 +456,37 @@
</div>
{#if !hideTarget}
<Section label="Runnable">
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered<Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={fixedScriptPath != '' || !can_write}
initialPath={fixedScriptPath || initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
href={itemKind === 'flow' ? '/flows/add?hub=66' : '/scripts/add?hub=hub%2F19663'}
target="_blank"
>
Create from template
</Button>
{/if}
</div>
{#if fixedScriptPath != ''}
<PipelineLockedRunnableInfo path={fixedScriptPath} />
{:else}
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered<Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={!can_write}
initialPath={initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
href={itemKind === 'flow' ? '/flows/add?hub=66' : '/scripts/add?hub=hub%2F19663'}
target="_blank"
>
Create from template
</Button>
{/if}
</div>
{/if}
</Section>
{/if}
@@ -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(
@@ -5,6 +5,7 @@
import Path from '$lib/components/Path.svelte'
import Required from '$lib/components/Required.svelte'
import ScriptPicker from '$lib/components/ScriptPicker.svelte'
import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
import {
PostgresTriggerService,
type ErrorHandler,
@@ -230,7 +231,8 @@
export async function openEdit(
ePath: string,
isFlow: boolean,
defaultConfig?: Record<string, any>
defaultConfig?: Record<string, any>,
fixedScriptPath_?: string
) {
let loadingTimeout = setTimeout(() => {
showLoading = true
@@ -249,6 +251,7 @@
relations = []
transaction_to_track = []
tab = 'basic'
fixedScriptPath = fixedScriptPath_ ?? ''
await loadTrigger(defaultConfig)
if (!defaultConfig) {
initialConfig = structuredClone($state.snapshot(getSaveCfg()))
@@ -629,43 +632,47 @@
</Label>
{#if !hideTarget}
<Section label="Runnable">
<p class="text-xs text-primary">
Pick a script or flow to be triggered <Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={fixedScriptPath != '' || !can_write}
initialPath={fixedScriptPath || initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if fixedScriptPath != ''}
<PipelineLockedRunnableInfo path={fixedScriptPath} />
{:else}
<p class="text-xs text-primary">
Pick a script or flow to be triggered <Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={!can_write}
initialPath={initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path) && is_flow === false}
<div class="flex">
<Button
disabled={!can_write}
btnClasses="ml-4"
variant="accent"
size="xs"
on:click={getTemplateScript}
target="_blank"
{loading}
>Create from template
<Tooltip light>
The conversion requires a <strong>database resource</strong> and at least one
<strong>schema</strong>
to be set.<br />
Please ensure these conditions are met before proceeding.
</Tooltip>
</Button>
</div>
{/if}
</div>
{#if emptyString(script_path) && is_flow === false}
<div class="flex">
<Button
disabled={!can_write}
btnClasses="ml-4"
variant="accent"
size="xs"
on:click={getTemplateScript}
target="_blank"
{loading}
>Create from template
<Tooltip light>
The conversion requires a <strong>database resource</strong> and at least one
<strong>schema</strong>
to be set.<br />
Please ensure these conditions are met before proceeding.
</Tooltip>
</Button>
</div>
{/if}
</div>
{/if}
</Section>
{/if}
@@ -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()
@@ -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<string, any> = $state({})
let loading = $state(false)
@@ -141,7 +146,12 @@
deployed: () => initialConfig
})
export async function openEdit(ePath: string, isFlow: boolean, defaultCfg?: Record<string, any>) {
export async function openEdit(
ePath: string,
isFlow: boolean,
defaultCfg?: Record<string, any>,
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 @@
<Section label="Runnable">
{#if !hideTarget}
{#if !edit}
{#if fixedScriptPath != ''}
<PipelineLockedRunnableInfo path={fixedScriptPath} />
{:else if !edit}
<p class="text-xs mb-1 text-secondary">
Pick a script or flow to be triggered by the schedule<Required required={true} />
</p>
@@ -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(
@@ -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<string, any>
defaultConfig?: Record<string, any>,
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}
<Section label="Runnable">
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered <Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={fixedScriptPath != '' || !can_write}
initialPath={fixedScriptPath || initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
{#if fixedScriptPath != ''}
<PipelineLockedRunnableInfo path={fixedScriptPath} />
{:else}
<p class="text-xs mb-1 text-primary">
Pick a script or flow to be triggered <Required required={true} />
</p>
<div class="flex flex-row mb-2">
<ScriptPicker
disabled={!can_write}
href={itemKind === 'flow' ? '/flows/add?hub=59' : '/scripts/add?hub=hub%2F19657'}
target="_blank"
>
Create from template
</Button>
{/if}
</div>
initialPath={initialScriptPath}
kinds={['script']}
allowFlow={true}
bind:itemKind
bind:scriptPath={script_path}
allowRefresh={can_write}
allowEdit={!$userStore?.operator}
clearable
/>
{#if emptyString(script_path)}
<Button
btnClasses="ml-4"
variant="accent"
size="xs"
disabled={!can_write}
href={itemKind === 'flow' ? '/flows/add?hub=59' : '/scripts/add?hub=hub%2F19657'}
target="_blank"
>
Create from template
</Button>
{/if}
</div>
{/if}
</Section>
{/if}
@@ -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<string, Draft>()
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<Map<string, Array<{ kind: AssetKind; path: string }>>>(
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<Map<string, Array<{ kind: AssetKind; path: string }>>>(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<Map<string, string>>(new Map())
// Sibling cache: the parsed asset usages from `inferAssets` (wasm), one
// pass per body. Same only-add semantics as `bodiesByPath`.
let inferredAssetsByPath = $state<Map<string, AssetWithAltAccessType[]>>(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<Map<string, Set<NativeTriggerKind>>>(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<string, Array<{ kind: AssetKind; path: string }>>()
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<string, Array<{ kind: AssetKind; path: string }>>()
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<string, Set<NativeTriggerKind>>()
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<NativeTriggerKind>
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 @@
<PipelinePickerModal bind:open={pickerModalOpen} currentFolder={folder} />
<ConfirmationModal
open={triggerDeleteOpen}
loading={triggerDeleteLoading}
title={triggerDeleteTarget ? `Delete ${triggerDeleteTarget.kind} trigger` : 'Delete trigger'}
confirmationText="Delete"
onConfirmed={confirmDeleteAttachedTrigger}
onCanceled={() => {
if (!triggerDeleteLoading) triggerDeleteTarget = undefined
}}
>
{#if triggerDeleteTarget}
<p>
Delete <code class="font-mono">{triggerDeleteTarget.path}</code>? The
<code class="font-mono">// on {triggerDeleteTarget.kind}</code> annotation on the script
stays — the trigger will read as missing on the canvas until you recreate it or remove
the annotation.
</p>
{/if}
</ConfirmationModal>
<!-- Native trigger editors mounted off-screen. Each only renders its
inner drawer when `open=true` (set by openNew/openEdit), so this
adds ~zero render cost while idle. `onUpdate` refreshes the graph
@@ -1429,6 +1772,7 @@
<SqsTriggerEditor bind:this={sqsEditor} onUpdate={() => graphRes.refetch()} />
<GcpTriggerEditor bind:this={gcpEditor} onUpdate={() => graphRes.refetch()} />
<EmailTriggerEditor bind:this={emailEditor} onUpdate={() => graphRes.refetch()} />
<ScheduleEditor bind:this={scheduleEditor} onUpdate={() => graphRes.refetch()} />
{#if leaveModalOpen}
<!-- Three-button leave guard. Built inline rather than reusing