diff --git a/backend/.sqlx/query-02e526146f3584cd599dec708e1be48db3b0cd1c74adbfa2e4039377daa016f0.json b/backend/.sqlx/query-02e526146f3584cd599dec708e1be48db3b0cd1c74adbfa2e4039377daa016f0.json new file mode 100644 index 0000000000..5ca75fe782 --- /dev/null +++ b/backend/.sqlx/query-02e526146f3584cd599dec708e1be48db3b0cd1c74adbfa2e4039377daa016f0.json @@ -0,0 +1,38 @@ +{ + "db_name": "PostgreSQL", + "query": "\n DELETE FROM asset\n WHERE (workspace_id, path, kind) IN (\n SELECT workspace_id, path, kind FROM (\n SELECT a.workspace_id, a.path, a.kind, a.usage_kind, ROW_NUMBER() OVER (\n PARTITION BY a.workspace_id, a.path, a.kind\n ORDER BY a.created_at DESC\n ) as rn,\n limits.max_n\n FROM asset a\n INNER JOIN (\n SELECT * FROM UNNEST(\n $1::varchar[], \n $2::varchar[], \n $3::asset_kind[],\n $4::int[]\n ) AS t(workspace_id, path, kind, max_n)\n ) limits\n ON a.workspace_id = limits.workspace_id \n AND a.path = limits.path \n AND a.kind = limits.kind\n WHERE a.usage_kind = 'job'\n ) ranked\n WHERE rn > max_n\n )", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "VarcharArray", + "VarcharArray", + { + "Custom": { + "name": "asset_kind[]", + "kind": { + "Array": { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + } + } + } + }, + "Int4Array" + ] + }, + "nullable": [] + }, + "hash": "02e526146f3584cd599dec708e1be48db3b0cd1c74adbfa2e4039377daa016f0" +} diff --git a/backend/.sqlx/query-04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835.json b/backend/.sqlx/query-04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835.json new file mode 100644 index 0000000000..9c88e54c21 --- /dev/null +++ b/backend/.sqlx/query-04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835.json @@ -0,0 +1,78 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n flow_version.id AS version,\n flow_version.value->>'early_return' as early_return,\n flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor,\n flow_version.value->>'failure_module' IS NOT NULL as has_failure_module,\n (flow_version.value->>'chat_input_enabled')::boolean as chat_input_enabled,\n flow.tag,\n flow.dedicated_worker,\n flow.on_behalf_of_email,\n flow.edited_by,\n flow.labels\n FROM\n flow_version\n INNER JOIN flow\n ON flow.path = flow_version.path AND\n flow.workspace_id = flow_version.workspace_id\n WHERE\n flow_version.workspace_id = $1 AND\n flow_version.path = $2 AND\n flow_version.id = $3\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "early_return", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "has_preprocessor", + "type_info": "Bool" + }, + { + "ordinal": 3, + "name": "has_failure_module", + "type_info": "Bool" + }, + { + "ordinal": 4, + "name": "chat_input_enabled", + "type_info": "Bool" + }, + { + "ordinal": 5, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 6, + "name": "dedicated_worker", + "type_info": "Bool" + }, + { + "ordinal": 7, + "name": "on_behalf_of_email", + "type_info": "Text" + }, + { + "ordinal": 8, + "name": "edited_by", + "type_info": "Varchar" + }, + { + "ordinal": 9, + "name": "labels", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [ + false, + null, + null, + null, + null, + true, + true, + true, + false, + true + ] + }, + "hash": "04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835" +} diff --git a/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json b/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json index 79ef0c0a81..a779aa0e95 100644 --- a/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json +++ b/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json @@ -34,7 +34,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c.json b/backend/.sqlx/query-3194753fa26b1395fdbe3a7a434fe061db12e3bc4cb810e05110cd8115e1ca07.json similarity index 58% rename from backend/.sqlx/query-2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c.json rename to backend/.sqlx/query-3194753fa26b1395fdbe3a7a434fe061db12e3bc4cb810e05110cd8115e1ca07.json index 8153ca9932..2f21964011 100644 --- a/backend/.sqlx/query-2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c.json +++ b/backend/.sqlx/query-3194753fa26b1395fdbe3a7a434fe061db12e3bc4cb810e05110cd8115e1ca07.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "DELETE FROM schedule\n WHERE workspace_id = $1\n AND managed_by_runnable_path = $2", + "query": "DELETE FROM schedule\n WHERE workspace_id = $1\n AND script_path = $2\n AND managed", "describe": { "columns": [], "parameters": { @@ -11,5 +11,5 @@ }, "nullable": [] }, - "hash": "2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c" + "hash": "3194753fa26b1395fdbe3a7a434fe061db12e3bc4cb810e05110cd8115e1ca07" } diff --git a/backend/.sqlx/query-3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9.json b/backend/.sqlx/query-3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9.json new file mode 100644 index 0000000000..843efefd4f --- /dev/null +++ b/backend/.sqlx/query-3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO schedule (\n workspace_id, path, schedule, timezone, edited_by, script_path,\n is_flow, enabled, email, permissioned_as,\n ws_error_handler_muted, no_flow_overlap, cron_version,\n managed\n )\n VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', true)\n ON CONFLICT (workspace_id, path) DO UPDATE\n SET schedule = EXCLUDED.schedule,\n edited_at = now(),\n edited_by = EXCLUDED.edited_by,\n managed = true\n WHERE schedule.managed\n OR schedule.script_path = EXCLUDED.script_path\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar", + "Varchar", + "Bool", + "Varchar", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9" +} diff --git a/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json b/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json index 2a6930755a..7d950f6d8f 100644 --- a/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json +++ b/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json @@ -79,7 +79,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 36ddb8ab9f..713ccb9dd3 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - true + null ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-67e25a7c19ea0ffaf7ea5303fcd04af5a7eb488c76f783e690af0c2153b1d6a8.json b/backend/.sqlx/query-67e25a7c19ea0ffaf7ea5303fcd04af5a7eb488c76f783e690af0c2153b1d6a8.json index 50e7b53387..08ebe6bba5 100644 --- a/backend/.sqlx/query-67e25a7c19ea0ffaf7ea5303fcd04af5a7eb488c76f783e690af0c2153b1d6a8.json +++ b/backend/.sqlx/query-67e25a7c19ea0ffaf7ea5303fcd04af5a7eb488c76f783e690af0c2153b1d6a8.json @@ -160,7 +160,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-756f82b72af07fd690f37b2e16ed2d390604f4fc4cb330842a88d5764cbcf0c6.json b/backend/.sqlx/query-756f82b72af07fd690f37b2e16ed2d390604f4fc4cb330842a88d5764cbcf0c6.json index 8bc7e78a58..df721b4cde 100644 --- a/backend/.sqlx/query-756f82b72af07fd690f37b2e16ed2d390604f4fc4cb330842a88d5764cbcf0c6.json +++ b/backend/.sqlx/query-756f82b72af07fd690f37b2e16ed2d390604f4fc4cb330842a88d5764cbcf0c6.json @@ -127,7 +127,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json b/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json index fd32ba2753..ad9e57801e 100644 --- a/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json +++ b/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json @@ -34,7 +34,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json b/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json index f88fbc8a47..9a21f228ea 100644 --- a/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json +++ b/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json @@ -190,7 +190,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08.json b/backend/.sqlx/query-ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce.json similarity index 86% rename from backend/.sqlx/query-c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08.json rename to backend/.sqlx/query-ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce.json index 8376b31a9b..153532085e 100644 --- a/backend/.sqlx/query-c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08.json +++ b/backend/.sqlx/query-ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT\n runnable_kind AS \"runnable_kind!: AssetUsageKind\",\n runnable_path AS \"runnable_path!\",\n trigger_kind::text AS \"trigger_kind!\",\n trigger_ref AS \"trigger_ref!\"\n FROM script_trigger\n WHERE workspace_id = $1\n AND ($2::text IS NULL OR runnable_path LIKE $2)\n ", + "query": "\n SELECT\n runnable_kind AS \"runnable_kind!: AssetUsageKind\",\n runnable_path AS \"runnable_path!\",\n trigger_kind::text AS \"trigger_kind!\",\n trigger_ref AS \"trigger_ref!\"\n FROM script_trigger\n WHERE workspace_id = $1\n AND trigger_kind IN ('asset', 'schedule')\n AND ($2::text IS NULL OR runnable_path LIKE $2)\n ", "describe": { "columns": [ { @@ -48,5 +48,5 @@ false ] }, - "hash": "c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08" + "hash": "ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce" } diff --git a/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json b/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json index ca95a3bca8..3efa843923 100644 --- a/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json +++ b/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json @@ -165,7 +165,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json b/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json index 045d470de5..f6ff25a4bf 100644 --- a/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json +++ b/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json @@ -79,7 +79,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json b/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json index a8bb03e1b3..8dc66064dc 100644 --- a/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json +++ b/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json @@ -110,7 +110,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df.json b/backend/.sqlx/query-c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df.json deleted file mode 100644 index 2a122623fc..0000000000 --- a/backend/.sqlx/query-c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO schedule (\n workspace_id, path, schedule, timezone, edited_by, script_path,\n is_flow, enabled, email, permissioned_as,\n ws_error_handler_muted, no_flow_overlap, cron_version,\n managed_by_runnable_path\n )\n VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', $2)\n ON CONFLICT (workspace_id, path) DO UPDATE\n SET schedule = EXCLUDED.schedule,\n edited_at = now(),\n edited_by = EXCLUDED.edited_by,\n managed_by_runnable_path = EXCLUDED.managed_by_runnable_path\n WHERE schedule.managed_by_runnable_path = EXCLUDED.managed_by_runnable_path\n OR schedule.managed_by_runnable_path IS NULL AND schedule.script_path = EXCLUDED.script_path\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Varchar", - "Varchar", - "Bool", - "Varchar", - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df" -} diff --git a/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json b/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json index 5ed2e53367..d5365ffe94 100644 --- a/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json +++ b/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json @@ -110,7 +110,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-d4211392e174a0e8f89c7fcebdf120e5b0f629f9f04e08a2982df33ff23ac7a9.json b/backend/.sqlx/query-d4211392e174a0e8f89c7fcebdf120e5b0f629f9f04e08a2982df33ff23ac7a9.json index a27bea8b2a..33a5534b42 100644 --- a/backend/.sqlx/query-d4211392e174a0e8f89c7fcebdf120e5b0f629f9f04e08a2982df33ff23ac7a9.json +++ b/backend/.sqlx/query-d4211392e174a0e8f89c7fcebdf120e5b0f629f9f04e08a2982df33ff23ac7a9.json @@ -250,7 +250,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-e3ee812acd5bb9d5af39ca7dc61481ddd56a641e8e6f6c7a145f2cd5f3dc4602.json b/backend/.sqlx/query-e3ee812acd5bb9d5af39ca7dc61481ddd56a641e8e6f6c7a145f2cd5f3dc4602.json new file mode 100644 index 0000000000..2acb3c771e --- /dev/null +++ b/backend/.sqlx/query-e3ee812acd5bb9d5af39ca7dc61481ddd56a641e8e6f6c7a145f2cd5f3dc4602.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job SET args = $1, preprocessed = TRUE WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Jsonb", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "e3ee812acd5bb9d5af39ca7dc61481ddd56a641e8e6f6c7a145f2cd5f3dc4602" +} diff --git a/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json b/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json index 1a4cc407e0..470c651020 100644 --- a/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json +++ b/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json @@ -190,7 +190,8 @@ "google", "ci_test", "github", - "azure" + "azure", + "asset" ] } } diff --git a/backend/.sqlx/query-818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05.json b/backend/.sqlx/query-edf2814409bd480ac7c61140ec8d91b461782169bad6d47127c83bc5992d30ec.json similarity index 61% rename from backend/.sqlx/query-818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05.json rename to backend/.sqlx/query-edf2814409bd480ac7c61140ec8d91b461782169bad6d47127c83bc5992d30ec.json index 6d84a4158f..1a6c650e8f 100644 --- a/backend/.sqlx/query-818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05.json +++ b/backend/.sqlx/query-edf2814409bd480ac7c61140ec8d91b461782169bad6d47127c83bc5992d30ec.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "DELETE FROM schedule\n WHERE workspace_id = $1\n AND managed_by_runnable_path = $2", + "query": "DELETE FROM schedule\n WHERE workspace_id = $1\n AND script_path = $2\n AND managed", "describe": { "columns": [], "parameters": { @@ -11,5 +11,5 @@ }, "nullable": [] }, - "hash": "818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05" + "hash": "edf2814409bd480ac7c61140ec8d91b461782169bad6d47127c83bc5992d30ec" } diff --git a/backend/.sqlx/query-f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3.json b/backend/.sqlx/query-f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3.json new file mode 100644 index 0000000000..34b3be608b --- /dev/null +++ b/backend/.sqlx/query-f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3.json @@ -0,0 +1,41 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT kind, path, script_path, is_flow FROM (\n SELECT 'email' AS kind, path, script_path, is_flow FROM email_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'kafka', path, script_path, is_flow FROM kafka_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'mqtt', path, script_path, is_flow FROM mqtt_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'nats', path, script_path, is_flow FROM nats_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'postgres', path, script_path, is_flow FROM postgres_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'sqs', path, script_path, is_flow FROM sqs_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'gcp', path, script_path, is_flow FROM gcp_trigger\n WHERE workspace_id = $1\n ) t\n WHERE ($2::text IS NULL OR script_path LIKE $2)\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "kind", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "script_path", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "is_flow", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null, + null, + null, + null + ] + }, + "hash": "f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3" +} diff --git a/backend/migrations/20260430172050_pipeline_managed_schedules.down.sql b/backend/migrations/20260430172050_pipeline_managed_schedules.down.sql index 4a38a1921c..cab4b13a2f 100644 --- a/backend/migrations/20260430172050_pipeline_managed_schedules.down.sql +++ b/backend/migrations/20260430172050_pipeline_managed_schedules.down.sql @@ -1,2 +1,2 @@ -DROP INDEX IF EXISTS idx_schedule_managed_by_runnable_path; -ALTER TABLE schedule DROP COLUMN IF EXISTS managed_by_runnable_path; +DROP INDEX IF EXISTS idx_schedule_managed; +ALTER TABLE schedule DROP COLUMN IF EXISTS managed; diff --git a/backend/migrations/20260430172050_pipeline_managed_schedules.up.sql b/backend/migrations/20260430172050_pipeline_managed_schedules.up.sql index 8e0c223f33..0002ff9a98 100644 --- a/backend/migrations/20260430172050_pipeline_managed_schedules.up.sql +++ b/backend/migrations/20260430172050_pipeline_managed_schedules.up.sql @@ -1,12 +1,14 @@ --- Track schedules auto-created from a materializer's `// schedule ""` +-- Flag schedules auto-created from a pipeline script's `// schedule ""` -- annotation so reconciliation can update / drop them on subsequent deploys --- without touching schedules a user created manually. Null for all --- pre-existing rows. -ALTER TABLE schedule ADD COLUMN IF NOT EXISTS managed_by_runnable_path VARCHAR(255) DEFAULT NULL; +-- without touching schedules a user created manually. Defaults to false for +-- pre-existing rows. `script_path` already tells us which script owns the +-- row — this is just a boolean discriminator. +ALTER TABLE schedule ADD COLUMN IF NOT EXISTS managed BOOLEAN NOT NULL DEFAULT false; --- One managed schedule per (workspace, runnable). Index supports the --- reconciliation lookup ("does this script already have a managed schedule?") --- and the cleanup-on-delete query. -CREATE INDEX IF NOT EXISTS idx_schedule_managed_by_runnable_path - ON schedule (workspace_id, managed_by_runnable_path) - WHERE managed_by_runnable_path IS NOT NULL; +-- Partial index for the two hot reconciliation queries: +-- * "does this script already have a managed schedule?" (script_path lookup) +-- * "drop any managed schedules for this deleted script" (same lookup) +-- The boolean predicate keeps the index narrow (only managed rows are stored). +CREATE INDEX IF NOT EXISTS idx_schedule_managed + ON schedule (workspace_id, script_path) + WHERE managed; diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 1cfbbb898b..201069c8cd 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -115,35 +115,20 @@ pub enum TriggerSpec { 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, - }, + // `// on ` — marker-only declaration that this script wants to be + // triggered by a native trigger of the given kind. No path: the binding + // is the trigger row's own `script_path` field (set when the user creates + // the kafka/mqtt/… trigger in its dedicated UI). The graph endpoint + // discovers attached triggers by `WHERE script_path = ` and + // surfaces a "missing" placeholder when an annotation has no matching row. + Webhook, + Email, + Kafka, + Mqtt, + Nats, + Postgres, + Sqs, + Gcp, } impl TriggerSpec { @@ -620,33 +605,38 @@ fn parse_partitioned_spec(s: &str) -> Option { // webhook | email | kafka | mqtt | nats | postgres | sqs | gcp // (e.g. s3://bucket/key, $res:f/foo) // +// 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 { - // ` ` — 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 }), + // Marker-only native trigger keywords. The match table keeps the + // annotation set in lockstep with `TriggerSpec`. + const NATIVE_KINDS: &[(&str, TriggerSpec)] = &[ + ("webhook", TriggerSpec::Webhook), + ("email", TriggerSpec::Email), + ("kafka", TriggerSpec::Kafka), + ("mqtt", TriggerSpec::Mqtt), + ("nats", TriggerSpec::Nats), + ("postgres", TriggerSpec::Postgres), + ("sqs", TriggerSpec::Sqs), + ("gcp", TriggerSpec::Gcp), ]; - for (kw, ctor) in KINDS { + for (kw, spec) in NATIVE_KINDS { if let Some(rest) = s.strip_prefix(kw) { - if !rest.starts_with(|c: char| c.is_whitespace()) { + // Must be a complete word — `kafkalike` doesn't match `kafka`. + // Trailing whitespace alone is fine; any non-empty trailing + // content is treated as malformed (the annotation is marker-only). + if !rest.is_empty() && !rest.starts_with(|c: char| c.is_whitespace()) { continue; } - let path = rest.trim(); - if path.is_empty() { + if !rest.trim().is_empty() { return None; } - return Some(ctor(path.to_string())); + return Some(spec.clone()); } } @@ -845,12 +835,43 @@ mod pipeline_annotation_tests { } #[test] - fn on_kv_split_preserves_non_asset_and_spaced_refs() { - // ` ` ref with a trailing opt still parses; the opt is - // simply not carried for non-asset triggers. - let out = parse_pipeline_annotations("// on webhook f/foo debounce=30s"); + fn native_trigger_keywords_are_marker_only() { + // Marker form: `// on kafka` parses to the unit variant. + let out = parse_pipeline_annotations("// on kafka"); assert_eq!(out.triggers.len(), 1); - assert!(matches!(out.triggers[0], TriggerSpec::Webhook { .. })); + assert!(matches!(out.triggers[0], TriggerSpec::Kafka)); + + // Old path-bearing form is rejected (no path on native markers). + let out = parse_pipeline_annotations("// on webhook f/foo"); + assert!(out.triggers.is_empty()); + + // Trailing key=value opts are silently dropped by the line-level + // KV splitter before parse_trigger_spec sees them — same behaviour + // for both asset and native kinds. The opts have no meaning for a + // marker, but the marker still parses. + let out = parse_pipeline_annotations("// on mqtt debounce=30s"); + assert_eq!(out.triggers.len(), 1); + assert!(matches!(out.triggers[0], TriggerSpec::Mqtt)); + + // `kafkalike` mustn't match `kafka`. + let out = parse_pipeline_annotations("// on kafkalike"); + assert!(out.triggers.is_empty()); + } + + #[test] + fn all_native_marker_keywords_parse() { + let code = "// on webhook\n// on email\n// on kafka\n// on mqtt\n\ + // on nats\n// on postgres\n// on sqs\n// on gcp"; + let out = parse_pipeline_annotations(code); + assert_eq!(out.triggers.len(), 8); + assert!(matches!(out.triggers[0], TriggerSpec::Webhook)); + assert!(matches!(out.triggers[1], TriggerSpec::Email)); + assert!(matches!(out.triggers[2], TriggerSpec::Kafka)); + assert!(matches!(out.triggers[3], TriggerSpec::Mqtt)); + assert!(matches!(out.triggers[4], TriggerSpec::Nats)); + assert!(matches!(out.triggers[5], TriggerSpec::Postgres)); + assert!(matches!(out.triggers[6], TriggerSpec::Sqs)); + assert!(matches!(out.triggers[7], TriggerSpec::Gcp)); } #[test] diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 656e9e1f64..417ef4a969 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -407,9 +407,14 @@ struct GraphEdge { } // 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. +// 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. +// +// `webhook` is parsed as an annotation marker but has no dedicated trigger +// table — every script gets an implicit webhook endpoint — so no variant +// here. The frontend renders the marker from the source annotations alone. #[derive(Serialize, Debug)] #[serde(tag = "trigger_kind", rename_all = "lowercase")] enum TriggerEdge { @@ -424,11 +429,6 @@ enum TriggerEdge { runnable_kind: AssetUsageKind, runnable_path: String, }, - Webhook { - path: String, - runnable_kind: AssetUsageKind, - runnable_path: String, - }, Email { path: String, runnable_kind: AssetUsageKind, @@ -518,9 +518,11 @@ 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). + // 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 + // discovered below by querying each native trigger table directly. let trigger_rows = sqlx::query!( r#" SELECT @@ -530,6 +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 ($2::text IS NULL OR runnable_path LIKE $2) "#, &w_id, @@ -538,6 +541,43 @@ async fn asset_graph( .fetch_all(&mut *tx) .await?; + // Native triggers in scope. Each native trigger table stores its + // 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. + 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 + WHERE workspace_id = $1 + UNION ALL + SELECT 'kafka', path, script_path, is_flow FROM kafka_trigger + WHERE workspace_id = $1 + UNION ALL + SELECT 'mqtt', path, script_path, is_flow FROM mqtt_trigger + WHERE workspace_id = $1 + UNION ALL + SELECT 'nats', path, script_path, is_flow FROM nats_trigger + WHERE workspace_id = $1 + UNION ALL + SELECT 'postgres', path, script_path, is_flow FROM postgres_trigger + WHERE workspace_id = $1 + UNION ALL + SELECT 'sqs', path, script_path, is_flow FROM sqs_trigger + WHERE workspace_id = $1 + UNION ALL + SELECT 'gcp', path, script_path, is_flow FROM gcp_trigger + WHERE workspace_id = $1 + ) t + WHERE ($2::text IS NULL OR script_path LIKE $2) + "#, + &w_id, + folder_filter.as_deref(), + ) + .fetch_all(&mut *tx) + .await?; + // Which scripts in scope are pipeline members (have `// pipeline`). let pipeline_member_paths = sqlx::query!( r#" @@ -584,7 +624,8 @@ async fn asset_graph( }); } - let mut triggers: Vec = Vec::with_capacity(trigger_rows.len()); + let mut triggers: Vec = + Vec::with_capacity(trigger_rows.len() + native_trigger_rows.len()); for t in trigger_rows { runnable_set.insert((t.runnable_kind, t.runnable_path.clone())); match t.trigger_kind.as_str() { @@ -610,53 +651,36 @@ async fn asset_graph( 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. + _ => {} // Native kinds 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. + for t in native_trigger_rows { + let kind = t.kind.unwrap_or_default(); + let path = t.path.unwrap_or_default(); + let script_path = t.script_path.unwrap_or_default(); + let runnable_kind = if t.is_flow.unwrap_or(false) { + AssetUsageKind::Flow + } else { + AssetUsageKind::Script + }; + runnable_set.insert((runnable_kind, script_path.clone())); + let edge = match kind.as_str() { + "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 }, + "nats" => TriggerEdge::Nats { path, runnable_kind, runnable_path: script_path }, + "postgres" => TriggerEdge::Postgres { path, runnable_kind, runnable_path: script_path }, + "sqs" => TriggerEdge::Sqs { path, runnable_kind, runnable_path: script_path }, + "gcp" => TriggerEdge::Gcp { path, runnable_kind, runnable_path: script_path }, + _ => continue, + }; + triggers.push(edge); + } + let mut assets: Vec = asset_set .into_iter() .map(|(kind, path)| GraphAssetNode { kind, path }) diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 76952c68ad..961f5f7f0f 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -1566,10 +1566,15 @@ async fn create_script_internal<'c>( } // Pipeline trigger edges: wipe-and-reinsert per deploy so removing an - // `// on ...` annotation drops the edge. + // `// on ...` annotation drops the edge. Only Asset / Schedule produce + // a row — native trigger marker annotations (`// on kafka`, etc.) are + // discovered by the graph endpoint directly from the per-kind trigger + // tables, so `trigger_spec_to_row` returns None for those. 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); + let Some((trigger_kind, trigger_ref)) = trigger_spec_to_row(spec) else { + continue; + }; // Effective debounce for this edge: per-`// on debounce=` wins, // else the script-level `// debounce` default. Debounce only // applies to asset-cascade edges; other trigger kinds get none. diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 6a6a3ff617..7808e722b5 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -107,7 +107,7 @@ pub async fn clear_script_triggers<'e>( // `// schedule ""` annotation. Idempotent: each call brings the // `schedule` row in line with the annotation as of *this* deploy. // -// The schedule lives at the same path as the runnable. `managed_by_runnable_path` +// 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). @@ -135,16 +135,16 @@ pub async fn reconcile_pipeline_schedule<'e>( 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_by_runnable_path + managed ) - VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', $2) + 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_by_runnable_path = EXCLUDED.managed_by_runnable_path - WHERE schedule.managed_by_runnable_path = EXCLUDED.managed_by_runnable_path - OR schedule.managed_by_runnable_path IS NULL AND schedule.script_path = EXCLUDED.script_path + managed = true + WHERE schedule.managed + OR schedule.script_path = EXCLUDED.script_path "#, workspace_id, runnable_path, @@ -159,12 +159,13 @@ pub async fn reconcile_pipeline_schedule<'e>( } None => { // Drop any prior managed schedule for this runnable. Manual - // schedules at the same path keep `managed_by_runnable_path = - // NULL` and are unaffected. + // schedules at the same path keep `managed = false` and are + // unaffected. sqlx::query!( r#"DELETE FROM schedule WHERE workspace_id = $1 - AND managed_by_runnable_path = $2"#, + AND script_path = $2 + AND managed"#, workspace_id, runnable_path, ) @@ -186,7 +187,8 @@ pub async fn delete_managed_pipeline_schedule<'e>( sqlx::query!( r#"DELETE FROM schedule WHERE workspace_id = $1 - AND managed_by_runnable_path = $2"#, + AND script_path = $2 + AND managed"#, workspace_id, runnable_path, ) @@ -289,7 +291,12 @@ pub fn parse_asset_trigger_ref(s: &str) -> Option<(AssetKind, 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) { +// +// Returns `None` for native trigger kinds (Kafka, Mqtt, Postgres, …) — +// those annotations are marker-only and don't produce a `script_trigger` +// row. The actual binding lives on the trigger row's own `script_path` +// column; the graph endpoint looks it up directly per kind. +pub fn trigger_spec_to_row(spec: &TriggerSpec) -> Option<(ScriptTriggerKind, String)> { match spec { TriggerSpec::Asset { asset_kind, path, .. } => { let prefix = match asset_kind { @@ -299,17 +306,17 @@ pub fn trigger_spec_to_row(spec: &TriggerSpec) -> (ScriptTriggerKind, String) { windmill_parser::asset_parser::AssetKind::DataTable => "datatable://", windmill_parser::asset_parser::AssetKind::Volume => "volume://", }; - (ScriptTriggerKind::Asset, format!("{}{}", prefix, path)) + Some((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()), + TriggerSpec::Schedule { cron } => Some((ScriptTriggerKind::Schedule, cron.clone())), + TriggerSpec::Webhook + | TriggerSpec::Email + | TriggerSpec::Kafka + | TriggerSpec::Mqtt + | TriggerSpec::Nats + | TriggerSpec::Postgres + | TriggerSpec::Sqs + | TriggerSpec::Gcp => None, } } diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte index e72a9e0bd4..bc40f30670 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte @@ -15,7 +15,7 @@ import AddNode from './AddNode.svelte' import AssetGraphEdge from './AssetGraphEdge.svelte' import { layoutAssetGraph } from './assetGraphLayout' - import type { AssetGraphResponse, AssetGraphSelection } from './types' + import type { AssetGraphResponse, AssetGraphSelection, NativeTriggerKind } from './types' import type { RunnableRunState } from './activeRunnables.svelte' import type { AssetKind } from '$lib/gen' import { NODE } from '$lib/components/graph/util' @@ -51,10 +51,7 @@ path: string, source: | { kind: 'schedule'; cron: string } - | { - kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp' - path: string | undefined - }, + | { kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp' }, outputKind: import('./pipelineTemplates').PipelineOutputKind, aiPrompt?: string ) => void @@ -105,6 +102,11 @@ // a small badge on each runnable node. Same poll source as // `activeRunnableIds`; persists the last status while idle. runStates?: ReadonlyMap + // Click handler for a "missing trigger" placeholder. The page wires + // this to its native trigger drawer set so clicking the red node + // opens the matching editor with `script_path` pre-filled — no + // navigation, drafts stay intact. + onCreateMissingTrigger?: (kind: NativeTriggerKind, scriptPath: string) => void } let { graph, @@ -119,7 +121,8 @@ onRunnableMenuRemove, activeRunnable, activeRunnableIds, - runStates + runStates, + onCreateMissingTrigger }: Props = $props() const ADD_NODE_ID = '__add__' @@ -140,6 +143,10 @@ | 'trigger-native' | 'add-anchor' unsaved?: boolean + // Edge from a missing-trigger placeholder — styled red dashed to + // signal "this script declared `// on kafka` but no trigger row + // targets it; create one or remove the annotation". + missing?: boolean } // Graph-id of the script the user just launched (zero-latency hint), @@ -315,18 +322,33 @@ } // Non-asset triggers (schedule + native) are rendered as source nodes - // above the pipeline script. 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. + // above the pipeline script. Real (non-missing) nodes are + // deduplicated per (kind, ref) tuple so a single schedule shared + // across multiple scripts shows as one node with N outgoing edges. + // "missing" placeholders are scoped per-(kind, script) — each script + // gets its own placeholder so the prompt "create / delete" tells + // the user which script the annotation lives on. const triggerSourceNodes = new Map< string, - { allUnsaved: boolean; kind: TriggerNodeKind; ref: string } + { + allUnsaved: boolean + kind: TriggerNodeKind + ref: string + missing: boolean + runnable_path?: string + } >() - function recordSourceTrigger(id: string, kind: TriggerNodeKind, ref: string, unsaved: boolean) { + function recordSourceTrigger( + id: string, + kind: TriggerNodeKind, + ref: string, + unsaved: boolean, + missing: boolean, + runnable_path?: string + ) { const prev = triggerSourceNodes.get(id) if (!prev) { - triggerSourceNodes.set(id, { allUnsaved: unsaved, kind, ref }) + triggerSourceNodes.set(id, { allUnsaved: unsaved, kind, ref, missing, runnable_path }) } else { prev.allUnsaved = prev.allUnsaved && unsaved } @@ -345,22 +367,45 @@ }) continue } - const ref = t.trigger_kind === 'schedule' ? (t as any).cron : (t as any).path + 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 sourceId = `trigger:${t.trigger_kind}:${ref}` - recordSourceTrigger(sourceId, t.trigger_kind, ref, !!t.unsaved) + recordSourceTrigger( + sourceId, + t.trigger_kind, + ref, + !!t.unsaved, + isMissing, + isMissing ? t.runnable_path : undefined + ) 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 + unsaved: t.unsaved, + missing: isMissing }) } for (const [id, info] of triggerSourceNodes) { nodes.push({ id, type: 'trigger', - data: { kind: info.kind, ref: info.ref, unsaved: info.allUnsaved } + data: { + kind: info.kind, + ref: info.ref, + unsaved: info.allUnsaved, + missing: info.missing, + runnable_path: info.runnable_path, + onCreateMissingTrigger + } }) } @@ -517,6 +562,18 @@ style = `${style} opacity: 0.7;` if (label) label = `${label} (unsaved)` } + // Missing-trigger edge: overrides the per-kind stroke colour + // with red so the entire "annotated but no row" branch reads + // as broken at a glance. Composes with `unsaved` if both + // (red dashed dimmed — fresh draft annotation that also has + // no matching row, which is the common case). + if (e.missing) { + style = 'stroke: rgb(239 68 68); stroke-width: 2px;' + strokeDasharray = '3 3' + markerColor = 'rgb(239 68 68)' + label = 'missing trigger' + labelStyle = 'fill: rgb(239 68 68); font-size: 10px; font-weight: 600;' + } if (strokeDasharray) { style = `${style} stroke-dasharray: ${strokeDasharray};` } diff --git a/frontend/src/lib/components/assets/AssetGraph/TriggerNode.svelte b/frontend/src/lib/components/assets/AssetGraph/TriggerNode.svelte index 30ab68f82d..cc40d418c0 100644 --- a/frontend/src/lib/components/assets/AssetGraph/TriggerNode.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/TriggerNode.svelte @@ -110,36 +110,117 @@ import { Handle, Position } from '@xyflow/svelte' import { NODE } from '$lib/components/graph/util' import { twMerge } from 'tailwind-merge' + import { AlertTriangle } from 'lucide-svelte' interface Props { // `ref` is the cron expression for schedules, the trigger-path for - // every other kind. Rendered verbatim — no formatting per kind. - data: { kind: TriggerNodeKind; ref: string; unsaved?: boolean } + // attached native triggers, and a synthetic `missing:
-
- -
- - {style.label}{data.unsaved ? ' · unsaved' : ''} - - {data.ref} + {#if canCreate} + + + {:else} +
+ +
+ + {style.label}{data.missing ? ' · missing' : data.unsaved ? ' · unsaved' : ''} + + + {data.missing ? 'no trigger row' : data.ref} + +
-
+ {/if}
diff --git a/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts b/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts index f05c359f35..2976b58551 100644 --- a/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts +++ b/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts @@ -40,7 +40,10 @@ const NATIVE_TRIGGER_KEYWORDS: NativeTriggerKind[] = [ export const PARTITION_TOKEN = '{partition}' export type PipelineTriggerAsset = { kind: AssetKind; path: string } -export type PipelineNativeTrigger = { kind: NativeTriggerKind; path: string } +// Marker-only — native trigger annotations carry no path. The binding lives +// on the trigger row's own `script_path` field; the graph endpoint resolves +// it by querying the per-kind trigger tables. +export type PipelineNativeTrigger = { kind: NativeTriggerKind } export type PartitionKind = | { kind: 'daily' } @@ -130,15 +133,22 @@ type ParsedTriggerSpec = // Parse a single `on ` right-hand side. The top-level `// schedule` // is handled separately at the line level (not via `on`). +// +// Native trigger keywords (kafka, mqtt, …) are marker-only — `// on kafka` +// without a trailing path. Trailing content makes the line malformed and +// is rejected. Asset triggers always carry an `` ref. function parseTriggerSpec(s: string): ParsedTriggerSpec | undefined { for (const kw of NATIVE_TRIGGER_KEYWORDS) { if (s.startsWith(kw)) { const after = s.slice(kw.length) - // Require whitespace so `kafkalike` doesn't match `kafka`. - if (after.length === 0 || !/\s/.test(after[0])) continue - const path = after.trim() - if (!path) return undefined - return { kind: 'native', value: { kind: kw, path } } + // `kafka` must end the line (modulo whitespace) — `kafkalike` + // is not `kafka`. Anything trailing makes it malformed. + if (after.length === 0) { + return { kind: 'native', value: { kind: kw } } + } + if (!/\s/.test(after[0])) continue + if (after.trim().length > 0) return undefined + return { kind: 'native', value: { kind: kw } } } } const asset = parseAssetSyntax(s) @@ -262,9 +272,7 @@ export function parsePipelineAnnotations(code: string): PipelineAnnotations { out.triggerAssets.push(spec.value) } } else { - if ( - !out.nativeTriggers.some((n) => n.kind === spec.value.kind && n.path === spec.value.path) - ) { + if (!out.nativeTriggers.some((n) => n.kind === spec.value.kind)) { out.nativeTriggers.push(spec.value) } } diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts index 2d94a84c93..cdf32e202d 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts @@ -234,10 +234,11 @@ 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' - path: string | undefined - } + // Native trigger marker — the annotation is path-less (`// on kafka`). + // The binding lives on the trigger row's own `script_path`; the user + // creates that row via the drawer the canvas opens on the missing- + // trigger placeholder. + | { kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp' } export type TemplateContext = { language: ScriptLang @@ -262,7 +263,8 @@ function header(language: ScriptLang, triggers: DraftTriggerSource[]): string { case 'asset': return `${p} on ${t.ref}` default: - return `${p} on ${t.kind} ${t.path ?? ''}` + // Native triggers: marker-only — no trailing path. + return `${p} on ${t.kind}` } }) // Discoverability hint — the three annotations users most often miss diff --git a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts index f05628d98f..386c84e871 100644 --- a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts @@ -27,6 +27,7 @@ const input = (over: Partial = {}): ResolveGraphInput => ({ liveAnnotations: { scriptPath: undefined, annotations: ann() }, inferredWritesByPath: new Map(), inferredReadsByPath: new Map(), + annotatedNativeKindsByPath: new Map(), ...over }) diff --git a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts index d0f5c46bd9..f4aff5612b 100644 --- a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts +++ b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts @@ -1,4 +1,4 @@ -import type { AssetGraphResponse } from './types' +import type { AssetGraphResponse, NativeTriggerKind } from './types' import { parsePipelineAnnotations, type PipelineAnnotations } from './parsePipelineAnnotations' import { extractWrites, @@ -25,6 +25,13 @@ export type ResolveGraphInput = { /** Sticky session caches of inferred body writes/reads per script path. */ inferredWritesByPath: Map> inferredReadsByPath: Map> + /** + * Sticky cache of native trigger kinds declared via `// on ` in + * each script's deployed source. Filled by the load-time prefetch + * sweep. Used here to emit "missing" placeholders for scripts whose + * annotation has no matching trigger row in `base.triggers`. + */ + annotatedNativeKindsByPath: Map> } /** @@ -48,7 +55,8 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse { liveBodyAssets, liveAnnotations, inferredWritesByPath, - inferredReadsByPath + inferredReadsByPath, + annotatedNativeKindsByPath } = input // Every draft contributes: a runnable, an output asset, a write edge, @@ -136,13 +144,17 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse { const hasTriggerAsset = assets.some((x) => x.kind === a.kind && x.path === a.path) if (!hasTriggerAsset) assets.push({ kind: a.kind, path: a.path }) } + // Native trigger annotations on a draft are always "missing" until + // the user creates the matching trigger row — drafts can't carry a + // real trigger row since the script isn't deployed yet. Surface a + // red placeholder so the user knows to wire it up. for (const n of parsed.nativeTriggers) { extraTriggers.push({ trigger_kind: n.kind, - path: n.path, runnable_kind: 'script', runnable_path: path, - unsaved: true + unsaved: true, + missing: true }) } } @@ -206,11 +218,13 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse { unsaved: true }) } - // Persisted native triggers keyed by `:`, used to - // suppress duplicate overlay for already-saved `// on ` - // annotations. trigger_kind is narrower than the union so we - // cast through string. - const persistedNativeKeys = new Set( + // 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 + // at this script, the source node is already on the canvas — no + // overlay needed. Otherwise emit a "missing" placeholder so the + // user can either create the trigger row or remove the annotation. + const persistedNativeKinds = new Set( base.triggers .filter( (t) => @@ -219,17 +233,49 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse { t.runnable_kind === 'script' && t.runnable_path === livePath ) - .map((t) => `${t.trigger_kind}:${(t as { path: string }).path}`) + .map((t) => t.trigger_kind) ) for (const n of liveAnnotations.annotations.nativeTriggers) { - const key = `${n.kind}:${n.path}` - if (persistedNativeKeys.has(key)) continue + if (persistedNativeKinds.has(n.kind)) continue extraTriggers.push({ trigger_kind: n.kind, - path: n.path, runnable_kind: 'script', runnable_path: livePath, - unsaved: true + unsaved: true, + missing: true + }) + } + } + + // Cross-check for already-deployed scripts (not the open buffer): if a + // script's persisted body declares `// on kafka` but no matching + // kafka_trigger row points at it, surface a red placeholder. The + // annotated-kinds map is filled by the page-level prefetch sweep + // (one read per script in the folder); drafts and the active editor + // are handled by the loops above. Scripts that haven't been swept + // yet contribute nothing here — they'll surface on the next refetch. + const livePathExcl = livePath + for (const [scriptPath, kinds] of annotatedNativeKindsByPath) { + if (drafts.has(scriptPath)) continue + if (scriptPath === livePathExcl) continue + const attachedKinds = new Set( + base.triggers + .filter( + (t) => + t.trigger_kind !== 'asset' && + t.trigger_kind !== 'schedule' && + t.runnable_kind === 'script' && + t.runnable_path === scriptPath + ) + .map((t) => t.trigger_kind) + ) + for (const kind of kinds) { + if (attachedKinds.has(kind)) continue + extraTriggers.push({ + trigger_kind: kind, + runnable_kind: 'script', + runnable_path: scriptPath, + missing: true }) } } diff --git a/frontend/src/lib/components/assets/AssetGraph/types.ts b/frontend/src/lib/components/assets/AssetGraph/types.ts index 7de140482f..d2306146d4 100644 --- a/frontend/src/lib/components/assets/AssetGraph/types.ts +++ b/frontend/src/lib/components/assets/AssetGraph/types.ts @@ -70,10 +70,17 @@ export type AssetGraphTrigger = } | { trigger_kind: NativeTriggerKind - path: string + // path of the matching trigger row (kafka_trigger.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 runnable_path: string unsaved?: boolean + // Annotation declared but no matching trigger row was found — + // the canvas renders a red placeholder with a "Create trigger" + // affordance instead of a fully-wired source. + missing?: boolean } export interface AssetGraphResponse { diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index e75dead45a..513234bf7a 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -16,9 +16,13 @@ } from '$lib/components/assets/lib' import type { AssetGraphResponse, - AssetGraphSelection + AssetGraphSelection, + NativeTriggerKind } from '$lib/components/assets/AssetGraph/types' - import type { PipelineAnnotations } from '$lib/components/assets/AssetGraph/parsePipelineAnnotations' + import { + parsePipelineAnnotations, + type PipelineAnnotations + } from '$lib/components/assets/AssetGraph/parsePipelineAnnotations' import { resolveGraph } from '$lib/components/assets/AssetGraph/resolveGraph' import { generatePipelineDraft, @@ -56,6 +60,13 @@ import Popover from '$lib/components/meltComponents/Popover.svelte' import HideButton from '$lib/components/apps/editor/settingsPanel/HideButton.svelte' import { inferArgs, inferAssets } from '$lib/infer' + import KafkaTriggerEditor from '$lib/components/triggers/kafka/KafkaTriggerEditor.svelte' + import MqttTriggerEditor from '$lib/components/triggers/mqtt/MqttTriggerEditor.svelte' + import NatsTriggerEditor from '$lib/components/triggers/nats/NatsTriggerEditor.svelte' + import PostgresTriggerEditor from '$lib/components/triggers/postgres/PostgresTriggerEditor.svelte' + 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' // Variables and resources are declarative config, not pipeline assets — // they're hub-shaped (referenced by most runnables) and would swamp the @@ -274,6 +285,12 @@ // persisted asset rows. let inferredReadsByPath = $state>>(new Map()) + // Sticky cache of native trigger kinds declared via `// on kafka` etc. + // in each deployed script's source. Filled by the prefetch sweep below. + // resolveGraph uses this to flag scripts whose annotation has no + // matching trigger row — red placeholder on the canvas. + let annotatedNativeKindsByPath = $state>>(new Map()) + // Build a runnable Script from picked language / triggers / output. // Delegates to the shared template generator (pipelineTemplates.ts) so // the same logic is reachable from anywhere a draft is needed. @@ -678,7 +695,8 @@ liveBodyAssets, liveAnnotations, inferredWritesByPath, - inferredReadsByPath + inferredReadsByPath, + annotatedNativeKindsByPath }) ) @@ -799,6 +817,41 @@ // folder…" entry in the dropdown otherwise. let pickerModalOpen = $state(false) + // Native trigger editors mounted inline so clicking a "missing" + // placeholder opens the matching drawer with `script_path` pre-filled + // — keeps pipeline drafts intact instead of navigating away. Each + // editor's wrapper lazily mounts its Inner only when `open=true`, so + // holding refs to all seven is cheap. + let kafkaEditor: KafkaTriggerEditor | undefined = $state() + let mqttEditor: MqttTriggerEditor | undefined = $state() + let natsEditor: NatsTriggerEditor | undefined = $state() + let postgresEditor: PostgresTriggerEditor | undefined = $state() + let sqsEditor: SqsTriggerEditor | undefined = $state() + let gcpEditor: GcpTriggerEditor | undefined = $state() + let emailEditor: EmailTriggerEditor | undefined = $state() + + function openMissingTriggerDrawer(kind: NativeTriggerKind, scriptPath: string) { + switch (kind) { + case 'kafka': + return kafkaEditor?.openNew(false, scriptPath) + case 'mqtt': + return mqttEditor?.openNew(false, scriptPath) + case 'nats': + return natsEditor?.openNew(false, scriptPath) + case 'postgres': + return postgresEditor?.openNew(false, scriptPath) + case 'sqs': + return sqsEditor?.openNew(false, scriptPath) + case 'gcp': + return gcpEditor?.openNew(false, scriptPath) + case 'email': + return emailEditor?.openNew(false, scriptPath) + // webhook has no dedicated editor; schedule is inline-managed. + default: + return + } + } + // Reuse the empty AssetGraphResponse so we can still render the canvas // (layout, controls, mini-map) on a fresh pipeline. const EMPTY_GRAPH: AssetGraphResponse = { @@ -888,7 +941,11 @@ .filter((r) => r.usage_kind === 'script') .map((r) => r.path) .filter( - (p) => !drafts.has(p) && !inferredWritesByPath.has(p) && !inferredReadsByPath.has(p) + (p) => + !drafts.has(p) && + !inferredWritesByPath.has(p) && + !inferredReadsByPath.has(p) && + !annotatedNativeKindsByPath.has(p) ) ) if (targets.length === 0) return @@ -900,11 +957,18 @@ try { const s = await ScriptService.getScriptByPath({ workspace: ws, path }) if (gen !== assetPrefetchGen) return - const res = await inferAssets(s.language, s.content ?? '') + const content = s.content ?? '' + const res = await inferAssets(s.language, content) if (gen !== assetPrefetchGen) 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)) { @@ -917,6 +981,13 @@ next.set(path, reads) inferredReadsByPath = next } + // 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. @@ -1085,6 +1156,7 @@ {pathPrefix} defaultPathSuffix={DEFAULT_PATH_SUFFIX} defaultScheduleCron={DEFAULT_SCHEDULE_CRON} + onCreateMissingTrigger={openMissingTriggerDrawer} onselect={(s) => { // Clicking a draft runnable node re-opens it in the pane; // clicking anything else selects it normally and detaches @@ -1346,6 +1418,18 @@ + + graphRes.refetch()} /> + graphRes.refetch()} /> + graphRes.refetch()} /> + graphRes.refetch()} /> + graphRes.refetch()} /> + graphRes.refetch()} /> + graphRes.refetch()} /> + {#if leaveModalOpen}