From a689c0b0f127482cea86df72b79c575a17abd57d Mon Sep 17 00:00:00 2001 From: Diego Imbert <70353967+diegoimbert@users.noreply.github.com> Date: Mon, 14 Apr 2025 16:25:41 +0200 Subject: [PATCH] feat: Batch re-run (#5553) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * runs on svelte 5 * Line component from svelte-chartjs * Replaced all svelte-chartjs occurrences with custom wrapper * Fix props mistake * Fix illegal table structures * self-closing-tags fix * aria labels * Fixed trivial warnings and errors * @tanstack/svelte-table fix * upgrade to vite 6 * svelte-kit sync before running svelte-check * Remove on:clear which is actually on:removeAll and already handled by on:change * fix worker tags not displaying in Autoscaling * Try to fix svelte-kit sync not working during CI * remove warnings * Fix add flow page crashing * access worldStore before assignment fix * fix infinite recursions in App Editor * Replaced JSON.stringify with proper deepEqual * component mount api changed (no longer classes) * fix ci errors * Fix infinite loops in background runnable panel * factored effect on deep equal logic in onObjChange * fix "Add" not working in AgGrid Table * Replaced legacy component.$set api * Fix multiselect infinite value reaction * Fix flow input fields resetting when opening their edit tab * fix date input resetting when typing year * Remove !p-0 affecting subgrid dotted borders * fix missing debounceTemplate causing hundreds of updates * Fix AgGrid action refreshes and disppearing * resolve getItems generating random ids every rerun * fix cannot access items before init * fix sort lambda arguments being undefined * Revert "Remove !p-0 affecting subgrid dotted borders" This reverts commit c62809bb45d682a48376b071680645ed4e1c601b. * fix input not updating in decision tree editor * Update frontend/src/lib/components/schema/EditableSchemaWrapper.svelte Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com> * Re-added padding affecting subgrid dotted borders (#5479) * remove !p-0 in preset components * removed extra padding on accordion tabs subgrid * Fix non-reactive SchemaForm * dirty fix for the oneOf bug * feat: add nu-lang support (#5217) * feat: add nu (nushell) support * add worker tests * deactivate tables and non-any types below top-level full support will come in V1 for V0 it's better to keep things minimal and simple * add syntax highlighting used python's grammar, since nushell isn't supported by monaco nor svelte-highlights for V1 nu will get it`s own grammar * add logo * partially implement plugin support * change logo + ability to deploy + nsjail draft * static variables + get_resource + get_variable * lsp/dev.nu + initial nu lsp (not working yet) * make it work with nsjail * nullguard * Much more flexible signature parsing and better error-messages * add init script * rename nulsp to nu * install nu to dockerfile * fix merge * implement Default for MainArgSignature * stage NU_CACHE_DIR * improve dockerfiles * dev.nu for parser-wasm + flake.nix * update code for windows * add nushell to flake * upload Cargo.lock * make build.sh work on nixos * build wasm cli parsers * add docs to README_DEV.md * add helper script docker/dev.nu * improve docker/dev.nu * fix windows * commit frontend/package(lock).json * update cargo.lock * correctly update cargo.lock * remove lsp * update flake.nix to include svelte server and nushell * Revert base.sql to main * remove PLUGIN_USE_RE * make CARGO_PATH private * add nu to cli * Change flags to build wasm-nu-parser * remove flake.nix from parser-wasm * update wasm-build target * remove unused import * add cli support for nu * update github workflows * wasm-build 0.17 -> 0.19 * update build script * update cargo.lock * Fix typographical error * update Cargo.lock * update ENV_SETTINGS * use published nu parser * update package.lock * rewrite parser in tree-sitter * implement parser from scratch * polishing * change init script to match new parser * fix imports * fix cli build * fix cli build * merge * update wasm * use MiniPulledJob * update cli * change cli wasm schema * change cli * update deno.json * make wasm modules load lazily * regenerate parsers * remove leftover * update cargo.lock * clean up dnt.ts * add docs to cli/test.nu * add schema validation option * add Nu to try_validate_schema * reference frontend to new parser version * feat: unsafe parameters for sql queries (table names, column names) (#5488) * Make schema validation struct Schema Validation rules that are constructed from the schema or from the MainArgSig(TODO). * Make other validator builder * Fail dependency job like with lockfile failing for schema validator * Add last types + tests * Remove unused dependency * fix typos * Migration ID was colliding with another, changed it manually * Add Oneof + other fixes * fix: cache for querying scripts correclty handles ScriptMetadata * Add cache for schema validation from main arg sig * Prepare sqlx * Remove default features * Feature flags * WIP: unsafe sql params for sql langauges * Fix down migration table name * cleanup: put validation logic inside a function * Refactor to cache the should_validate boolean Changed the schemavalidators cache to take in an Option, effectively storing the `should_validate_schema` information. Also pass the schema when avaialble to construct the schema validator * Add other job kinds to u8 cache key just in case * Change sql languages to all get arguments as Values instead of RawValue * Only cache if not preview * Add last sql languages and some CI fixes * Rename after typo on `sanitized` * Finish rename * Remove unused import * Fix wrong test * Add newly published regex parser version * Remove default features from cargo.toml * Change to a cleaner syntax for the interpolated args * Update republished parser * fix win build (#5494) * add sysinfoapi feature flag for winapi dependency * add ff * add ff at the right place * fix(frontend): use stable path for capture tables + nits (#5495) * add missing capture move on first time deploy (#5496) * avoid regen client as build step * perf: cache workspace env variables to avoid one query (#5499) * perf: optimize number of queries needed for job run (#5504) * optPerf * update sqlx * update sqlx * fix: improve cancel for flows with many substeps * feat: list references upon renaming a script or a flow (#5487) * Refactored flow_workspace_runnables to more generic workspace_runnable_dependencies * list flows referencing an item upon renaming it * Refactor with two exclusive columns to avoid breaking FK constraints * Show apps depending on item upon renaming * sqlx prepare * list-disc instead of • * on delete and on update cascade * displayPathChangedWarning oneOf check instead noneOf * combine migrations + add "on update cascade" to flow fk * unique index on app dependencies to avoid duplicates * create new workspace_runnable_dependencies instead of renaming old table * Add "looking for references" loading msg * Revert "create new workspace_runnable_dependencies instead of renaming old table" This reverts commit 015c38ca8f9fbd8b4a4e817f1ad105434bfd68c5. * flow_workspace_runnables view for backwards compatibility * Add warning for script imports on rename * support import dependency tracking in deno * number of using scripts / flows / apps tooltip * forgot sqlx prepare * delete app-related rows in down migration * Made selection more generic * RunsBatchActionsDropdown refactor * started BatchReRunOptionsPane * fix overflow quirk * fetch schema * refactor to group jobs by (kind, path) * auto select * computePropertyMap * InputTransformForm works * Pickable properties * remove PropPickerWrapper and make it optional in InputTransformForm * hide help btn * available expressions info alert * extraLib for editor linting * fix selected group not updating * nit * Refactor async logic in script tag * persist changes in state * correct typing * count for each (path, kind) group * support flows * use dot operator when possible * count jobs and fix wrong number * fix selectedJobs recomputing periodically * (v0) individual api requests to re-run jobs * move batchReRunChangedArgs state upwards * Support static arg * mistake * Single confirmation modal + removed unnecessary state * change confirmation modal color * use runes in confirmation modal * listSelectedJobsSchema API endpoint * refactored batch rerun pane for listSelectedJobsSchema * eliminated selectedJobs * batch rerun works backend (v0 same args) * Static input transforms * simpler list_selected_jobs_schemas sql query with coalesce * use latest schema UI + refactor * run latest version in backend * add deno_core dependency to windmill-api * stream jobs from db * basic js evaluation * sqlx prepare * add id path and hash in editor lint * js works with job object! * moved deno_core logic to separate function * openapi yaml mistake * unnecessary bind * fix date as string * Stream re-ran uuids * handle SSE multiple values at once * don't select all by default on batch action * nit ui * check that schema has property backend * Better JobGroup query + cache * handle multi type properties * Notify user on error * stupid mistake * Fix warnings and update svelte-exmarkdown for svelte 5 * regen package-lock to fix crash on vite preview * batch re-run all filtered jobs * merge schemas to common type * more explicit tooltips * changed sse counter ui * typos * fix tutorial first part * nit mistake * package lock + elipsis nit * fix: latest_schema option still checked on the job original schema * always gotta forget sqlx prepare * fix flashing loading screen * fix batch re-run select all filtered * better tooltip * fix batch actions btn growing on wide screen * revert disableBatchActions * fix selectable step jobs --------- Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com> Co-authored-by: pyranota <92104930+pyranota@users.noreply.github.com> Co-authored-by: wendrul <53628737+wendrul@users.noreply.github.com> Co-authored-by: Alexander Petric Co-authored-by: HugoCasa Co-authored-by: Ruben Fiszel --- ...4cdb6eae2e71e56bc8e5215559af967c81bd8.json | 79 ++++ ...953e3b32a9a7a73784219d48f8e8932cdb0c8.json | 23 + ...53be3a93dbe2a67bdc4d4342c84f439b5eaf9.json | 85 ++++ ...3b077c957da8455ea747b0680923d8282425b.json | 73 +++ ...7e6b3cb1d60f15fed9a24e3f05b1bd60c0fc8.json | 73 +++ ...b1ec3fc844ed286ff563f0c9a5c4778cfb718.json | 67 +++ ...bd8f537e68fabceb094fe535693fbd146eff2.json | 23 + backend/Cargo.lock | 4 + backend/Cargo.toml | 4 +- backend/windmill-api/Cargo.toml | 9 +- backend/windmill-api/openapi.yaml | 310 +++++++++---- backend/windmill-api/src/jobs.rs | 367 ++++++++++++++- .../src/lib/components/FieldHeader.svelte | 10 +- .../lib/components/InputTransformForm.svelte | 72 +-- .../src/lib/components/S3FilePicker.svelte | 4 +- .../ConfirmationModal.svelte | 54 ++- .../src/lib/components/jobs/batchReruns.ts | 110 +++++ .../runs/BatchReRunOptionsPane.svelte | 275 ++++++++++++ .../src/lib/components/runs/RunRow.svelte | 11 +- .../runs/RunsBatchActionsDropdown.svelte | 92 ++++ .../src/lib/components/runs/RunsTable.svelte | 22 +- frontend/src/lib/schema.ts | 18 +- frontend/src/lib/utils.ts | 20 +- .../(logged)/runs/[...path]/+page.svelte | 422 ++++++++++-------- 24 files changed, 1872 insertions(+), 355 deletions(-) create mode 100644 backend/.sqlx/query-3376b42d273c2499f3517c7754f4cdb6eae2e71e56bc8e5215559af967c81bd8.json create mode 100644 backend/.sqlx/query-5ea1d8c87a17690942ca9e70e0b953e3b32a9a7a73784219d48f8e8932cdb0c8.json create mode 100644 backend/.sqlx/query-ad88831c9021b79c9037a925c2b53be3a93dbe2a67bdc4d4342c84f439b5eaf9.json create mode 100644 backend/.sqlx/query-b61a00e6a8ca0a1d24e64fdc9223b077c957da8455ea747b0680923d8282425b.json create mode 100644 backend/.sqlx/query-c4f382045e5c47986e02f1e57667e6b3cb1d60f15fed9a24e3f05b1bd60c0fc8.json create mode 100644 backend/.sqlx/query-ccd4c7fe5fbdf1ab4bbbbd7d2a9b1ec3fc844ed286ff563f0c9a5c4778cfb718.json create mode 100644 backend/.sqlx/query-d1fdfabd4f3bbe93cf9667e8e76bd8f537e68fabceb094fe535693fbd146eff2.json create mode 100644 frontend/src/lib/components/jobs/batchReruns.ts create mode 100644 frontend/src/lib/components/runs/BatchReRunOptionsPane.svelte create mode 100644 frontend/src/lib/components/runs/RunsBatchActionsDropdown.svelte diff --git a/backend/.sqlx/query-3376b42d273c2499f3517c7754f4cdb6eae2e71e56bc8e5215559af967c81bd8.json b/backend/.sqlx/query-3376b42d273c2499f3517c7754f4cdb6eae2e71e56bc8e5215559af967c81bd8.json new file mode 100644 index 0000000000..7229d4053f --- /dev/null +++ b/backend/.sqlx/query-3376b42d273c2499f3517c7754f4cdb6eae2e71e56bc8e5215559af967c81bd8.json @@ -0,0 +1,79 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT \n j.id,\n j.kind AS \"kind: _\",\n COALESCE(s.path, f.path) AS \"script_path!\",\n COALESCE(s.hash, f.id) AS \"script_hash!: _\",\n COALESCE(jc.started_at, jq.scheduled_for, make_date(1970, 1, 1)) AS \"scheduled_for!: _\",\n args AS input\n FROM v2_job j\n LEFT JOIN script s ON j.runnable_id = s.hash AND j.kind = 'script'\n LEFT JOIN flow_version f ON j.runnable_id = f.id AND j.runnable_path = f.path AND j.kind = 'flow'\n LEFT JOIN v2_job_completed jc ON jc.id = j.id\n LEFT JOIN v2_job_queue jq ON jq.id = j.id\n WHERE j.id = ANY($1)\n AND j.workspace_id = $2\n AND COALESCE(s.hash, f.id) IS NOT NULL\n AND COALESCE(s.path, f.path) IS NOT NULL", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "kind: _", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlescriptflow", + "flowscript", + "flownode", + "appscript" + ] + } + } + } + }, + { + "ordinal": 2, + "name": "script_path!", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "script_hash!: _", + "type_info": "Int8" + }, + { + "ordinal": 4, + "name": "scheduled_for!: _", + "type_info": "Timestamptz" + }, + { + "ordinal": 5, + "name": "input", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "UuidArray", + "Text" + ] + }, + "nullable": [ + false, + false, + null, + null, + null, + true + ] + }, + "hash": "3376b42d273c2499f3517c7754f4cdb6eae2e71e56bc8e5215559af967c81bd8" +} diff --git a/backend/.sqlx/query-5ea1d8c87a17690942ca9e70e0b953e3b32a9a7a73784219d48f8e8932cdb0c8.json b/backend/.sqlx/query-5ea1d8c87a17690942ca9e70e0b953e3b32a9a7a73784219d48f8e8932cdb0c8.json new file mode 100644 index 0000000000..304814543b --- /dev/null +++ b/backend/.sqlx/query-5ea1d8c87a17690942ca9e70e0b953e3b32a9a7a73784219d48f8e8932cdb0c8.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT jsonb_build_object(\n 'kind', jb.kind,\n 'script_path', jb.runnable_path,\n 'latest_schema', COALESCE(\n (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),\n (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = jb.runnable_path AND jb.kind = 'flow')\n ),\n 'schemas', ARRAY(\n SELECT jsonb_build_object(\n 'script_hash', LPAD(TO_HEX(COALESCE(s.hash, f.id)), 16, '0'),\n 'job_ids', ARRAY_AGG(DISTINCT j.id),\n 'schema', ANY_VALUE(COALESCE(s.schema, f.schema))\n ) FROM v2_job j\n LEFT JOIN script s ON s.hash = j.runnable_id AND j.kind = 'script'\n LEFT JOIN flow_version f ON f.id = j.runnable_id AND j.kind = 'flow'\n WHERE j.id = ANY(ARRAY_AGG(jb.id))\n GROUP BY COALESCE(s.hash, f.id)\n )\n ) FROM v2_job jb\n WHERE (jb.kind = 'flow' OR jb.kind = 'script')\n AND jb.workspace_id = $1 AND jb.id = ANY($2)\n GROUP BY jb.kind, jb.runnable_path", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "jsonb_build_object", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "UuidArray" + ] + }, + "nullable": [ + null + ] + }, + "hash": "5ea1d8c87a17690942ca9e70e0b953e3b32a9a7a73784219d48f8e8932cdb0c8" +} diff --git a/backend/.sqlx/query-ad88831c9021b79c9037a925c2b53be3a93dbe2a67bdc4d4342c84f439b5eaf9.json b/backend/.sqlx/query-ad88831c9021b79c9037a925c2b53be3a93dbe2a67bdc4d4342c84f439b5eaf9.json new file mode 100644 index 0000000000..8d51b798e9 --- /dev/null +++ b/backend/.sqlx/query-ad88831c9021b79c9037a925c2b53be3a93dbe2a67bdc4d4342c84f439b5eaf9.json @@ -0,0 +1,85 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT \n j.id,\n j.kind AS \"kind: _\",\n COALESCE(s.path, f.path) AS \"script_path!\",\n COALESCE(s.hash, f.id) AS \"script_hash!: _\",\n COALESCE(jc.started_at, jq.scheduled_for, make_date(1970, 1, 1)) AS \"scheduled_for!: _\",\n args AS input,\n COALESCE(s.schema, f.schema) AS \"schema: _\"\n FROM v2_job j\n LEFT JOIN script s ON j.runnable_id = s.hash AND j.kind = 'script'\n LEFT JOIN flow_version f ON j.runnable_id = f.id AND j.runnable_path = f.path AND j.kind = 'flow'\n LEFT JOIN v2_job_completed jc ON jc.id = j.id\n LEFT JOIN v2_job_queue jq ON jq.id = j.id\n WHERE j.id = ANY($1)\n AND j.workspace_id = $2\n AND COALESCE(s.hash, f.id) IS NOT NULL\n AND COALESCE(s.path, f.path) IS NOT NULL", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "kind: _", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlescriptflow", + "flowscript", + "flownode", + "appscript" + ] + } + } + } + }, + { + "ordinal": 2, + "name": "script_path!", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "script_hash!: _", + "type_info": "Int8" + }, + { + "ordinal": 4, + "name": "scheduled_for!: _", + "type_info": "Timestamptz" + }, + { + "ordinal": 5, + "name": "input", + "type_info": "Jsonb" + }, + { + "ordinal": 6, + "name": "schema: _", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "UuidArray", + "Text" + ] + }, + "nullable": [ + false, + false, + null, + null, + null, + true, + null + ] + }, + "hash": "ad88831c9021b79c9037a925c2b53be3a93dbe2a67bdc4d4342c84f439b5eaf9" +} diff --git a/backend/.sqlx/query-b61a00e6a8ca0a1d24e64fdc9223b077c957da8455ea747b0680923d8282425b.json b/backend/.sqlx/query-b61a00e6a8ca0a1d24e64fdc9223b077c957da8455ea747b0680923d8282425b.json new file mode 100644 index 0000000000..80f0549739 --- /dev/null +++ b/backend/.sqlx/query-b61a00e6a8ca0a1d24e64fdc9223b077c957da8455ea747b0680923d8282425b.json @@ -0,0 +1,73 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n j.kind AS \"kind!: JobKind\",\n j.runnable_id AS \"script_hash: _\",\n j.runnable_path AS script_path,\n COUNT(*) AS \"count!\",\n ANY_VALUE(COALESCE(f.schema, s.schema)) AS schema\n FROM v2_job j\n LEFT JOIN script s ON s.hash = j.runnable_id AND j.kind = 'script'\n LEFT JOIN flow_version f ON f.id = j.runnable_id AND f.path = j.runnable_path AND j.kind = 'flow'\n WHERE COALESCE(s.hash, f.id) IS NOT NULL\n AND COALESCE(s.path, f.path) IS NOT NULL\n AND j.workspace_id = $1 AND j.id = ANY($2)\n GROUP BY j.runnable_id, j.runnable_path, j.kind", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "kind!: JobKind", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlescriptflow", + "flowscript", + "flownode", + "appscript" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "script_hash: _", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "script_path", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "count!", + "type_info": "Int8" + }, + { + "ordinal": 4, + "name": "schema", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Text", + "UuidArray" + ] + }, + "nullable": [ + false, + true, + true, + null, + null + ] + }, + "hash": "b61a00e6a8ca0a1d24e64fdc9223b077c957da8455ea747b0680923d8282425b" +} diff --git a/backend/.sqlx/query-c4f382045e5c47986e02f1e57667e6b3cb1d60f15fed9a24e3f05b1bd60c0fc8.json b/backend/.sqlx/query-c4f382045e5c47986e02f1e57667e6b3cb1d60f15fed9a24e3f05b1bd60c0fc8.json new file mode 100644 index 0000000000..e976cd0dd9 --- /dev/null +++ b/backend/.sqlx/query-c4f382045e5c47986e02f1e57667e6b3cb1d60f15fed9a24e3f05b1bd60c0fc8.json @@ -0,0 +1,73 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT ON (j.runnable_path, j.kind) \n j.kind AS \"kind!: JobKind\",\n j.runnable_path AS script_path,\n NULL as \"script_hash: _\",\n -1::bigint as \"count!: _\",\n COALESCE(f.schema, s.schema) AS schema\n FROM v2_job j\n LEFT JOIN script s ON s.path = j.runnable_path AND j.kind = 'script'\n LEFT JOIN flow_version f ON f.path = j.runnable_path AND j.kind = 'flow'\n WHERE COALESCE(s.hash, f.id) IS NOT NULL\n AND j.workspace_id = $1 AND j.id = ANY($2)\n ORDER BY j.runnable_path, j.kind, COALESCE(f.created_at, s.created_at) DESC", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "kind!: JobKind", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlescriptflow", + "flowscript", + "flownode", + "appscript" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "script_path", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "script_hash: _", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "count!: _", + "type_info": "Int8" + }, + { + "ordinal": 4, + "name": "schema", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Text", + "UuidArray" + ] + }, + "nullable": [ + false, + true, + null, + null, + null + ] + }, + "hash": "c4f382045e5c47986e02f1e57667e6b3cb1d60f15fed9a24e3f05b1bd60c0fc8" +} diff --git a/backend/.sqlx/query-ccd4c7fe5fbdf1ab4bbbbd7d2a9b1ec3fc844ed286ff563f0c9a5c4778cfb718.json b/backend/.sqlx/query-ccd4c7fe5fbdf1ab4bbbbd7d2a9b1ec3fc844ed286ff563f0c9a5c4778cfb718.json new file mode 100644 index 0000000000..a457b8b2bf --- /dev/null +++ b/backend/.sqlx/query-ccd4c7fe5fbdf1ab4bbbbd7d2a9b1ec3fc844ed286ff563f0c9a5c4778cfb718.json @@ -0,0 +1,67 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT j.kind AS \"kind: _\", COALESCE(s.path, f.path) AS \"script_path!\", COALESCE(s.hash, f.id) AS \"script_hash!: _\", args\n FROM v2_job j\n LEFT JOIN script s ON j.runnable_id = s.hash AND j.kind = 'script'\n LEFT JOIN flow_version f ON j.runnable_id = f.id AND j.runnable_path = f.path AND j.kind = 'flow'\n WHERE j.id = ANY($1)\n AND j.workspace_id = $2\n AND COALESCE(s.hash, f.id) IS NOT NULL\n AND COALESCE(s.path, f.path) IS NOT NULL", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "kind: _", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlescriptflow", + "flowscript", + "flownode", + "appscript" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "script_path!", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "script_hash!: _", + "type_info": "Int8" + }, + { + "ordinal": 3, + "name": "args", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "UuidArray", + "Text" + ] + }, + "nullable": [ + false, + null, + null, + true + ] + }, + "hash": "ccd4c7fe5fbdf1ab4bbbbd7d2a9b1ec3fc844ed286ff563f0c9a5c4778cfb718" +} diff --git a/backend/.sqlx/query-d1fdfabd4f3bbe93cf9667e8e76bd8f537e68fabceb094fe535693fbd146eff2.json b/backend/.sqlx/query-d1fdfabd4f3bbe93cf9667e8e76bd8f537e68fabceb094fe535693fbd146eff2.json new file mode 100644 index 0000000000..5ebf571855 --- /dev/null +++ b/backend/.sqlx/query-d1fdfabd4f3bbe93cf9667e8e76bd8f537e68fabceb094fe535693fbd146eff2.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COALESCE(\n (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),\n (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = jb.runnable_path AND jb.kind = 'flow')\n ) FROM v2_job jb\n WHERE jb.id = $1 AND jb.workspace_id = $2\n GROUP BY jb.kind, jb.runnable_path", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "coalesce", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "d1fdfabd4f3bbe93cf9667e8e76bd8f537e68fabceb094fe535693fbd146eff2" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 4737d318f6..0f9558966f 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -13941,6 +13941,7 @@ dependencies = [ "tikv-jemalloc-sys", "tikv-jemallocator", "tokio", + "tokio-stream", "tracing", "url", "uuid", @@ -13983,6 +13984,8 @@ dependencies = [ "cookie 0.17.0", "cron", "datafusion", + "deno_core", + "deno_error", "futures", "git-version", "google-cloud-googleapis", @@ -14034,6 +14037,7 @@ dependencies = [ "tokio", "tokio-native-tls", "tokio-postgres 0.7.11", + "tokio-stream", "tokio-tar", "tokio-tungstenite", "tokio-util", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index ef4d7c5712..ed4155f495 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -64,6 +64,7 @@ cloud = ["windmill-queue/cloud", "windmill-worker/cloud", "windmill-common/cloud jemalloc = ["windmill-common/jemalloc", "dep:tikv-jemallocator", "dep:tikv-jemalloc-sys", "dep:tikv-jemalloc-ctl"] tantivy = ["dep:windmill-indexer", "windmill-api/tantivy", "windmill-indexer/enterprise", "windmill-indexer/parquet", "windmill-common/tantivy", "enterprise", "parquet"] sqlx = ["windmill-worker/sqlx"] +deno_core = ["windmill-worker/deno_core", "windmill-api/deno_core", "dep:deno_core", "dep:v8"] kafka = ["windmill-api/kafka"] nats = ["windmill-api/nats"] otel = ["windmill-common/otel", "windmill-worker/otel"] @@ -82,7 +83,6 @@ static_frontend = ["windmill-api/static_frontend"] scoped_cache = ["windmill-common/scoped_cache"] # Languages python = ["windmill-worker/python"] -deno_core = ["windmill-worker/deno_core", "dep:deno_core", "dep:v8"] rust = ["windmill-worker/rust"] mysql = ["windmill-worker/mysql"] oracledb = ["windmill-worker/oracledb"] @@ -98,6 +98,7 @@ all_languages = [ "python", "deno_core", "rust", "mysql", "oracledb", "mssql", " [dependencies] anyhow.workspace = true tokio.workspace = true +tokio-stream.workspace = true dotenv.workspace = true windmill-queue.workspace = true windmill-common = { workspace = true, default-features = false } @@ -181,6 +182,7 @@ axum = { version = "^0.7", features = ["multipart"] } headers = "^0" hyper = { version = "^1", features = ["full"] } tokio = { version = "^1.42.0", features = ["full", "tracing", "time"] } +tokio-stream = { version = "0.1.17" } tower = "^0" tower-http = { version = "^0.6", features = ["trace", "cors"] } tower-cookies = "^0.10" diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index c3d695f96e..0996b5956d 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -32,6 +32,7 @@ static_frontend = ["dep:rust-embed"] postgres_trigger = ["dep:rust-postgres", "dep:pg_escape", "dep:byteorder", "dep:thiserror", "dep:rust_decimal", "dep:rust-postgres-native-tls"] mqtt_trigger = ["dep:thiserror", "dep:rumqttc"] sqs_trigger = ["dep:aws-sdk-sqs", "dep:thiserror", "dep:aws-config"] +deno_core = ["dep:deno_core", "dep:deno_error"] gcp_trigger = ["dep:thiserror", "dep:google-cloud-pubsub", "dep:google-cloud-googleapis", "dep:tonic"] cloud = ["windmill-common/cloud"] @@ -47,6 +48,7 @@ windmill-git-sync.workspace = true windmill-indexer = { workspace = true, optional = true } windmill-worker.workspace = true tokio.workspace = true +tokio-stream.workspace = true anyhow.workspace = true argon2.workspace = true axum.workspace = true @@ -133,4 +135,9 @@ aws-sdk-sqs = { workspace = true, optional = true } aws-config = { workspace = true, optional = true } google-cloud-pubsub = { workspace = true, optional = true } google-cloud-googleapis = { workspace = true , optional = true } -tonic = { workspace = true, optional = true } \ No newline at end of file +tonic = { workspace = true, optional = true } +deno_error = { workspace = true, optional = true } +deno_core = { workspace = true, optional = true } + +[build-dependencies] +deno_core = { workspace = true, optional = true } \ No newline at end of file diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index f1c3e0a195..5ed6537931 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -1378,7 +1378,7 @@ paths: format: int64 description: The ID of the GitHub installation to delete responses: - '200': + "200": description: Installation successfully deleted /w/{workspace}/github_app/export/{installationId}: @@ -1400,7 +1400,7 @@ paths: schema: type: integer responses: - '200': + "200": description: Successfully exported the JWT token content: application/json: @@ -1435,7 +1435,7 @@ paths: jwt_token: type: string responses: - '200': + "200": description: Successfully imported the installation /users/accept_invite: @@ -1791,7 +1791,7 @@ paths: schema: $ref: "#/components/schemas/OperatorSettings" responses: - '200': + "200": description: Operator settings updated successfully content: text/plain: @@ -1997,7 +1997,7 @@ paths: required: - premium - owner - + /w/{workspace}/workspaces/threshold_alert: get: summary: get threshold alert info @@ -2368,7 +2368,6 @@ paths: application/json: schema: $ref: "#/components/schemas/AIConfig" - /w/{workspace}/workspaces/edit_error_handler: post: @@ -3538,14 +3537,14 @@ paths: tags: - teams responses: - '200': + "200": description: Teams information successfully synchronized content: application/json: schema: type: array items: - $ref: '#/components/schemas/TeamInfo' + $ref: "#/components/schemas/TeamInfo" /teams/activities: post: @@ -3579,7 +3578,7 @@ paths: description: The card block to be sent in the Teams card responses: - '200': + "200": description: Activity processed successfully /w/{workspace}/resources/create: @@ -4946,8 +4945,7 @@ paths: /scripts_u/tokened_raw/{workspace}/{token}/{path}: get: - summary: - raw script by path with a token (mostly used by lsp to be used with + summary: raw script by path with a token (mostly used by lsp to be used with import maps to resolve scripts) operationId: rawScriptByPathTokened tags: @@ -5041,6 +5039,63 @@ paths: lock_error_logs: type: string + /w/{workspace}/jobs/list_selected_job_groups: + # We use post because sending a huge array as a query param can produce + # URLs that may be too long + post: + summary: list selected jobs script/flow schemas grouped by (kind, path) + operationId: listSelectedJobGroups + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: script args + required: true + content: + application/json: + schema: + type: array + items: + type: string + format: uuid + responses: + "200": + description: result + content: + text/plain: + schema: + type: array + items: + type: object + properties: + kind: + type: string + enum: ["script", "flow"] + script_path: + type: string + latest_schema: + type: object + schemas: + type: array + items: + type: object + properties: + schema: + type: object + script_hash: + type: string + job_ids: + type: array + items: + type: string + required: [schema, script_hash, job_ids] + required: + - kind + - script_path + - latest_schema + - schemas + /w/{workspace}/jobs/run/p/{path}: post: summary: run script by path @@ -5525,7 +5580,6 @@ paths: lock_error_logs: type: string - /w/{workspace}/flows/get_triggers_count/{path}: get: summary: get triggers count of flow @@ -6523,7 +6577,6 @@ paths: in: query schema: type: boolean - requestBody: description: flow args required: true @@ -6531,7 +6584,6 @@ paths: application/json: schema: $ref: "#/components/schemas/ScriptArgs" - responses: "201": description: job created @@ -6541,6 +6593,60 @@ paths: type: string format: uuid + /w/{workspace}/jobs/run/batch_rerun_jobs: + post: + summary: re-run multiple jobs + operationId: batchReRunJobs + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: list of job ids to re run and arg tranforms + required: true + content: + application/json: + schema: + type: object + required: [job_ids, script_options_by_path, flow_options_by_path] + properties: + job_ids: + type: array + items: + type: string + script_options_by_path: + type: object + additionalProperties: + type: object + properties: + input_transforms: + type: object + additionalProperties: + $ref: "#/components/schemas/InputTransform" + use_latest_version: boolean + flow_options_by_path: + type: object + additionalProperties: + type: object + properties: + input_transforms: + type: object + additionalProperties: + $ref: "#/components/schemas/InputTransform" + use_latest_version: boolean + responses: + "201": + description: stream of created job uuids separated by \n. Lines may start with 'Error:' + example: | + a1a74c0d-708e-4539-9768-e8b3d37996bd + f0949132-5b30-48fe-bac8-873f047df810 + Error: Could not re-run 0b885808-ae89-4458-af95-c1ca3a13b0a5 + 52b9c01d-1125-4bbb-8bee-d41f26b70066 + content: + text/event-stream: + schema: + type: string + /w/{workspace}/jobs/restart/f/{id}/from/{step_id}/{branch_or_iteration_n}: post: summary: restart a completed flow at a given step @@ -6557,8 +6663,7 @@ paths: schema: type: string - name: branch_or_iteration_n - description: - for branchall or loop, the iteration at which the flow should + description: for branchall or loop, the iteration at which the flow should restart required: true in: path @@ -6926,10 +7031,82 @@ paths: schema: type: integer - /w/{workspace}/jobs/queue/list_filtered_uuids: + /w/{workspace}/jobs/list_filtered_uuids: get: summary: get the ids of all jobs matching the given filters - operationId: listFilteredUuids + operationId: listFilteredJobsUuids + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/CreatedBy" + - $ref: "#/components/parameters/Label" + - $ref: "#/components/parameters/Worker" + - $ref: "#/components/parameters/ParentJob" + - $ref: "#/components/parameters/ScriptExactPath" + - $ref: "#/components/parameters/ScriptStartPath" + - $ref: "#/components/parameters/SchedulePath" + - $ref: "#/components/parameters/ScriptExactHash" + - $ref: "#/components/parameters/StartedBefore" + - $ref: "#/components/parameters/StartedAfter" + - $ref: "#/components/parameters/CreatedBefore" + - $ref: "#/components/parameters/CreatedAfter" + - $ref: "#/components/parameters/CreatedOrStartedBefore" + - $ref: "#/components/parameters/Running" + - $ref: "#/components/parameters/ScheduledForBeforeNow" + - $ref: "#/components/parameters/CreatedOrStartedAfter" + - $ref: "#/components/parameters/CreatedOrStartedAfterCompletedJob" + - $ref: "#/components/parameters/JobKinds" + - $ref: "#/components/parameters/Suspended" + - $ref: "#/components/parameters/ArgsFilter" + - $ref: "#/components/parameters/Tag" + - $ref: "#/components/parameters/ResultFilter" + - $ref: "#/components/parameters/Page" + - $ref: "#/components/parameters/PerPage" + - name: is_skipped + description: is the job skipped + in: query + schema: + type: boolean + - name: is_flow_step + description: is the job a flow step + in: query + schema: + type: boolean + - name: has_null_parent + description: has null parent + in: query + schema: + type: boolean + - name: success + description: filter on successful jobs + in: query + schema: + type: boolean + - name: all_workspaces + description: get jobs from all workspaces (only valid if request come from the `admins` workspace) + in: query + schema: + type: boolean + - name: is_not_schedule + description: is not a scheduled job + in: query + schema: + type: boolean + responses: + "200": + description: uuids of jobs + content: + application/json: + schema: + type: array + items: + type: string + + /w/{workspace}/jobs/queue/list_filtered_uuids: + get: + summary: get the ids of all queued jobs matching the given filters + operationId: listFilteredQueueUuids tags: - job parameters: @@ -8739,7 +8916,7 @@ paths: summary: delete nats trigger operationId: deleteNatsTrigger tags: - - nats_trigger + - nats_trigger parameters: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/Path" @@ -8768,7 +8945,6 @@ paths: schema: $ref: "#/components/schemas/NatsTrigger" - /w/{workspace}/nats_triggers/list: get: summary: list nats triggers @@ -8803,7 +8979,6 @@ paths: items: $ref: "#/components/schemas/NatsTrigger" - /w/{workspace}/nats_triggers/exists/{path}: get: summary: does nats trigger exists @@ -8850,7 +9025,6 @@ paths: schema: type: string - /w/{workspace}/nats_triggers/test: post: summary: test NATS connection @@ -9523,8 +9697,8 @@ paths: tags: - postgres_trigger parameters: - - $ref: "#/components/parameters/WorkspaceId" - - $ref: "#/components/parameters/Path" + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/Path" responses: "200": description: boolean that indicates if postgres is set to logical level or not @@ -9727,7 +9901,6 @@ paths: schema: type: string - /w/{workspace}/postgres_triggers/publication/delete/{publication}/{path}: delete: summary: delete postgres publication @@ -10840,7 +11013,8 @@ paths: required: true schema: type: string - enum: [ + enum: + [ script, group_, resource, @@ -11078,7 +11252,6 @@ paths: items: $ref: "#/components/schemas/Capture" - /w/{workspace}/capture/move/{runnable_kind}/{path}: post: summary: move captures and configs for a script or flow @@ -11336,8 +11509,7 @@ paths: /w/{workspace}/job_helpers/duckdb_connection_settings: post: - summary: - Converts an S3 resource to the set of instructions necessary to connect + summary: Converts an S3 resource to the set of instructions necessary to connect DuckDB to an S3 bucket operationId: duckdbConnectionSettings tags: @@ -11366,8 +11538,7 @@ paths: type: string /w/{workspace}/job_helpers/v2/duckdb_connection_settings: post: - summary: - Converts an S3 resource to the set of instructions necessary to connect + summary: Converts an S3 resource to the set of instructions necessary to connect DuckDB to an S3 bucket operationId: duckdbConnectionSettingsV2 tags: @@ -11403,8 +11574,7 @@ paths: /w/{workspace}/job_helpers/polars_connection_settings: post: - summary: - Converts an S3 resource to the set of arguments necessary to connect + summary: Converts an S3 resource to the set of arguments necessary to connect Polars to an S3 bucket operationId: polarsConnectionSettings tags: @@ -11448,8 +11618,7 @@ paths: - client_kwargs /w/{workspace}/job_helpers/v2/polars_connection_settings: post: - summary: - Converts an S3 resource to the set of arguments necessary to connect + summary: Converts an S3 resource to the set of arguments necessary to connect Polars to an S3 bucket operationId: polarsConnectionSettingsV2 tags: @@ -11526,8 +11695,7 @@ paths: parameters: - $ref: "#/components/parameters/WorkspaceId" requestBody: - description: - S3 resource path to use. If empty, the S3 resource defined in the + description: S3 resource path to use. If empty, the S3 resource defined in the workspace settings will be used required: true content: @@ -12571,8 +12739,7 @@ components: type: string ParentJob: name: parent_job - description: - The parent job that is at the origin and responsible for the execution + description: The parent job that is at the origin and responsible for the execution of this script if any in: query schema: @@ -12592,8 +12759,7 @@ components: type: string NewJobId: name: job_id - description: - The job id to assign to the created job. if missing, job is chosen + description: The job id to assign to the created job. if missing, job is chosen randomly using the ULID scheme. If a job id already exists in the queue or as a completed job, the request to create one will fail (Bad Request) in: query @@ -12684,8 +12850,7 @@ components: format: date-time CreatedOrStartedAfter: name: created_or_started_after - description: - filter on created_at for non non started job and started_at otherwise + description: filter on created_at for non non started job and started_at otherwise after (exclusive) timestamp in: query schema: @@ -12693,8 +12858,7 @@ components: format: date-time CreatedOrStartedAfterCompletedJob: name: created_or_started_after_completed_jobs - description: - filter on created_at for non non started job and started_at otherwise + description: filter on created_at for non non started job and started_at otherwise after (exclusive) timestamp but only for the completed jobs in: query schema: @@ -12702,8 +12866,7 @@ components: format: date-time CreatedOrStartedBefore: name: created_or_started_before - description: - filter on created_at for non non started job and started_at otherwise + description: filter on created_at for non non started job and started_at otherwise before (inclusive) timestamp in: query schema: @@ -12791,8 +12954,7 @@ components: enum: [Create, Update, Delete, Execute] JobKinds: name: job_kinds - description: - filter on job kind (values 'preview', 'script', 'dependencies', 'flow') + description: filter on job kind (values 'preview', 'script', 'dependencies', 'flow') separated by, in: query schema: @@ -13200,7 +13362,7 @@ components: "singlescriptflow", "flowscript", "flownode", - "appscript", + "appscript" ] schedule_path: type: string @@ -13307,7 +13469,7 @@ components: "singlescriptflow", "flowscript", "flownode", - "appscript", + "appscript" ] schedule_path: type: string @@ -13757,7 +13919,7 @@ components: "bytes", "dict", "datetime", - "sql", + "sql" ] - type: object properties: @@ -13797,7 +13959,7 @@ components: "bytes", "dict", "datetime", - "sql", + "sql" ] - type: object properties: @@ -13823,7 +13985,7 @@ components: "bytes", "dict", "datetime", - "sql", + "sql" ] - type: object properties: @@ -13855,8 +14017,7 @@ components: ScriptLang: type: string - enum: - [ + enum: [ python3, deno, go, @@ -14293,16 +14454,15 @@ components: - edited_at - is_flow - AuthenticationMethod: type: string enum: - - none - - windmill - - api_key - - basic_http - - custom_script - - signature + - none + - windmill + - api_key + - basic_http + - custom_script + - signature HttpTrigger: allOf: @@ -14644,7 +14804,7 @@ components: MqttQoS: type: string - enum: ['qos0', 'qos1', 'qos2'] + enum: ["qos0", "qos1", "qos2"] MqttV3Config: type: object @@ -15024,7 +15184,6 @@ components: - is_flow - enabled - Slot: type: object properties: @@ -15119,8 +15278,8 @@ components: error: type: string last_server_ping: - type: string - format: date-time + type: string + format: date-time required: - enabled - postgres_resource_path @@ -15916,13 +16075,7 @@ components: properties: type: type: string - enum: - [ - "S3Storage", - "AzureBlobStorage", - "AzureWorkloadIdentity", - "S3AwsOidc", - ] + enum: ["S3Storage", "AzureBlobStorage", "AzureWorkloadIdentity", "S3AwsOidc"] s3_resource_path: type: string azure_blob_resource_path: @@ -15937,12 +16090,7 @@ components: type: type: string enum: - [ - "S3Storage", - "AzureBlobStorage", - "AzureWorkloadIdentity", - "S3AwsOidc", - ] + ["S3Storage", "AzureBlobStorage", "AzureWorkloadIdentity", "S3AwsOidc"] s3_resource_path: type: string azure_blob_resource_path: @@ -16422,7 +16570,7 @@ components: type: array description: List of channels within the team items: - $ref: '#/components/schemas/ChannelInfo' + $ref: "#/components/schemas/ChannelInfo" ChannelInfo: type: object diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 4111f28b6b..be015f5929 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -8,7 +8,9 @@ use axum::body::Body; use axum::http::HeaderValue; -use futures::TryFutureExt; +#[cfg(feature = "deno_core")] +use deno_core::{op2, serde_v8, v8, JsRuntime, OpState}; +use futures::{StreamExt, TryFutureExt}; use http::{HeaderMap, HeaderName}; use itertools::Itertools; use quick_cache::sync::Cache; @@ -140,6 +142,13 @@ pub fn workspaced_service() -> Router { .layer(cors.clone()) .layer(ce_headers.clone()), ) + .route( + "/run/batch_rerun_jobs", + post(batch_rerun_jobs) + .head(|| async { "" }) + .layer(cors.clone()) + .layer(ce_headers.clone()), + ) .route( "/run/workflow_as_code/:job_id/:entrypoint", post(run_workflow_as_code) @@ -203,6 +212,13 @@ pub fn workspaced_service() -> Router { "/list", get(list_jobs).layer(Extension(api_list_jobs_query_duration)), ) + .route( + "/list_selected_job_groups", + // We use post because sending a huge array as a query param can produce + // URLs that may be too long + post(list_selected_job_groups), + ) + .route("/list_filtered_uuids", get(list_filtered_job_uuids)) .route("/queue/list", get(list_queue_jobs)) .route("/queue/count", get(count_queue_jobs)) .route("/queue/list_filtered_uuids", get(list_filtered_uuids)) @@ -645,6 +661,48 @@ async fn get_flow_job_debug_info( } } +async fn list_selected_job_groups( + authed: ApiAuthed, + Extension(user_db): Extension, + Path(w_id): Path, + Json(uuids): Json>, +) -> error::Result { + let mut tx = user_db.begin(&authed).await?; + + let results = sqlx::query_scalar!( + r#"SELECT jsonb_build_object( + 'kind', jb.kind, + 'script_path', jb.runnable_path, + 'latest_schema', COALESCE( + (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC), + (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = jb.runnable_path AND jb.kind = 'flow') + ), + 'schemas', ARRAY( + SELECT jsonb_build_object( + 'script_hash', LPAD(TO_HEX(COALESCE(s.hash, f.id)), 16, '0'), + 'job_ids', ARRAY_AGG(DISTINCT j.id), + 'schema', ANY_VALUE(COALESCE(s.schema, f.schema)) + ) FROM v2_job j + LEFT JOIN script s ON s.hash = j.runnable_id AND j.kind = 'script' + LEFT JOIN flow_version f ON f.id = j.runnable_id AND j.kind = 'flow' + WHERE j.id = ANY(ARRAY_AGG(jb.id)) + GROUP BY COALESCE(s.hash, f.id) + ) + ) FROM v2_job jb + WHERE (jb.kind = 'flow' OR jb.kind = 'script') + AND jb.workspace_id = $1 AND jb.id = ANY($2) + GROUP BY jb.kind, jb.runnable_path"#, + &w_id, + &uuids + ) + .fetch_all(&mut *tx) + .await?; + + tx.commit().await?; + + Ok(Json(results).into_response()) +} + #[derive(Deserialize)] struct GetJobQuery { pub no_logs: Option, @@ -1745,6 +1803,37 @@ async fn cancel_selection( cancel_jobs(jobs_to_cancel, &db, authed.username.as_str(), w_id.as_str()).await } +async fn list_filtered_job_uuids( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + Query(lq): Query, +) -> error::JsonResult> { + require_admin(authed.is_admin, &authed.username)?; + check_scopes(&authed, || format!("jobs:listjobs"))?; + + let mut sqlb = list_completed_jobs_query( + w_id.as_str(), + None, + 0, + &lq, + &["v2_job.id"], + false, + get_scope_tags(&authed), + ); + let sqlb2 = list_queue_jobs_query( + w_id.as_str(), + &lq.into(), + &["v2_job.id"], + Pagination { page: None, per_page: None }, + false, + get_scope_tags(&authed), + ); + let query = sqlb.union_all(sqlb2.subquery()?).subquery()?; + let ids = sqlx::query_scalar(query.as_str()).fetch_all(&db).await?; + Ok(Json(ids)) +} + async fn list_filtered_uuids( authed: ApiAuthed, Extension(db): Extension, @@ -1901,7 +1990,7 @@ async fn list_jobs( let sqlc = if lq.running.is_none() { Some(list_completed_jobs_query( &w_id, - per_page + offset, + Some(per_page + offset), 0, &ListCompletedQuery { order_desc: Some(true), ..lqc }, UnifiedJob::completed_job_fields(), @@ -1940,7 +2029,9 @@ async fn list_jobs( } else { if sqlc.is_none() { return Err(error::Error::BadRequest( - "cannot specify success, label, created_or_started_before, or started_before with running".to_string(), + "cannot specify success, label, created_or_started_before, or starte + d_before with running" + .to_string(), )); } sqlc.unwrap().limit(per_page).offset(offset).query()? @@ -3152,6 +3243,268 @@ pub async fn check_license_key_valid() -> error::Result<()> { Ok(()) } +use windmill_common::flows::InputTransform; + +#[derive(Deserialize)] +struct BatchReRunJobsBodyArgs { + job_ids: Vec, + script_options_by_path: HashMap, + flow_options_by_path: HashMap, +} + +#[derive(Deserialize)] +struct BatchReRunOptions { + input_transforms: Option>, + use_latest_version: Option, +} + +#[derive(sqlx::FromRow, Serialize, Clone)] +struct BatchReRunQueryReturnType { + id: Uuid, + kind: JobKind, + script_path: String, + script_hash: ScriptHash, + input: serde_json::Value, + scheduled_for: chrono::DateTime, + #[serde(skip_serializing_if = "Option::is_none")] + schema: Option, +} + +#[cfg(feature = "deno_core")] +#[op2] +#[string] +fn get_deno_core_job_value(state: &mut OpState) -> Option { + let obj = state.borrow::(); + let str = serde_json::to_string(&obj).ok()?; + Some(str) +} + +#[cfg(feature = "deno_core")] +async fn batch_rerun_compute_js_expression( + expr: String, + job: BatchReRunQueryReturnType, +) -> error::Result> { + let ext = deno_core::Extension { + name: "batch_rerun_arg_transform_ext", + ops: vec![get_deno_core_job_value()].into(), + ..Default::default() + }; + let mut isolate = + JsRuntime::new(deno_core::RuntimeOptions { extensions: vec![ext], ..Default::default() }); + + { + let op_state = isolate.op_state(); + let mut op_state = op_state.borrow_mut(); + op_state.put(BatchReRunQueryReturnType { schema: None, ..job }); + } + isolate + .execute_script( + "", + "let job = JSON.parse(Deno.core.ops.get_deno_core_job_value());", + ) + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + + // Run user expr + let result = isolate + .execute_script("", expr) + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + let mut scope = isolate.handle_scope(); + let result = v8::Local::new(&mut scope, result); + let result: serde_json::Value = + serde_v8::from_v8(&mut scope, result).map_err(|e| Error::ExecutionErr(e.to_string()))?; + let result = JsonRawValue::from_string(result.to_string())?; + Ok(result) +} + +async fn batch_rerun_jobs( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Path(w_id): Path, + Json(body): Json, +) -> Response { + let stream = batch_rerun_jobs_inner(authed, db, user_db, w_id, body); + + let body = axum::body::Body::from_stream(stream.map(Result::<_, std::convert::Infallible>::Ok)); + + Response::builder() + .status(201) + .header("Content-Type", "text/event-stream") + .header("Cache-Control", "no-cache") + .body(body) + .unwrap() +} + +fn batch_rerun_jobs_inner( + authed: ApiAuthed, + db: DB, + user_db: UserDB, + w_id: String, + body: BatchReRunJobsBodyArgs, +) -> impl futures::Stream { + let (tx, rx) = tokio::sync::mpsc::channel(10); + tokio::spawn(async move { + let mut job_stream = sqlx::query_as!( + BatchReRunQueryReturnType, + r#"SELECT + j.id, + j.kind AS "kind: _", + COALESCE(s.path, f.path) AS "script_path!", + COALESCE(s.hash, f.id) AS "script_hash!: _", + COALESCE(jc.started_at, jq.scheduled_for, make_date(1970, 1, 1)) AS "scheduled_for!: _", + args AS input, + COALESCE(s.schema, f.schema) AS "schema: _" + FROM v2_job j + LEFT JOIN script s ON j.runnable_id = s.hash AND j.kind = 'script' + LEFT JOIN flow_version f ON j.runnable_id = f.id AND j.runnable_path = f.path AND j.kind = 'flow' + LEFT JOIN v2_job_completed jc ON jc.id = j.id + LEFT JOIN v2_job_queue jq ON jq.id = j.id + WHERE j.id = ANY($1) + AND j.workspace_id = $2 + AND COALESCE(s.hash, f.id) IS NOT NULL + AND COALESCE(s.path, f.path) IS NOT NULL"#, + &body.job_ids, + w_id + ).fetch(&db); + while let Some(Ok(job)) = job_stream.next().await { + let job_result = + batch_rerun_handle_job(&job, &authed, &db, &user_db, &w_id, &body).await; + let send_to_stream_result = tx + .send(match job_result { + Ok(uuid) => format!("{}\n", uuid), + Err(err) => format!("Error: {}\n", err.to_string()), + }) + .await; + match send_to_stream_result { + Ok(_) => {} + Err(e) => tracing::error!("Couldn't re-run job {}: {}", job.id, e.to_string()), + } + } + }); + tokio_stream::wrappers::ReceiverStream::new(rx) +} + +async fn batch_rerun_handle_job( + job: &BatchReRunQueryReturnType, + authed: &ApiAuthed, + db: &DB, + user_db: &UserDB, + w_id: &String, + body: &BatchReRunJobsBodyArgs, +) -> error::Result { + let options = if matches!(job.kind, JobKind::Script) { + &body.script_options_by_path + } else { + &body.flow_options_by_path + } + .get(&job.script_path); + + let mut args: HashMap> = serde_json::from_value(job.input.clone())?; + let use_latest_version = options.and_then(|o| o.use_latest_version).unwrap_or(false); + let input_transforms = options + .and_then(|o| o.input_transforms.as_ref()) + .map(|t| t.iter()) + .into_iter() + .flatten(); + + let latest_schema; + let schema = if use_latest_version { + latest_schema = sqlx::query_scalar!( + r#"SELECT COALESCE( + (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC), + (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = jb.runnable_path AND jb.kind = 'flow') + ) FROM v2_job jb + WHERE jb.id = $1 AND jb.workspace_id = $2 + GROUP BY jb.kind, jb.runnable_path"#, + &job.id, + &w_id + ).fetch_optional(db).await?.flatten(); + latest_schema.as_ref() + } else { + job.schema.as_ref() + }; + let schema = schema + .and_then(serde_json::Value::as_object) + .and_then(|s| s.get("properties")) + .and_then(serde_json::Value::as_object); + for (property_name, transform) in input_transforms { + let schema_has_key = schema + .map(|s| s.contains_key(property_name)) + .unwrap_or(false); + if !schema_has_key { + continue; + } + match transform { + InputTransform::Static { value } => { + args.insert(property_name.clone(), value.clone()); + } + InputTransform::Javascript { expr } => { + #[cfg(not(feature = "deno_core"))] + tracing::error!("deno_core feature is not activated, cannot evaluate: {expr}"); + #[cfg(feature = "deno_core")] + args.insert( + property_name.clone(), + batch_rerun_compute_js_expression(expr.clone(), job.clone()).await?, + ); + } + } + } + + // Call appropriate function to push job to queue + match job.kind { + JobKind::Flow => { + let result = run_flow_by_path_inner( + authed.clone(), + db.clone(), + user_db.clone(), + w_id.clone(), + StripPath(job.script_path.clone()), + RunJobQuery { ..Default::default() }, + PushArgsOwned { extra: None, args }, + None, + ) + .await; + if let Ok((_, uuid)) = result { + return Ok(uuid); + } + } + JobKind::Script => { + let result = if use_latest_version { + run_script_by_path_inner( + authed.clone(), + db.clone(), + user_db.clone(), + w_id.clone(), + StripPath(job.script_path.clone()), + RunJobQuery { ..Default::default() }, + PushArgsOwned { extra: None, args }, + None, + ) + .await + } else { + run_job_by_hash_inner( + authed.clone(), + db.clone(), + user_db.clone(), + w_id.clone(), + job.script_hash, + RunJobQuery { ..Default::default() }, + PushArgsOwned { extra: None, args }, + None, + ) + .await + }; + if let Ok((_, uuid)) = result { + return Ok(uuid); + } + } + _ => {} + } + Err(error::Error::ExecutionErr( + format!("Couldn't re-run job {}", job.id).to_string(), + )) +} + pub async fn run_flow_by_path( authed: ApiAuthed, Extension(db): Extension, @@ -5609,7 +5962,7 @@ pub fn filter_list_completed_query( pub fn list_completed_jobs_query( w_id: &str, - per_page: usize, + per_page: Option, offset: usize, lq: &ListCompletedQuery, fields: &[&str], @@ -5620,8 +5973,10 @@ pub fn list_completed_jobs_query( .fields(fields) .order_by("v2_job.created_at", lq.order_desc.unwrap_or(true)) .offset(offset) - .limit(per_page) .clone(); + if let Some(per_page) = per_page { + sqlb.limit(per_page); + } if let Some(tags) = tags { sqlb.and_where_in( @@ -5682,7 +6037,7 @@ async fn list_completed_jobs( let sql = list_completed_jobs_query( &w_id, - per_page, + Some(per_page), offset, &lq, &[ diff --git a/frontend/src/lib/components/FieldHeader.svelte b/frontend/src/lib/components/FieldHeader.svelte index 2bbbacf198..9c465f73fe 100644 --- a/frontend/src/lib/components/FieldHeader.svelte +++ b/frontend/src/lib/components/FieldHeader.svelte @@ -1,8 +1,9 @@
@@ -51,8 +54,9 @@ {/if} {#if !emptyString(simpleTooltip)} - - + + + {simpleTooltip} diff --git a/frontend/src/lib/components/InputTransformForm.svelte b/frontend/src/lib/components/InputTransformForm.svelte index 4d67f10ba3..67f88534c5 100644 --- a/frontend/src/lib/components/InputTransformForm.svelte +++ b/frontend/src/lib/components/InputTransformForm.svelte @@ -29,7 +29,7 @@ import type { InputTransform } from '$lib/gen' import TemplateEditor from './TemplateEditor.svelte' import { setInputCat as computeInputCat, isCodeInjection } from '$lib/utils' - import { FunctionSquare } from 'lucide-svelte' + import { FunctionSquare, InfoIcon } from 'lucide-svelte' import { getResourceTypes } from './resourceTypesStore' import type { FlowCopilotContext } from './copilot/flow' import StepInputGen from './copilot/StepInputGen.svelte' @@ -39,6 +39,9 @@ export let schema: Schema | { properties?: Record; required?: string[] } export let arg: InputTransform | any export let argName: string + export let headerTooltip: string | undefined = undefined + export let headerTooltipIconClass = '' + export let HeaderTooltipIcon = InfoIcon export let extraLib: string = 'missing extraLib' export let inputCheck: boolean = true export let previousModuleId: string | undefined @@ -49,6 +52,7 @@ export let argExtra: Record = {} export let pickableProperties: PickableProperties | undefined = undefined export let enableAi = false + export let hideHelpButton = false let monaco: SimpleEditor | undefined = undefined let monacoTemplate: TemplateEditor | undefined = undefined @@ -70,8 +74,9 @@ const { shouldUpdatePropertyType, exprsToSet } = getContext('FlowCopilotContext') || {} - const { inputMatches, focusProp, propPickerConfig, clearFocus } = + const propPickerWrapperContext: PropPickerWrapperContext | undefined = getContext('PropPickerWrapper') + const { inputMatches, focusProp, propPickerConfig, clearFocus } = propPickerWrapperContext ?? {} function setExpr() { const newArg = $exprsToSet?.[argName] @@ -221,7 +226,7 @@ function onFocus() { focused = true if (isStaticTemplate(inputCat)) { - focusProp(argName, 'append', (path) => { + focusProp?.(argName, 'append', (path) => { const toAppend = `\$\{${path}}` arg.value = `${arg.value ?? ''}${toAppend}` monacoTemplate?.setCode(arg.value) @@ -230,7 +235,7 @@ return false }) } else { - focusProp(argName, 'insert', (path) => { + focusProp?.(argName, 'insert', (path) => { arg.expr = path arg.type = 'javascript' propertyType = 'javascript' @@ -253,7 +258,7 @@ if (propertyType == 'static') { setPropertyType(arg?.value) codeInjectionDetected = checkCodeInjection(arg?.value) != undefined - } else if (propertyType == 'javascript' && focused) { + } else if (propertyType == 'javascript' && focused && inputMatches) { // setPropertyType(arg?.expr) $inputMatches = checkCodeInjection(arg?.expr) } @@ -266,7 +271,7 @@ } function updateFocused(newFocused: boolean) { - if (focusedPrev && !newFocused) { + if (focusedPrev && !newFocused && inputMatches) { $inputMatches = undefined } focusedPrev = focused @@ -304,6 +309,9 @@
- { - if ( - $propPickerConfig?.propName == argName && - $propPickerConfig?.insertionMode == 'connect' - ) { - clearFocus() - } else { - focusProp?.(argName, 'connect', (path) => { - connectProperty(path) - dispatch('change', { argName }) - return true - }) - } - }} - /> + {#if propPickerWrapperContext} + { + if ( + $propPickerConfig?.propName == argName && + $propPickerConfig?.insertionMode == 'connect' + ) { + clearFocus() + } else { + focusProp?.(argName, 'connect', (path) => { + connectProperty(path) + dispatch('change', { argName }) + return true + }) + } + }} + /> + {/if}
{/if} @@ -503,7 +513,7 @@ bind:code={arg.value} fontSize={14} on:change={() => { - dispatch('change', { argName }) + dispatch('change', { argName, arg }) }} /> {/if} @@ -520,7 +530,7 @@ }} shouldDispatchChanges on:change={() => { - dispatch('change', { argName }) + dispatch('change', { argName, arg }) }} label={argName} bind:editor={monaco} @@ -554,20 +564,20 @@ bind:this={monaco} bind:code={arg.expr} on:change={() => { - dispatch('change', { argName }) + dispatch('change', { argName, arg }) }} {extraLib} lang="javascript" shouldBindKey={false} on:focus={() => { focused = true - focusProp(argName, 'insert', (path) => { + focusProp?.(argName, 'insert', (path) => { monaco?.insertAtCursor(path) return false }) }} on:change={() => { - dispatch('change', { argName }) + dispatch('change', { argName, arg }) }} on:blur={() => { focused = false @@ -575,7 +585,9 @@ autoHeight /> - + {#if !hideHelpButton} + + {/if}
{:else} Not recognized input type {argName} ({arg.expr}, {propertyType}) diff --git a/frontend/src/lib/components/S3FilePicker.svelte b/frontend/src/lib/components/S3FilePicker.svelte index 3fd2e3a309..6a184e5ab9 100644 --- a/frontend/src/lib/components/S3FilePicker.svelte +++ b/frontend/src/lib/components/S3FilePicker.svelte @@ -799,7 +799,7 @@ deleteFileFromS3(fileMetadata?.fileKey) }} keyListen={false} - bind:loading={fileDeletionInProgress} + loading={fileDeletionInProgress} >
diff --git a/frontend/src/lib/components/common/confirmationModal/ConfirmationModal.svelte b/frontend/src/lib/components/common/confirmationModal/ConfirmationModal.svelte index 49f6089a63..d334dc8c4c 100644 --- a/frontend/src/lib/components/common/confirmationModal/ConfirmationModal.svelte +++ b/frontend/src/lib/components/common/confirmationModal/ConfirmationModal.svelte @@ -3,14 +3,26 @@ import { createEventDispatcher } from 'svelte' import { fade } from 'svelte/transition' import Button from '../button/Button.svelte' - import { AlertTriangle, CornerDownLeft, Loader2 } from 'lucide-svelte' + import { AlertTriangle, CornerDownLeft, Loader2, RefreshCcw } from 'lucide-svelte' - export let title: string - export let confirmationText: string - export let keyListen: boolean = true - export let loading: boolean = false + type Props = { + title: string + confirmationText: string + keyListen?: boolean + loading?: boolean + open?: boolean + type?: 'danger' | 'reload' + } - export let open: boolean = false + const { + title, + confirmationText, + keyListen = true, + loading = false, + open = false, + type: _type + }: Props = $props() + const type = $derived(_type ?? 'danger') const dispatch = createEventDispatcher() @@ -31,6 +43,27 @@ function fadeFast(node: HTMLElement) { return fade(node, { duration: 100 }) } + + const theme = { + danger: { + Icon: AlertTriangle, + color: 'red', + classes: { + icon: 'text-red-500 dark:text-red-400', + iconWrapper: 'bg-red-100 dark:bg-red-800/50' + } + }, + + reload: { + Icon: RefreshCcw, + color: 'dark', + classes: { + icon: 'text-blue-500 dark:text-blue-400', + iconWrapper: 'bg-blue-100 dark:bg-blue-800/50' + } + } + } satisfies { [type in typeof type]: any } + const Icon = $derived(theme[type].Icon ?? AlertTriangle) @@ -60,15 +93,16 @@ >
- +

{title}

+
@@ -77,14 +111,14 @@ + {/each} + {/await} +
+ + + + + {#if selected} +
+ + Use the job object to access data about the original job + +
+ { + if (!selected) return + ;(options[selected.kind][selected.script_path] ??= {}).use_latest_version = + e.detail as boolean + }} + size="sm" + options={{ + right: 'Always use latest version', + rightTooltip: + selected.kind === 'flow' + ? 'Flow jobs will always run on the latest version of the flow' + : 'Run all jobs with the latest version of the script even if they originally ran an older version' + }} + /> + + + + {@const displayedSchema = selectedUsesLatestSchema + ? (selected.latest_schema as Schema) + : mergeSchemasForBatchReruns(selected.schemas.map((s) => s.schema as Schema))} + {@const extraLib = buildExtraLibForBatchReruns({ + schemas: selected.schemas, + script_path: selected.script_path + })} +
+ {#key [selected, displayedSchema]} + {#each Object.keys(displayedSchema.properties) as propertyName} + { + if (!selected) return + const newArg = e.detail.arg as InputTransform + ;((options[selected.kind][selected.script_path] ??= {}).input_transforms ??= + {})[propertyName] = newArg + }} + argName={propertyName} + schema={displayedSchema} + {extraLib} + previousModuleId={undefined} + pickablepropertyMap={{ + hasResume: false, + previousId: undefined, + priorIds: {}, + flow_input: {} + }} + hideHelpButton + {...propertyAlwaysExists(propertyName, selected) + ? {} + : { + headerTooltip: + 'This property does not exist on all versions of the script. You can handle different cases in the code below', + HeaderTooltipIcon: TriangleAlert, + headerTooltipIconClass: 'text-orange-500' + }} + {...propertyAlwaysHasSameType(propertyName, selected) + ? {} + : { + headerTooltip: + 'This property does not always have the same type depending on the version of the script. You can handle different cases in the code below', + HeaderTooltipIcon: TriangleAlert, + headerTooltipIconClass: 'text-orange-500' + }} + /> + {/each} + {/key} +
+ {/if} + + + +
+
diff --git a/frontend/src/lib/components/runs/RunRow.svelte b/frontend/src/lib/components/runs/RunRow.svelte index 4430708091..32c2e16f3a 100644 --- a/frontend/src/lib/components/runs/RunRow.svelte +++ b/frontend/src/lib/components/runs/RunRow.svelte @@ -7,9 +7,9 @@ msToReadableTime, truncateHash, truncateRev, - isJobCancelable, isFlowPreview, - isScriptPreview + isScriptPreview, + isJobSelectable } from '$lib/utils' import { Badge, Button } from '../common' import ScheduleEditor from '../ScheduleEditor.svelte' @@ -33,6 +33,7 @@ import Portal from '$lib/components/Portal.svelte' import WaitTimeWarning from '../common/waitTimeWarning/WaitTimeWarning.svelte' + import type { RunsSelectionMode } from './RunsBatchActionsDropdown.svelte' const dispatch = createEventDispatcher() @@ -41,7 +42,7 @@ export let containerWidth: number = 0 export let containsLabel: boolean = false export let activeLabel: string | null - export let isSelectingJobsToCancel: boolean = false + export let selectionMode: RunsSelectionMode | false = false let scheduleEditor: ScheduleEditor @@ -61,13 +62,13 @@ )} style="width: {containerWidth}px" on:click={() => { - if (!isSelectingJobsToCancel || isJobCancelable(job)) { + if (!selectionMode || isJobSelectable(selectionMode)(job)) { dispatch('select') } }} >
- {#if isSelectingJobsToCancel && isJobCancelable(job)} + {#if selectionMode && isJobSelectable(selectionMode)(job)}
diff --git a/frontend/src/lib/components/runs/RunsBatchActionsDropdown.svelte b/frontend/src/lib/components/runs/RunsBatchActionsDropdown.svelte new file mode 100644 index 0000000000..1f682c4fa5 --- /dev/null +++ b/frontend/src/lib/components/runs/RunsBatchActionsDropdown.svelte @@ -0,0 +1,92 @@ + + + + +{#if isLoading} + +{:else if selectionMode} +
+ + {/if} + {#if selectionMode == 're-run'} + + {/if} +
+{:else} + onSetSelectionMode('cancel') + }, + ...($userStore?.is_admin || $superadmin + ? [{ displayName: 'Cancel all jobs matching filters', action: onCancelFilteredJobs }] + : []), + { + displayName: 'Select jobs to re-run', + action: () => onSetSelectionMode('re-run') + }, + ...($userStore?.is_admin || $superadmin + ? [{ displayName: 'Re-run all jobs matching filters', action: onReRunFilteredJobs }] + : []) + ]} + > + +
+ Batch actions + +
+ + +{/if} diff --git a/frontend/src/lib/components/runs/RunsTable.svelte b/frontend/src/lib/components/runs/RunsTable.svelte index 112a9aa5e6..17171bb669 100644 --- a/frontend/src/lib/components/runs/RunsTable.svelte +++ b/frontend/src/lib/components/runs/RunsTable.svelte @@ -8,14 +8,15 @@ import Popover from '../Popover.svelte' import { workspaceStore } from '$lib/stores' import { twMerge } from 'tailwind-merge' - import { isJobCancelable } from '$lib/utils' + import { isJobSelectable } from '$lib/utils' + import type { RunsSelectionMode } from './RunsBatchActionsDropdown.svelte' //import InfiniteLoading from 'svelte-infinite-loading' export let jobs: Job[] | undefined = undefined export let externalJobs: Job[] = [] export let omittedObscuredJobs: boolean export let showExternalJobs: boolean = false - export let isSelectingJobsToCancel: boolean = false + export let selectionMode: RunsSelectionMode | false = false export let selectedIds: string[] = [] export let selectedWorkspace: string | undefined = undefined export let activeLabel: string | null = null @@ -143,17 +144,20 @@ let allSelected: boolean = false function selectAll() { + if (!selectionMode) return if (allSelected) { allSelected = false selectedIds = [] } else { allSelected = true - selectedIds = jobs?.filter(isJobCancelable).map((j) => j.id) ?? [] + selectedIds = jobs?.filter(isJobSelectable(selectionMode)).map((j) => j.id) ?? [] } } - let cancelableJobCount: number = 0 - $: isSelectingJobsToCancel && (allSelected = selectedIds.length === cancelableJobCount) - $: isSelectingJobsToCancel && (cancelableJobCount = jobs?.filter(isJobCancelable).length ?? 0) + $: selectionMode && (allSelected = selectedIds.length === selectableJobCount) + + let selectableJobCount: number = 0 + $: selectionMode && + (selectableJobCount = jobs?.filter(isJobSelectable(selectionMode)).length ?? 0) function jobCountString(jobCount: number | undefined, lastFetchWentToEnd: boolean): string { if (jobCount === undefined) { @@ -195,7 +199,7 @@ bind:clientWidth={containerWidth} >
- {#if isSelectingJobsToCancel && cancelableJobCount != 0} + {#if selectionMode && selectableJobCount}
{ const jobId = jobOrDate.job.id - if (isSelectingJobsToCancel) { + if (selectionMode) { if (selectedIds.includes(jobOrDate.job.id)) { selectedIds = selectedIds.filter((id) => id != jobId) } else { diff --git a/frontend/src/lib/schema.ts b/frontend/src/lib/schema.ts index ae23a678ba..b01c18a628 100644 --- a/frontend/src/lib/schema.ts +++ b/frontend/src/lib/schema.ts @@ -1,15 +1,17 @@ -import type { Schema } from './common' +import type { Schema, SchemaProperty } from './common' -export function schemaToTsType(schema: Schema): string { - if (!schema || !schema.properties) { +export function schemaToTsType(schema: Schema | SchemaProperty): string { + const schemaProperties = schema.properties + const schemaRequired = schema.required + if (!schema || !schemaProperties) { return 'any' } - const propKeys = Object.keys(schema.properties) + const propKeys = Object.keys(schemaProperties) const types = propKeys .map((key: string) => { - const prop = schema.properties[key] - const isOptional = !schema.required.includes(key) + const prop = schemaProperties[key] + const isOptional = !schemaRequired?.includes(key) const prefix = `${key}${isOptional ? '?' : ''}` let type: string = 'any' if (prop.type === 'string') { @@ -19,11 +21,13 @@ export function schemaToTsType(schema: Schema): string { } else if (prop.type === 'boolean') { type = 'boolean' } else if (prop.type === 'array') { - let type = prop.items?.type ?? 'any' + type = prop.items?.type ?? 'any' if (type === 'integer') { type = 'number' } type = `${type}[]` + } else if (prop.type === 'object' && prop.properties) { + type = schemaToTsType(prop) } return `${prefix}: ${type}` diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts index 1e0d9cf6dc..fd1c26bb69 100644 --- a/frontend/src/lib/utils.ts +++ b/frontend/src/lib/utils.ts @@ -16,10 +16,22 @@ import type { EnumType, SchemaProperty } from './common' import type { Schema } from './common' export { sendUserToast } import type { AnyMeltElement } from '@melt-ui/svelte' +import type { RunsSelectionMode } from './components/runs/RunsBatchActionsDropdown.svelte' export function isJobCancelable(j: Job): boolean { return j.type === 'QueuedJob' && !j.schedule_path && !j.canceled } +export function isJobReRunnable(j: Job): boolean { + return (j.job_kind === 'script' || j.job_kind === 'flow') && j.parent_job === undefined +} + +export function isJobSelectable(selectionType: RunsSelectionMode) { + const f: (j: Job) => boolean = { + cancel: isJobCancelable, + 're-run': isJobReRunnable + }[selectionType] + return f +} export function validateUsername(username: string): string { if (username != '' && !/^[a-zA-Z]\w+$/.test(username)) { @@ -85,7 +97,7 @@ export function displayDate( ? { day: 'numeric', month: 'numeric' - } + } : {} return date.toLocaleString(undefined, { ...timeChoices, @@ -229,7 +241,7 @@ export function clickOutside( } } - const capture = typeof options === 'boolean' ? options : options?.capture ?? true + const capture = typeof options === 'boolean' ? options : (options?.capture ?? true) document.addEventListener('click', handleClick, capture ?? true) return { @@ -605,6 +617,10 @@ export function pluralize(quantity: number, word: string, customPlural?: string) } } +export function addDeterminant(word: string): string { + return (/^[aeiou]/i.test(word) ? 'an ' : 'a ') + word +} + export function capitalize(word: string): string { return word ? word.charAt(0).toUpperCase() + word.slice(1) : '' } diff --git a/frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte b/frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte index 11fc0732ac..4f57d3d1ed 100644 --- a/frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte @@ -7,12 +7,13 @@ FolderService, ScriptService, FlowService, - type ExtendedJobs + type ExtendedJobs, + OpenAPI } from '$lib/gen' import { page } from '$app/stores' import { sendUserToast } from '$lib/toast' - import { superadmin, userStore, workspaceStore, userWorkspaces } from '$lib/stores' + import { userStore, workspaceStore, userWorkspaces } from '$lib/stores' import { Button, Drawer, DrawerContent, Skeleton } from '$lib/components/common' import RunChart from '$lib/components/RunChart.svelte' @@ -31,19 +32,28 @@ import { twMerge } from 'tailwind-merge' import ManuelDatePicker from '$lib/components/runs/ManuelDatePicker.svelte' import JobLoader from '$lib/components/runs/JobLoader.svelte' - import { AlertTriangle, Calendar, Check, ChevronDown, Clock, X } from 'lucide-svelte' + import { AlertTriangle, Calendar, ChevronDown, Clock } from 'lucide-svelte' import ConcurrentJobsChart from '$lib/components/ConcurrentJobsChart.svelte' import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte' import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte' import DropdownV2 from '$lib/components/DropdownV2.svelte' import { goto } from '$app/navigation' import { base } from '$app/paths' - import { isJobCancelable } from '$lib/utils' + import type { RunsSelectionMode } from '$lib/components/runs/RunsBatchActionsDropdown.svelte' + import RunsBatchActionsDropdown from '$lib/components/runs/RunsBatchActionsDropdown.svelte' + import { isJobSelectable } from '$lib/utils' + import BatchReRunOptionsPane, { + type BatchReRunOptions + } from '$lib/components/runs/BatchReRunOptionsPane.svelte' let jobs: Job[] | undefined let selectedIds: string[] = [] + let loadingSelectedIds = false + $: loadingSelectedIds && selectedIds.length && setTimeout(() => (loadingSelectedIds = false), 250) let selectedWorkspace: string | undefined = undefined + let batchReRunOptions: BatchReRunOptions = { flow: {}, script: {} } + // All Filters // Filter by let path: string | null = $page.params.path @@ -158,9 +168,17 @@ let selectedManualDate = 0 let autoRefresh: boolean = getAutoRefresh() let runDrawer: Drawer - let isCancelingVisibleJobs = false - let isCancelingFilteredJobs = false let lookback: number = 1 + let askingForConfirmation: + | undefined + | { + title: string + confirmBtnText: string + loading?: boolean + preContent?: string + onConfirm?: () => void + type?: ConfirmationModal['$$prop_def']['type'] + } = undefined function getAutoRefresh() { try { @@ -353,8 +371,8 @@ lastFetchWentToEnd = false selectedManualDate = 0 selectedIds = [] - jobIdsToCancel = [] - isSelectingJobsToCancel = false + batchReRunOptions = { flow: {}, script: {} } + selectionMode = false selectedWorkspace = undefined jobLoader?.loadJobs(minTs, maxTs, true) } @@ -484,22 +502,29 @@ } } - let jobIdsToCancel: string[] = [] - let isSelectingJobsToCancel = false - let fetchingFilteredJobs = false - let selectedFiltersString: string | undefined = undefined + let selectionMode: RunsSelectionMode | false = false - async function cancelVisibleJobs() { - isSelectingJobsToCancel = true - selectedIds = jobs?.filter(isJobCancelable).map((j) => j.id) ?? [] - if (selectedIds.length === 0) { - sendUserToast('There are no visible jobs that can be canceled', true) + async function onSetSelectionMode(mode: RunsSelectionMode | false) { + selectionMode = mode + if (!mode) { + selectedIds = [] + batchReRunOptions = { flow: {}, script: {} } + return + } + const selectableIds = jobs?.filter(isJobSelectable(mode)).map((j) => j.id) ?? [] + selectedIds = [] + + if (!selectableIds?.length) { + sendUserToast( + 'There are no visible jobs that can be ' + + { cancel: 'cancelled', 're-run': 're-ran' }[mode], + true + ) } } - async function cancelFilteredJobs() { - isCancelingFilteredJobs = true - fetchingFilteredJobs = true - const selectedFilters = { + + function getSelectedFilters() { + return { workspace: $workspaceStore ?? '', startedBefore: maxTs, startedAfter: minTs, @@ -536,19 +561,148 @@ allWorkspaces: allWorkspaces ? true : undefined, allowWildcards: allowWildcards ? true : undefined } - - selectedFiltersString = JSON.stringify(selectedFilters, null, 4) - jobIdsToCancel = await JobService.listFilteredUuids(selectedFilters) - fetchingFilteredJobs = false } - async function cancelSelectedJobs() { - jobIdsToCancel = selectedIds - isCancelingVisibleJobs = true + async function cancelJobs(uuidsToCancel: string[]) { + const uuids = await JobService.cancelSelection({ + workspace: $workspaceStore ?? '', + requestBody: uuidsToCancel + }) + selectedIds = [] + jobLoader?.loadJobs(minTs, maxTs, true, true) + sendUserToast(`Canceled ${uuids.length} jobs`) + selectionMode = false } - function jobCountString(count: number) { - return `${count} ${count == 1 ? 'job' : 'jobs'}` + async function onCancelFilteredJobs() { + askingForConfirmation = { + title: 'Confirm cancelling all jobs corresponding to the selected filters', + confirmBtnText: 'Loading...', + loading: true + } + + const selectedFilters = getSelectedFilters() + const selectedFiltersString = JSON.stringify(selectedFilters, null, 4) + const jobIdsToCancel = await JobService.listFilteredQueueUuids(selectedFilters) + + askingForConfirmation = { + title: `Confirm cancelling all jobs corresponding to the selected filters (${jobIdsToCancel.length} jobs)`, + confirmBtnText: `Cancel ${jobIdsToCancel.length} jobs that matched the filters`, + preContent: selectedFiltersString, + onConfirm: () => { + cancelJobs(jobIdsToCancel) + } + } + } + + async function onCancelSelectedJobs() { + askingForConfirmation = { + confirmBtnText: `Cancel ${selectedIds.length} jobs`, + title: 'Confirm cancelling the selected jobs', + onConfirm: () => { + cancelJobs(selectedIds) + } + } + } + + async function reRunJobs(jobIdsToReRun: string[]) { + if (!$workspaceStore) return + + if (askingForConfirmation) { + askingForConfirmation.loading = true + } + + const body: Parameters[0]['requestBody'] = { + job_ids: jobIdsToReRun, + script_options_by_path: batchReRunOptions.script, + flow_options_by_path: batchReRunOptions.flow + } + + // workaround because EventSource does not support POST requests + // https://medium.com/@david.richards.tech/sse-server-sent-events-using-a-post-request-without-eventsource-1c0bd6f14425 + const response = await fetch(`${OpenAPI.BASE}/w/${$workspaceStore}/jobs/run/batch_rerun_jobs`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(body) + }) + await new Promise(async (resolve) => { + const reader = response?.body?.pipeThrough(new TextDecoderStream()).getReader() + let reRanUuids: string[] = [] + if (reader) { + while (true) { + const { value, done } = await reader.read() + if (value) { + // It is possible get multiple values at once in case of buffering + const receivedUuids: string[] = [] + for (const line of value.split('\n')) { + if (!line) continue + else if (line.startsWith('Error:')) { + console.error(line) + } else { + receivedUuids.push(line) + } + } + if (receivedUuids.length) { + reRanUuids.push(...receivedUuids) + if (askingForConfirmation) { + askingForConfirmation.confirmBtnText = `${reRanUuids.length}/${jobIdsToReRun.length}` + } + } + } + + if (done || !value) { + if (reRanUuids.length) { + sendUserToast(`Re-ran ${reRanUuids.length}/${jobIdsToReRun.length} jobs`) + } + if (reRanUuids.length !== jobIdsToReRun.length) { + sendUserToast( + `Failed to re-run ${jobIdsToReRun.length - reRanUuids.length} jobs. Check console for details`, + true + ) + // We do not get explicit error from backend if the job script don't exist + for (const jobId of jobIdsToReRun) { + if (reRanUuids.includes(jobId)) continue + console.error('Could not re-run job ' + jobId) + } + } + break + } + } + } + resolve(undefined) + }) + + selectedIds = [] + batchReRunOptions = { flow: {}, script: {} } + jobLoader?.loadJobs(minTs, maxTs, true, true) + selectionMode = false + } + + async function onReRunFilteredJobs() { + const selectedFilters = getSelectedFilters() + selectedIds = [] + loadingSelectedIds = true + + if (jobKindsCat !== 'runs') { + sendUserToast('Batch re-run is only supported for scripts and flows', true) + } + selectedIds = await JobService.listFilteredJobsUuids({ + ...selectedFilters, + jobKinds: 'script,flow' + }) + selectionMode = 're-run' + } + + async function onReRunSelectedJobs() { + const jobIdsToReRun = selectedIds + askingForConfirmation = { + title: `Confirm re-running the selected jobs`, + confirmBtnText: `Re-run ${jobIdsToReRun.length} jobs`, + type: 'reload', + onConfirm: async () => { + await reRunJobs(jobIdsToReRun) + } + } } function setLookback(lookbackInDays: number) { @@ -628,50 +782,25 @@ /> { - isCancelingFilteredJobs = false - let uuids = await JobService.cancelSelection({ - workspace: $workspaceStore ?? '', - requestBody: jobIdsToCancel - }) - jobIdsToCancel = [] - selectedIds = [] - jobLoader?.loadJobs(minTs, maxTs, true, true) - sendUserToast(`Canceled ${uuids.length} jobs`) - isSelectingJobsToCancel = false + const func = askingForConfirmation?.onConfirm + await func?.() + askingForConfirmation = undefined }} - loading={fetchingFilteredJobs} + type={askingForConfirmation?.type} + loading={askingForConfirmation?.loading} on:canceled={() => { - isCancelingFilteredJobs = false + askingForConfirmation = undefined }} > - 
{selectedFiltersString}
+ {#if askingForConfirmation?.preContent} + 
{askingForConfirmation.preContent}
+ {/if} - { - isCancelingVisibleJobs = false - let uuids = await JobService.cancelSelection({ - workspace: $workspaceStore ?? '', - requestBody: jobIdsToCancel - }) - jobIdsToCancel = [] - selectedIds = [] - jobLoader?.loadJobs(minTs, maxTs, true, true) - sendUserToast(`Canceled ${uuids.length} jobs`) - isSelectingJobsToCancel = false - }} - on:canceled={() => { - isCancelingVisibleJobs = false - }} -/> - {#if selectedIds.length === 1} @@ -815,7 +944,7 @@ -
- {#if isSelectingJobsToCancel} -
- -
- {:else if !$userStore?.is_admin && !$superadmin} - - -
- Cancel jobs - -
- - - {:else} - - -
- Cancel jobs - -
- - - {/if} -
+
{/if} - - {#if selectedIds.length === 1} + + {#if selectionMode === 're-run'} + + {:else if selectedIds.length === 1} {#if selectedIds[0] === '-'}
There is no information available for this job
{:else} @@ -1195,7 +1273,7 @@ {/if}
- {#if isSelectingJobsToCancel} -
- -
- {:else if !$userStore?.is_admin && !$superadmin} - - -
- Cancel jobs - -
- - - {:else} - - -
- Cancel jobs - -
- - - {/if} +
@@ -1427,13 +1453,13 @@ externalJobs={externalJobs ?? []} omittedObscuredJobs={extendedJobs?.omitted_obscured_jobs ?? false} showExternalJobs={!graphIsRunsChart} - {isSelectingJobsToCancel} + {selectionMode} bind:selectedIds bind:selectedWorkspace bind:lastFetchWentToEnd on:loadExtra={loadExtra} on:select={() => { - if (!isSelectingJobsToCancel) runDrawer.openDrawer() + if (!selectionMode) runDrawer.openDrawer() }} on:filterByPath={filterByPath} on:filterByUser={filterByUser}