starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-20 00:02:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: on behalf of (#5058)

Browse Source 
* feat: on behalf of

* fix

* revert mistakke

* temp ee ref

* renaming + CLI

* sqlx

* nits + ee ref
This commit is contained in:
HugoCasa
2025-01-14 18:32:25 +01:00
committed by GitHub
parent 9152b237c5
commit 175379fbe9
56 changed files with 947 additions and 361 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0a56301b5aaf57339cb2904c8f617366b74e891034d32f2867ccb019da869fc8.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow\n (workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at) \n SELECT workspace_id, REGEXP_REPLACE(path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1'), summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at\n FROM flow \n WHERE path LIKE ('u/' || $2 || '/%') AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "0a56301b5aaf57339cb2904c8f617366b74e891034d32f2867ccb019da869fc8"
}
backend/.sqlx/query-13358ffeb0917dd9dff9f8527a59dfee63bb704c3f712af179732dc281411917.json
Generated
-15
View File
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow \n (workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at) \n SELECT $1, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at\n FROM flow WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "13358ffeb0917dd9dff9f8527a59dfee63bb704c3f712af179732dc281411917"
}
backend/.sqlx/query-25155e44372aecbb38d042bfc2772ed0c01a0bb974488530cd713b834f537f4a.json
Generated
-16
View File
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow\n (workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at) \n SELECT workspace_id, REGEXP_REPLACE(path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1'), summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at\n FROM flow \n WHERE path LIKE ('u/' || $2 || '/%') AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "25155e44372aecbb38d042bfc2772ed0c01a0bb974488530cd713b834f537f4a"
}
backend/.sqlx/query-2f440ab6083764b49e309c1f8dde0c5508da110f54bc5f465f1892afdf851af0.json
Generated
-77
View File
@@ -1,77 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, created_at, trigger_kind as \"trigger_kind: _\", payload as \"payload: _\", trigger_extra as \"trigger_extra: _\"\n FROM capture\n WHERE workspace_id = $1\n AND path = $2 AND is_flow = $3\n AND ($4::trigger_kind IS NULL OR trigger_kind = $4)\n ORDER BY created_at DESC",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 2,
"name": "trigger_kind: _",
"type_info": {
"Custom": {
"name": "trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats"
]
}
}
}
},
{
"ordinal": 3,
"name": "payload: _",
"type_info": "Jsonb"
},
{
"ordinal": 4,
"name": "trigger_extra: _",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Bool",
{
"Custom": {
"name": "trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats"
]
}
}
}
]
},
"nullable": [
false,
false,
false,
false,
true
]
},
"hash": "2f440ab6083764b49e309c1f8dde0c5508da110f54bc5f465f1892afdf851af0"
}
backend/.sqlx/query-3fedf12ee5d44e8af9c3864cc70a8f607dccc015ce2d5150c56c68074cd0f84d.json
Generated
-35
View File
@@ -1,35 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT tag, dedicated_worker, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor \n FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 and flow.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "has_preprocessor",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true,
true,
null
]
},
"hash": "3fedf12ee5d44e8af9c3864cc70a8f607dccc015ce2d5150c56c68074cd0f84d"
}
backend/.sqlx/query-9e64c6b6db2155ad8e6e514b08da92d80caafd1a217813bcd354aa264810f690.json → backend/.sqlx/query-4205d237c123d8d1c9ff2d61118027a80ccc8cd75e9703cb1d014b45f57c2be6.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow (workspace_id, path, summary, description, dependency_job, draft_only, tag, dedicated_worker, visible_to_runner_only, value, schema, edited_by, edited_at) \n VALUES ($1, $2, $3, $4, NULL, $5, $6, $7, $8, $9, $10::text::json, $11, now())",
"query": "INSERT INTO flow (workspace_id, path, summary, description, dependency_job, draft_only, tag, dedicated_worker, visible_to_runner_only, on_behalf_of_email, value, schema, edited_by, edited_at) \n VALUES ($1, $2, $3, $4, NULL, $5, $6, $7, $8, $9, $10, $11::text::json, $12, now())",
"describe": {
"columns": [],
"parameters": {
@@ -13,6 +13,7 @@
"Varchar",
"Bool",
"Bool",
"Text",
"Jsonb",
"Text",
"Varchar"
@@ -20,5 +21,5 @@
},
"nullable": []
},
"hash": "9e64c6b6db2155ad8e6e514b08da92d80caafd1a217813bcd354aa264810f690"
"hash": "4205d237c123d8d1c9ff2d61118027a80ccc8cd75e9703cb1d014b45f57c2be6"
}
backend/.sqlx/query-544a00afb5c72d2aa24ab4ccfe68d1e383aeda9a5a71183b7fbaa41deca4e333.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow \n (workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at) \n SELECT workspace_id, $1, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at\n FROM flow\n WHERE path = $2 AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "544a00afb5c72d2aa24ab4ccfe68d1e383aeda9a5a71183b7fbaa41deca4e333"
}
backend/.sqlx/query-fc7ac95d8d5f4532a2b0e50f84c1e273d25543401639a68f868e1e4fd8b0640a.json → backend/.sqlx/query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json
Generated
+10 -10
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)",
"query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)",
"describe": {
"columns": [
{
@@ -76,18 +76,18 @@
},
{
"ordinal": 9,
"name": "delete_after_use",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 10,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 11,
"name": "has_preprocessor",
"type_info": "Bool"
"name": "created_by",
"type_info": "Varchar"
}
],
"parameters": {
@@ -108,8 +108,8 @@
true,
true,
true,
true
false
]
},
"hash": "fc7ac95d8d5f4532a2b0e50f84c1e273d25543401639a68f868e1e4fd8b0640a"
"hash": "61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4"
}
backend/.sqlx/query-6d36da815795d5cac2e76b4d34a7f1f9f836bd5b6866d3db26b25428d39c0b23.json
Generated
+29
View File
@@ -0,0 +1,29 @@
{
"db_name": "PostgreSQL",
"query": "SELECT on_behalf_of_email, edited_by FROM flow WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "edited_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true,
false
]
},
"hash": "6d36da815795d5cac2e76b4d34a7f1f9f836bd5b6866d3db26b25428d39c0b23"
}
backend/.sqlx/query-64ce3eabed6de7a4f99884640c21218ac19ab2a4bc5f8b42deb2476f85d0deac.json → backend/.sqlx/query-75451b6d48e4c26812ae64981d0d968b8fb0bf4374a2fccc167fa879bad7078f.json
Generated
+15 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select path, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor from script where hash = $1 AND workspace_id = $2",
"query": "select path, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by from script where hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
@@ -88,6 +88,16 @@
"ordinal": 11,
"name": "has_preprocessor",
"type_info": "Bool"
},
{
"ordinal": 12,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 13,
"name": "created_by",
"type_info": "Varchar"
}
],
"parameters": {
@@ -108,8 +118,10 @@
true,
true,
true,
true
true,
true,
false
]
},
"hash": "64ce3eabed6de7a4f99884640c21218ac19ab2a4bc5f8b42deb2476f85d0deac"
"hash": "75451b6d48e4c26812ae64981d0d968b8fb0bf4374a2fccc167fa879bad7078f"
}
backend/.sqlx/query-81a9d976ac5c1a78c83b95a1164995e78878a4a4ff6894a04e6626cdd98c24e4.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow \n (workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at) \n SELECT $1, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at\n FROM flow WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "81a9d976ac5c1a78c83b95a1164995e78878a4a4ff6894a04e6626cdd98c24e4"
}
backend/.sqlx/query-8929873f308f1d131b2afda17267f82bac6e34a35b94765463c12e282fba6daa.json
Generated
+20
View File
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "SELECT CURRENT_USER = 'windmill_user' as \"is_windmill_user!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "is_windmill_user!",
"type_info": "Bool"
}
],
"parameters": {
"Left": []
},
"nullable": [
null
]
},
"hash": "8929873f308f1d131b2afda17267f82bac6e34a35b94765463c12e282fba6daa"
}
backend/.sqlx/query-9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418.json → backend/.sqlx/query-89de3ff8ab32e545efcbcda05f994cb1a32c4991cbd25046282d34272587d2de.json
Generated
+27 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout FROM script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)",
"query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)",
"describe": {
"columns": [
{
@@ -76,8 +76,28 @@
},
{
"ordinal": 9,
"name": "delete_after_use",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 11,
"name": "has_preprocessor",
"type_info": "Bool"
},
{
"ordinal": 12,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 13,
"name": "created_by",
"type_info": "Varchar"
}
],
"parameters": {
@@ -96,8 +116,12 @@
false,
true,
true,
true
true,
true,
true,
true,
false
]
},
"hash": "9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418"
"hash": "89de3ff8ab32e545efcbcda05f994cb1a32c4991cbd25046282d34272587d2de"
}
backend/.sqlx/query-8fb2581a439c26391e66ae7fac32c6cd2932f28ab6490ace027ed3a790b2a0f7.json
Generated
+28
View File
@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "SELECT edited_by, on_behalf_of_email FROM flow WHERE path = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "on_behalf_of_email",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
true
]
},
"hash": "8fb2581a439c26391e66ae7fac32c6cd2932f28ab6490ace027ed3a790b2a0f7"
}
backend/.sqlx/query-98081ee9a4a4a52ae70d74d26e8608a010722d7ea9dc343ae8f03e38978ffaba.json → backend/.sqlx/query-a0833b9899833166891c5de926f78632fae1123e736d728bf92cb2de004b6826.json
Generated
+15 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT tag, dedicated_worker, flow_version.value->>'early_return' as early_return, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor\n FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 and flow.workspace_id = $2",
"query": "SELECT tag, dedicated_worker, flow_version.value->>'early_return' as early_return, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, on_behalf_of_email, edited_by\n FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 and flow.workspace_id = $2",
"describe": {
"columns": [
{
@@ -22,6 +22,16 @@
"ordinal": 3,
"name": "has_preprocessor",
"type_info": "Bool"
},
{
"ordinal": 4,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "edited_by",
"type_info": "Varchar"
}
],
"parameters": {
@@ -34,8 +44,10 @@
true,
true,
null,
null
null,
true,
false
]
},
"hash": "98081ee9a4a4a52ae70d74d26e8608a010722d7ea9dc343ae8f03e38978ffaba"
"hash": "a0833b9899833166891c5de926f78632fae1123e736d728bf92cb2de004b6826"
}
backend/.sqlx/query-ab3f8ebcebf0088c1104be4014c43dab1c08d88c9bcc40b62f2e6bbbbc234ff3.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "SET LOCAL ROLE NONE",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "ab3f8ebcebf0088c1104be4014c43dab1c08d88c9bcc40b62f2e6bbbbc234ff3"
}
backend/.sqlx/query-b53007a988c6336015dbe50a207d4996b379d436566ee697bbe32361c859be64.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "SET LOCAL ROLE windmill_user",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "b53007a988c6336015dbe50a207d4996b379d436566ee697bbe32361c859be64"
}
backend/.sqlx/query-899a406cc03659e29bad831cc4d3d1dcda1a39b826a03136353344ccae29871b.json → backend/.sqlx/query-ba8bde5018fdf7b12f85cd3a6557c4accfc78bf160c1277f35d9d8ddcd056963.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE flow SET path = $1, summary = $2, description = $3,dependency_job = NULL, draft_only = NULL, tag = $4, dedicated_worker = $5, visible_to_runner_only = $6, value = $7, schema = $8::text::json, edited_by = $9, edited_at = now()\n WHERE path = $10 AND workspace_id = $11",
"query": "UPDATE flow SET path = $1, summary = $2, description = $3,dependency_job = NULL, draft_only = NULL, tag = $4, dedicated_worker = $5, visible_to_runner_only = $6, on_behalf_of_email = $7, value = $8, schema = $9::text::json, edited_by = $10, edited_at = now()\n WHERE path = $11 AND workspace_id = $12",
"describe": {
"columns": [],
"parameters": {
@@ -11,6 +11,7 @@
"Varchar",
"Bool",
"Bool",
"Text",
"Jsonb",
"Text",
"Varchar",
@@ -20,5 +21,5 @@
},
"nullable": []
},
"hash": "899a406cc03659e29bad831cc4d3d1dcda1a39b826a03136353344ccae29871b"
"hash": "ba8bde5018fdf7b12f85cd3a6557c4accfc78bf160c1277f35d9d8ddcd056963"
}
backend/.sqlx/query-b49b1eaf58e62063c7a03039d0c36e65991b1d21a3306390fb1c7ca38babafe3.json → backend/.sqlx/query-bc00efe12901bc7b49c7892d2d9da675e0eaf5e6cc013149740bc778987a1006.json
Generated
+4 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31)",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32)",
"describe": {
"columns": [],
"parameters": {
@@ -75,10 +75,11 @@
"Bool",
"Bool",
"Varchar",
"Bool"
"Bool",
"Text"
]
},
"nullable": []
},
"hash": "b49b1eaf58e62063c7a03039d0c36e65991b1d21a3306390fb1c7ca38babafe3"
"hash": "bc00efe12901bc7b49c7892d2d9da675e0eaf5e6cc013149740bc778987a1006"
}
backend/.sqlx/query-c794ff5e14429a1bc56f5e70cfd10e2160d1b7e123c7ab226dc77d1789f164f9.json
Generated
+47
View File
@@ -0,0 +1,47 @@
{
"db_name": "PostgreSQL",
"query": "SELECT tag, dedicated_worker, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, on_behalf_of_email, edited_by\n FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 and flow.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "has_preprocessor",
"type_info": "Bool"
},
{
"ordinal": 3,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "edited_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true,
true,
null,
true,
false
]
},
"hash": "c794ff5e14429a1bc56f5e70cfd10e2160d1b7e123c7ab226dc77d1789f164f9"
}
backend/.sqlx/query-d860cd90e583e4666beb37b283bd990dbcb91b781728b5b6253051decfda83d6.json
Generated
-16
View File
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow \n (workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at) \n SELECT workspace_id, $1, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at\n FROM flow\n WHERE path = $2 AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "d860cd90e583e4666beb37b283bd990dbcb91b781728b5b6253051decfda83d6"
}
backend/.sqlx/query-e08dddf6af2656b561c453460ff928e7a158bd29d9035bee136f9952fb96bad4.json
Generated
+4 -2
View File
@@ -25,7 +25,8 @@
"http",
"websocket",
"kafka",
"email"
"email",
"nats"
]
}
}
@@ -56,7 +57,8 @@
"http",
"websocket",
"kafka",
"email"
"email",
"nats"
]
}
}
backend/.sqlx/query-1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c.json → backend/.sqlx/query-ec7836df5f9056ec70015800b7f4feaeb1b671120f5f8c98fca8c89c6587fc35.json
Generated
+15 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
"query": "select tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, on_behalf_of_email, created_by from script where hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
@@ -78,6 +78,16 @@
"ordinal": 9,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 10,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 11,
"name": "created_by",
"type_info": "Varchar"
}
],
"parameters": {
@@ -96,8 +106,10 @@
true,
true,
true,
true
true,
true,
false
]
},
"hash": "1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c"
"hash": "ec7836df5f9056ec70015800b7f4feaeb1b671120f5f8c98fca8c89c6587fc35"
}
backend/.sqlx/query-f44f1a794ab217be05a9feb21e57b3409d9166a028701981cfc280aaab3c4436.json
Generated
+41
View File
@@ -0,0 +1,41 @@
{
"db_name": "PostgreSQL",
"query": "SELECT tag, dedicated_worker, on_behalf_of_email, edited_by from flow WHERE path = $1 and workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "edited_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true,
true,
true,
false
]
},
"hash": "f44f1a794ab217be05a9feb21e57b3409d9166a028701981cfc280aaab3c4436"
}
backend/.sqlx/query-fea19da213230401161cbf2ee1de521040caa6f6cc745afb7de66f4442e76cba.json
Generated
+20
View File
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "SELECT current_setting('session.groups')",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "current_setting",
"type_info": "Text"
}
],
"parameters": {
"Left": []
},
"nullable": [
null
]
},
"hash": "fea19da213230401161cbf2ee1de521040caa6f6cc745afb7de66f4442e76cba"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
9cea7e44f1b6401662e22088b41e8a00d7e67b23
a515264a3c1c138e289da1ef3bb21d260aead486
backend/migrations/20250113154322_runnable_on_behalf_of.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
alter table script drop column on_behalf_of_email;
alter table flow drop column on_behalf_of_email;
backend/migrations/20250113154322_runnable_on_behalf_of.up.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add up migration script here
ALTER TABLE script ADD COLUMN on_behalf_of_email TEXT;
ALTER TABLE flow ADD COLUMN on_behalf_of_email TEXT;
backend/tests/worker.rs
+2
View File
@@ -2874,6 +2874,7 @@ async fn test_flow_lock_all(db: Pool<Postgres>) {
dedicated_worker: None,
timeout: None,
visible_to_runner_only: None,
on_behalf_of_email: None,
},
draft_only: None,
deployment_message: None,
@@ -3473,6 +3474,7 @@ async fn run_deployed_relative_imports(
no_main_func: None,
codebase: None,
has_preprocessor: None,
on_behalf_of_email: None,
},
)
.await
backend/windmill-api/openapi.yaml
+8
View File
@@ -11043,6 +11043,8 @@ components:
type: string
has_preprocessor:
type: boolean
on_behalf_of_email:
type: string
required:
- hash
@@ -11144,6 +11146,8 @@ components:
type: string
has_preprocessor:
type: boolean
on_behalf_of_email:
type: string
required:
- path
- summary
@@ -13193,6 +13197,8 @@ components:
type: number
visible_to_runner_only:
type: boolean
on_behalf_of_email:
type: string
required:
- path
- edited_by
@@ -13219,6 +13225,8 @@ components:
type: number
visible_to_runner_only:
type: boolean
on_behalf_of_email:
type: string
required:
- path
backend/windmill-api/src/apps.rs
+13 -3
View File
@@ -1436,20 +1436,30 @@ async fn execute_component(
)
.await?;
let (job_payload, tag) = match (payload.path, payload.raw_code, payload.id) {
let (job_payload, tag, on_behalf_of) = match (payload.path, payload.raw_code, payload.id) {
// flow or script:
(Some(path), None, None) => get_payload_tag_from_prefixed_path(&path, &db, &w_id).await?,
// inline script: in "preview" mode or without entry in the `app_script` table.
(None, Some(raw_code), None) => (JobPayload::Code(raw_code), None),
(None, Some(raw_code), None) => (JobPayload::Code(raw_code), None, None),
// inline script: in "run" mode and with an entry in the `app_script` table.
(None, Some(RawCode { language, path, cache_ttl, .. }), Some(id)) => (
JobPayload::AppScript { id: AppScriptId(id), cache_ttl, language, path },
None,
None,
),
_ => unreachable!(),
};
let tx = PushIsolationLevel::IsolatedRoot(db.clone());
let (email, permissioned_as) = if let Some(on_behalf_of) = on_behalf_of.as_ref() {
(
on_behalf_of.email.as_str(),
on_behalf_of.permissioned_as.clone(),
)
} else {
(email.as_str(), permissioned_as)
};
let (uuid, tx) = push(
&db,
tx,
@@ -1457,7 +1467,7 @@ async fn execute_component(
job_payload,
PushArgs { args: &args.args, extra: args.extra },
&username,
&email,
email,
permissioned_as,
None,
None,
backend/windmill-api/src/flows.rs
+24 -12
View File
@@ -356,8 +356,8 @@ async fn create_flow(
sqlx::query!(
"INSERT INTO flow (workspace_id, path, summary, description, \
dependency_job, draft_only, tag, dedicated_worker, visible_to_runner_only, value, schema, edited_by, edited_at)
VALUES ($1, $2, $3, $4, NULL, $5, $6, $7, $8, $9, $10::text::json, $11, now())",
dependency_job, draft_only, tag, dedicated_worker, visible_to_runner_only, on_behalf_of_email, value, schema, edited_by, edited_at)
VALUES ($1, $2, $3, $4, NULL, $5, $6, $7, $8, $9, $10, $11::text::json, $12, now())",
w_id,
nf.path,
nf.summary,
@@ -366,6 +366,11 @@ async fn create_flow(
nf.tag,
nf.dedicated_worker,
nf.visible_to_runner_only.unwrap_or(false),
if nf.on_behalf_of_email.is_some() {
Some(&authed.email)
} else {
None
},
nf.value,
schema_str,
&authed.username,
@@ -571,7 +576,7 @@ async fn get_flow_version(
let mut tx = user_db.begin(&authed).await?;
let flow = sqlx::query_as::<_, Flow>(
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by
FROM flow
LEFT JOIN flow_version ON flow_version.path = flow.path AND flow_version.workspace_id = flow.workspace_id
WHERE flow.path = $1 AND flow.workspace_id = $2 AND flow_version.id = $3",
@@ -678,15 +683,20 @@ async fn update_flow(
sqlx::query!(
"UPDATE flow SET path = $1, summary = $2, description = $3,\
dependency_job = NULL, draft_only = NULL, tag = $4, dedicated_worker = $5, visible_to_runner_only = $6, \
value = $7, schema = $8::text::json, edited_by = $9, edited_at = now()
WHERE path = $10 AND workspace_id = $11",
dependency_job = NULL, draft_only = NULL, tag = $4, dedicated_worker = $5, visible_to_runner_only = $6, on_behalf_of_email = $7, \
value = $8, schema = $9::text::json, edited_by = $10, edited_at = now()
WHERE path = $11 AND workspace_id = $12",
if is_new_path { flow_path } else { &nf.path }, // if new path, do not rename directly (to avoid flow_version foreign key constraint)
nf.summary,
nf.description.unwrap_or_else(String::new),
nf.tag,
nf.dedicated_worker,
nf.visible_to_runner_only.unwrap_or(false),
if nf.on_behalf_of_email.is_some() {
Some(&authed.email)
} else {
None
},
nf.value,
schema_str,
&authed.username,
@@ -700,8 +710,8 @@ async fn update_flow(
// if new path, must clone flow to new path and delete old flow for flow_version foreign key constraint
sqlx::query!(
"INSERT INTO flow
(workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at)
SELECT workspace_id, $1, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at
(workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at)
SELECT workspace_id, $1, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at
FROM flow
WHERE path = $2 AND workspace_id = $3",
nf.path,
@@ -749,7 +759,7 @@ async fn update_flow(
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE capture SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS TRUE",
nf.path,
@@ -947,7 +957,7 @@ async fn get_flow_by_path(
let flow_o = if query.with_starred_info.unwrap_or(false) {
sqlx::query_as::<_, FlowWithStarred>(
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by, favorite.path IS NOT NULL as starred
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by, favorite.path IS NOT NULL as starred
FROM flow
LEFT JOIN favorite
ON favorite.favorite_kind = 'flow'
@@ -964,7 +974,7 @@ async fn get_flow_by_path(
.await?
} else {
sqlx::query_as::<_, FlowWithStarred>(
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by, NULL as starred
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by, NULL as starred
FROM flow
LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
WHERE flow.path = $1 AND flow.workspace_id = $2"
@@ -1000,6 +1010,8 @@ pub struct FlowWDraft {
pub dedicated_worker: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub visible_to_runner_only: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_behalf_of_email: Option<String>,
}
async fn get_flow_by_path_w_draft(
@@ -1011,7 +1023,7 @@ async fn get_flow_by_path_w_draft(
let mut tx = user_db.begin(&authed).await?;
let flow_o = sqlx::query_as::<_, FlowWDraft>(
"SELECT flow.path, flow.summary, flow,description, flow_version.schema, flow_version.value, flow.extra_perms, flow.draft_only, flow.ws_error_handler_muted, flow.dedicated_worker, draft.value as draft, flow.tag, flow.visible_to_runner_only
"SELECT flow.path, flow.summary, flow,description, flow_version.schema, flow_version.value, flow.extra_perms, flow.draft_only, flow.ws_error_handler_muted, flow.dedicated_worker, draft.value as draft, flow.tag, flow.visible_to_runner_only, flow.on_behalf_of_email
FROM flow
LEFT JOIN draft
ON flow.path = draft.path AND draft.workspace_id = $2 AND draft.typ = 'flow'
backend/windmill-api/src/jobs.rs
+177 -44
View File
@@ -561,9 +561,11 @@ pub async fn get_path_tag_limits_cache_for_hash(
Option<bool>,
Option<i32>,
Option<bool>,
Option<String>,
String,
)> {
let script = sqlx::query!(
"select path, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor from script where hash = $1 AND workspace_id = $2",
"select path, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by from script where hash = $1 AND workspace_id = $2",
hash,
w_id
)
@@ -589,6 +591,8 @@ pub async fn get_path_tag_limits_cache_for_hash(
script.delete_after_use,
script.timeout,
script.has_preprocessor,
script.on_behalf_of_email,
script.created_by,
))
}
@@ -3038,8 +3042,8 @@ pub async fn run_flow_by_path_inner(
let flow_path = flow_path.to_path();
check_scopes(&authed, || format!("run:flow/{flow_path}"))?;
let (tag, dedicated_worker, has_preprocessor) = sqlx::query!(
"SELECT tag, dedicated_worker, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor
let (tag, dedicated_worker, has_preprocessor, on_behalf_of_email, edited_by) = sqlx::query!(
"SELECT tag, dedicated_worker, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, on_behalf_of_email, edited_by
FROM flow
LEFT JOIN flow_version
ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
@@ -3049,7 +3053,7 @@ pub async fn run_flow_by_path_inner(
)
.fetch_optional(&db)
.await?
.map(|x| (x.tag, x.dedicated_worker, x.has_preprocessor))
.map(|x| (x.tag, x.dedicated_worker, x.has_preprocessor, x.on_behalf_of_email, x.edited_by))
.ok_or_else(|| {
Error::NotFound(format!(
"flow not found at path {flow_path} in workspace {w_id}"
@@ -3060,7 +3064,24 @@ pub async fn run_flow_by_path_inner(
check_tag_available_for_workspace(&w_id, &tag, &authed).await?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());
let (email, permissioned_as, push_authed, tx) =
if let Some(on_behalf_of_email) = on_behalf_of_email.as_ref() {
(
on_behalf_of_email,
username_to_permissioned_as(&edited_by),
None,
PushIsolationLevel::IsolatedRoot(db.clone()),
)
} else {
(
&authed.email,
username_to_permissioned_as(&authed.username),
Some(authed.clone().into()),
PushIsolationLevel::Isolated(user_db, authed.clone().into()),
)
};
let (uuid, tx) = push(
&db,
tx,
@@ -3075,8 +3096,8 @@ pub async fn run_flow_by_path_inner(
&label_prefix
.map(|x| x + authed.display_username())
.unwrap_or_else(|| authed.display_username().to_string()),
&authed.email,
username_to_permissioned_as(&authed.username),
email,
permissioned_as,
scheduled_for,
None,
run_query.parent_job,
@@ -3090,7 +3111,7 @@ pub async fn run_flow_by_path_inner(
None,
None,
None,
Some(&authed.clone().into()),
push_authed.as_ref(),
)
.await?;
tx.commit().await?;
@@ -3227,14 +3248,29 @@ pub async fn run_script_by_path_inner(
check_scopes(&authed, || format!("run:script/{script_path}"))?;
let (job_payload, tag, _delete_after_use, timeout) =
let (job_payload, tag, _delete_after_use, timeout, on_behalf_of) =
script_path_to_payload(script_path, &db, &w_id, run_query.skip_preprocessor).await?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag, &authed).await?;
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());
let (email, permissioned_as, push_authed, tx) =
if let Some(on_behalf_of) = on_behalf_of.as_ref() {
(
on_behalf_of.email.as_str(),
on_behalf_of.permissioned_as.clone(),
None,
PushIsolationLevel::IsolatedRoot(db.clone()),
)
} else {
(
authed.email.as_str(),
username_to_permissioned_as(&authed.username),
Some(authed.clone().into()),
PushIsolationLevel::Isolated(user_db, authed.clone().into()),
)
};
let (uuid, tx) = push(
&db,
@@ -3245,8 +3281,8 @@ pub async fn run_script_by_path_inner(
&label_prefix
.map(|x| x + authed.display_username())
.unwrap_or_else(|| authed.display_username().to_string()),
&authed.email,
username_to_permissioned_as(&authed.username),
email,
permissioned_as,
scheduled_for,
None,
run_query.parent_job,
@@ -3260,7 +3296,7 @@ pub async fn run_script_by_path_inner(
timeout,
None,
None,
Some(&authed.clone().into()),
push_authed.as_ref(),
)
.await?;
tx.commit().await?;
@@ -3309,7 +3345,7 @@ pub async fn run_workflow_as_code(
let job = not_found_if_none(job, "Queued Job", &job_id.to_string())?;
let JobExtended { inner: job, raw_code, raw_lock, .. } = job;
let (job_payload, tag, _delete_after_use, timeout) = match job.job_kind {
let (job_payload, tag, _delete_after_use, timeout, on_behalf_of) = match job.job_kind {
JobKind::Preview => (
JobPayload::Code(RawCode {
hash: None,
@@ -3328,6 +3364,7 @@ pub async fn run_workflow_as_code(
Some(job.tag.clone()),
None,
run_query.timeout,
None,
),
JobKind::Script => {
script_path_to_payload(job.script_path(), &db, &w_id, run_query.skip_preprocessor)
@@ -3354,13 +3391,28 @@ pub async fn run_workflow_as_code(
i += 1;
}
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
i += 1;
}
let (email, permissioned_as, push_authed, tx) =
if let Some(on_behalf_of) = on_behalf_of.as_ref() {
(
on_behalf_of.email.as_str(),
on_behalf_of.permissioned_as.clone(),
None,
PushIsolationLevel::IsolatedRoot(db.clone()),
)
} else {
(
authed.email.as_str(),
username_to_permissioned_as(&authed.username),
Some(authed.clone().into()),
PushIsolationLevel::Isolated(user_db, authed.clone().into()),
)
};
let (uuid, mut tx) = push(
&db,
tx,
@@ -3368,8 +3420,8 @@ pub async fn run_workflow_as_code(
job_payload,
PushArgs { args: &args.args, extra: args.extra },
authed.display_username(),
&authed.email,
username_to_permissioned_as(&authed.username),
email,
permissioned_as,
scheduled_for,
None,
Some(job_id),
@@ -3383,7 +3435,7 @@ pub async fn run_workflow_as_code(
timeout,
None,
None,
Some(&authed.clone().into()),
push_authed.as_ref(),
)
.await?;
@@ -3800,13 +3852,28 @@ pub async fn run_wait_result_job_by_path_get(
let script_path = script_path.to_path();
check_scopes(&authed, || format!("run:script/{script_path}"))?;
let (job_payload, tag, delete_after_use, timeout) =
let (job_payload, tag, delete_after_use, timeout, on_behalf_authed) =
script_path_to_payload(script_path, &db, &w_id, run_query.skip_preprocessor).await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag, &authed).await?;
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());
let (email, permissioned_as, push_authed, tx) =
if let Some(on_behalf_of) = on_behalf_authed.as_ref() {
(
on_behalf_of.email.as_str(),
on_behalf_of.permissioned_as.clone(),
None,
PushIsolationLevel::IsolatedRoot(db.clone()),
)
} else {
(
authed.email.as_str(),
username_to_permissioned_as(&authed.username),
Some(authed.clone().into()),
PushIsolationLevel::Isolated(user_db, authed.clone().into()),
)
};
let (uuid, tx) = push(
&db,
@@ -3815,8 +3882,8 @@ pub async fn run_wait_result_job_by_path_get(
job_payload,
PushArgs { args: &args.args, extra: args.extra },
authed.display_username(),
&authed.email,
username_to_permissioned_as(&authed.username),
email,
permissioned_as,
None,
None,
run_query.parent_job,
@@ -3830,7 +3897,7 @@ pub async fn run_wait_result_job_by_path_get(
timeout,
None,
None,
Some(&authed.clone().into()),
push_authed.as_ref(),
)
.await?;
tx.commit().await?;
@@ -3921,13 +3988,28 @@ pub async fn run_wait_result_script_by_path_internal(
let script_path = script_path.to_path();
check_scopes(&authed, || format!("run:script/{script_path}"))?;
let (job_payload, tag, delete_after_use, timeout) =
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) =
script_path_to_payload(script_path, &db, &w_id, run_query.skip_preprocessor).await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag, &authed).await?;
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());
let (email, permissioned_as, push_authed, tx) =
if let Some(on_behalf_of) = on_behalf_of.as_ref() {
(
on_behalf_of.email.as_str(),
on_behalf_of.permissioned_as.clone(),
None,
PushIsolationLevel::IsolatedRoot(db.clone()),
)
} else {
(
authed.email.as_str(),
username_to_permissioned_as(&authed.username),
Some(authed.clone().into()),
PushIsolationLevel::Isolated(user_db, authed.clone().into()),
)
};
let (uuid, tx) = push(
&db,
@@ -3938,8 +4020,8 @@ pub async fn run_wait_result_script_by_path_internal(
&label_prefix
.map(|x| x + authed.display_username())
.unwrap_or_else(|| authed.display_username().to_string()),
&authed.email,
username_to_permissioned_as(&authed.username),
email,
permissioned_as,
None,
None,
run_query.parent_job,
@@ -3953,7 +4035,7 @@ pub async fn run_wait_result_script_by_path_internal(
timeout,
None,
None,
Some(&authed.clone().into()),
push_authed.as_ref(),
)
.await?;
tx.commit().await?;
@@ -3994,6 +4076,8 @@ pub async fn run_wait_result_script_by_hash(
delete_after_use,
timeout,
has_preprocessor,
on_behalf_of_email,
created_by,
) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
if let Some(run_query_cache_ttl) = run_query.cache_ttl {
cache_ttl = Some(run_query_cache_ttl);
@@ -4003,7 +4087,22 @@ pub async fn run_wait_result_script_by_hash(
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag, &authed).await?;
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());
let (email, permissioned_as, push_authed, tx) = if let Some(email) = on_behalf_of_email.as_ref()
{
(
email,
username_to_permissioned_as(created_by.as_str()),
None,
PushIsolationLevel::IsolatedRoot(db.clone()),
)
} else {
(
&authed.email,
username_to_permissioned_as(&authed.username),
Some(authed.clone().into()),
PushIsolationLevel::Isolated(user_db, authed.clone().into()),
)
};
let (uuid, tx) = push(
&db,
@@ -4024,8 +4123,8 @@ pub async fn run_wait_result_script_by_hash(
},
PushArgs { args: &args.args, extra: args.extra },
authed.display_username(),
&authed.email,
username_to_permissioned_as(&authed.username),
email,
permissioned_as,
None,
None,
run_query.parent_job,
@@ -4039,7 +4138,7 @@ pub async fn run_wait_result_script_by_hash(
timeout,
None,
None,
Some(&authed.clone().into()),
push_authed.as_ref(),
)
.await?;
tx.commit().await?;
@@ -4087,8 +4186,8 @@ pub async fn run_wait_result_flow_by_path_internal(
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let (tag, dedicated_worker, early_return, has_preprocessor) = sqlx::query!(
"SELECT tag, dedicated_worker, flow_version.value->>'early_return' as early_return, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor
let (tag, dedicated_worker, early_return, has_preprocessor, on_behalf_of_email, edited_by) = sqlx::query!(
"SELECT tag, dedicated_worker, flow_version.value->>'early_return' as early_return, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, on_behalf_of_email, edited_by
FROM flow
LEFT JOIN flow_version
ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
@@ -4098,7 +4197,7 @@ pub async fn run_wait_result_flow_by_path_internal(
)
.fetch_optional(&db)
.await?
.map(|x| (x.tag, x.dedicated_worker, x.early_return, x.has_preprocessor))
.map(|x| (x.tag, x.dedicated_worker, x.early_return, x.has_preprocessor, x.on_behalf_of_email, x.edited_by))
.ok_or_else(|| {
Error::NotFound(format!(
"flow not found at path {flow_path} in workspace {w_id}"
@@ -4108,7 +4207,22 @@ pub async fn run_wait_result_flow_by_path_internal(
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag, &authed).await?;
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());
let (email, permissioned_as, push_authed, tx) =
if let Some(on_behalf_of_email) = on_behalf_of_email.as_ref() {
(
on_behalf_of_email,
username_to_permissioned_as(&edited_by),
None,
PushIsolationLevel::IsolatedRoot(db.clone()),
)
} else {
(
&authed.email,
username_to_permissioned_as(&authed.username),
Some(authed.clone().into()),
PushIsolationLevel::Isolated(user_db, authed.clone().into()),
)
};
let (uuid, tx) = push(
&db,
@@ -4124,8 +4238,8 @@ pub async fn run_wait_result_flow_by_path_internal(
&label_prefix
.map(|x| x + authed.display_username())
.unwrap_or_else(|| authed.display_username().to_string()),
&authed.email,
username_to_permissioned_as(&authed.username),
email,
permissioned_as,
scheduled_for,
None,
run_query.parent_job,
@@ -4139,7 +4253,7 @@ pub async fn run_wait_result_flow_by_path_internal(
None,
None,
None,
Some(&authed.clone().into()),
push_authed.as_ref(),
)
.await?;
tx.commit().await?;
@@ -4567,6 +4681,8 @@ async fn add_batch_jobs(
_delete_after_use,
timeout,
_,
_, // TODO: consider on_behalf_of_email and created_by for batch jobs
_, // ------------------------------------------
) = get_latest_deployed_hash_for_path(&db, &w_id, &path).await?;
(
Some(script_hash),
@@ -4852,6 +4968,8 @@ pub async fn run_job_by_hash_inner(
_delete_after_use, // not taken into account in async endpoints
timeout,
has_preprocessor,
on_behalf_of_email,
created_by,
) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
check_scopes(&authed, || format!("run:script/{path}"))?;
if let Some(run_query_cache_ttl) = run_query.cache_ttl {
@@ -4862,7 +4980,22 @@ pub async fn run_job_by_hash_inner(
check_tag_available_for_workspace(&w_id, &tag, &authed).await?;
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());
let (email, permissioned_as, push_authed, tx) = if let Some(email) = on_behalf_of_email.as_ref()
{
(
email,
username_to_permissioned_as(created_by.as_str()),
None,
PushIsolationLevel::IsolatedRoot(db.clone()),
)
} else {
(
&authed.email,
username_to_permissioned_as(&authed.username),
Some(authed.clone().into()),
PushIsolationLevel::Isolated(user_db, authed.clone().into()),
)
};
let (uuid, tx) = push(
&db,
@@ -4885,8 +5018,8 @@ pub async fn run_job_by_hash_inner(
&label_prefix
.map(|x| x + authed.display_username())
.unwrap_or_else(|| authed.display_username().to_string()),
&authed.email,
username_to_permissioned_as(&authed.username),
email,
permissioned_as,
scheduled_for,
None,
run_query.parent_job,
@@ -4900,7 +5033,7 @@ pub async fn run_job_by_hash_inner(
timeout,
None,
None,
Some(&authed.clone().into()),
push_authed.as_ref(),
)
.await?;
tx.commit().await?;
backend/windmill-api/src/scripts.rs
+12 -5
View File
@@ -7,13 +7,13 @@
*/
use crate::{
auth::AuthCache,
db::{ApiAuthed, DB},
schedule::clear_schedule,
triggers::{
get_triggers_count_internal, list_tokens_internal, TriggersCount, TruncatedTokenWithEmail,
},
users::{maybe_refresh_folders, require_owner_of_path},
auth::AuthCache,
utils::WithStarredInfoQuery,
webhook_util::{WebhookMessage, WebhookShared},
HTTP_CLIENT,
@@ -109,6 +109,8 @@ pub struct ScriptWDraft {
pub no_main_func: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub has_preprocessor: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_behalf_of_email: Option<String>,
}
pub fn global_service() -> Router {
@@ -616,8 +618,8 @@ async fn create_script_internal<'c>(
content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \
dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \
delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31)",
delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32)",
&w_id,
&hash.0,
ns.path,
@@ -649,6 +651,11 @@ async fn create_script_internal<'c>(
ns.no_main_func,
codebase,
ns.has_preprocessor,
if ns.on_behalf_of_email.is_some() {
Some(&authed.email)
} else {
None
}
)
.execute(&mut *tx)
.await?;
@@ -670,7 +677,7 @@ async fn create_script_internal<'c>(
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE capture SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS FALSE",
ns.path,
@@ -917,7 +924,7 @@ async fn get_script_by_path_w_draft(
let mut tx = user_db.begin(&authed).await?;
let script_o = sqlx::query_as::<_, ScriptWDraft>(
"SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, has_preprocessor FROM script LEFT JOIN draft ON
"SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, has_preprocessor, on_behalf_of_email FROM script LEFT JOIN draft ON
script.path = draft.path AND script.workspace_id = draft.workspace_id AND draft.typ = 'script'
WHERE script.path = $1 AND script.workspace_id = $2 \
AND script.created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND \
backend/windmill-api/src/users.rs
+2 -2
View File
@@ -2488,8 +2488,8 @@ async fn update_username_in_workpsace<'c>(
// ---- flows ----
sqlx::query!(
r#"INSERT INTO flow
(workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at)
SELECT workspace_id, REGEXP_REPLACE(path,'u/' || $2 || '/(.*)','u/' || $1 || '/\1'), summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at
(workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at)
SELECT workspace_id, REGEXP_REPLACE(path,'u/' || $2 || '/(.*)','u/' || $1 || '/\1'), summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at
FROM flow
WHERE path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#,
new_username,
backend/windmill-api/src/workspaces_export.rs
+4 -1
View File
@@ -83,6 +83,8 @@ struct ScriptMetadata {
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub has_preprocessor: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_behalf_of_email: Option<String>,
}
pub fn is_none_or_false(val: &Option<bool>) -> bool {
@@ -379,6 +381,7 @@ pub(crate) async fn tarball_workspace(
codebase: script.codebase,
concurrency_key: script.concurrency_key,
has_preprocessor: script.has_preprocessor,
on_behalf_of_email: script.on_behalf_of_email,
};
let metadata_str = serde_json::to_string_pretty(&metadata).unwrap();
archive
@@ -426,7 +429,7 @@ pub(crate) async fn tarball_workspace(
{
let flows = sqlx::query_as::<_, Flow>(
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by
FROM flow
LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
WHERE flow.workspace_id = $1 AND flow.archived = false",
backend/windmill-api/src/workspaces_extra.rs
+2 -2
View File
@@ -153,8 +153,8 @@ pub(crate) async fn change_workspace_id(
sqlx::query!(
"INSERT INTO flow
(workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at)
SELECT $1, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at
(workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at)
SELECT $1, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at
FROM flow WHERE workspace_id = $2",
&rw.new_id,
&old_id
backend/windmill-common/src/auth.rs
+1 -1
View File
@@ -75,7 +75,7 @@ pub async fn is_super_admin_email(db: &DB, email: &str) -> Result<bool> {
pub async fn is_devops_email(db: &DB, email: &str) -> Result<bool> {
if is_super_admin_email(db, email).await? {
return Ok(true)
return Ok(true);
}
let is_devops = sqlx::query_scalar!("SELECT devops FROM password WHERE email = $1", email)
backend/windmill-common/src/flows.rs
+3
View File
@@ -49,6 +49,8 @@ pub struct Flow {
pub timeout: Option<i32>,
#[serde(skip_serializing_if = "is_none_or_false")]
pub visible_to_runner_only: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_behalf_of_email: Option<String>,
}
#[derive(Serialize, sqlx::FromRow)]
@@ -98,6 +100,7 @@ pub struct NewFlow {
pub timeout: Option<i32>,
pub deployment_message: Option<String>,
pub visible_to_runner_only: Option<bool>,
pub on_behalf_of_email: Option<String>,
}
#[derive(Deserialize, Serialize, Debug, Clone, Default)]
backend/windmill-common/src/jobs.rs
+84 -45
View File
@@ -20,6 +20,7 @@ use crate::{
flows::{FlowNodeId, FlowValue, Retry},
get_latest_deployed_hash_for_path,
scripts::{ScriptHash, ScriptLang},
users::username_to_permissioned_as,
worker::{to_raw_value, TMP_DIR},
};
@@ -370,54 +371,87 @@ type Tag = String;
pub type DB = Pool<Postgres>;
pub async fn script_path_to_payload<'e, E: sqlx::Executor<'e, Database = Postgres>>(
#[derive(Clone, Debug)]
pub struct OnBehalfOf {
pub email: String,
pub permissioned_as: String,
}
pub async fn script_path_to_payload(
script_path: &str,
db: E,
db: &DB,
w_id: &str,
skip_preprocessor: Option<bool>,
) -> error::Result<(JobPayload, Option<Tag>, Option<bool>, Option<i32>)> {
let (job_payload, tag, delete_after_use, script_timeout) = if script_path.starts_with("hub/") {
(
JobPayload::ScriptHub { path: script_path.to_owned() },
None,
None,
None,
)
} else {
let (
script_hash,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
language,
dedicated_worker,
priority,
delete_after_use,
script_timeout,
has_preprocessor,
) = get_latest_deployed_hash_for_path(db, w_id, script_path).await?;
(
JobPayload::ScriptHash {
hash: script_hash,
path: script_path.to_owned(),
) -> error::Result<(
JobPayload,
Option<Tag>,
Option<bool>,
Option<i32>,
Option<OnBehalfOf>,
)> {
let (job_payload, tag, delete_after_use, script_timeout, on_behalf_of) =
if script_path.starts_with("hub/") {
(
JobPayload::ScriptHub { path: script_path.to_owned() },
None,
None,
None,
None,
)
} else {
let (
script_hash,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl: cache_ttl,
cache_ttl,
language,
dedicated_worker,
priority,
apply_preprocessor: !skip_preprocessor.unwrap_or(false)
&& has_preprocessor.unwrap_or(false),
},
tag,
delete_after_use,
script_timeout,
)
};
Ok((job_payload, tag, delete_after_use, script_timeout))
delete_after_use,
script_timeout,
has_preprocessor,
on_behalf_of_email,
created_by,
) = get_latest_deployed_hash_for_path(db, w_id, script_path).await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf {
email,
permissioned_as: username_to_permissioned_as(created_by.as_str()),
})
} else {
None
};
(
JobPayload::ScriptHash {
hash: script_hash,
path: script_path.to_owned(),
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl: cache_ttl,
language,
dedicated_worker,
priority,
apply_preprocessor: !skip_preprocessor.unwrap_or(false)
&& has_preprocessor.unwrap_or(false),
},
tag,
delete_after_use,
script_timeout,
on_behalf_of,
)
};
Ok((
job_payload,
tag,
delete_after_use,
script_timeout,
on_behalf_of,
))
}
pub async fn script_hash_to_tag_and_limits<'c>(
@@ -435,9 +469,11 @@ pub async fn script_hash_to_tag_and_limits<'c>(
Option<i16>,
Option<bool>,
Option<i32>,
Option<String>,
String,
)> {
let script = sqlx::query!(
"select tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
"select tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, on_behalf_of_email, created_by from script where hash = $1 AND workspace_id = $2",
script_hash.0,
w_id
)
@@ -459,15 +495,17 @@ pub async fn script_hash_to_tag_and_limits<'c>(
script.priority,
script.delete_after_use,
script.timeout,
script.on_behalf_of_email,
script.created_by,
))
}
pub async fn get_payload_tag_from_prefixed_path<'e, E: sqlx::Executor<'e, Database = Postgres>>(
pub async fn get_payload_tag_from_prefixed_path(
path: &str,
db: E,
db: &DB,
w_id: &str,
) -> Result<(JobPayload, Option<String>), Error> {
let (payload, tag, _, _) = if path.starts_with("script/") {
) -> Result<(JobPayload, Option<String>, Option<OnBehalfOf>), Error> {
let (payload, tag, _, _, on_behalf_of) = if path.starts_with("script/") {
script_path_to_payload(path.strip_prefix("script/").unwrap(), db, w_id, Some(true)).await?
} else if path.starts_with("flow/") {
let path = path.strip_prefix("flow/").unwrap().to_string();
@@ -486,6 +524,7 @@ pub async fn get_payload_tag_from_prefixed_path<'e, E: sqlx::Executor<'e, Databa
tag,
None,
None,
None,
)
} else {
return Err(Error::BadRequest(format!(
@@ -493,7 +532,7 @@ pub async fn get_payload_tag_from_prefixed_path<'e, E: sqlx::Executor<'e, Databa
path
)));
};
Ok((payload, tag))
Ok((payload, tag, on_behalf_of))
}
#[derive(Deserialize)]
backend/windmill-common/src/lib.rs
+10 -2
View File
@@ -302,9 +302,11 @@ pub async fn get_latest_deployed_hash_for_path<'e, E: sqlx::Executor<'e, Databas
Option<bool>,
Option<i32>,
Option<bool>,
Option<String>,
String,
)> {
let r_o = sqlx::query!(
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor from script where path = $1 AND workspace_id = $2 AND
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by from script where path = $1 AND workspace_id = $2 AND
created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND
deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)",
script_path,
@@ -328,6 +330,8 @@ pub async fn get_latest_deployed_hash_for_path<'e, E: sqlx::Executor<'e, Databas
script.delete_after_use,
script.timeout,
script.has_preprocessor,
script.on_behalf_of_email,
script.created_by,
))
}
@@ -346,9 +350,11 @@ pub async fn get_latest_hash_for_path<'c>(
Option<bool>,
Option<i16>,
Option<i32>,
Option<String>,
String,
)> {
let r_o = sqlx::query!(
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout FROM script where path = $1 AND workspace_id = $2 AND
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script where path = $1 AND workspace_id = $2 AND
created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND
deleted = false AND archived = false)",
script_path,
@@ -370,5 +376,7 @@ pub async fn get_latest_hash_for_path<'c>(
script.dedicated_worker,
script.priority,
script.timeout,
script.on_behalf_of_email,
script.created_by,
))
}
backend/windmill-common/src/scripts.rs
+3
View File
@@ -197,6 +197,8 @@ pub struct Script {
pub codebase: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub has_preprocessor: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_behalf_of_email: Option<String>,
}
#[derive(Serialize, sqlx::FromRow)]
@@ -294,6 +296,7 @@ pub struct NewScript {
pub no_main_func: Option<bool>,
pub codebase: Option<String>,
pub has_preprocessor: Option<bool>,
pub on_behalf_of_email: Option<String>,
}
fn lock_deserialize<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
backend/windmill-queue/src/jobs.rs
+54 -19
View File
@@ -1514,7 +1514,7 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>(
} else {
w_id
};
let (payload, tag) =
let (payload, tag, on_behalf_of) =
get_payload_tag_from_prefixed_path(on_failure_path, db, handler_w_id).await?;
let mut extra = HashMap::new();
@@ -1543,6 +1543,25 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>(
let result = sanitize_result(result);
let (email, permissioned_as) = if let Some(on_behalf_of) = on_behalf_of.as_ref() {
(
on_behalf_of.email.as_str(),
on_behalf_of.permissioned_as.clone(),
)
} else if is_global_error_handler {
(SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SECRET_EMAIL.to_string())
} else if is_schedule_error_handler {
(
SCHEDULE_ERROR_HANDLER_USER_EMAIL,
ERROR_HANDLER_USER_GROUP.to_string(),
)
} else {
(
ERROR_HANDLER_USER_EMAIL,
ERROR_HANDLER_USER_GROUP.to_string(),
)
};
let tx = PushIsolationLevel::IsolatedRoot(db.clone());
let (uuid, tx) = push(
&db,
@@ -1557,18 +1576,8 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>(
} else {
ERROR_HANDLER_USERNAME
},
if is_global_error_handler {
SUPERADMIN_SECRET_EMAIL
} else if is_schedule_error_handler {
SCHEDULE_ERROR_HANDLER_USER_EMAIL
} else {
ERROR_HANDLER_USER_EMAIL
},
if is_global_error_handler {
SUPERADMIN_SECRET_EMAIL.to_string()
} else {
ERROR_HANDLER_USER_GROUP.to_string()
},
email,
permissioned_as,
None,
None,
Some(job_id),
@@ -1619,7 +1628,8 @@ async fn handle_recovered_schedule<'a, 'c, T: Serialize + Send + Sync>(
successful_job_started_at: DateTime<Utc>,
extra_args: Option<Json<Box<RawValue>>>,
) -> windmill_common::error::Result<()> {
let (payload, tag) = get_payload_tag_from_prefixed_path(on_recovery_path, db, w_id).await?;
let (payload, tag, on_behalf_of) =
get_payload_tag_from_prefixed_path(on_recovery_path, db, w_id).await?;
let mut extra = HashMap::new();
extra.insert(
@@ -1656,6 +1666,18 @@ async fn handle_recovered_schedule<'a, 'c, T: Serialize + Send + Sync>(
.and_then(|x| serde_json::from_str::<HashMap<String, Box<RawValue>>>(x.0.get()).ok())
.unwrap_or_else(HashMap::new);
let (email, permissioned_as) = if let Some(on_behalf_of) = on_behalf_of.as_ref() {
(
on_behalf_of.email.as_str(),
on_behalf_of.permissioned_as.clone(),
)
} else {
(
SCHEDULE_RECOVERY_HANDLER_USER_EMAIL,
ERROR_HANDLER_USER_GROUP.to_string(),
)
};
let tx = PushIsolationLevel::Transaction(tx);
let (uuid, tx) = push(
&db,
@@ -1664,8 +1686,8 @@ async fn handle_recovered_schedule<'a, 'c, T: Serialize + Send + Sync>(
payload,
PushArgs { extra: Some(extra), args: &args },
SCHEDULE_RECOVERY_HANDLER_USERNAME,
SCHEDULE_RECOVERY_HANDLER_USER_EMAIL,
ERROR_HANDLER_USER_GROUP.to_string(),
email,
permissioned_as,
None,
None,
Some(job_id),
@@ -1704,7 +1726,8 @@ async fn handle_successful_schedule<'a, 'c, T: Serialize + Send + Sync>(
successful_job_started_at: DateTime<Utc>,
extra_args: Option<Json<Box<RawValue>>>,
) -> windmill_common::error::Result<()> {
let (payload, tag) = get_payload_tag_from_prefixed_path(on_success_path, db, w_id).await?;
let (payload, tag, on_behalf_of) =
get_payload_tag_from_prefixed_path(on_success_path, db, w_id).await?;
let mut extra = HashMap::new();
extra.insert("schedule_path".to_string(), to_raw_value(&schedule_path));
@@ -1731,6 +1754,18 @@ async fn handle_successful_schedule<'a, 'c, T: Serialize + Send + Sync>(
}
}
let (email, permissioned_as) = if let Some(on_behalf_of) = on_behalf_of.as_ref() {
(
on_behalf_of.email.as_str(),
on_behalf_of.permissioned_as.clone(),
)
} else {
(
SCHEDULE_RECOVERY_HANDLER_USER_EMAIL,
ERROR_HANDLER_USER_GROUP.to_string(),
)
};
let tx = PushIsolationLevel::IsolatedRoot(db.clone());
let (uuid, tx) = push(
&db,
@@ -1739,8 +1774,8 @@ async fn handle_successful_schedule<'a, 'c, T: Serialize + Send + Sync>(
payload,
PushArgs { extra: Some(extra), args: &HashMap::new() },
SCHEDULE_RECOVERY_HANDLER_USERNAME,
SCHEDULE_RECOVERY_HANDLER_USER_EMAIL,
ERROR_HANDLER_USER_GROUP.to_string(),
email,
permissioned_as,
None,
None,
Some(job_id),
backend/windmill-queue/src/schedule.rs
+50 -9
View File
@@ -103,17 +103,17 @@ pub async fn push_scheduled_job<'c>(
}
}
let (payload, tag, timeout) = if schedule.is_flow {
let (payload, tag, timeout, on_behalf_of_email, created_by) = if schedule.is_flow {
let r = sqlx::query!(
"SELECT tag, dedicated_worker from flow WHERE path = $1 and workspace_id = $2",
"SELECT tag, dedicated_worker, on_behalf_of_email, edited_by from flow WHERE path = $1 and workspace_id = $2",
&schedule.script_path,
&schedule.workspace_id,
)
.fetch_optional(&mut *tx)
.await?;
let (tag, dedicated_worker) = r
.map(|x| (x.tag, x.dedicated_worker))
.unwrap_or_else(|| (None, None));
let (tag, dedicated_worker, on_behalf_of_email, edited_by) = r
.map(|x| (x.tag, x.dedicated_worker, x.on_behalf_of_email, x.edited_by))
.unwrap_or_else(|| (None, None, None, "".to_string()));
(
JobPayload::Flow {
path: schedule.script_path.clone(),
@@ -122,6 +122,8 @@ pub async fn push_scheduled_job<'c>(
},
tag,
None,
on_behalf_of_email,
edited_by,
)
} else {
let (
@@ -135,6 +137,8 @@ pub async fn push_scheduled_job<'c>(
dedicated_worker,
priority,
timeout,
on_behalf_of_email,
created_by,
) = windmill_common::get_latest_hash_for_path(
&mut tx,
&schedule.workspace_id,
@@ -174,6 +178,8 @@ pub async fn push_scheduled_job<'c>(
tag
},
timeout,
on_behalf_of_email,
created_by,
)
} else {
(
@@ -195,6 +201,8 @@ pub async fn push_scheduled_job<'c>(
tag
},
timeout,
on_behalf_of_email,
created_by,
)
}
};
@@ -214,16 +222,43 @@ pub async fn push_scheduled_job<'c>(
);
};
let (email, permissioned_as, push_authed, revert_to_windmill_user) = if let Some(email) =
on_behalf_of_email.as_ref()
{
let is_windmill_user =
sqlx::query_scalar!("SELECT CURRENT_USER = 'windmill_user' as \"is_windmill_user!\"")
.fetch_one(&mut *tx)
.await?;
if is_windmill_user {
sqlx::query!("SET LOCAL ROLE NONE")
.execute(&mut *tx)
.await?;
}
(
email,
username_to_permissioned_as(&created_by),
None,
is_windmill_user,
)
} else {
(
&schedule.email,
username_to_permissioned_as(&schedule.edited_by),
authed,
false,
)
};
let tx = PushIsolationLevel::Transaction(tx);
let (_, tx) = push(
let (_, mut tx) = push(
&db,
tx,
&schedule.workspace_id,
payload,
crate::PushArgs { args: &args, extra: None },
&schedule_to_user(&schedule.path),
&schedule.email,
username_to_permissioned_as(&schedule.edited_by),
email,
permissioned_as,
Some(next),
Some(schedule.path.clone()),
None,
@@ -237,10 +272,16 @@ pub async fn push_scheduled_job<'c>(
timeout,
None,
None,
authed,
push_authed,
)
.await?;
if revert_to_windmill_user {
sqlx::query!("SET LOCAL ROLE windmill_user")
.execute(&mut *tx)
.await?;
}
Ok(tx) // TODO: Bubble up pushed UUID from here
}
backend/windmill-worker/src/worker_flow.rs
+59 -9
View File
@@ -40,9 +40,10 @@ use windmill_common::flow_status::{
use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId};
use windmill_common::jobs::{
script_hash_to_tag_and_limits, script_path_to_payload, BranchResults, JobKind, JobPayload,
QueuedJob, RawCode, ENTRYPOINT_OVERRIDE,
OnBehalfOf, QueuedJob, RawCode, ENTRYPOINT_OVERRIDE,
};
use windmill_common::scripts::ScriptHash;
use windmill_common::users::username_to_permissioned_as;
use windmill_common::utils::WarnAfterExt;
use windmill_common::worker::to_raw_value;
use windmill_common::{
@@ -2547,6 +2548,13 @@ async fn push_next_flow_job(
} else {
Some(flow_job.tag.clone())
};
let (email, permissioned_as) = if let Some(on_behalf_of) = payload_tag.on_behalf_of.as_ref()
{
(&on_behalf_of.email, on_behalf_of.permissioned_as.clone())
} else {
(&flow_job.email, flow_job.permissioned_as.to_owned())
};
let tx2 = PushIsolationLevel::Transaction(tx);
let (uuid, mut inner_tx) = push(
&db,
@@ -2555,8 +2563,8 @@ async fn push_next_flow_job(
payload_tag.payload.clone(),
push_args,
&flow_job.created_by,
&flow_job.email,
flow_job.permissioned_as.to_owned(),
email,
permissioned_as,
scheduled_for_o,
flow_job.schedule_path.clone(),
Some(flow_job.id),
@@ -2913,6 +2921,7 @@ struct JobPayloadWithTag {
tag: Option<String>,
delete_after_use: bool,
timeout: Option<i32>,
on_behalf_of: Option<OnBehalfOf>,
}
enum ContinuePayload {
SingleJob(JobPayloadWithTag),
@@ -3010,6 +3019,7 @@ async fn compute_next_flow_transform(
tag: None,
delete_after_use: false,
timeout: None,
on_behalf_of: None,
}),
NextStatus::NextStep,
));
@@ -3021,6 +3031,7 @@ async fn compute_next_flow_transform(
tag: None,
delete_after_use: false,
timeout: None,
on_behalf_of: None,
}),
NextStatus::NextStep,
))
@@ -3035,7 +3046,8 @@ async fn compute_next_flow_transform(
match module.get_value()? {
FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity),
FlowModuleValue::Flow { path, .. } => {
let payload = flow_to_payload(path, delete_after_use);
let payload =
flow_to_payload(path, delete_after_use, &flow_job.workspace_id, db).await?;
Ok(NextFlowTransform::Continue(
ContinuePayload::SingleJob(payload),
NextStatus::NextStep,
@@ -3105,6 +3117,7 @@ async fn compute_next_flow_transform(
tag: tag.clone(),
delete_after_use,
timeout: module.timeout,
on_behalf_of: None,
};
Ok(NextFlowTransform::Continue(
ContinuePayload::SingleJob(payload),
@@ -3229,6 +3242,7 @@ async fn compute_next_flow_transform(
tag: None,
delete_after_use,
timeout: None,
on_behalf_of: None,
})
})
.collect::<Vec<_>>();
@@ -3323,6 +3337,7 @@ async fn compute_next_flow_transform(
tag: None,
delete_after_use,
timeout: None,
on_behalf_of: None,
}),
NextStatus::BranchChosen(branch),
))
@@ -3356,6 +3371,7 @@ async fn compute_next_flow_transform(
tag: None,
delete_after_use,
timeout: None,
on_behalf_of: None,
})
})
.collect::<Vec<_>>();
@@ -3427,6 +3443,7 @@ async fn compute_next_flow_transform(
tag: None,
delete_after_use,
timeout: None,
on_behalf_of: None,
}),
NextStatus::NextBranchStep(NextBranch {
status: branch_status,
@@ -3488,6 +3505,7 @@ async fn next_loop_iteration(
tag: None,
delete_after_use,
timeout: None,
on_behalf_of: None,
}),
NextStatus::NextLoopIteration { next: ns, simple_input_transforms: None },
))
@@ -3659,7 +3677,9 @@ async fn payload_from_simple_module(
) -> Result<JobPayloadWithTag, Error> {
let delete_after_use = module.delete_after_use.unwrap_or(false);
Ok(match value {
FlowModuleValue::Flow { path, .. } => flow_to_payload(path, delete_after_use),
FlowModuleValue::Flow { path, .. } => {
flow_to_payload(path, delete_after_use, &flow_job.workspace_id, db).await?
}
FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => {
script_to_payload(script_hash, script_path, db, flow_job, module, tag_override).await?
}
@@ -3707,6 +3727,7 @@ async fn payload_from_simple_module(
tag,
delete_after_use,
timeout: module.timeout,
on_behalf_of: None,
},
_ => unreachable!("is simple flow"),
})
@@ -3740,12 +3761,31 @@ fn raw_script_to_payload(
tag,
delete_after_use,
timeout: module.timeout,
on_behalf_of: None,
}
}
fn flow_to_payload(path: String, delete_after_use: bool) -> JobPayloadWithTag {
async fn flow_to_payload(
path: String,
delete_after_use: bool,
w_id: &str,
db: &DB,
) -> Result<JobPayloadWithTag, Error> {
let record = sqlx::query!(
"SELECT on_behalf_of_email, edited_by FROM flow WHERE path = $1 AND workspace_id = $2",
path,
w_id,
)
.fetch_one(db)
.await
.map_err(|e| Error::NotFound(format!("fetching flow: {e:#}")))?;
let on_behalf_of = if let Some(email) = record.on_behalf_of_email {
Some(OnBehalfOf { email, permissioned_as: username_to_permissioned_as(&record.edited_by) })
} else {
None
};
let payload = JobPayload::Flow { path, dedicated_worker: None, apply_preprocessor: false };
JobPayloadWithTag { payload, tag: None, delete_after_use, timeout: None }
Ok(JobPayloadWithTag { payload, tag: None, delete_after_use, timeout: None, on_behalf_of })
}
async fn script_to_payload(
@@ -3761,14 +3801,15 @@ async fn script_to_payload(
} else {
tag_override
};
let (payload, tag, delete_after_use, script_timeout) = if script_hash.is_none() {
let (jp, tag, delete_after_use, script_timeout) =
let (payload, tag, delete_after_use, script_timeout, on_behalf_of) = if script_hash.is_none() {
let (jp, tag, delete_after_use, script_timeout, on_behalf_of) =
script_path_to_payload(&script_path, db, &flow_job.workspace_id, Some(true)).await?;
(
jp,
tag_override.to_owned().or(tag),
delete_after_use,
script_timeout,
on_behalf_of,
)
} else {
let hash = script_hash.unwrap();
@@ -3784,7 +3825,14 @@ async fn script_to_payload(
priority,
delete_after_use,
script_timeout,
on_behalf_of_email,
created_by,
) = script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf { email, permissioned_as: username_to_permissioned_as(&created_by) })
} else {
None
};
(
JobPayload::ScriptHash {
hash,
@@ -3801,6 +3849,7 @@ async fn script_to_payload(
tag_override.to_owned().or(tag),
delete_after_use,
script_timeout,
on_behalf_of,
)
};
// the module value overrides the value set at the script level. Defaults to false if both are unset.
@@ -3812,6 +3861,7 @@ async fn script_to_payload(
tag,
delete_after_use: final_delete_after_user,
timeout: flow_step_timeout,
on_behalf_of,
})
}
backend/windmill-worker/src/worker_lockfiles.rs
+10 -9
View File
@@ -444,20 +444,21 @@ async fn trigger_dependents_to_recompute_dependencies(
let kind = s.importer_kind.clone().unwrap_or_default();
let job_payload = if kind == "script" {
let r = get_latest_deployed_hash_for_path(db, w_id, s.importer_path.as_str()).await;
if let Ok(r) = r {
JobPayload::Dependencies {
match r {
Ok(r) => JobPayload::Dependencies {
path: s.importer_path.clone(),
hash: r.0,
language: r.6,
dedicated_worker: r.7,
},
Err(err) => {
tracing::error!(
"error getting latest deployed hash for path {path}: {err}",
path = s.importer_path,
err = err
);
continue;
}
} else {
tracing::error!(
"error getting latest deployed hash for path {path}: {err}",
path = s.importer_path,
err = r.unwrap_err()
);
continue;
}
} else if kind == "flow" {
args.insert(
cli/script.ts
+3 -1
View File
@@ -309,6 +309,7 @@ export async function handleFile(
//@ts-ignore
codebase: codebase?.digest,
timeout: typed?.timeout,
on_behalf_of_email: typed?.on_behalf_of_email,
};
if (remote) {
@@ -343,7 +344,8 @@ export async function handleFile(
typed.timeout == remote.timeout &&
//@ts-ignore
typed.concurrency_key == remote["concurrency_key"] &&
typed.codebase == remote.codebase)
typed.codebase == remote.codebase &&
typed.on_behalf_of_email == remote.on_behalf_of_email)
) {
log.info(colors.green(`Script ${remotePath} is up to date`));
return true;
frontend/src/lib/components/FlowBuilder.svelte
+4 -1
View File
@@ -228,7 +228,8 @@
tag: flow.tag,
draft_only: true,
ws_error_handler_muted: flow.ws_error_handler_muted,
visible_to_runner_only: flow.visible_to_runner_only
visible_to_runner_only: flow.visible_to_runner_only,
on_behalf_of_email: flow.on_behalf_of_email
}
})
}
@@ -359,6 +360,7 @@
tag: flow.tag,
dedicated_worker: flow.dedicated_worker,
visible_to_runner_only: flow.visible_to_runner_only,
on_behalf_of_email: flow.on_behalf_of_email,
deployment_message: deploymentMsg || undefined
}
})
@@ -432,6 +434,7 @@
dedicated_worker: flow.dedicated_worker,
ws_error_handler_muted: flow.ws_error_handler_muted,
visible_to_runner_only: flow.visible_to_runner_only,
on_behalf_of_email: flow.on_behalf_of_email,
deployment_message: deploymentMsg || undefined
}
})
frontend/src/lib/components/MoveDrawer.svelte
+2 -1
View File
@@ -54,7 +54,8 @@
tag: flow.tag,
dedicated_worker: flow.dedicated_worker,
ws_error_handler_muted: flow.ws_error_handler_muted,
visible_to_runner_only: flow.visible_to_runner_only
visible_to_runner_only: flow.visible_to_runner_only,
on_behalf_of_email: flow.on_behalf_of_email
}
})
} else if (kind == 'script') {
frontend/src/lib/components/ScheduleEditorInner.svelte
-1
View File
@@ -584,7 +584,6 @@
{#if drawerLoading}
<Loader2 class="animate-spin" />
{:else}
{can_write}
<div class="flex flex-col gap-12">
<div class="flex flex-col gap-4">
<div>
frontend/src/lib/components/ScriptBuilder.svelte
+28 -2
View File
@@ -387,7 +387,8 @@
visible_to_runner_only: script.visible_to_runner_only,
no_main_func: script.no_main_func,
has_preprocessor: script.has_preprocessor,
deployment_message: deploymentMsg || undefined
deployment_message: deploymentMsg || undefined,
on_behalf_of_email: script.on_behalf_of_email
}
})
@@ -527,7 +528,8 @@
: script.concurrency_key,
visible_to_runner_only: script.visible_to_runner_only,
no_main_func: script.no_main_func,
has_preprocessor: script.has_preprocessor
has_preprocessor: script.has_preprocessor,
on_behalf_of_email: script.on_behalf_of_email
}
})
}
@@ -1194,6 +1196,30 @@
/>
</div>
</Section>
<Section label="On behalf of last editor">
<svelte:fragment slot="header">
<Tooltip>
When this option is enabled, the script will be run with the permissions of
the last editor.
</Tooltip>
</svelte:fragment>
<div class="flex gap-2 shrink flex-col">
<Toggle
size="sm"
checked={Boolean(script.on_behalf_of_email)}
on:change={() => {
if (script.on_behalf_of_email) {
script.on_behalf_of_email = undefined
} else {
script.on_behalf_of_email = $userStore?.email
}
}}
options={{
right: 'Run on behalf of last editor'
}}
/>
</div>
</Section>
{#if !isCloudHosted()}
<Section label="Custom env variables">
<svelte:fragment slot="header">
frontend/src/lib/components/flows/content/FlowSettings.svelte
+23 -1
View File
@@ -5,7 +5,7 @@
import { Alert, Button, SecondsInput } from '$lib/components/common'
import { getContext } from 'svelte'
import type { FlowEditorContext } from '../types'
import { enterpriseLicense, workspaceStore } from '$lib/stores'
import { enterpriseLicense, userStore, workspaceStore } from '$lib/stores'
import { isCloudHosted } from '$lib/cloud'
import Tooltip from '$lib/components/Tooltip.svelte'
import SimpleEditor from '$lib/components/SimpleEditor.svelte'
@@ -46,6 +46,7 @@
{ name: 'Early Return', active: Boolean($flowStore.value.early_return) },
{ name: 'Dedicated Worker', active: Boolean($flowStore.dedicated_worker) },
{ name: 'Concurrent Limit', active: Boolean($flowStore.value.concurrent_limit) },
{ name: 'Run on Behalf of Last Editor', active: Boolean($flowStore.on_behalf_of_email) },
{ name: 'Worker Tag', active: displayWorkerTagPicker }
]
@@ -348,6 +349,27 @@
class="py-1"
/>
<!-- On behalf of last editor section -->
<Toggle
textClass="font-normal text-sm"
color="nord"
size="xs"
checked={Boolean($flowStore.on_behalf_of_email)}
on:change={() => {
if ($flowStore.on_behalf_of_email) {
$flowStore.on_behalf_of_email = undefined
} else {
$flowStore.on_behalf_of_email = $userStore?.email
}
}}
options={{
right: 'Run on behalf of last editor',
rightTooltip:
'When this option is enabled, the flow will be run with the permissions of the last editor.'
}}
class="py-1"
/>
<!-- Error Handler Section -->
<div class="flex flex-row items-center py-1">
<ErrorHandlerToggleButtonV2
frontend/src/lib/components/flows/content/ScriptEditorDrawer.svelte
+2
View File
@@ -48,6 +48,7 @@
ws_error_handler_muted?: boolean
dedicated_worker?: boolean
visible_to_runner_only?: boolean
on_behalf_of_email?: string
no_main_func?: boolean
has_preprocessor?: boolean
}
@@ -68,6 +69,7 @@
ws_error_handler_muted?: boolean
dedicated_worker?: boolean
visible_to_runner_only?: boolean
on_behalf_of_email?: string
no_main_func?: boolean
has_preprocessor?: boolean
}
frontend/src/lib/components/flows/header/FlowYamlEditor.svelte
+3
View File
@@ -38,6 +38,9 @@
if (parsed['visible_to_runner_only'] !== undefined) {
$flowStore.visible_to_runner_only = parsed['visible_to_runner_only']
}
if (parsed['on_behalf_of_email'] !== undefined) {
$flowStore.on_behalf_of_email = parsed['on_behalf_of_email']
}
$flowStore.value = parsed.value
$flowStore.schema = parsed.schema
$flowStore.tag = parsed.tag
frontend/src/lib/components/flows/types.ts
+1
View File
@@ -23,6 +23,7 @@ export type ExtendedOpenFlow = OpenFlow & {
ws_error_handler_muted?: boolean
dedicated_worker?: boolean
visible_to_runner_only?: boolean
on_behalf_of_email?: string
}
export type FlowInputEditorState = {
frontend/src/lib/components/flows/utils.ts
+4
View File
@@ -85,6 +85,9 @@ export function filteredContentForExport(flow: ExtendedOpenFlow) {
if (flow.visible_to_runner_only) {
o['visible_to_runner_only'] = flow.visible_to_runner_only
}
if (flow.on_behalf_of_email) {
o['on_behalf_of_email'] = flow.on_behalf_of_email
}
if (flow.ws_error_handler_muted) {
o['ws_error_handler_muted'] = flow.ws_error_handler_muted
}
@@ -99,6 +102,7 @@ export function cleanInputs(flow: OpenFlow | any): OpenFlow & {
ws_error_handler_muted?: boolean
dedicated_worker?: boolean
visible_to_runner_only?: boolean
on_behalf_of_email?: string
} {
const newFlow: Flow = JSON.parse(JSON.stringify(flow))
newFlow.value.modules.forEach((mod) => {