diff --git a/backend/migrations/20260421205023_add_asset_materializer.down.sql b/backend/migrations/20260421205023_add_asset_materializer.down.sql new file mode 100644 index 0000000000..318351178e --- /dev/null +++ b/backend/migrations/20260421205023_add_asset_materializer.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS asset_materializer_history; +DROP TABLE IF EXISTS asset_materializer; diff --git a/backend/migrations/20260421205023_add_asset_materializer.up.sql b/backend/migrations/20260421205023_add_asset_materializer.up.sql new file mode 100644 index 0000000000..91f661c770 --- /dev/null +++ b/backend/migrations/20260421205023_add_asset_materializer.up.sql @@ -0,0 +1,33 @@ +-- Current materializer: exactly one runnable owns each (workspace, kind, path) asset. +-- Updated via last-deploy-wins on `// materialize ` annotations. +CREATE TABLE asset_materializer ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + asset_kind ASSET_KIND NOT NULL, + asset_path VARCHAR(255) NOT NULL, + runnable_kind ASSET_USAGE_KIND NOT NULL, + runnable_path VARCHAR(255) NOT NULL, + deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + deployed_by VARCHAR(50) NOT NULL, + PRIMARY KEY (workspace_id, asset_kind, asset_path) +); + +CREATE INDEX idx_asset_materializer_runnable + ON asset_materializer (workspace_id, runnable_kind, runnable_path); + +-- Append-only history. Every claim (including takeovers) gets a row. +-- Revocations (redeploy without annotation) are represented by absence in +-- `asset_materializer` rather than a row here, so "who last materialized X" +-- is the most recent history row for X regardless of current state. +CREATE TABLE asset_materializer_history ( + id BIGSERIAL PRIMARY KEY, + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + asset_kind ASSET_KIND NOT NULL, + asset_path VARCHAR(255) NOT NULL, + runnable_kind ASSET_USAGE_KIND NOT NULL, + runnable_path VARCHAR(255) NOT NULL, + deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + deployed_by VARCHAR(50) NOT NULL +); + +CREATE INDEX idx_asset_materializer_history_asset + ON asset_materializer_history (workspace_id, asset_kind, asset_path, deployed_at DESC); diff --git a/backend/migrations/20260421212900_pipeline_folders_index.down.sql b/backend/migrations/20260421212900_pipeline_folders_index.down.sql new file mode 100644 index 0000000000..6c5f740b1d --- /dev/null +++ b/backend/migrations/20260421212900_pipeline_folders_index.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS idx_script_materializer_path; diff --git a/backend/migrations/20260421212900_pipeline_folders_index.up.sql b/backend/migrations/20260421212900_pipeline_folders_index.up.sql new file mode 100644 index 0000000000..053a635b45 --- /dev/null +++ b/backend/migrations/20260421212900_pipeline_folders_index.up.sql @@ -0,0 +1,8 @@ +-- Fast lookups for: +-- 1. "does folder F have a pipeline?" (exists check on prefix) +-- 2. "list all folders with a pipeline" (distinct folder from path) +-- The partial predicate keeps the index tiny on workspaces with few +-- materializer scripts, and text_pattern_ops lets 'f/foo/%' LIKE scans use it. +CREATE INDEX IF NOT EXISTS idx_script_materializer_path + ON script (workspace_id, path text_pattern_ops) + WHERE auto_kind = 'materializer' AND archived = false AND deleted = false; diff --git a/backend/migrations/20260422041559_pipeline_refactor_triggers.down.sql b/backend/migrations/20260422041559_pipeline_refactor_triggers.down.sql new file mode 100644 index 0000000000..b96eaf8041 --- /dev/null +++ b/backend/migrations/20260422041559_pipeline_refactor_triggers.down.sql @@ -0,0 +1,30 @@ +DROP TABLE IF EXISTS script_trigger; +DROP TYPE IF EXISTS SCRIPT_TRIGGER_KIND; + +-- Recreate the materializer tables so rolling back to the pre-refactor +-- backend code boots. Data is gone either way (irrecoverable). +CREATE TABLE asset_materializer ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + asset_kind ASSET_KIND NOT NULL, + asset_path VARCHAR(255) NOT NULL, + runnable_kind ASSET_USAGE_KIND NOT NULL, + runnable_path VARCHAR(255) NOT NULL, + deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + deployed_by VARCHAR(50) NOT NULL, + PRIMARY KEY (workspace_id, asset_kind, asset_path) +); +CREATE INDEX idx_asset_materializer_runnable + ON asset_materializer (workspace_id, runnable_kind, runnable_path); + +CREATE TABLE asset_materializer_history ( + id BIGSERIAL PRIMARY KEY, + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + asset_kind ASSET_KIND NOT NULL, + asset_path VARCHAR(255) NOT NULL, + runnable_kind ASSET_USAGE_KIND NOT NULL, + runnable_path VARCHAR(255) NOT NULL, + deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + deployed_by VARCHAR(50) NOT NULL +); +CREATE INDEX idx_asset_materializer_history_asset + ON asset_materializer_history (workspace_id, asset_kind, asset_path, deployed_at DESC); diff --git a/backend/migrations/20260422041559_pipeline_refactor_triggers.up.sql b/backend/migrations/20260422041559_pipeline_refactor_triggers.up.sql new file mode 100644 index 0000000000..aea9c7ebbd --- /dev/null +++ b/backend/migrations/20260422041559_pipeline_refactor_triggers.up.sql @@ -0,0 +1,31 @@ +-- Simplified pipeline model: `// materialize` is a bare opt-in marker (sets +-- auto_kind='materializer') and the per-asset ownership tracking goes away. +-- Writes are already tracked in the `asset` table via the parser, so the +-- `asset_materializer*` tables no longer earn their keep. +DROP TABLE IF EXISTS asset_materializer_history; +DROP TABLE IF EXISTS asset_materializer; + +-- Execution DAG edges declared via `// on ` annotations. +-- For `trigger_kind='asset'`: trigger_ref is `://` (kind from +-- parse_asset_syntax, so downstream lookups match the `asset` table). +-- For `trigger_kind='schedule'`: trigger_ref is the raw cron expression. +CREATE TYPE SCRIPT_TRIGGER_KIND AS ENUM ('asset', 'schedule'); + +CREATE TABLE script_trigger ( + id BIGSERIAL PRIMARY KEY, + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + runnable_kind ASSET_USAGE_KIND NOT NULL, + runnable_path VARCHAR(255) NOT NULL, + trigger_kind SCRIPT_TRIGGER_KIND NOT NULL, + trigger_ref TEXT NOT NULL +); + +-- Per-runnable lookup (wipe-on-deploy, list-triggers-for-script). +CREATE INDEX idx_script_trigger_runnable + ON script_trigger (workspace_id, runnable_kind, runnable_path); + +-- Reverse lookup: "which scripts are triggered by asset X?" (the asset → script +-- edges in the graph). trigger_ref is unbounded text so can't share the +-- asset_kind btree, but this covers the common prefix-scan use case. +CREATE INDEX idx_script_trigger_ref + ON script_trigger (workspace_id, trigger_kind, trigger_ref); diff --git a/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.down.sql b/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.down.sql new file mode 100644 index 0000000000..c2b2966e7d --- /dev/null +++ b/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.down.sql @@ -0,0 +1,12 @@ +-- Postgres doesn't support removing enum values in-place. The only safe +-- rollback is to recreate the type with the original set and rewrite the +-- column, deleting any rows using values introduced in the up migration. +DELETE FROM script_trigger + WHERE trigger_kind NOT IN ('asset', 'schedule'); + +CREATE TYPE SCRIPT_TRIGGER_KIND_OLD AS ENUM ('asset', 'schedule'); +ALTER TABLE script_trigger + ALTER COLUMN trigger_kind TYPE SCRIPT_TRIGGER_KIND_OLD + USING trigger_kind::text::SCRIPT_TRIGGER_KIND_OLD; +DROP TYPE SCRIPT_TRIGGER_KIND; +ALTER TYPE SCRIPT_TRIGGER_KIND_OLD RENAME TO SCRIPT_TRIGGER_KIND; diff --git a/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.up.sql b/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.up.sql new file mode 100644 index 0000000000..6d440b782c --- /dev/null +++ b/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.up.sql @@ -0,0 +1,11 @@ +-- Expand the pipeline trigger enum to cover every non-integration +-- (i.e. non-native) trigger kind Windmill supports. Each value mirrors a +-- keyword the annotation parser recognises in `// on ` lines. +ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'webhook'; +ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'email'; +ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'kafka'; +ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'mqtt'; +ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'nats'; +ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'postgres'; +ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'sqs'; +ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'gcp'; diff --git a/backend/parsers/windmill-parser-py-asset/src/lib.rs b/backend/parsers/windmill-parser-py-asset/src/lib.rs index 10b93a998b..9607092b10 100644 --- a/backend/parsers/windmill-parser-py-asset/src/lib.rs +++ b/backend/parsers/windmill-parser-py-asset/src/lib.rs @@ -2,8 +2,8 @@ use rustpython_ast::{Constant, Expr, ExprConstant, Visitor}; use rustpython_parser::{ast::Suite, Parse}; use std::collections::HashMap; use windmill_parser::asset_parser::{ - asset_was_used, merge_assets, parse_asset_syntax, AssetKind, AssetUsageAccessType, - ParseAssetsOutput, ParseAssetsResult, + asset_was_used, merge_assets, parse_asset_syntax, parse_pipeline_annotations, AssetKind, + AssetUsageAccessType, ParseAssetsOutput, ParseAssetsResult, }; use AssetUsageAccessType::*; @@ -28,7 +28,13 @@ pub fn parse_assets(input: &str) -> anyhow::Result { } } - Ok(ParseAssetsOutput { assets: merge_assets(assets_finder.assets), ..Default::default() }) + let (is_materializer, triggers) = parse_pipeline_annotations(input); + Ok(ParseAssetsOutput { + assets: merge_assets(assets_finder.assets), + is_materializer, + triggers, + ..Default::default() + }) } type VarAssetName = String; diff --git a/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs b/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs index 002ed78fc0..131acfdd4d 100644 --- a/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs +++ b/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs @@ -9,8 +9,8 @@ use sqlparser::{ parser::Parser, }; use windmill_parser::asset_parser::{ - asset_was_used, merge_assets, parse_asset_syntax, AssetKind, AssetUsageAccessType, - ParseAssetsOutput, ParseAssetsResult, + asset_was_used, merge_assets, parse_asset_syntax, parse_pipeline_annotations, AssetKind, + AssetUsageAccessType, ParseAssetsOutput, ParseAssetsResult, }; use AssetUsageAccessType::*; @@ -33,7 +33,13 @@ pub fn parse_assets(input: &str) -> anyhow::Result { } } - Ok(ParseAssetsOutput { assets: merge_assets(collector.assets), ..Default::default() }) + let (is_materializer, triggers) = parse_pipeline_annotations(input); + Ok(ParseAssetsOutput { + assets: merge_assets(collector.assets), + is_materializer, + triggers, + ..Default::default() + }) } /// Visitor that collects S3 asset literals from SQL statements diff --git a/backend/parsers/windmill-parser-ts-asset/src/lib.rs b/backend/parsers/windmill-parser-ts-asset/src/lib.rs index bc31a16c99..42d18ac2c9 100644 --- a/backend/parsers/windmill-parser-ts-asset/src/lib.rs +++ b/backend/parsers/windmill-parser-ts-asset/src/lib.rs @@ -5,8 +5,8 @@ use swc_ecma_ast::{CallExpr, Expr, Lit, MemberExpr, MemberProp, Str}; use swc_ecma_parser::{lexer::Lexer, Parser, StringInput, Syntax, TsSyntax}; use swc_ecma_visit::{Visit, VisitWith}; use windmill_parser::asset_parser::{ - asset_was_used, merge_assets, parse_asset_syntax, AssetKind, AssetUsageAccessType, - ParseAssetsOutput, ParseAssetsResult, SqlQueryDetails, + asset_was_used, merge_assets, parse_asset_syntax, parse_pipeline_annotations, AssetKind, + AssetUsageAccessType, ParseAssetsOutput, ParseAssetsResult, SqlQueryDetails, }; use AssetUsageAccessType::*; @@ -38,9 +38,12 @@ pub fn parse_assets(code: &str) -> anyhow::Result { let mut assets_finder = AssetsFinder { assets: vec![], sql_queries: vec![], var_identifiers: HashMap::new() }; assets_finder.visit_module_items(&ast); + let (is_materializer, triggers) = parse_pipeline_annotations(code); Ok(ParseAssetsOutput { assets: merge_assets(assets_finder.assets), sql_queries: assets_finder.sql_queries, + is_materializer, + triggers, }) } diff --git a/backend/parsers/windmill-parser-yaml/src/asset_parser.rs b/backend/parsers/windmill-parser-yaml/src/asset_parser.rs index 7e67d563ff..e4243b4d01 100644 --- a/backend/parsers/windmill-parser-yaml/src/asset_parser.rs +++ b/backend/parsers/windmill-parser-yaml/src/asset_parser.rs @@ -1,5 +1,6 @@ use windmill_parser::asset_parser::{ - merge_assets, AssetKind, AssetUsageAccessType, ParseAssetsOutput, ParseAssetsResult, + merge_assets, parse_pipeline_annotations, AssetKind, AssetUsageAccessType, ParseAssetsOutput, + ParseAssetsResult, }; use crate::{parse_ansible_reqs, ResourceOrVariablePath}; @@ -39,5 +40,11 @@ pub fn parse_assets(input: &str) -> anyhow::Result { } } - Ok(ParseAssetsOutput { assets: merge_assets(assets), ..Default::default() }) + let (is_materializer, triggers) = parse_pipeline_annotations(input); + Ok(ParseAssetsOutput { + assets: merge_assets(assets), + is_materializer, + triggers, + ..Default::default() + }) } diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 4239c1854c..606e1d9490 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -47,6 +47,42 @@ pub struct SqlQueryDetails { pub struct ParseAssetsOutput { pub assets: Vec, pub sql_queries: Vec, + // Bare `// materialize` (or `#` / `--`) anywhere in the source — opt-in + // marker that sets auto_kind='materializer' and includes the script in + // its folder's pipeline. Does NOT declare what is materialized: the + // parser-detected `w`/`rw` usages in `assets` are the outputs. + #[serde(skip_serializing_if = "std::ops::Not::not", default)] + pub is_materializer: bool, + // `// on ` annotations — execution DAG edges. Each is an + // independent OR (any fires the script). Empty = script has no + // automatic triggers (still runnable manually / via existing cron + // triggers). + #[serde(skip_serializing_if = "Vec::is_empty", default)] + pub triggers: Vec, +} + +#[derive(Serialize, Debug, PartialEq, Clone)] +#[serde(tag = "kind", rename_all = "lowercase")] +pub enum TriggerSpec { + // Refresh when `` changes. Kind comes from parse_asset_syntax so + // it matches the `asset` table. + Asset { asset_kind: AssetKind, path: String }, + // Refresh on cron. The raw expression is passed through as-is so the + // existing schedule subsystem can validate it. + Schedule { cron: String }, + // `// on ` style. The `path` is a workspace-relative + // reference to a trigger row already configured in the corresponding + // trigger table (http_trigger, email_trigger, kafka_trigger, …). Keeps + // the annotation terse; auth/broker/topic details live in the trigger's + // own UI. + Webhook { path: String }, + Email { path: String }, + Kafka { path: String }, + Mqtt { path: String }, + Nats { path: String }, + Postgres { path: String }, + Sqs { path: String }, + Gcp { path: String }, } #[derive(Debug, Clone, Serialize)] @@ -153,3 +189,181 @@ pub const ASSET_KINDS: &[(&str, AssetKind)] = &[ ("datatable://", AssetKind::DataTable), ("volume://", AssetKind::Volume), ]; + +// Scan raw source for pipeline annotations. Language-agnostic: any line +// whose first non-whitespace tokens are a comment prefix (`//`, `#`, or +// `--`) followed by either: +// - bare `materialize` (no arguments) → is_materializer = true +// - `on ` → one TriggerSpec entry +// +// Trailing content on a `// materialize` line (beyond whitespace) is ignored +// so the old `// materialize s3://foo` style is forward-compatible: the +// bare marker still fires even if users have stale per-asset arguments. +pub fn parse_pipeline_annotations(code: &str) -> (bool, Vec) { + let mut is_materializer = false; + let mut triggers: Vec = vec![]; + + for raw_line in code.lines() { + let line = raw_line.trim_start(); + let rest = if let Some(r) = line.strip_prefix("//") { + r + } else if let Some(r) = line.strip_prefix("--") { + r + } else if let Some(r) = line.strip_prefix('#') { + r + } else { + continue; + }; + let rest = rest.trim_start(); + + if let Some(after_kw) = rest.strip_prefix("materialize") { + // Bare marker: must be end-of-line, whitespace, or nothing. + // Rejects `materializer`, `materialized`, etc. + if after_kw.is_empty() || after_kw.starts_with(|c: char| c.is_whitespace()) { + is_materializer = true; + } + continue; + } + + if let Some(after_kw) = rest.strip_prefix("on") { + if !after_kw.starts_with(|c: char| c.is_whitespace()) { + continue; + } + let spec_text = after_kw.trim(); + if spec_text.is_empty() { + continue; + } + if let Some(trig) = parse_trigger_spec(spec_text) { + if !triggers.contains(&trig) { + triggers.push(trig); + } + } + } + } + + (is_materializer, triggers) +} + +// Parse a single `on ` right-hand side. Accepted forms: +// schedule "" +// — where is one of +// webhook | email | kafka | mqtt | nats | postgres | sqs | gcp +// (e.g. s3://bucket/key, $res:f/foo) +fn parse_trigger_spec(s: &str) -> Option { + if let Some(rest) = s.strip_prefix("schedule") { + let rest = rest.trim_start(); + // Accept either double or single quotes around the cron expression. + let cron = rest + .strip_prefix('"') + .and_then(|r| r.strip_suffix('"')) + .or_else(|| rest.strip_prefix('\'').and_then(|r| r.strip_suffix('\'')))?; + if cron.trim().is_empty() { + return None; + } + return Some(TriggerSpec::Schedule { cron: cron.to_string() }); + } + + // ` ` — delegate to a tiny table so the annotation set + // stays in lockstep with `TriggerSpec`. + type Ctor = fn(String) -> TriggerSpec; + const KINDS: &[(&str, Ctor)] = &[ + ("webhook", |p| TriggerSpec::Webhook { path: p }), + ("email", |p| TriggerSpec::Email { path: p }), + ("kafka", |p| TriggerSpec::Kafka { path: p }), + ("mqtt", |p| TriggerSpec::Mqtt { path: p }), + ("nats", |p| TriggerSpec::Nats { path: p }), + ("postgres", |p| TriggerSpec::Postgres { path: p }), + ("sqs", |p| TriggerSpec::Sqs { path: p }), + ("gcp", |p| TriggerSpec::Gcp { path: p }), + ]; + for (kw, ctor) in KINDS { + if let Some(rest) = s.strip_prefix(kw) { + if !rest.starts_with(|c: char| c.is_whitespace()) { + continue; + } + let path = rest.trim(); + if path.is_empty() { + return None; + } + return Some(ctor(path.to_string())); + } + } + + let (kind, path) = parse_asset_syntax(s, false)?; + Some(TriggerSpec::Asset { asset_kind: kind, path: path.to_string() }) +} + +#[cfg(test)] +mod pipeline_annotation_tests { + use super::*; + + #[test] + fn bare_materialize_marker() { + let (is_m, triggers) = parse_pipeline_annotations("// materialize\nconsole.log('hi')"); + assert!(is_m); + assert!(triggers.is_empty()); + } + + #[test] + fn bare_materialize_with_trailing_noise_still_fires() { + // Forward-compat with old `// materialize s3://foo` style. + let (is_m, _) = parse_pipeline_annotations("// materialize s3://legacy"); + assert!(is_m); + } + + #[test] + fn rejects_materializer_keyword_variants() { + let (is_m, _) = parse_pipeline_annotations("// materializer\n# materialized"); + assert!(!is_m); + } + + #[test] + fn on_schedule() { + let (_, triggers) = parse_pipeline_annotations("// on schedule \"0 */6 * * *\""); + assert_eq!(triggers.len(), 1); + assert_eq!( + triggers[0], + TriggerSpec::Schedule { cron: "0 */6 * * *".to_string() } + ); + } + + #[test] + fn on_asset_ts_py_sql() { + let code = "// on s3://a/b\n# on datatable://main\n-- on $res:f/foo"; + let (_, triggers) = parse_pipeline_annotations(code); + assert_eq!(triggers.len(), 3); + assert!(matches!( + triggers[0], + TriggerSpec::Asset { asset_kind: AssetKind::S3Object, .. } + )); + assert!(matches!( + triggers[1], + TriggerSpec::Asset { asset_kind: AssetKind::DataTable, .. } + )); + assert!(matches!( + triggers[2], + TriggerSpec::Asset { asset_kind: AssetKind::Resource, .. } + )); + } + + #[test] + fn on_deduplicates() { + let code = "// on s3://a/b\n# on s3://a/b"; + let (_, triggers) = parse_pipeline_annotations(code); + assert_eq!(triggers.len(), 1); + } + + #[test] + fn rejects_unknown_trigger_spec() { + let (_, triggers) = parse_pipeline_annotations("// on unknown://nope\n# on schedule"); + assert!(triggers.is_empty()); + } + + #[test] + fn combined() { + let code = "// materialize\n// on s3://in.csv\n// on schedule \"0 0 * * *\""; + let (is_m, triggers) = parse_pipeline_annotations(code); + assert!(is_m); + assert_eq!(triggers.len(), 2); + } +} diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 5f5b287dc3..0abe04d6eb 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use sqlx::Row; use windmill_common::{ - assets::{AssetKind, AssetUsageKind}, + assets::{parse_asset_trigger_ref, AssetKind, AssetUsageKind}, db::UserDB, error::JsonResult, utils::escape_ilike_pattern, @@ -21,6 +21,7 @@ pub fn workspaced_service() -> Router { .route("/list_by_usages", post(list_assets_by_usages)) .route("/list_favorites", get(list_favorites)) .route("/graph", get(asset_graph)) + .route("/pipelines", get(list_pipeline_folders)) } #[derive(Deserialize)] @@ -388,8 +389,14 @@ struct GraphAssetNode { struct GraphRunnableNode { path: String, usage_kind: AssetUsageKind, + // True iff the script was deployed with `// materialize` — drives the + // pipeline-member visual state on the frontend. + #[serde(skip_serializing_if = "std::ops::Not::not", default)] + is_materializer: bool, } +// Lineage edge from parsed r/w usages. One per (runnable, asset, access_type) +// tuple. Informational — not the DAG execution edges. #[derive(Serialize, Debug)] struct GraphEdge { runnable_path: String, @@ -399,11 +406,72 @@ struct GraphEdge { access_type: Option, } +// Declared `// on ` trigger edge — the actual execution DAG. +// For the eight non-native, non-schedule trigger kinds the variant carries +// just the trigger's workspace path; the config (broker, topic, auth, …) +// lives in its own trigger table and UI. +#[derive(Serialize, Debug)] +#[serde(tag = "trigger_kind", rename_all = "lowercase")] +enum TriggerEdge { + Asset { + asset_kind: AssetKind, + asset_path: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, + Schedule { + cron: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, + Webhook { + path: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, + Email { + path: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, + Kafka { + path: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, + Mqtt { + path: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, + Nats { + path: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, + Postgres { + path: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, + Sqs { + path: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, + Gcp { + path: String, + runnable_kind: AssetUsageKind, + runnable_path: String, + }, +} + #[derive(Serialize, Debug)] struct AssetGraphResponse { assets: Vec, runnables: Vec, edges: Vec, + triggers: Vec, } async fn asset_graph( @@ -450,12 +518,60 @@ async fn asset_graph( .fetch_all(&mut *tx) .await?; + // Pipeline triggers attached to scripts in scope. 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). + let trigger_rows = sqlx::query!( + r#" + SELECT + runnable_kind AS "runnable_kind!: AssetUsageKind", + runnable_path AS "runnable_path!", + trigger_kind::text AS "trigger_kind!", + trigger_ref AS "trigger_ref!" + FROM script_trigger + WHERE workspace_id = $1 + AND ($2::text IS NULL OR runnable_path LIKE $2) + "#, + &w_id, + folder_filter.as_deref(), + ) + .fetch_all(&mut *tx) + .await?; + + // Which scripts in scope are pipeline members (have `// materialize`). + let materializer_paths = sqlx::query!( + r#" + SELECT path AS "path!" + FROM script + WHERE workspace_id = $1 + AND auto_kind = 'materializer' + AND archived = false + AND deleted = false + AND ($2::text IS NULL OR path LIKE $2) + "#, + &w_id, + folder_filter.as_deref(), + ) + .fetch_all(&mut *tx) + .await?; + tx.commit().await?; + let materializer_script_paths: std::collections::HashSet = + materializer_paths.into_iter().map(|r| r.path).collect(); + let mut edges = Vec::with_capacity(rows.len()); let mut asset_set: std::collections::HashSet<(AssetKind, String)> = Default::default(); let mut runnable_set: std::collections::HashSet<(AssetUsageKind, String)> = Default::default(); + // Every materializer in scope goes into the graph, even when the parser + // didn't detect any asset r/w and the script has no triggers yet. Without + // this, a freshly-saved materializer whose template body hasn't been + // filled in would vanish from the pipeline view on graph refetch. + for path in &materializer_script_paths { + runnable_set.insert((AssetUsageKind::Script, path.clone())); + } + for r in rows { asset_set.insert((r.asset_kind, r.asset_path.clone())); runnable_set.insert((r.usage_kind, r.usage_path.clone())); @@ -468,6 +584,79 @@ async fn asset_graph( }); } + let mut triggers: Vec = Vec::with_capacity(trigger_rows.len()); + for t in trigger_rows { + runnable_set.insert((t.runnable_kind, t.runnable_path.clone())); + match t.trigger_kind.as_str() { + "asset" => { + // trigger_ref is `` — parse back out so both + // endpoints match what the frontend uses for node ids. + if let Some((asset_kind, asset_path)) = parse_asset_trigger_ref(&t.trigger_ref) { + // Make sure the source asset has a node even if nothing + // reads/writes it in this folder. + asset_set.insert((asset_kind, asset_path.clone())); + triggers.push(TriggerEdge::Asset { + asset_kind, + asset_path, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }); + } + } + "schedule" => { + triggers.push(TriggerEdge::Schedule { + cron: t.trigger_ref, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }); + } + // One-liners for the ` ` trigger variants. Kept as a + // flat match rather than a helper — each arm's variant ctor is + // different and we don't benefit from abstracting it. + "webhook" => triggers.push(TriggerEdge::Webhook { + path: t.trigger_ref, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }), + "email" => triggers.push(TriggerEdge::Email { + path: t.trigger_ref, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }), + "kafka" => triggers.push(TriggerEdge::Kafka { + path: t.trigger_ref, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }), + "mqtt" => triggers.push(TriggerEdge::Mqtt { + path: t.trigger_ref, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }), + "nats" => triggers.push(TriggerEdge::Nats { + path: t.trigger_ref, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }), + "postgres" => triggers.push(TriggerEdge::Postgres { + path: t.trigger_ref, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }), + "sqs" => triggers.push(TriggerEdge::Sqs { + path: t.trigger_ref, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }), + "gcp" => triggers.push(TriggerEdge::Gcp { + path: t.trigger_ref, + runnable_kind: t.runnable_kind, + runnable_path: t.runnable_path, + }), + _ => {} // Unknown trigger_kind — forward-compat. + } + } + let mut assets: Vec = asset_set .into_iter() .map(|(kind, path)| GraphAssetNode { kind, path }) @@ -476,9 +665,66 @@ async fn asset_graph( let mut runnables: Vec = runnable_set .into_iter() - .map(|(usage_kind, path)| GraphRunnableNode { path, usage_kind }) + .map(|(usage_kind, path)| { + let is_materializer = + usage_kind == AssetUsageKind::Script && materializer_script_paths.contains(&path); + GraphRunnableNode { path, usage_kind, is_materializer } + }) .collect(); runnables.sort_by(|a, b| a.path.cmp(&b.path)); - Ok(Json(AssetGraphResponse { assets, runnables, edges })) + Ok(Json(AssetGraphResponse { + assets, + runnables, + edges, + triggers, + })) +} + +// ------------------------------------------------------------------ +// GET /w/:workspace/assets/pipelines +// ------------------------------------------------------------------ +// Distinct folder names that contain at least one materializer script +// (auto_kind='materializer'). Used by the pipeline-editor folder picker +// and the "Pipeline" entry in folder views. Keyed by the partial index +// on `script (workspace_id, path) WHERE auto_kind='materializer' ...` +// so this is effectively O(matches). + +#[derive(Serialize, Debug)] +struct PipelineFolder { + folder: String, + script_count: i64, +} + +async fn list_pipeline_folders( + authed: ApiAuthed, + Path(w_id): Path, + Extension(user_db): Extension, +) -> JsonResult> { + let mut tx = user_db.begin(&authed).await?; + let rows = sqlx::query!( + r#" + SELECT + substring(path from '^f/([^/]+)/') AS "folder!", + COUNT(*) AS "script_count!" + FROM script + WHERE workspace_id = $1 + AND auto_kind = 'materializer' + AND archived = false + AND deleted = false + AND path LIKE 'f/%' + GROUP BY substring(path from '^f/([^/]+)/') + ORDER BY substring(path from '^f/([^/]+)/') + "#, + &w_id, + ) + .fetch_all(&mut *tx) + .await?; + tx.commit().await?; + + Ok(Json( + rows.into_iter() + .map(|r| PipelineFolder { folder: r.folder, script_count: r.script_count }) + .collect(), + )) } diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index c4b928bcb5..597ede8b8e 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -43,8 +43,9 @@ use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap; use windmill_common::{ assets::{ - clear_static_asset_usage, clear_static_asset_usage_by_script_hash, - insert_static_asset_usage, AssetUsageKind, AssetWithAltAccessType, + clear_script_triggers, clear_static_asset_usage, clear_static_asset_usage_by_script_hash, + insert_script_trigger, insert_static_asset_usage, parse_pipeline_annotations, + trigger_spec_to_row, AssetUsageKind, AssetWithAltAccessType, }, error::{self, to_anyhow}, min_version::{MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2}, @@ -966,7 +967,14 @@ async fn create_script_internal<'c>( let ci_test_refs = windmill_common::schema::parse_ci_test_annotation(&ns.content, &lang.as_comment_lit()); - let auto_kind = if ci_test_refs.is_some() { + // `materializer` wins over `test` and any client-supplied auto_kind. The + // bare `// materialize` marker is the opt-in signal for pipeline + // membership; parsed writes tell us what is produced (we don't record + // them in auto_kind itself). + let (is_materializer, pipeline_triggers) = parse_pipeline_annotations(&ns.content); + let auto_kind = if is_materializer { + Some("materializer".to_string()) + } else if ci_test_refs.is_some() { Some("test".to_string()) } else { auto_kind @@ -1277,6 +1285,22 @@ async fn create_script_internal<'c>( .await?; } + // Pipeline trigger edges: wipe-and-reinsert per deploy so removing an + // `// on ...` annotation drops the edge. + clear_script_triggers(&mut *tx, &w_id, &ns.path, AssetUsageKind::Script).await?; + for spec in &pipeline_triggers { + let (trigger_kind, trigger_ref) = trigger_spec_to_row(spec); + insert_script_trigger( + &mut *tx, + &w_id, + AssetUsageKind::Script, + &ns.path, + trigger_kind, + &trigger_ref, + ) + .await?; + } + let permissioned_as = username_to_permissioned_as(&authed.username); if let Some(parent_hash) = ns.parent_hash { tracing::info!( diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 2968493ecc..88be6dc939 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -2,8 +2,24 @@ use sqlx::PgExecutor; use crate::{error, scripts::ScriptHash}; +pub use windmill_parser::asset_parser::{parse_pipeline_annotations, TriggerSpec}; pub use windmill_types::assets::*; +#[derive(sqlx::Type, Debug, Clone, Copy, PartialEq)] +#[sqlx(type_name = "SCRIPT_TRIGGER_KIND", rename_all = "lowercase")] +pub enum ScriptTriggerKind { + Asset, + Schedule, + Webhook, + Email, + Kafka, + Mqtt, + Nats, + Postgres, + Sqs, + Gcp, +} + pub async fn insert_static_asset_usage<'e>( executor: impl PgExecutor<'e>, workspace_id: &str, @@ -66,6 +82,87 @@ pub async fn clear_static_asset_usage_by_script_hash<'e>( Ok(()) } +// Wipe all pipeline trigger declarations held by the given runnable. Used at +// deploy time: redeploying a script wipes its prior `// on` annotations so +// removing them implicitly un-declares those edges. +pub async fn clear_script_triggers<'e>( + executor: impl PgExecutor<'e>, + workspace_id: &str, + runnable_path: &str, + runnable_kind: AssetUsageKind, +) -> error::Result<()> { + sqlx::query!( + r#"DELETE FROM script_trigger + WHERE workspace_id = $1 AND runnable_kind = $2 AND runnable_path = $3"#, + workspace_id, + runnable_kind as AssetUsageKind, + runnable_path, + ) + .execute(executor) + .await?; + Ok(()) +} + +// Insert a single trigger declaration. Caller is expected to wipe first. +pub async fn insert_script_trigger<'e>( + executor: impl PgExecutor<'e>, + workspace_id: &str, + runnable_kind: AssetUsageKind, + runnable_path: &str, + trigger_kind: ScriptTriggerKind, + trigger_ref: &str, +) -> error::Result<()> { + sqlx::query!( + r#"INSERT INTO script_trigger + (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref) + VALUES ($1, $2, $3, $4, $5)"#, + workspace_id, + runnable_kind as AssetUsageKind, + runnable_path, + trigger_kind as ScriptTriggerKind, + trigger_ref, + ) + .execute(executor) + .await?; + Ok(()) +} + +// Inverse of trigger_spec_to_row for the Asset variant: parses a stored +// trigger_ref (e.g. `s3://foo`, `$res:bar`) back into the (kind, path) pair +// used as a graph node id. Returns None for refs that don't match any known +// asset prefix — callers should skip those edges. +pub fn parse_asset_trigger_ref(s: &str) -> Option<(AssetKind, String)> { + let (kind, path) = windmill_parser::asset_parser::parse_asset_syntax(s, false)?; + Some((asset_kind_from_parser(kind), path.to_string())) +} + +// Convert a parser TriggerSpec into the `(kind, ref)` pair stored in +// script_trigger. Asset refs get their canonical prefix back so the +// trigger_ref matches what downstream lookups expect. +pub fn trigger_spec_to_row(spec: &TriggerSpec) -> (ScriptTriggerKind, String) { + match spec { + TriggerSpec::Asset { asset_kind, path } => { + let prefix = match asset_kind { + windmill_parser::asset_parser::AssetKind::S3Object => "s3://", + windmill_parser::asset_parser::AssetKind::Resource => "$res:", + windmill_parser::asset_parser::AssetKind::Ducklake => "ducklake://", + windmill_parser::asset_parser::AssetKind::DataTable => "datatable://", + windmill_parser::asset_parser::AssetKind::Volume => "volume://", + }; + (ScriptTriggerKind::Asset, format!("{}{}", prefix, path)) + } + TriggerSpec::Schedule { cron } => (ScriptTriggerKind::Schedule, cron.clone()), + TriggerSpec::Webhook { path } => (ScriptTriggerKind::Webhook, path.clone()), + TriggerSpec::Email { path } => (ScriptTriggerKind::Email, path.clone()), + TriggerSpec::Kafka { path } => (ScriptTriggerKind::Kafka, path.clone()), + TriggerSpec::Mqtt { path } => (ScriptTriggerKind::Mqtt, path.clone()), + TriggerSpec::Nats { path } => (ScriptTriggerKind::Nats, path.clone()), + TriggerSpec::Postgres { path } => (ScriptTriggerKind::Postgres, path.clone()), + TriggerSpec::Sqs { path } => (ScriptTriggerKind::Sqs, path.clone()), + TriggerSpec::Gcp { path } => (ScriptTriggerKind::Gcp, path.clone()), + } +} + pub fn asset_kind_from_parser(parser_kind: windmill_parser::asset_parser::AssetKind) -> AssetKind { match parser_kind { windmill_parser::asset_parser::AssetKind::S3Object => AssetKind::S3Object, diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte index 7ee5e22f13..24565a809d 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte @@ -2,7 +2,6 @@ import '@xyflow/svelte/dist/base.css' import { SvelteFlow, - Background, Controls, MiniMap, ConnectionLineType, @@ -12,36 +11,132 @@ } from '@xyflow/svelte' import AssetNode from './AssetNode.svelte' import RunnableNode from './RunnableNode.svelte' + import TriggerNode, { type TriggerNodeKind } from './TriggerNode.svelte' + import AddNode from './AddNode.svelte' import AssetGraphEdge from './AssetGraphEdge.svelte' import { layoutAssetGraph } from './assetGraphLayout' import type { AssetGraphResponse, AssetGraphSelection } from './types' + import type { AssetKind } from '$lib/gen' + import { NODE } from '$lib/components/graph/util' + + // Width of the + node's rendered DOM element. Sugiyama allocates a full + // NODE.width slot for every node, so the small round button ends up + // left-aligned in its slot. We compensate by shifting the + node right + // by half the difference so its visual center matches the slot center. + const ADD_NODE_WIDTH = 40 interface Props { graph: AssetGraphResponse selection?: AssetGraphSelection | undefined onselect?: (selection: AssetGraphSelection | undefined) => void + // Called when the user clicks the per-asset + button (consumer-script + // entry). Kept optional so the canvas stays usable outside the + // pipeline editor. + onAddScriptForAsset?: ( + asset: { kind: AssetKind; path: string }, + language: import('$lib/gen').ScriptLang, + scriptPath: string + ) => void + // Pipeline-wide + node shown at the top of the graph. Picking any + // kind from the menu invokes this one callback with the chosen + // trigger source — the page uses it to seed the draft's annotation. + onAddMaterializer?: ( + language: import('$lib/gen').ScriptLang, + path: string, + source: + | { kind: 'schedule'; cron: string } + | { + kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp' + path: string | undefined + } + ) => void + // Folder-scoped prefix shown as a read-only chip in the insert menu + // path input (e.g. `f/{folder}/`). Shared across top + and per-asset +. + pathPrefix?: string + // Seeded editable suffix (e.g. `new_materializer`). + defaultPathSuffix?: string + // Default cron expression seeded into the top + template. + defaultScheduleCron?: string } - let { graph, selection, onselect }: Props = $props() + let { + graph, + selection, + onselect, + onAddScriptForAsset, + onAddMaterializer, + pathPrefix = '', + defaultPathSuffix = '', + defaultScheduleCron = '' + }: Props = $props() - // Producers (w/rw) point runnable → asset; consumers (r) point asset → - // runnable. The sugiyama layout (assetGraphLayout.ts) renders this as a - // top-down DAG: producers above, assets in the middle, consumers below. + const ADD_NODE_ID = '__add__' + + type BuiltEdge = { + id: string + source: string + target: string + // 'add-anchor' edges connect the + node to every otherwise-root node in + // the DAG. They force sugiyama to put + at layer 0 (top) and center + // it horizontally over the roots — same mechanism the flow editor + // uses for its Trigger node. Filtered out of rendered edges. + kind: + | 'lineage-write' + | 'lineage-read' + | 'trigger-asset' + | 'trigger-schedule' + | 'trigger-native' + | 'add-anchor' + unsaved?: boolean + } + + // Lineage edges (parsed r/w usages): writer → asset, asset → reader. + // Trigger edges (`// on `): asset → script or schedule → script. The + // lineage subgraph is informational; the trigger subgraph is executable. function build(g: AssetGraphResponse) { - const nodes: Array<{ id: string; type: 'asset' | 'runnable'; data: any }> = [] - const edges: Array<{ id: string; source: string; target: string; access: string | null }> = [] + const nodes: Array<{ + id: string + type: 'asset' | 'runnable' | 'trigger' | 'add' + data: any + }> = [] + const edges: BuiltEdge[] = [] + + const hasAddNode = onAddMaterializer != null + if (hasAddNode) { + nodes.push({ + id: ADD_NODE_ID, + type: 'add', + data: { + onAddMaterializer: onAddMaterializer!, + pathPrefix, + defaultPathSuffix, + defaultScheduleCron + } + }) + } for (const a of g.assets) { + const assetId = `asset:${a.kind}:${a.path}` nodes.push({ - id: `asset:${a.kind}:${a.path}`, + id: assetId, type: 'asset', - data: { asset_kind: a.kind, path: a.path } + data: { + asset_kind: a.kind, + path: a.path, + onAddScript: onAddScriptForAsset, + pathPrefix, + defaultPathSuffix + } }) } for (const r of g.runnables) { nodes.push({ id: `${r.usage_kind}:${r.path}`, type: 'runnable', - data: { runnable_kind: r.usage_kind, path: r.path } + data: { + runnable_kind: r.usage_kind, + path: r.path, + is_materializer: r.is_materializer ?? false + } }) } @@ -54,7 +149,8 @@ id: `prod:${runnableId}->${assetId}`, source: runnableId, target: assetId, - access + kind: 'lineage-write', + unsaved: e.unsaved }) } if (access === 'r' || access === 'rw') { @@ -62,10 +158,83 @@ id: `cons:${assetId}->${runnableId}`, source: assetId, target: runnableId, - access + kind: 'lineage-read', + unsaved: e.unsaved }) } } + + // Non-asset triggers (schedule + native) are rendered as source nodes + // above the materializer. Nodes are deduplicated per (kind, ref) + // tuple so a single schedule/webhook shared across multiple scripts + // shows as one node with N outgoing edges. A trigger node is + // considered unsaved if every attachment referencing it is unsaved. + const triggerSourceNodes = new Map< + string, + { allUnsaved: boolean; kind: TriggerNodeKind; ref: string } + >() + function recordSourceTrigger(id: string, kind: TriggerNodeKind, ref: string, unsaved: boolean) { + const prev = triggerSourceNodes.get(id) + if (!prev) { + triggerSourceNodes.set(id, { allUnsaved: unsaved, kind, ref }) + } else { + prev.allUnsaved = prev.allUnsaved && unsaved + } + } + + for (const t of g.triggers ?? []) { + const runnableId = `${t.runnable_kind}:${t.runnable_path}` + if (t.trigger_kind === 'asset') { + const assetId = `asset:${t.asset_kind}:${t.asset_path}` + edges.push({ + id: `trig-a:${assetId}->${runnableId}`, + source: assetId, + target: runnableId, + kind: 'trigger-asset', + unsaved: t.unsaved + }) + continue + } + const ref = t.trigger_kind === 'schedule' ? (t as any).cron : (t as any).path + const sourceId = `trigger:${t.trigger_kind}:${ref}` + recordSourceTrigger(sourceId, t.trigger_kind, ref, !!t.unsaved) + edges.push({ + id: `trig-${t.trigger_kind}:${sourceId}->${runnableId}`, + source: sourceId, + target: runnableId, + kind: t.trigger_kind === 'schedule' ? 'trigger-schedule' : 'trigger-native', + unsaved: t.unsaved + }) + } + for (const [id, info] of triggerSourceNodes) { + nodes.push({ + id, + type: 'trigger', + data: { kind: info.kind, ref: info.ref, unsaved: info.allUnsaved } + }) + } + + // Anchor the + to layer 0 by making it a parent of every node that + // would otherwise be a root (no incoming edge), *including* schedule + // nodes. Must run AFTER all real edges are added so we catch the + // current set of roots. Sugiyama then places + on layer 0 and + // centers it over whatever's below. + if (hasAddNode) { + const hasIncoming = new Set() + for (const e of edges) hasIncoming.add(e.target) + for (const n of nodes) { + if (n.id === ADD_NODE_ID) continue + if (!hasIncoming.has(n.id)) { + edges.push({ + id: `add-anchor:${n.id}`, + source: ADD_NODE_ID, + target: n.id, + kind: 'add-anchor' + }) + } + } + } + return { nodes, edges } } @@ -78,36 +247,125 @@ : `${selection.runnable_kind}:${selection.path}` }) + // Pane width drives horizontal centering (same pattern FlowGraphV2 uses). + // Bound on the outer wrapper; updates on pane resize via $state. + let paneWidth = $state(800) + 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()) { + 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((n) => { const p = positions.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 return { id: n.id, type: n.type, - position: { x: p.x, y: p.y }, + position: { x: p.x + xCenter + xShift, y: p.y + 40 }, data: n.data, - selected: n.id === selectedId + selected: n.id === selectedId, + // 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. + draggable: false, + selectable: n.id !== ADD_NODE_ID && n.type !== 'trigger' } }) }) let flowEdges = $derived.by(() => - model.edges.map((e) => ({ - id: e.id, - source: e.source, - target: e.target, - type: 'asset', - animated: e.access === 'rw', - style: - e.access === 'w' || e.access === 'rw' - ? 'stroke: rgb(59 130 246); stroke-width: 1.5px;' - : 'stroke: rgb(107 114 128); stroke-width: 1.25px;', - markerEnd: { type: MarkerType.ArrowClosed, width: 14, height: 14 } - })) + model.edges + // Anchor edges are layout-only. + .filter((e) => e.kind !== 'add-anchor') + .map((e) => { + let style: string + let animated = false + let markerColor: string | undefined = undefined + let strokeDasharray: string | undefined = undefined + let label: string | undefined = undefined + let labelStyle: string | undefined = undefined + switch (e.kind) { + case 'lineage-write': + style = 'stroke: rgb(59 130 246); stroke-width: 1.5px;' + break + case 'lineage-read': + style = 'stroke: rgb(107 114 128); stroke-width: 1.25px;' + break + case 'trigger-asset': + style = 'stroke: rgb(16 185 129); stroke-width: 2px;' + animated = true + markerColor = 'rgb(16 185 129)' + 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 + // say "this fires it" distinctly from lineage. + style = 'stroke: rgb(100 116 139); stroke-width: 2px;' + strokeDasharray = '6 3' + markerColor = 'rgb(100 116 139)' + label = 'triggers' + labelStyle = 'fill: rgb(100 116 139); font-size: 10px; font-weight: 600;' + break + default: + style = '' + } + // Unsaved edges (trigger or lineage) get a distinct dashed + // pattern + dimmed opacity so the user sees they're live- + // parsed, not persisted. Lineage edges from base graph data + // don't carry `unsaved`, only those synthesized by the draft + // overlay (e.g. the random output asset). + if (e.unsaved) { + strokeDasharray = '3 3' + style = `${style} opacity: 0.7;` + animated = false + if (label) label = `${label} (unsaved)` + } + if (strokeDasharray) { + style = `${style} stroke-dasharray: ${strokeDasharray};` + } + return { + id: e.id, + source: e.source, + target: e.target, + type: 'asset', + animated, + label, + labelStyle, + labelBgStyle: label ? 'fill: rgb(255 255 255 / 0.9);' : undefined, + style, + markerEnd: { + type: MarkerType.ArrowClosed, + width: 14, + height: 14, + color: markerColor + } + } + }) ) let nodes = $state.raw([]) @@ -121,7 +379,9 @@ const nodeTypes = { asset: AssetNode as any, - runnable: RunnableNode as any + runnable: RunnableNode as any, + trigger: TriggerNode as any, + add: AddNode as any } const edgeTypes = { @@ -133,19 +393,19 @@ const data = node.data as any if (node.type === 'asset') { onselect({ kind: 'asset', asset_kind: data.asset_kind, path: data.path }) - } else { + } else if (node.type === 'runnable') { onselect({ kind: 'runnable', runnable_kind: data.runnable_kind, path: data.path }) } + // 'schedule' doesn't produce a selection. } -
+
onselect?.(undefined)} --background-color={false} > -
- +
(n.type === 'asset' ? 'rgb(96 165 250 / 0.5)' : 'rgb(52 211 153 / 0.5)')} + nodeColor={(n) => + n.type === 'asset' + ? 'rgb(96 165 250 / 0.5)' + : n.type === 'trigger' + ? 'rgb(251 191 36 / 0.5)' + : 'rgb(52 211 153 / 0.5)'} nodeStrokeColor="transparent" maskColor="rgb(0 0 0 / 0.2)" /> diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index d8f4ae98f3..25603c23fd 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -7,39 +7,87 @@ import WorkerTagSelect from '$lib/components/WorkerTagSelect.svelte' import AssetGenericIcon from '$lib/components/icons/AssetGenericIcon.svelte' import { formatAssetKind } from '$lib/components/assets/lib' - import { Code2, ExternalLink, GitBranch, Loader2, Save, X } from 'lucide-svelte' + import { Code2, ExternalLink, GitBranch, Loader2, Save, Trash2, X } from 'lucide-svelte' import { inferArgs } from '$lib/infer' import { emptySchema, sendUserToast } from '$lib/utils' import type { AssetGraphSelection } from './types' + import { parsePipelineAnnotations, type PipelineAnnotations } from './parsePipelineAnnotations' interface Props { - selection: AssetGraphSelection + // Regular selection — loads the script by path for inline editing. + selection?: AssetGraphSelection | undefined + // Pre-built draft script not yet persisted. Takes precedence over + // `selection` when present. Used by the pipeline + menu so a new + // materializer opens inline instead of navigating to /scripts/add. + draftScript?: Script | undefined workspace: string onclose: () => void + // Called after a draft is saved for the first time so the page can + // refetch the graph and clear its local draft. + onDraftSaved?: (savedPath: string) => void + // Called when the user hits "Discard" on the draft header. Separate + // from `onclose` so the page can drop the draft from its map vs + // just dismissing the pane. + onDiscard?: () => void + // Emits live-parsed pipeline annotations from the open script so the + // canvas can overlay unsaved schedule / trigger nodes in real time. + // Fires whenever the script content changes. + onAnnotationsChange?: (scriptPath: string | undefined, annotations: PipelineAnnotations) => void } - let { selection, workspace, onclose }: Props = $props() + let { + selection, + draftScript, + workspace, + onclose, + onDraftSaved, + onDiscard, + onAnnotationsChange + }: Props = $props() - // Mirror the workspace-script editing flow from ScriptEditorDrawer: load - // the script by path, hand a mutable copy to , and on save - // re-infer args then POST a new version with parent_hash chained. + // When `draftScript` is provided we bypass the fetch entirely and edit + // it locally; saving calls ScriptService.createScript to deploy it. let scriptRes = resource( - [() => workspace, () => selection], - async ([ws, sel], _prev, { signal }) => { - if (sel.kind !== 'runnable' || sel.runnable_kind !== 'script') return undefined + [() => workspace, () => selection, () => draftScript], + async ([ws, sel, draft], _prev, { signal }) => { + if (draft) return undefined + if (!sel || sel.kind !== 'runnable' || sel.runnable_kind !== 'script') return undefined return await ScriptService.getScriptByPath({ workspace: ws, path: sel.path }, signal as any) } ) - // Local mutable copy that ScriptEditor binds to. Reset whenever the - // underlying resource yields a new script (selection change or refetch). + // Local mutable copy. For drafts: seeded once from the incoming prop + // (subsequent typing stays in `script`, not `draftScript`). For fetched + // scripts: reset whenever the resource yields new data. let script = $state
@@ -81,7 +138,15 @@ class="flex items-center justify-between gap-2 px-3 py-2 border-b shrink-0 min-h-10 whitespace-nowrap" >
- {#if selection.kind === 'asset'} + {#if isDraft && script} + +
+ + Draft materializer + + {script.path} +
+ {:else if selection?.kind === 'asset'} {selection.path}
- {:else if selection.runnable_kind === 'script'} + {:else if selection?.kind === 'runnable' && selection.runnable_kind === 'script'}
Script {selection.path}
- {:else} + {:else if selection?.kind === 'runnable' && selection.runnable_kind === 'flow'}
Flow @@ -108,7 +173,17 @@ {/if}
- {#if selection.kind === 'runnable' && selection.runnable_kind === 'script' && script} + {#if isDraft && onDiscard} + {/if} - {#if selection.kind === 'runnable'} + {#if !isDraft && selection?.kind === 'runnable'} + {/snippet} + +
+ {/if}
diff --git a/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte b/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte index f2db913452..aa11c4a5ce 100644 --- a/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte @@ -1,33 +1,44 @@
- -
+ +
{label} {data.path}
+ {#if data.is_materializer} +
+ +
+ {/if}
diff --git a/frontend/src/lib/components/assets/AssetGraph/types.ts b/frontend/src/lib/components/assets/AssetGraph/types.ts index 799cb7025d..90462db53c 100644 --- a/frontend/src/lib/components/assets/AssetGraph/types.ts +++ b/frontend/src/lib/components/assets/AssetGraph/types.ts @@ -10,20 +10,66 @@ export interface AssetGraphAssetNode { export interface AssetGraphRunnableNode { path: string usage_kind: GraphUsageKind + // Script has `// materialize` annotation. Drives the pipeline-member + // visual state; unrelated to what the script actually writes (that's + // parsed separately into lineage edges). + is_materializer?: boolean } +// Lineage edge from parsed r/w usages — informational only, not the +// execution DAG. `unsaved: true` for edges synthesized by a draft overlay +// (e.g. the random output asset attached at draft creation). export interface AssetGraphEdge { runnable_path: string runnable_kind: GraphUsageKind asset_kind: AssetKind asset_path: string access_type: 'r' | 'w' | 'rw' | null + unsaved?: boolean } +// Declared `// on ` — the actual execution DAG edges. +// `unsaved: true` marks overlays computed live from editor buffer that +// haven't been persisted to script_trigger yet. +export type NativeTriggerKind = + | 'webhook' + | 'email' + | 'kafka' + | 'mqtt' + | 'nats' + | 'postgres' + | 'sqs' + | 'gcp' + +export type AssetGraphTrigger = + | { + trigger_kind: 'asset' + asset_kind: AssetKind + asset_path: string + runnable_kind: GraphUsageKind + runnable_path: string + unsaved?: boolean + } + | { + trigger_kind: 'schedule' + cron: string + runnable_kind: GraphUsageKind + runnable_path: string + unsaved?: boolean + } + | { + trigger_kind: NativeTriggerKind + path: string + runnable_kind: GraphUsageKind + runnable_path: string + unsaved?: boolean + } + export interface AssetGraphResponse { assets: AssetGraphAssetNode[] runnables: AssetGraphRunnableNode[] edges: AssetGraphEdge[] + triggers: AssetGraphTrigger[] } export type AssetGraphNodeData = diff --git a/frontend/src/lib/components/home/ItemsList.svelte b/frontend/src/lib/components/home/ItemsList.svelte index f8bb688636..4b7ebbc9d1 100644 --- a/frontend/src/lib/components/home/ItemsList.svelte +++ b/frontend/src/lib/components/home/ItemsList.svelte @@ -9,8 +9,10 @@ type Script, ScriptService, type Flow, - type ListableRawApp + type ListableRawApp, + OpenAPI } from '$lib/gen' + import { resource } from 'runed' import { userStore, workspaceStore } from '$lib/stores' import type uFuzzy from '@leeoniya/ufuzzy' import { @@ -70,6 +72,25 @@ type TableApp = TableItem type TableRawApp = TableItem + // Folders with ≥1 materializer script (auto_kind='materializer'). Used by + // TreeView to surface a "Pipeline" entry inside those folders. Cheap + // thanks to the partial index on script.auto_kind. + let pipelineFoldersRes = resource( + () => $workspaceStore, + async (ws, _prev, { signal }) => { + if (!ws) return new Set() + const base_url = OpenAPI.BASE ?? '' + const res = await fetch(`${base_url}/w/${ws}/assets/pipelines`, { + credentials: 'include', + signal + }) + if (!res.ok) return new Set() + const rows = (await res.json()) as Array<{ folder: string }> + return new Set(rows.map((r) => r.folder)) + } + ) + let pipelineFolders = $derived(pipelineFoldersRes.current ?? new Set()) + let scripts: TableScript[] | undefined = $state() let flows: TableFlow[] | undefined = $state() let apps: TableApp[] | undefined = $state() @@ -557,6 +578,7 @@ {items} {nbDisplayed} {collapseAll} + {pipelineFolders} isSearching={filter !== ''} on:scriptChanged={() => loadScripts(includeWithoutMain)} on:flowChanged={loadFlows} diff --git a/frontend/src/lib/components/home/TreeView.svelte b/frontend/src/lib/components/home/TreeView.svelte index 93081cae5d..651f3fe6f9 100644 --- a/frontend/src/lib/components/home/TreeView.svelte +++ b/frontend/src/lib/components/home/TreeView.svelte @@ -1,11 +1,12 @@ - - - Asset graph — Windmill - - -{#if $userStore?.operator} -
Page not available for operators.
-{:else} -
-
-
-
-
- -
-
- -
- {#if graphRes.loading && !graphRes.current} -
- - Loading graph… -
- {:else if graphRes.error} -
- Failed to load graph: {graphRes.error.message} -
- {:else if graphRes.current && graphRes.current.assets.length === 0 && graphRes.current.runnables.length === 0} -
- No assets are referenced by scripts or flows in this workspace yet. -
- {:else if graphRes.current} - - - (selection = s)} - /> - - {#if selection && $workspaceStore} - - (selection = undefined)} - /> - - {/if} - - {/if} -
-
-{/if} diff --git a/frontend/src/routes/(root)/(logged)/scripts/add/+page.svelte b/frontend/src/routes/(root)/(logged)/scripts/add/+page.svelte index 8f43dde796..c482598b16 100644 --- a/frontend/src/routes/(root)/(logged)/scripts/add/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/scripts/add/+page.svelte @@ -33,6 +33,14 @@ const collabLang = page.url.searchParams.get('lang') as ScriptLang | null const wacParam = page.url.searchParams.get('wac') const importParam = page.url.searchParams.get('import') + // Pipeline opt-in marker. Any truthy value prefills ` + // materialize` at the top of the new script so the bare marker fires on + // deploy. ScriptBuilder's initialCode path only runs when content is + // empty, so this survives. + const materializeParam = page.url.searchParams.get('materialize') + // Optional `// on ` trigger. Value is the full asset-syntax string + // (e.g. s3://bucket/key) — written verbatim after the `materialize` line. + const onAssetParam = page.url.searchParams.get('on_asset') let initialArgs = urlArgs ? decodeState(urlArgs) : (get(initialArgsStore) ?? {}) if (get(initialArgsStore)) $initialArgsStore = undefined @@ -53,19 +61,61 @@ } } + // Language → comment prefix recognized by the pipeline annotation + // parser. Any of //, #, -- is accepted. + function commentPrefix(lang: ScriptLang): string { + switch (lang) { + case 'python3': + case 'bash': + case 'powershell': + case 'nu': + case 'ansible': + return '#' + case 'postgresql': + case 'mysql': + case 'bigquery': + case 'snowflake': + case 'mssql': + case 'oracledb': + case 'duckdb': + return '--' + default: + return '//' + } + } + + // Compose the prefix block for a new pipeline script: bare `// materialize` + // marker plus optional `// on ` trigger. Placed at the very top so + // comments precede any language-specific preamble the builder injects. + function pipelinePreamble(lang: ScriptLang): string { + const p = commentPrefix(lang) + const lines: string[] = [] + if (materializeParam) lines.push(`${p} materialize`) + if (onAssetParam) lines.push(`${p} on ${onAssetParam}`) + return lines.length ? lines.join('\n') + '\n' : '' + } + function defaultScript(): Script { + const language = ((wacParam === 'python' + ? 'python3' + : wacParam === 'typescript' + ? 'bun' + : null) ?? + collabLang ?? + $defaultScripts?.order?.filter( + (x) => $defaultScripts?.hidden == undefined || !$defaultScripts.hidden.includes(x) + )?.[0] ?? + 'bun') as ScriptLang return { hash: '', path: path ?? '', summary: '', - content: '', + content: pipelinePreamble(language), description: '', schema: schema, is_template: false, extra_perms: {}, - language: (wacParam === 'python' ? 'python3' : wacParam === 'typescript' ? 'bun' : null) ?? collabLang ?? ($defaultScripts?.order?.filter( - (x) => $defaultScripts?.hidden == undefined || !$defaultScripts.hidden.includes(x) - )?.[0] ?? 'bun') as ScriptLang, + language, kind: 'script' } } @@ -139,8 +189,7 @@ extra_perms: {} } if (isWac) { - importedWacTemplate = - imported.language === 'python3' ? 'wac_python' : 'wac_typescript' + importedWacTemplate = imported.language === 'python3' ? 'wac_python' : 'wac_typescript' sendUserToast('WAC script loaded from YAML/JSON') } else { sendUserToast('Script loaded from YAML/JSON') @@ -159,7 +208,12 @@ {initialArgs} bind:this={scriptBuilder} lockedLanguage={templatePath != null || hubPath != null} - template={importedWacTemplate ?? (wacParam === 'python' ? 'wac_python' : wacParam === 'typescript' ? 'wac_typescript' : 'script')} + template={importedWacTemplate ?? + (wacParam === 'python' + ? 'wac_python' + : wacParam === 'typescript' + ? 'wac_typescript' + : 'script')} onDeploy={(e) => { goto(`/scripts/get/${e.hash}?workspace=${$workspaceStore}`) }}