From 175379fbe9ca2ede49c399891267caab5ce9b180 Mon Sep 17 00:00:00 2001 From: HugoCasa Date: Tue, 14 Jan 2025 18:32:25 +0100 Subject: [PATCH] feat: on behalf of (#5058) * feat: on behalf of * fix * revert mistakke * temp ee ref * renaming + CLI * sqlx * nits + ee ref --- ...17366b74e891034d32f2867ccb019da869fc8.json | 16 ++ ...9dfee63bb704c3f712af179732dc281411917.json | 15 -- ...72ed0c01a0bb974488530cd713b834f537f4a.json | 16 -- ...e0c5508da110f54bc5f465f1892afdf851af0.json | 77 ------ ...a8f607dccc015ce2d5150c56c68074cd0f84d.json | 35 --- ...27a80ccc8cd75e9703cb1d014b45f57c2be6.json} | 5 +- ...8d1e383aeda9a5a71183b7fbaa41deca4e333.json | 16 ++ ...b9b96854e5bfd752e45922c358c42ebea0c4.json} | 20 +- ...7f1f9f836bd5b6866d3db26b25428d39c0b23.json | 29 +++ ...968b8fb0bf4374a2fccc167fa879bad7078f.json} | 18 +- ...995e78878a4a4ff6894a04e6626cdd98c24e4.json | 15 ++ ...7f82bac6e34a35b94765463c12e282fba6daa.json | 20 ++ ...4cb1a32c4991cbd25046282d34272587d2de.json} | 30 ++- ...2c6cd2932f28ab6490ace027ed3a790b2a0f7.json | 28 +++ ...8632fae1123e736d728bf92cb2de004b6826.json} | 18 +- ...43dab1c08d88c9bcc40b62f2e6bbbbc234ff3.json | 12 + ...d4996b379d436566ee697bbe32361c859be64.json | 12 + ...c4accfc78bf160c1277f35d9d8ddcd056963.json} | 5 +- ...a675e0eaf5e6cc013149740bc778987a1006.json} | 7 +- ...10e2160d1b7e123c7ab226dc77d1789f164f9.json | 47 ++++ ...d990dbcb91b781728b5b6253051decfda83d6.json | 16 -- ...928e7a158bd29d9035bee136f9952fb96bad4.json | 6 +- ...feaeb1b671120f5f8c98fca8c89c6587fc35.json} | 18 +- ...7b3409d9166a028701981cfc280aaab3c4436.json | 41 ++++ ...e521040caa6f6cc745afb7de66f4442e76cba.json | 20 ++ backend/ee-repo-ref.txt | 2 +- ...50113154322_runnable_on_behalf_of.down.sql | 3 + ...0250113154322_runnable_on_behalf_of.up.sql | 3 + backend/tests/worker.rs | 2 + backend/windmill-api/openapi.yaml | 8 + backend/windmill-api/src/apps.rs | 16 +- backend/windmill-api/src/flows.rs | 36 ++- backend/windmill-api/src/jobs.rs | 221 ++++++++++++++---- backend/windmill-api/src/scripts.rs | 17 +- backend/windmill-api/src/users.rs | 4 +- backend/windmill-api/src/workspaces_export.rs | 5 +- backend/windmill-api/src/workspaces_extra.rs | 4 +- backend/windmill-common/src/auth.rs | 2 +- backend/windmill-common/src/flows.rs | 3 + backend/windmill-common/src/jobs.rs | 129 ++++++---- backend/windmill-common/src/lib.rs | 12 +- backend/windmill-common/src/scripts.rs | 3 + backend/windmill-queue/src/jobs.rs | 73 ++++-- backend/windmill-queue/src/schedule.rs | 59 ++++- backend/windmill-worker/src/worker_flow.rs | 68 +++++- .../windmill-worker/src/worker_lockfiles.rs | 19 +- cli/script.ts | 4 +- .../src/lib/components/FlowBuilder.svelte | 5 +- frontend/src/lib/components/MoveDrawer.svelte | 3 +- .../lib/components/ScheduleEditorInner.svelte | 1 - .../src/lib/components/ScriptBuilder.svelte | 30 ++- .../flows/content/FlowSettings.svelte | 24 +- .../flows/content/ScriptEditorDrawer.svelte | 2 + .../flows/header/FlowYamlEditor.svelte | 3 + frontend/src/lib/components/flows/types.ts | 1 + frontend/src/lib/components/flows/utils.ts | 4 + 56 files changed, 947 insertions(+), 361 deletions(-) create mode 100644 backend/.sqlx/query-0a56301b5aaf57339cb2904c8f617366b74e891034d32f2867ccb019da869fc8.json delete mode 100644 backend/.sqlx/query-13358ffeb0917dd9dff9f8527a59dfee63bb704c3f712af179732dc281411917.json delete mode 100644 backend/.sqlx/query-25155e44372aecbb38d042bfc2772ed0c01a0bb974488530cd713b834f537f4a.json delete mode 100644 backend/.sqlx/query-2f440ab6083764b49e309c1f8dde0c5508da110f54bc5f465f1892afdf851af0.json delete mode 100644 backend/.sqlx/query-3fedf12ee5d44e8af9c3864cc70a8f607dccc015ce2d5150c56c68074cd0f84d.json rename backend/.sqlx/{query-9e64c6b6db2155ad8e6e514b08da92d80caafd1a217813bcd354aa264810f690.json => query-4205d237c123d8d1c9ff2d61118027a80ccc8cd75e9703cb1d014b45f57c2be6.json} (63%) create mode 100644 backend/.sqlx/query-544a00afb5c72d2aa24ab4ccfe68d1e383aeda9a5a71183b7fbaa41deca4e333.json rename backend/.sqlx/{query-fc7ac95d8d5f4532a2b0e50f84c1e273d25543401639a68f868e1e4fd8b0640a.json => query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json} (80%) create mode 100644 backend/.sqlx/query-6d36da815795d5cac2e76b4d34a7f1f9f836bd5b6866d3db26b25428d39c0b23.json rename backend/.sqlx/{query-64ce3eabed6de7a4f99884640c21218ac19ab2a4bc5f8b42deb2476f85d0deac.json => query-75451b6d48e4c26812ae64981d0d968b8fb0bf4374a2fccc167fa879bad7078f.json} (84%) create mode 100644 backend/.sqlx/query-81a9d976ac5c1a78c83b95a1164995e78878a4a4ff6894a04e6626cdd98c24e4.json create mode 100644 backend/.sqlx/query-8929873f308f1d131b2afda17267f82bac6e34a35b94765463c12e282fba6daa.json rename backend/.sqlx/{query-9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418.json => query-89de3ff8ab32e545efcbcda05f994cb1a32c4991cbd25046282d34272587d2de.json} (69%) create mode 100644 backend/.sqlx/query-8fb2581a439c26391e66ae7fac32c6cd2932f28ab6490ace027ed3a790b2a0f7.json rename backend/.sqlx/{query-98081ee9a4a4a52ae70d74d26e8608a010722d7ea9dc343ae8f03e38978ffaba.json => query-a0833b9899833166891c5de926f78632fae1123e736d728bf92cb2de004b6826.json} (56%) create mode 100644 backend/.sqlx/query-ab3f8ebcebf0088c1104be4014c43dab1c08d88c9bcc40b62f2e6bbbbc234ff3.json create mode 100644 backend/.sqlx/query-b53007a988c6336015dbe50a207d4996b379d436566ee697bbe32361c859be64.json rename backend/.sqlx/{query-899a406cc03659e29bad831cc4d3d1dcda1a39b826a03136353344ccae29871b.json => query-ba8bde5018fdf7b12f85cd3a6557c4accfc78bf160c1277f35d9d8ddcd056963.json} (61%) rename backend/.sqlx/{query-b49b1eaf58e62063c7a03039d0c36e65991b1d21a3306390fb1c7ca38babafe3.json => query-bc00efe12901bc7b49c7892d2d9da675e0eaf5e6cc013149740bc778987a1006.json} (84%) create mode 100644 backend/.sqlx/query-c794ff5e14429a1bc56f5e70cfd10e2160d1b7e123c7ab226dc77d1789f164f9.json delete mode 100644 backend/.sqlx/query-d860cd90e583e4666beb37b283bd990dbcb91b781728b5b6253051decfda83d6.json rename backend/.sqlx/{query-1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c.json => query-ec7836df5f9056ec70015800b7f4feaeb1b671120f5f8c98fca8c89c6587fc35.json} (81%) create mode 100644 backend/.sqlx/query-f44f1a794ab217be05a9feb21e57b3409d9166a028701981cfc280aaab3c4436.json create mode 100644 backend/.sqlx/query-fea19da213230401161cbf2ee1de521040caa6f6cc745afb7de66f4442e76cba.json create mode 100644 backend/migrations/20250113154322_runnable_on_behalf_of.down.sql create mode 100644 backend/migrations/20250113154322_runnable_on_behalf_of.up.sql diff --git a/backend/.sqlx/query-0a56301b5aaf57339cb2904c8f617366b74e891034d32f2867ccb019da869fc8.json b/backend/.sqlx/query-0a56301b5aaf57339cb2904c8f617366b74e891034d32f2867ccb019da869fc8.json new file mode 100644 index 0000000000..159789f439 --- /dev/null +++ b/backend/.sqlx/query-0a56301b5aaf57339cb2904c8f617366b74e891034d32f2867ccb019da869fc8.json @@ -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" +} diff --git a/backend/.sqlx/query-13358ffeb0917dd9dff9f8527a59dfee63bb704c3f712af179732dc281411917.json b/backend/.sqlx/query-13358ffeb0917dd9dff9f8527a59dfee63bb704c3f712af179732dc281411917.json deleted file mode 100644 index d49a1b9d44..0000000000 --- a/backend/.sqlx/query-13358ffeb0917dd9dff9f8527a59dfee63bb704c3f712af179732dc281411917.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-25155e44372aecbb38d042bfc2772ed0c01a0bb974488530cd713b834f537f4a.json b/backend/.sqlx/query-25155e44372aecbb38d042bfc2772ed0c01a0bb974488530cd713b834f537f4a.json deleted file mode 100644 index f459a72957..0000000000 --- a/backend/.sqlx/query-25155e44372aecbb38d042bfc2772ed0c01a0bb974488530cd713b834f537f4a.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-2f440ab6083764b49e309c1f8dde0c5508da110f54bc5f465f1892afdf851af0.json b/backend/.sqlx/query-2f440ab6083764b49e309c1f8dde0c5508da110f54bc5f465f1892afdf851af0.json deleted file mode 100644 index 986ab59f4d..0000000000 --- a/backend/.sqlx/query-2f440ab6083764b49e309c1f8dde0c5508da110f54bc5f465f1892afdf851af0.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-3fedf12ee5d44e8af9c3864cc70a8f607dccc015ce2d5150c56c68074cd0f84d.json b/backend/.sqlx/query-3fedf12ee5d44e8af9c3864cc70a8f607dccc015ce2d5150c56c68074cd0f84d.json deleted file mode 100644 index 1d91cd4031..0000000000 --- a/backend/.sqlx/query-3fedf12ee5d44e8af9c3864cc70a8f607dccc015ce2d5150c56c68074cd0f84d.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-9e64c6b6db2155ad8e6e514b08da92d80caafd1a217813bcd354aa264810f690.json b/backend/.sqlx/query-4205d237c123d8d1c9ff2d61118027a80ccc8cd75e9703cb1d014b45f57c2be6.json similarity index 63% rename from backend/.sqlx/query-9e64c6b6db2155ad8e6e514b08da92d80caafd1a217813bcd354aa264810f690.json rename to backend/.sqlx/query-4205d237c123d8d1c9ff2d61118027a80ccc8cd75e9703cb1d014b45f57c2be6.json index 25296490e0..6d4744325c 100644 --- a/backend/.sqlx/query-9e64c6b6db2155ad8e6e514b08da92d80caafd1a217813bcd354aa264810f690.json +++ b/backend/.sqlx/query-4205d237c123d8d1c9ff2d61118027a80ccc8cd75e9703cb1d014b45f57c2be6.json @@ -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" } diff --git a/backend/.sqlx/query-544a00afb5c72d2aa24ab4ccfe68d1e383aeda9a5a71183b7fbaa41deca4e333.json b/backend/.sqlx/query-544a00afb5c72d2aa24ab4ccfe68d1e383aeda9a5a71183b7fbaa41deca4e333.json new file mode 100644 index 0000000000..bf1418bb0d --- /dev/null +++ b/backend/.sqlx/query-544a00afb5c72d2aa24ab4ccfe68d1e383aeda9a5a71183b7fbaa41deca4e333.json @@ -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" +} diff --git a/backend/.sqlx/query-fc7ac95d8d5f4532a2b0e50f84c1e273d25543401639a68f868e1e4fd8b0640a.json b/backend/.sqlx/query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json similarity index 80% rename from backend/.sqlx/query-fc7ac95d8d5f4532a2b0e50f84c1e273d25543401639a68f868e1e4fd8b0640a.json rename to backend/.sqlx/query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json index 8f459f96fd..5698ec9939 100644 --- a/backend/.sqlx/query-fc7ac95d8d5f4532a2b0e50f84c1e273d25543401639a68f868e1e4fd8b0640a.json +++ b/backend/.sqlx/query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json @@ -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" } diff --git a/backend/.sqlx/query-6d36da815795d5cac2e76b4d34a7f1f9f836bd5b6866d3db26b25428d39c0b23.json b/backend/.sqlx/query-6d36da815795d5cac2e76b4d34a7f1f9f836bd5b6866d3db26b25428d39c0b23.json new file mode 100644 index 0000000000..009c8a0e26 --- /dev/null +++ b/backend/.sqlx/query-6d36da815795d5cac2e76b4d34a7f1f9f836bd5b6866d3db26b25428d39c0b23.json @@ -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" +} diff --git a/backend/.sqlx/query-64ce3eabed6de7a4f99884640c21218ac19ab2a4bc5f8b42deb2476f85d0deac.json b/backend/.sqlx/query-75451b6d48e4c26812ae64981d0d968b8fb0bf4374a2fccc167fa879bad7078f.json similarity index 84% rename from backend/.sqlx/query-64ce3eabed6de7a4f99884640c21218ac19ab2a4bc5f8b42deb2476f85d0deac.json rename to backend/.sqlx/query-75451b6d48e4c26812ae64981d0d968b8fb0bf4374a2fccc167fa879bad7078f.json index b35a48638a..19b8734576 100644 --- a/backend/.sqlx/query-64ce3eabed6de7a4f99884640c21218ac19ab2a4bc5f8b42deb2476f85d0deac.json +++ b/backend/.sqlx/query-75451b6d48e4c26812ae64981d0d968b8fb0bf4374a2fccc167fa879bad7078f.json @@ -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" } diff --git a/backend/.sqlx/query-81a9d976ac5c1a78c83b95a1164995e78878a4a4ff6894a04e6626cdd98c24e4.json b/backend/.sqlx/query-81a9d976ac5c1a78c83b95a1164995e78878a4a4ff6894a04e6626cdd98c24e4.json new file mode 100644 index 0000000000..4295debb89 --- /dev/null +++ b/backend/.sqlx/query-81a9d976ac5c1a78c83b95a1164995e78878a4a4ff6894a04e6626cdd98c24e4.json @@ -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" +} diff --git a/backend/.sqlx/query-8929873f308f1d131b2afda17267f82bac6e34a35b94765463c12e282fba6daa.json b/backend/.sqlx/query-8929873f308f1d131b2afda17267f82bac6e34a35b94765463c12e282fba6daa.json new file mode 100644 index 0000000000..982632ee3d --- /dev/null +++ b/backend/.sqlx/query-8929873f308f1d131b2afda17267f82bac6e34a35b94765463c12e282fba6daa.json @@ -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" +} diff --git a/backend/.sqlx/query-9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418.json b/backend/.sqlx/query-89de3ff8ab32e545efcbcda05f994cb1a32c4991cbd25046282d34272587d2de.json similarity index 69% rename from backend/.sqlx/query-9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418.json rename to backend/.sqlx/query-89de3ff8ab32e545efcbcda05f994cb1a32c4991cbd25046282d34272587d2de.json index b955a01fb6..56dc319186 100644 --- a/backend/.sqlx/query-9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418.json +++ b/backend/.sqlx/query-89de3ff8ab32e545efcbcda05f994cb1a32c4991cbd25046282d34272587d2de.json @@ -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" } diff --git a/backend/.sqlx/query-8fb2581a439c26391e66ae7fac32c6cd2932f28ab6490ace027ed3a790b2a0f7.json b/backend/.sqlx/query-8fb2581a439c26391e66ae7fac32c6cd2932f28ab6490ace027ed3a790b2a0f7.json new file mode 100644 index 0000000000..0ae81ab6dd --- /dev/null +++ b/backend/.sqlx/query-8fb2581a439c26391e66ae7fac32c6cd2932f28ab6490ace027ed3a790b2a0f7.json @@ -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" +} diff --git a/backend/.sqlx/query-98081ee9a4a4a52ae70d74d26e8608a010722d7ea9dc343ae8f03e38978ffaba.json b/backend/.sqlx/query-a0833b9899833166891c5de926f78632fae1123e736d728bf92cb2de004b6826.json similarity index 56% rename from backend/.sqlx/query-98081ee9a4a4a52ae70d74d26e8608a010722d7ea9dc343ae8f03e38978ffaba.json rename to backend/.sqlx/query-a0833b9899833166891c5de926f78632fae1123e736d728bf92cb2de004b6826.json index 05cbaffc69..a837158a37 100644 --- a/backend/.sqlx/query-98081ee9a4a4a52ae70d74d26e8608a010722d7ea9dc343ae8f03e38978ffaba.json +++ b/backend/.sqlx/query-a0833b9899833166891c5de926f78632fae1123e736d728bf92cb2de004b6826.json @@ -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" } diff --git a/backend/.sqlx/query-ab3f8ebcebf0088c1104be4014c43dab1c08d88c9bcc40b62f2e6bbbbc234ff3.json b/backend/.sqlx/query-ab3f8ebcebf0088c1104be4014c43dab1c08d88c9bcc40b62f2e6bbbbc234ff3.json new file mode 100644 index 0000000000..ae03769424 --- /dev/null +++ b/backend/.sqlx/query-ab3f8ebcebf0088c1104be4014c43dab1c08d88c9bcc40b62f2e6bbbbc234ff3.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "SET LOCAL ROLE NONE", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "ab3f8ebcebf0088c1104be4014c43dab1c08d88c9bcc40b62f2e6bbbbc234ff3" +} diff --git a/backend/.sqlx/query-b53007a988c6336015dbe50a207d4996b379d436566ee697bbe32361c859be64.json b/backend/.sqlx/query-b53007a988c6336015dbe50a207d4996b379d436566ee697bbe32361c859be64.json new file mode 100644 index 0000000000..0f22a3079b --- /dev/null +++ b/backend/.sqlx/query-b53007a988c6336015dbe50a207d4996b379d436566ee697bbe32361c859be64.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "SET LOCAL ROLE windmill_user", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "b53007a988c6336015dbe50a207d4996b379d436566ee697bbe32361c859be64" +} diff --git a/backend/.sqlx/query-899a406cc03659e29bad831cc4d3d1dcda1a39b826a03136353344ccae29871b.json b/backend/.sqlx/query-ba8bde5018fdf7b12f85cd3a6557c4accfc78bf160c1277f35d9d8ddcd056963.json similarity index 61% rename from backend/.sqlx/query-899a406cc03659e29bad831cc4d3d1dcda1a39b826a03136353344ccae29871b.json rename to backend/.sqlx/query-ba8bde5018fdf7b12f85cd3a6557c4accfc78bf160c1277f35d9d8ddcd056963.json index dec3a1bb5a..e6564a96fa 100644 --- a/backend/.sqlx/query-899a406cc03659e29bad831cc4d3d1dcda1a39b826a03136353344ccae29871b.json +++ b/backend/.sqlx/query-ba8bde5018fdf7b12f85cd3a6557c4accfc78bf160c1277f35d9d8ddcd056963.json @@ -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" } diff --git a/backend/.sqlx/query-b49b1eaf58e62063c7a03039d0c36e65991b1d21a3306390fb1c7ca38babafe3.json b/backend/.sqlx/query-bc00efe12901bc7b49c7892d2d9da675e0eaf5e6cc013149740bc778987a1006.json similarity index 84% rename from backend/.sqlx/query-b49b1eaf58e62063c7a03039d0c36e65991b1d21a3306390fb1c7ca38babafe3.json rename to backend/.sqlx/query-bc00efe12901bc7b49c7892d2d9da675e0eaf5e6cc013149740bc778987a1006.json index 8b3824a2a2..6abc06cf41 100644 --- a/backend/.sqlx/query-b49b1eaf58e62063c7a03039d0c36e65991b1d21a3306390fb1c7ca38babafe3.json +++ b/backend/.sqlx/query-bc00efe12901bc7b49c7892d2d9da675e0eaf5e6cc013149740bc778987a1006.json @@ -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" } diff --git a/backend/.sqlx/query-c794ff5e14429a1bc56f5e70cfd10e2160d1b7e123c7ab226dc77d1789f164f9.json b/backend/.sqlx/query-c794ff5e14429a1bc56f5e70cfd10e2160d1b7e123c7ab226dc77d1789f164f9.json new file mode 100644 index 0000000000..c4810efe68 --- /dev/null +++ b/backend/.sqlx/query-c794ff5e14429a1bc56f5e70cfd10e2160d1b7e123c7ab226dc77d1789f164f9.json @@ -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" +} diff --git a/backend/.sqlx/query-d860cd90e583e4666beb37b283bd990dbcb91b781728b5b6253051decfda83d6.json b/backend/.sqlx/query-d860cd90e583e4666beb37b283bd990dbcb91b781728b5b6253051decfda83d6.json deleted file mode 100644 index 149c649285..0000000000 --- a/backend/.sqlx/query-d860cd90e583e4666beb37b283bd990dbcb91b781728b5b6253051decfda83d6.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-e08dddf6af2656b561c453460ff928e7a158bd29d9035bee136f9952fb96bad4.json b/backend/.sqlx/query-e08dddf6af2656b561c453460ff928e7a158bd29d9035bee136f9952fb96bad4.json index ff34048ea9..b2ff93f8d9 100644 --- a/backend/.sqlx/query-e08dddf6af2656b561c453460ff928e7a158bd29d9035bee136f9952fb96bad4.json +++ b/backend/.sqlx/query-e08dddf6af2656b561c453460ff928e7a158bd29d9035bee136f9952fb96bad4.json @@ -25,7 +25,8 @@ "http", "websocket", "kafka", - "email" + "email", + "nats" ] } } @@ -56,7 +57,8 @@ "http", "websocket", "kafka", - "email" + "email", + "nats" ] } } diff --git a/backend/.sqlx/query-1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c.json b/backend/.sqlx/query-ec7836df5f9056ec70015800b7f4feaeb1b671120f5f8c98fca8c89c6587fc35.json similarity index 81% rename from backend/.sqlx/query-1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c.json rename to backend/.sqlx/query-ec7836df5f9056ec70015800b7f4feaeb1b671120f5f8c98fca8c89c6587fc35.json index a82d32768e..9fb33e41d3 100644 --- a/backend/.sqlx/query-1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c.json +++ b/backend/.sqlx/query-ec7836df5f9056ec70015800b7f4feaeb1b671120f5f8c98fca8c89c6587fc35.json @@ -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" } diff --git a/backend/.sqlx/query-f44f1a794ab217be05a9feb21e57b3409d9166a028701981cfc280aaab3c4436.json b/backend/.sqlx/query-f44f1a794ab217be05a9feb21e57b3409d9166a028701981cfc280aaab3c4436.json new file mode 100644 index 0000000000..79ef441a8a --- /dev/null +++ b/backend/.sqlx/query-f44f1a794ab217be05a9feb21e57b3409d9166a028701981cfc280aaab3c4436.json @@ -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" +} diff --git a/backend/.sqlx/query-fea19da213230401161cbf2ee1de521040caa6f6cc745afb7de66f4442e76cba.json b/backend/.sqlx/query-fea19da213230401161cbf2ee1de521040caa6f6cc745afb7de66f4442e76cba.json new file mode 100644 index 0000000000..d956bf8a5f --- /dev/null +++ b/backend/.sqlx/query-fea19da213230401161cbf2ee1de521040caa6f6cc745afb7de66f4442e76cba.json @@ -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" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 6620861be8..1c88bcf6e6 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -9cea7e44f1b6401662e22088b41e8a00d7e67b23 \ No newline at end of file +a515264a3c1c138e289da1ef3bb21d260aead486 \ No newline at end of file diff --git a/backend/migrations/20250113154322_runnable_on_behalf_of.down.sql b/backend/migrations/20250113154322_runnable_on_behalf_of.down.sql new file mode 100644 index 0000000000..ec1da37b8f --- /dev/null +++ b/backend/migrations/20250113154322_runnable_on_behalf_of.down.sql @@ -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; diff --git a/backend/migrations/20250113154322_runnable_on_behalf_of.up.sql b/backend/migrations/20250113154322_runnable_on_behalf_of.up.sql new file mode 100644 index 0000000000..8e237c1dca --- /dev/null +++ b/backend/migrations/20250113154322_runnable_on_behalf_of.up.sql @@ -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; \ No newline at end of file diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index adb0e5ccff..7bb71416c7 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -2874,6 +2874,7 @@ async fn test_flow_lock_all(db: Pool) { 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 diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index a128abd003..7855dac778 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -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 diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 7218f98ec0..a8cf4a5bde 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -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, diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index b37dbd9d9a..1b283d43f3 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -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, #[serde(skip_serializing_if = "Option::is_none")] pub visible_to_runner_only: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub on_behalf_of_email: Option, } 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' diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 9687840877..379cafa9fb 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -561,9 +561,11 @@ pub async fn get_path_tag_limits_cache_for_hash( Option, Option, Option, + Option, + 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?; diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 15d95a031b..bec8c3b2c8 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -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, #[serde(skip_serializing_if = "Option::is_none")] pub has_preprocessor: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub on_behalf_of_email: Option, } 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 \ diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index e82212ce53..fdb40d97c5 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -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, diff --git a/backend/windmill-api/src/workspaces_export.rs b/backend/windmill-api/src/workspaces_export.rs index 774bc97053..7ab9481569 100644 --- a/backend/windmill-api/src/workspaces_export.rs +++ b/backend/windmill-api/src/workspaces_export.rs @@ -83,6 +83,8 @@ struct ScriptMetadata { pub concurrency_key: Option, #[serde(skip_serializing_if = "Option::is_none")] pub has_preprocessor: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub on_behalf_of_email: Option, } pub fn is_none_or_false(val: &Option) -> 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", diff --git a/backend/windmill-api/src/workspaces_extra.rs b/backend/windmill-api/src/workspaces_extra.rs index 77a11b75c1..f01b89b683 100644 --- a/backend/windmill-api/src/workspaces_extra.rs +++ b/backend/windmill-api/src/workspaces_extra.rs @@ -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 diff --git a/backend/windmill-common/src/auth.rs b/backend/windmill-common/src/auth.rs index fcd528e9b0..a70319c547 100644 --- a/backend/windmill-common/src/auth.rs +++ b/backend/windmill-common/src/auth.rs @@ -75,7 +75,7 @@ pub async fn is_super_admin_email(db: &DB, email: &str) -> Result { pub async fn is_devops_email(db: &DB, email: &str) -> Result { 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) diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 370b8a69ec..22458af9ba 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -49,6 +49,8 @@ pub struct Flow { pub timeout: Option, #[serde(skip_serializing_if = "is_none_or_false")] pub visible_to_runner_only: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub on_behalf_of_email: Option, } #[derive(Serialize, sqlx::FromRow)] @@ -98,6 +100,7 @@ pub struct NewFlow { pub timeout: Option, pub deployment_message: Option, pub visible_to_runner_only: Option, + pub on_behalf_of_email: Option, } #[derive(Deserialize, Serialize, Debug, Clone, Default)] diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index dc371d319f..760640b83a 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -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; -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, -) -> error::Result<(JobPayload, Option, Option, Option)> { - 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, + Option, + Option, + Option, +)> { + 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, Option, Option, + Option, + 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), Error> { - let (payload, tag, _, _) = if path.starts_with("script/") { +) -> Result<(JobPayload, Option, Option), 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)] diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 4140f6e58b..9fcd14111f 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -302,9 +302,11 @@ pub async fn get_latest_deployed_hash_for_path<'e, E: sqlx::Executor<'e, Databas Option, Option, Option, + Option, + 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, Option, Option, + Option, + 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, )) } diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 0fb05e6f45..2d7cd0a5e7 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -197,6 +197,8 @@ pub struct Script { pub codebase: Option, #[serde(skip_serializing_if = "Option::is_none")] pub has_preprocessor: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub on_behalf_of_email: Option, } #[derive(Serialize, sqlx::FromRow)] @@ -294,6 +296,7 @@ pub struct NewScript { pub no_main_func: Option, pub codebase: Option, pub has_preprocessor: Option, + pub on_behalf_of_email: Option, } fn lock_deserialize<'de, D>(deserializer: D) -> Result, D::Error> diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index b66bf6ef43..5b2df07a39 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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, extra_args: Option>>, ) -> 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::>>(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, extra_args: Option>>, ) -> 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), diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 9bb39cc044..ea9f476ab6 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -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 } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 3feec7830e..b320b9a445 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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, delete_after_use: bool, timeout: Option, + on_behalf_of: Option, } 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::>(); @@ -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::>(); @@ -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 { 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 { + 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, }) } diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 8d45ec9ca2..213eb2dca5 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -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( diff --git a/cli/script.ts b/cli/script.ts index 6d0b438c7b..0ee33553ac 100644 --- a/cli/script.ts +++ b/cli/script.ts @@ -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; diff --git a/frontend/src/lib/components/FlowBuilder.svelte b/frontend/src/lib/components/FlowBuilder.svelte index 4cdfceb49f..35ff13f46e 100644 --- a/frontend/src/lib/components/FlowBuilder.svelte +++ b/frontend/src/lib/components/FlowBuilder.svelte @@ -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 } }) diff --git a/frontend/src/lib/components/MoveDrawer.svelte b/frontend/src/lib/components/MoveDrawer.svelte index 96b9111270..b5cb5b2c3f 100644 --- a/frontend/src/lib/components/MoveDrawer.svelte +++ b/frontend/src/lib/components/MoveDrawer.svelte @@ -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') { diff --git a/frontend/src/lib/components/ScheduleEditorInner.svelte b/frontend/src/lib/components/ScheduleEditorInner.svelte index 8921acb2cf..261bbe9327 100644 --- a/frontend/src/lib/components/ScheduleEditorInner.svelte +++ b/frontend/src/lib/components/ScheduleEditorInner.svelte @@ -584,7 +584,6 @@ {#if drawerLoading} {:else} - {can_write}
diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index 0c5afc41ce..38c537b407 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -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 @@ />
+
+ + + When this option is enabled, the script will be run with the permissions of + the last editor. + + +
+ { + 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' + }} + /> +
+
{#if !isCloudHosted()}
diff --git a/frontend/src/lib/components/flows/content/FlowSettings.svelte b/frontend/src/lib/components/flows/content/FlowSettings.svelte index 403ac55f1b..7204d67dab 100644 --- a/frontend/src/lib/components/flows/content/FlowSettings.svelte +++ b/frontend/src/lib/components/flows/content/FlowSettings.svelte @@ -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" /> + + { + 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" + /> +
{