From 49e4b570317818ff744757dc584c3caf6e225ec6 Mon Sep 17 00:00:00 2001 From: Diego Imbert <70353967+diegoimbert@users.noreply.github.com> Date: Fri, 19 Sep 2025 19:30:52 +0200 Subject: [PATCH 1/4] Refresh button in flow picker quick (#6645) --- .../flows/content/FlowInputsQuick.svelte | 7 ++++-- .../flows/map/InsertModuleInner.svelte | 5 ++++ .../flows/pickers/PickHubScriptQuick.svelte | 22 +++++++++++------- .../pickers/WorkspaceScriptPickerQuick.svelte | 23 ++++++++++++++++--- 4 files changed, 44 insertions(+), 13 deletions(-) diff --git a/frontend/src/lib/components/flows/content/FlowInputsQuick.svelte b/frontend/src/lib/components/flows/content/FlowInputsQuick.svelte index a670f83bb6..682d44a7e4 100644 --- a/frontend/src/lib/components/flows/content/FlowInputsQuick.svelte +++ b/frontend/src/lib/components/flows/content/FlowInputsQuick.svelte @@ -38,6 +38,7 @@ kind: 'trigger' | 'script' | 'preprocessor' | 'failure' | 'approval' selectedKind?: 'script' | 'flow' | 'approval' | 'trigger' | 'preprocessor' | 'failure' displayPath?: boolean + refreshCount?: number } let { @@ -51,7 +52,8 @@ small = false, kind, selectedKind = kind, - displayPath = false + displayPath = false, + refreshCount = 0 }: Props = $props() type HubCompletion = { @@ -452,9 +454,9 @@ on:pickScript on:pickFlow {displayPath} + {refreshCount} /> {/if} - {#if selectedKind != 'preprocessor' && selectedKind != 'flow'} {#if (!selected || selected?.kind === 'integrations') && (preFilter === 'hub' || preFilter === 'all')} {#if !selected && preFilter !== 'hub'} @@ -475,6 +477,7 @@ on:pickScript bind:loading {displayPath} + {refreshCount} /> {/if} {/if} diff --git a/frontend/src/lib/components/flows/map/InsertModuleInner.svelte b/frontend/src/lib/components/flows/map/InsertModuleInner.svelte index 7c531a8d87..a027fe71f7 100644 --- a/frontend/src/lib/components/flows/map/InsertModuleInner.svelte +++ b/frontend/src/lib/components/flows/map/InsertModuleInner.svelte @@ -5,6 +5,7 @@ import type { FlowBuilderWhitelabelCustomUi } from '$lib/components/custom_ui' import ToggleHubWorkspaceQuick from '$lib/components/ToggleHubWorkspaceQuick.svelte' import TopLevelNode from '../pickers/TopLevelNode.svelte' + import RefreshButton from '$lib/components/common/button/RefreshButton.svelte' const dispatch = createEventDispatcher() interface Props { @@ -35,6 +36,8 @@ let width = $state(0) let height = $state(0) + let refreshCount = $state(0) + let displayPath = $derived(width > 650 || height > 400) @@ -64,6 +67,7 @@ {#if selectedKind != 'preprocessor' && selectedKind != 'flow'} {/if} + (refreshCount += 1)} />
@@ -165,6 +169,7 @@ {preFilter} {small} {displayPath} + {refreshCount} />
diff --git a/frontend/src/lib/components/flows/pickers/PickHubScriptQuick.svelte b/frontend/src/lib/components/flows/pickers/PickHubScriptQuick.svelte index e9e2686840..feaa82a792 100644 --- a/frontend/src/lib/components/flows/pickers/PickHubScriptQuick.svelte +++ b/frontend/src/lib/components/flows/pickers/PickHubScriptQuick.svelte @@ -1,7 +1,8 @@ @@ -51,6 +53,7 @@ }[] displayPath?: boolean apps?: string[] + refreshCount?: number } let { @@ -61,14 +64,17 @@ appFilter = undefined, items = $bindable([]), displayPath = false, - apps = $bindable([]) + apps = $bindable([]), + refreshCount = 0 }: Props = $props() let allApps: string[] = [] async function getAllApps(filterKind: typeof kind) { try { hubNotAvailable = false - allApps = (await listHubIntegrationsCached({ kind: filterKind })).map((x) => x.name) + allApps = (await listHubIntegrationsCached({ kind: filterKind, refreshCount })).map( + (x) => x.name + ) apps = allApps } catch (err) { console.error('Hub is not available') @@ -79,11 +85,11 @@ } let hubScriptsFilteredPromise = usePromise( - () => listHubScriptsCached({ appFilter, filter, kind }), + () => listHubScriptsCached({ appFilter, filter, kind, refreshCount }), { loadInit: false } ) $effect(() => { - ;[filter, kind, appFilter] + ;[filter, kind, appFilter, refreshCount] hubScriptsFilteredPromise.refresh() }) $effect(() => { @@ -131,7 +137,7 @@ } } $effect(() => { - kind + ;[kind, refreshCount] untrack(() => { getAllApps(kind) }) diff --git a/frontend/src/lib/components/flows/pickers/WorkspaceScriptPickerQuick.svelte b/frontend/src/lib/components/flows/pickers/WorkspaceScriptPickerQuick.svelte index d003af2c30..2f6da65b47 100644 --- a/frontend/src/lib/components/flows/pickers/WorkspaceScriptPickerQuick.svelte +++ b/frontend/src/lib/components/flows/pickers/WorkspaceScriptPickerQuick.svelte @@ -1,4 +1,5 @@ @@ -28,6 +40,7 @@ import BarsStaggered from '$lib/components/icons/BarsStaggered.svelte' import Popover from '$lib/components/Popover.svelte' import { usePromise } from '$lib/svelte5Utils.svelte' + import { get } from 'svelte/store' type Item = { path: string @@ -37,7 +50,8 @@ } let items = usePromise( - async () => await loadItemsCached({ workspace: $workspaceStore!, kind, isTemplate }), + async () => + await loadItemsCached({ workspace: $workspaceStore!, kind, isTemplate, refreshCount }), { loadInit: false } ) @@ -54,6 +68,7 @@ ownerFilter?: | { kind: 'inline' | 'owner' | 'integrations'; name: string | undefined } | undefined + refreshCount?: number } let { @@ -64,7 +79,8 @@ filteredWithOwner = $bindable(undefined), filter = '', owners = $bindable([]), - ownerFilter = $bindable(undefined) + ownerFilter = $bindable(undefined), + refreshCount = 0 }: Props = $props() const dispatch = createEventDispatcher() @@ -88,6 +104,7 @@ } } $effect(() => { + refreshCount $workspaceStore && kind && untrack(() => items.refresh()) }) $effect(() => { From f71f9b089438fc29dfb26a21ec2b8aa5867d0296 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 19 Sep 2025 19:32:09 +0200 Subject: [PATCH 2/4] feat: load for loop jobs timeline directly from for loop flow status (#6646) --- ...e09f3462c8e39cb3291f0d497742ad4763fa3.json | 19 + ...b8192924b4a938d249fae92624cd55e44f488.json | 17 - ...6e749d2c43af837f2b3f3edbf49f979e44082.json | 29 ++ ...9d06da7d81e7177a324e3f3f9af1076b3ab66.json | 27 ++ ...75a7b57946b430ad0839685797a10ac1adfd8.json | 28 ++ ...942addd4581dddc88663b6c2cbae87ec205fc.json | 25 -- ...5518225b1243a9d39ce659b95637c33d0954f.json | 27 ++ ...8254c9a832b4f660ec755f4fa14f6f7bb3353.json | 25 -- backend/windmill-api/openapi.yaml | 30 +- backend/windmill-api/src/jobs.rs | 28 ++ backend/windmill-common/src/flow_status.rs | 63 ++++ backend/windmill-queue/src/jobs.rs | 20 +- .../windmill-worker/src/result_processor.rs | 15 +- backend/windmill-worker/src/worker_flow.rs | 329 +++++++++++++----- cli/src/commands/sync/sync.ts | 136 ++++++-- frontend/openapi-ts-error-1758271586180.log | 28 ++ .../src/lib/components/FlowLogViewer.svelte | 15 +- .../components/FlowLogViewerWrapper.svelte | 12 +- .../lib/components/FlowStatusViewer.svelte | 4 - .../components/FlowStatusViewerInner.svelte | 321 +++++++++++------ .../src/lib/components/FlowTimeline.svelte | 165 +++++---- .../src/lib/components/FlowTimelineBar.svelte | 92 ++--- .../src/lib/components/QueuePosition.svelte | 8 +- .../lib/components/flows/flowModuleNextId.ts | 2 + frontend/src/lib/components/graph/model.ts | 9 +- openflow.openapi.yaml | 13 +- 26 files changed, 1046 insertions(+), 441 deletions(-) create mode 100644 backend/.sqlx/query-0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3.json delete mode 100644 backend/.sqlx/query-1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488.json create mode 100644 backend/.sqlx/query-140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082.json create mode 100644 backend/.sqlx/query-2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66.json create mode 100644 backend/.sqlx/query-8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8.json delete mode 100644 backend/.sqlx/query-96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc.json create mode 100644 backend/.sqlx/query-a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f.json delete mode 100644 backend/.sqlx/query-d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353.json create mode 100644 frontend/openapi-ts-error-1758271586180.log diff --git a/backend/.sqlx/query-0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3.json b/backend/.sqlx/query-0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3.json new file mode 100644 index 0000000000..5fa8d88b2e --- /dev/null +++ b/backend/.sqlx/query-0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3.json @@ -0,0 +1,19 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_status SET flow_status = \n CASE WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN\n JSONB_SET(JSONB_SET(JSONB_SET(\n flow_status,\n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT],\n $4\n ),\n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $5),\n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $6)\n ELSE\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4)\n END\n WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Text", + "Jsonb", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3" +} diff --git a/backend/.sqlx/query-1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488.json b/backend/.sqlx/query-1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488.json deleted file mode 100644 index 39b7179e5c..0000000000 --- a/backend/.sqlx/query-1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_status SET\n flow_status = JSONB_SET(\n flow_status,\n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT],\n $4\n )\n WHERE id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Uuid", - "Text", - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488" -} diff --git a/backend/.sqlx/query-140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082.json b/backend/.sqlx/query-140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082.json new file mode 100644 index 0000000000..1734a545fe --- /dev/null +++ b/backend/.sqlx/query-140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT flow_status->'modules'->$2::int->'flow_jobs_success' as \"flow_jobs_success: Json>>\", flow_status->'modules'->$2::int->'flow_jobs_duration' as \"flow_jobs_duration: Json\"\n FROM v2_job_status WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "flow_jobs_success: Json>>", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "flow_jobs_duration: Json", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Int4" + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082" +} diff --git a/backend/.sqlx/query-2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66.json b/backend/.sqlx/query-2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66.json new file mode 100644 index 0000000000..9deb40619a --- /dev/null +++ b/backend/.sqlx/query-2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_status SET flow_status = \n CASE \n WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN\n JSONB_SET(\n JSONB_SET(JSONB_SET(JSONB_SET(\n flow_status, \n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6),\n ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n )\n ELSE\n JSONB_SET(JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'iterator', 'index'],\n ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n ) END\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "int4", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Int4", + "Uuid", + "Text", + "Jsonb", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [ + null + ] + }, + "hash": "2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66" +} diff --git a/backend/.sqlx/query-8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8.json b/backend/.sqlx/query-8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8.json new file mode 100644 index 0000000000..0d8d788602 --- /dev/null +++ b/backend/.sqlx/query-8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT id, started_at FROM v2_job_queue WHERE id = ANY($1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "started_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [ + false, + true + ] + }, + "hash": "8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8" +} diff --git a/backend/.sqlx/query-96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc.json b/backend/.sqlx/query-96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc.json deleted file mode 100644 index 0f6659264a..0000000000 --- a/backend/.sqlx/query-96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_status SET\n flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'branchall', 'branch'],\n ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb\n )\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "int4", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int4", - "Uuid", - "Text", - "Jsonb" - ] - }, - "nullable": [ - null - ] - }, - "hash": "96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc" -} diff --git a/backend/.sqlx/query-a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f.json b/backend/.sqlx/query-a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f.json new file mode 100644 index 0000000000..1315df082f --- /dev/null +++ b/backend/.sqlx/query-a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_status SET flow_status = \n CASE \n WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN\n JSONB_SET(\n JSONB_SET(JSONB_SET(JSONB_SET(\n flow_status, \n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6),\n ARRAY['modules', $1::TEXT, 'branchall', 'branch'],\n ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb\n )\n ELSE\n JSONB_SET(JSONB_SET(\n flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'branchall', 'branch'],\n ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb\n )\n END\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "int4", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Int4", + "Uuid", + "Text", + "Jsonb", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f" +} diff --git a/backend/.sqlx/query-d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353.json b/backend/.sqlx/query-d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353.json deleted file mode 100644 index 09f24968f3..0000000000 --- a/backend/.sqlx/query-d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_status SET\n flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'iterator', 'index'],\n ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n )\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "int4", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int4", - "Uuid", - "Text", - "Jsonb" - ] - }, - "nullable": [ - null - ] - }, - "hash": "d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353" -} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index c83b1ca3ab..40e3ab25e7 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -8024,6 +8024,34 @@ paths: application/json: schema: {} + /w/{workspace}/jobs_u/queue/get_started_at_by_ids: + post: + summary: get started at by ids + operationId: getStartedAtByIds + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: ids + required: true + content: + application/json: + schema: + type: array + items: + type: string + responses: + "200": + description: started at by ids + content: + application/json: + schema: + type: array + items: + type: string + format: date-time + /w/{workspace}/jobs_u/getupdate/{id}: get: summary: get job updates @@ -18249,7 +18277,7 @@ components: allow: type: string # comma separated permissions : "read,write,delete,list" required: ["pattern", "allow"] - + GitRepositorySettings: type: object properties: diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index b00a08cdb5..1b75070217 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -312,6 +312,7 @@ pub fn workspace_unauthed_service() -> Router { get(get_completed_job_logs_tail), ) .route("/get_args/:id", get(get_args)) + .route("/queue/get_started_at_by_ids", post(get_started_at_by_ids)) .route("/get_flow_debug_info/:id", get(get_flow_job_debug_info)) .route("/completed/get/:id", get(get_completed_job)) .route("/completed/get_result/:id", get(get_completed_job_result)) @@ -1663,6 +1664,32 @@ async fn get_args( } } +async fn get_started_at_by_ids( + Extension(db): Extension, + Json(mut ids): Json>, +) -> JsonResult>>> { + ids.truncate(100); + + let started_at = sqlx::query!( + "SELECT id, started_at FROM v2_job_queue WHERE id = ANY($1)", + ids.as_slice() + ) + .fetch_all(&db) + .await?; + + let as_map = started_at + .iter() + .map(|x| (x.id, x.started_at)) + .collect::>(); + + let mut r = Vec::new(); + for id in ids { + r.push(as_map.get(&id).map(|x| x.clone()).unwrap_or_default()); + } + + Ok(Json(r)) +} + #[derive(Debug, sqlx::FromRow, Serialize)] pub struct ListableCompletedJob { pub r#type: String, @@ -6868,6 +6895,7 @@ fn start_job_update_sse_stream( update.log_offset = None; } } + if let Some(new_stream_offset) = update.stream_offset { if new_stream_offset != stream_offset.unwrap_or(0) { stream_offset = Some(new_stream_offset); diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index 8b16e6f427..8dde5606fd 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/backend/windmill-common/src/flow_status.rs @@ -113,6 +113,50 @@ pub struct FlowCleanupModule { pub flow_jobs_to_clean: Vec, } +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct FlowJobsDuration { + pub started_at: Vec>>, + pub duration_ms: Vec>, +} + +impl FlowJobsDuration { + pub fn set(&mut self, position: Option, value: &Option) { + if let Some(position) = position { + if position >= self.started_at.len() + || position >= self.duration_ms.len() + || value.is_none() + { + return; + } + let value = value.clone().unwrap(); + self.started_at[position] = Some(value.started_at); + self.duration_ms[position] = Some(value.duration_ms); + } + } + + pub fn push(&mut self, value: &Option) { + self.started_at.push(value.as_ref().map(|x| x.started_at)); + self.duration_ms.push(value.as_ref().map(|x| x.duration_ms)); + } + + pub fn new(n: usize) -> Self { + Self { started_at: vec![None; n], duration_ms: vec![None; n] } + } +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct FlowJobDuration { + pub started_at: chrono::DateTime, + pub duration_ms: i64, +} + +impl FlowJobsDuration { + pub fn truncate(&mut self, n: usize) { + self.started_at.truncate(n); + self.duration_ms.truncate(n); + } +} + #[derive(Deserialize)] struct UntaggedFlowStatusModule { #[serde(rename = "type")] @@ -124,6 +168,7 @@ struct UntaggedFlowStatusModule { iterator: Option, flow_jobs: Option>, flow_jobs_success: Option>>, + flow_jobs_duration: Option, branch_chosen: Option, branchall: Option, parallel: Option, @@ -169,6 +214,8 @@ pub enum FlowStatusModule { #[serde(skip_serializing_if = "Option::is_none")] flow_jobs_success: Option>>, #[serde(skip_serializing_if = "Option::is_none")] + flow_jobs_duration: Option, + #[serde(skip_serializing_if = "Option::is_none")] branch_chosen: Option, #[serde(skip_serializing_if = "Option::is_none")] branchall: Option, @@ -189,6 +236,8 @@ pub enum FlowStatusModule { #[serde(skip_serializing_if = "Option::is_none")] flow_jobs_success: Option>>, #[serde(skip_serializing_if = "Option::is_none")] + flow_jobs_duration: Option, + #[serde(skip_serializing_if = "Option::is_none")] branch_chosen: Option, #[serde(default)] #[serde(skip_serializing_if = "Vec::is_empty")] @@ -209,6 +258,8 @@ pub enum FlowStatusModule { #[serde(skip_serializing_if = "Option::is_none")] flow_jobs_success: Option>>, #[serde(skip_serializing_if = "Option::is_none")] + flow_jobs_duration: Option, + #[serde(skip_serializing_if = "Option::is_none")] branch_chosen: Option, #[serde(skip_serializing_if = "Vec::is_empty")] failed_retries: Vec, @@ -262,6 +313,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule { iterator: untagged.iterator, flow_jobs: untagged.flow_jobs, flow_jobs_success: untagged.flow_jobs_success, + flow_jobs_duration: untagged.flow_jobs_duration, branch_chosen: untagged.branch_chosen, branchall: untagged.branchall, parallel: untagged.parallel.unwrap_or(false), @@ -279,6 +331,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule { .ok_or_else(|| serde::de::Error::missing_field("job"))?, flow_jobs: untagged.flow_jobs, flow_jobs_success: untagged.flow_jobs_success, + flow_jobs_duration: untagged.flow_jobs_duration, branch_chosen: untagged.branch_chosen, approvers: untagged.approvers.unwrap_or_default(), failed_retries: untagged.failed_retries.unwrap_or_default(), @@ -295,6 +348,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule { .ok_or_else(|| serde::de::Error::missing_field("job"))?, flow_jobs: untagged.flow_jobs, flow_jobs_success: untagged.flow_jobs_success, + flow_jobs_duration: untagged.flow_jobs_duration, branch_chosen: untagged.branch_chosen, failed_retries: untagged.failed_retries.unwrap_or_default(), agent_actions: untagged.agent_actions, @@ -360,6 +414,15 @@ impl FlowStatusModule { } } + pub fn flow_jobs_duration(&self) -> Option { + match self { + FlowStatusModule::InProgress { flow_jobs_duration, .. } => flow_jobs_duration.clone(), + FlowStatusModule::Success { flow_jobs_duration, .. } => flow_jobs_duration.clone(), + FlowStatusModule::Failure { flow_jobs_duration, .. } => flow_jobs_duration.clone(), + _ => None, + } + } + pub fn job_result(&self) -> Option { self.flow_jobs() .map(JobResult::ListJob) diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 7a7815dc6c..7c6a3b3dcd 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -758,7 +758,7 @@ pub async fn add_completed_job( canceled_by: Option, flow_is_done: bool, duration: Option, -) -> Result { +) -> Result<(Uuid, i64), Error> { // tracing::error!("Start"); // let start = tokio::time::Instant::now(); @@ -770,7 +770,7 @@ pub async fn add_completed_job( } let result_columns = result_columns.as_ref(); - let (opt_uuid, _duration, _skip_downstream_error_handlers) = (|| { + let (opt_uuid, duration, _skip_downstream_error_handlers) = (|| { commit_completed_job( db, queued_job, @@ -799,11 +799,11 @@ pub async fn add_completed_job( // if scheduling next job failed, return the job_id early to ensure the job get retried after a timeout if let Some(job_id) = opt_uuid { - return Ok(job_id); + return Ok((job_id, duration)); } #[cfg(feature = "cloud")] - apply_completed_job_cloud_usage(db, queued_job, _duration); + apply_completed_job_cloud_usage(db, queued_job, duration); #[cfg(feature = "enterprise")] apply_completed_job_error_handlers( @@ -820,7 +820,7 @@ pub async fn add_completed_job( // tracing::error!("4 {:?}", start.elapsed()); - Ok(queued_job.id) + Ok((queued_job.id, duration)) } async fn commit_completed_job( @@ -4982,12 +4982,17 @@ async fn restarted_flows_resolution( if let Some(new_flow_jobs_success) = new_flow_jobs_success.as_mut() { new_flow_jobs_success.truncate(branch_or_iteration_n); } + let mut new_flow_jobs_timeline = module.flow_jobs_duration(); + if let Some(new_flow_jobs_timeline) = new_flow_jobs_timeline.as_mut() { + new_flow_jobs_timeline.truncate(branch_or_iteration_n); + } truncated_modules.push(FlowStatusModule::InProgress { id: module.id(), job: new_flow_jobs[new_flow_jobs.len() - 1], // set to last finished job from completed flow iterator: None, flow_jobs: Some(new_flow_jobs), flow_jobs_success: new_flow_jobs_success, + flow_jobs_duration: new_flow_jobs_timeline, branch_chosen: None, branchall: Some(BranchAllStatus { branch: branch_or_iteration_n - 1, // Doing minus one here as this variable reflects the latest finished job in the iteration @@ -5022,6 +5027,10 @@ async fn restarted_flows_resolution( if let Some(new_flow_jobs_success) = new_flow_jobs_success.as_mut() { new_flow_jobs_success.truncate(branch_or_iteration_n); } + let mut new_flow_jobs_timeline = module.flow_jobs_duration(); + if let Some(new_flow_jobs_timeline) = new_flow_jobs_timeline.as_mut() { + new_flow_jobs_timeline.truncate(branch_or_iteration_n); + } truncated_modules.push(FlowStatusModule::InProgress { id: module.id(), job: new_flow_jobs[new_flow_jobs.len() - 1], // set to last finished job from completed flow @@ -5031,6 +5040,7 @@ async fn restarted_flows_resolution( }), flow_jobs: Some(new_flow_jobs), flow_jobs_success: new_flow_jobs_success, + flow_jobs_duration: new_flow_jobs_timeline, branch_chosen: None, branchall: None, parallel, diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 591d6bf11f..144c8d9101 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -17,13 +17,7 @@ use windmill_common::otel_oss::FutureExt; use uuid::Uuid; use windmill_common::{ - add_time, - error::{self, Error}, - jobs::JobKind, - utils::WarnAfterExt, - worker::{to_raw_value, Connection, WORKER_GROUP}, - worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, - KillpillSender, DB, + add_time, error::{self, Error}, flow_status::{FlowJobDuration}, jobs::JobKind, utils::WarnAfterExt, worker::{to_raw_value, Connection, WORKER_GROUP}, worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, KillpillSender, DB }; #[cfg(feature = "benchmark")] @@ -342,6 +336,7 @@ pub fn start_background_processor( &w_id, success, Arc::new(result), + None, true, &same_worker_tx, &worker_dir, @@ -582,6 +577,7 @@ pub async fn process_completed_job( let parent_job = job.parent_job.clone(); let job_id = job.id.clone(); let workspace_id = job.workspace_id.clone(); + let started_at = job.started_at.clone(); if job.flow_step_id.as_deref() == Some("preprocessor") { // Do this before inserting to `v2_job_completed` for backwards compatibility @@ -614,7 +610,7 @@ pub async fn process_completed_job( add_time!(bench, "pre add_completed_job"); - add_completed_job( + let (_, duration) = add_completed_job( db, &job, true, @@ -642,6 +638,7 @@ pub async fn process_completed_job( &workspace_id, true, result, + started_at.map(|x| FlowJobDuration { started_at: x, duration_ms: duration }), false, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(), &worker_dir, @@ -682,6 +679,7 @@ pub async fn process_completed_job( &job.workspace_id, false, Arc::new(serde_json::value::to_raw_value(&result).unwrap()), + duration.map(|x| FlowJobDuration { started_at: job.started_at.unwrap(), duration_ms: x }), false, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(), &worker_dir, @@ -783,6 +781,7 @@ pub async fn handle_job_error( &job.workspace_id, false, Arc::new(serde_json::value::to_raw_value(&wrapped_error).unwrap()), + None, unrecoverable, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).clone(), worker_dir, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 201924e825..832af368d1 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -37,7 +37,8 @@ use windmill_common::cache::{self, RawData}; use windmill_common::client::AuthedClient; use windmill_common::db::Authed; use windmill_common::flow_status::{ - ApprovalConditions, FlowStatusModuleWParent, Iterator as FlowIterator, JobResult, + ApprovalConditions, FlowJobDuration, FlowJobsDuration, FlowStatusModuleWParent, + Iterator as FlowIterator, JobResult, }; use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId, StopAfterIf}; use windmill_common::jobs::{ @@ -79,6 +80,7 @@ pub async fn update_flow_status_after_job_completion( w_id: &str, success: bool, result: Arc>, + flow_job_duration: Option, unrecoverable: bool, same_worker_tx: &SameWorkerSender, worker_dir: &str, @@ -95,6 +97,7 @@ pub async fn update_flow_status_after_job_completion( job_id_for_status: job_id_for_status.clone(), success, result, + flow_job_duration, stop_early_override, has_triggered_error_handler: false, }; @@ -108,6 +111,7 @@ pub async fn update_flow_status_after_job_completion( &rec.job_id_for_status, w_id, rec.success, + rec.flow_job_duration.clone(), rec.result, unrecoverable, same_worker_tx, @@ -131,6 +135,7 @@ pub async fn update_flow_status_after_job_completion( &rec.job_id_for_status, w_id, false, + rec.flow_job_duration, Arc::new(to_raw_value(&Json(&WrappedError { error: json!(e.to_string()), }))), @@ -180,11 +185,13 @@ pub enum UpdateFlowStatusAfterJobCompletion { NonLastParallelBranch, PreprocessingStep, } + pub struct RecUpdateFlowStatusAfterJobCompletion { flow: uuid::Uuid, job_id_for_status: Uuid, success: bool, result: Arc>, + flow_job_duration: Option, stop_early_override: Option, has_triggered_error_handler: bool, } @@ -267,6 +274,7 @@ pub async fn update_flow_status_after_job_completion_internal( job_id_for_status: &Uuid, w_id: &str, mut success: bool, + mut flow_job_duration: Option, result: Arc>, unrecoverable: bool, same_worker_tx: &SameWorkerSender, @@ -508,6 +516,7 @@ pub async fn update_flow_status_after_job_completion_internal( parallel, flow_jobs: Some(jobs), flow_jobs_success, + flow_jobs_duration, .. } if *parallel => { let (nindex, len) = match (iterator, branchall) { @@ -517,21 +526,38 @@ pub async fn update_flow_status_after_job_completion_internal( } else { None }; - + tracing::error!( + "flow_job_duration: {:?}, position: {:?} flow: {:?}", + flow_job_duration, + position, + flow + ); let nindex = if let Some(position) = position { sqlx::query_scalar!( - "UPDATE v2_job_status SET - flow_status = JSONB_SET( - JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), + "UPDATE v2_job_status SET flow_status = + CASE + WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN + JSONB_SET( + JSONB_SET(JSONB_SET(JSONB_SET( + flow_status, + ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6), + ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb + ) + ELSE + JSONB_SET(JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb - ) + ) END WHERE id = $2 RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", old_status.step, flow, position as i32, - json!(success) + json!(success), + flow_job_duration.as_ref().map(|x| json!(x.started_at)), + flow_job_duration.as_ref().map(|x| json!(x.duration_ms)) ) } else { sqlx::query_scalar!( @@ -554,6 +580,19 @@ pub async fn update_flow_status_after_job_completion_internal( )) })? .ok_or_else(|| Error::internal_err(format!("requiring an index in InProgress")))?; + + // let status_for_debug = sqlx::query!( + // "SELECT flow_status FROM v2_job_status WHERE id = $1", + // flow + // ) + // .fetch_one(&mut *tx) + // .await + // .map_err(|e| { + // Error::internal_err(format!("error while fetching flow status: {e:#}")) + // })?; + + // tracing::error!("status_for_debug: {:?}", status_for_debug.flow_status); + tracing::info!( "parallel iteration {job_id_for_status} of flow {flow} update nindex: {nindex} len: {len}", nindex = nindex, @@ -570,18 +609,33 @@ pub async fn update_flow_status_after_job_completion_internal( let nindex = if let Some(position) = position { sqlx::query_scalar!( - "UPDATE v2_job_status SET - flow_status = JSONB_SET( - JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), + "UPDATE v2_job_status SET flow_status = + CASE + WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN + JSONB_SET( + JSONB_SET(JSONB_SET(JSONB_SET( + flow_status, + ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6), + ARRAY['modules', $1::TEXT, 'branchall', 'branch'], + ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb + ) + ELSE + JSONB_SET(JSONB_SET( + flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb ) + END WHERE id = $2 RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int", old_status.step, flow, position as i32, - json!(success) + json!(success), + flow_job_duration.as_ref().map(|x| json!(x.started_at)), + flow_job_duration.as_ref().map(|x| json!(x.duration_ms)) ) } else { sqlx::query_scalar!( @@ -616,7 +670,22 @@ pub async fn update_flow_status_after_job_completion_internal( .execute(&mut *tx) .await?; if nindex == len { - let mut flow_jobs_success = flow_jobs_success.clone(); + let success_and_durations = match sqlx::query!( + "SELECT flow_status->'modules'->$2::int->'flow_jobs_success' as \"flow_jobs_success: Json>>\", flow_status->'modules'->$2::int->'flow_jobs_duration' as \"flow_jobs_duration: Json\" + FROM v2_job_status WHERE id = $1", + flow, + old_status.step + ) + .fetch_one(&mut *tx) + .await { + Err(e) => { + tracing::error!("error while fetching success and durations: {e:#}"); + (flow_jobs_success.clone(), flow_jobs_duration.clone()) + } + Ok(x) => (x.flow_jobs_success.map(|x| x.0), x.flow_jobs_duration.map(|x| x.0)), + }; + + let mut flow_jobs_success = success_and_durations.0; if let Some(flow_job_success) = flow_jobs_success.as_mut() { let position = jobs.iter().position(|x| x == job_id_for_status); if let Some(position) = position { @@ -625,6 +694,11 @@ pub async fn update_flow_status_after_job_completion_internal( } } } + let mut flow_jobs_duration = success_and_durations.1; + if let Some(flow_jobs_duration) = flow_jobs_duration.as_mut() { + let position = jobs.iter().position(|x| x == job_id_for_status); + flow_jobs_duration.set(position, &flow_job_duration); + } let branches = current_module .and_then(|x| x.get_branches_skip_failures().ok()) @@ -698,6 +772,7 @@ pub async fn update_flow_status_after_job_completion_internal( job: job_id_for_status.clone(), flow_jobs: Some(jobs.clone()), flow_jobs_success: flow_jobs_success.clone(), + flow_jobs_duration: flow_jobs_duration.clone(), branch_chosen: None, approvers: vec![], failed_retries: vec![], @@ -712,6 +787,7 @@ pub async fn update_flow_status_after_job_completion_internal( job: job_id_for_status.clone(), flow_jobs: Some(jobs.clone()), flow_jobs_success: flow_jobs_success.clone(), + flow_jobs_duration: flow_jobs_duration.clone(), branch_chosen: None, failed_retries: vec![], agent_actions: None, @@ -797,13 +873,14 @@ pub async fn update_flow_status_after_job_completion_internal( && !stop_early => { if let Some(jobs) = flow_jobs { - set_success_in_flow_job_success( + set_success_and_duration_in_flow_job_success( flow_jobs_success, jobs, job_id_for_status, - &old_status, + old_status.step, flow, success, + flow_job_duration.clone(), &mut tx, ) .await?; @@ -821,13 +898,14 @@ pub async fn update_flow_status_after_job_completion_internal( && !stop_early => { if let Some(jobs) = flow_jobs { - set_success_in_flow_job_success( + set_success_and_duration_in_flow_job_success( flow_jobs_success, jobs, job_id_for_status, - &old_status, + old_status.step, flow, success, + flow_job_duration.clone(), &mut tx, ) .await?; @@ -872,6 +950,7 @@ pub async fn update_flow_status_after_job_completion_internal( let flow_jobs = module_status.flow_jobs(); let branch_chosen = module_status.branch_chosen(); let mut flow_jobs_success = module_status.flow_jobs_success(); + let mut flow_jobs_duration = module_status.flow_jobs_duration(); if let (Some(flow_job_success), Some(flow_jobs)) = (flow_jobs_success.as_mut(), flow_jobs.as_ref()) @@ -884,6 +963,13 @@ pub async fn update_flow_status_after_job_completion_internal( } } + if let (Some(flow_jobs_duration), Some(flow_jobs)) = + (flow_jobs_duration.as_mut(), flow_jobs.as_ref()) + { + let position = flow_jobs.iter().position(|x| x == job_id_for_status); + flow_jobs_duration.set(position, &flow_job_duration); + } + // if stop_early with error message, we want to set the job as failure and trigger the error handler if it exists if (success || (flow_jobs.is_some() && (skip_loop_failures || skip_seq_branch_failure))) @@ -912,6 +998,7 @@ pub async fn update_flow_status_after_job_completion_internal( job: job_id_for_status.clone(), flow_jobs, flow_jobs_success, + flow_jobs_duration, branch_chosen, approvers: vec![], failed_retries: old_status.retry.failed_jobs.clone(), @@ -953,6 +1040,7 @@ pub async fn update_flow_status_after_job_completion_internal( job: job_id_for_status.clone(), flow_jobs, flow_jobs_success, + flow_jobs_duration, branch_chosen, failed_retries: old_status.retry.failed_jobs.clone(), agent_actions: module_status.agent_actions(), @@ -1387,8 +1475,8 @@ pub async fn update_flow_status_after_job_completion_internal( let success = success && (!is_failure_step || result_has_recover_true(nresult.clone())); add_time!(bench, "flow status update 1"); - if success { - add_completed_job( + let duration = if success { + let (_, duration) = add_completed_job( db, &flow_job, true, @@ -1401,8 +1489,9 @@ pub async fn update_flow_status_after_job_completion_internal( None, ) .await?; + duration } else { - add_completed_job( + let (_, duration) = add_completed_job( db, &flow_job, false, @@ -1419,7 +1508,11 @@ pub async fn update_flow_status_after_job_completion_internal( None, ) .await?; - } + duration + }; + flow_job_duration = flow_job + .started_at + .map(|x| FlowJobDuration { started_at: x, duration_ms: duration }); } true } else { @@ -1469,6 +1562,7 @@ pub async fn update_flow_status_after_job_completion_internal( flow: parent_job, job_id_for_status: flow, success: success && !is_failure_step, + flow_job_duration: flow_job_duration.clone(), result: nresult.clone(), stop_early_override: if stop_early { Some(skip_if_stop_early) @@ -1490,35 +1584,46 @@ fn find_flow_job_index(flow_jobs: &Vec, job_id_for_status: &Uuid) -> Optio flow_jobs.iter().position(|x| x == job_id_for_status) } -async fn set_success_in_flow_job_success<'c>( +async fn set_success_and_duration_in_flow_job_success<'c>( flow_jobs_success: &Option>>, flow_jobs: &Vec, job_id_for_status: &Uuid, - old_status: &FlowStatus, + old_status_step: i32, flow: Uuid, success: bool, + flow_job_duration: Option, tx: &mut Transaction<'c, Postgres>, ) -> error::Result<()> { if flow_jobs_success.is_some() { let position = find_flow_job_index(flow_jobs, job_id_for_status); if let Some(position) = position { sqlx::query!( - "UPDATE v2_job_status SET - flow_status = JSONB_SET( + "UPDATE v2_job_status SET flow_status = + CASE WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN + JSONB_SET(JSONB_SET(JSONB_SET( flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4 - ) - WHERE id = $2", - old_status.step as i32, + ), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $5), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $6) + ELSE + JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4) + END + WHERE id = $2", + old_status_step as i32, flow, position as i32, - json!(success) + json!(success), + flow_job_duration.as_ref().map(|x| json!(x.duration_ms)), + flow_job_duration.as_ref().map(|x| json!(x.started_at)) ) .execute(&mut **tx) .await .map_err(|e| { - Error::internal_err(format!("error while setting flow_jobs_success: {e:#}")) + Error::internal_err(format!( + "error while setting flow_jobs_success/timeline: {e:#}" + )) })?; } } @@ -2753,6 +2858,11 @@ async fn push_next_flow_job( } else { Some(vec![]) }, + flow_jobs_duration: if branch_chosen.is_some() { + None + } else { + Some(FlowJobsDuration { started_at: vec![], duration_ms: vec![] }) + }, branch_chosen: branch_chosen, approvers: vec![], failed_retries: vec![], @@ -3148,6 +3258,7 @@ async fn push_next_flow_job( mut flow_jobs, while_loop, mut flow_jobs_success, + mut flow_jobs_duration, .. }, .. @@ -3159,11 +3270,15 @@ async fn push_next_flow_job( if let Some(flow_jobs_success) = &mut flow_jobs_success { flow_jobs_success.push(None); } + if let Some(flow_jobs_duration) = &mut flow_jobs_duration { + flow_jobs_duration.push(&None); + } FlowStatusModule::InProgress { job: uuid, iterator: Some(FlowIterator { index, itered }), flow_jobs: Some(flow_jobs), flow_jobs_success, + flow_jobs_duration, branch_chosen: None, branchall: None, id: status_module.id(), @@ -3179,6 +3294,7 @@ async fn push_next_flow_job( iterator, flow_jobs_success: Some(vec![None; uuids.len()]), flow_jobs: Some(uuids.clone()), + flow_jobs_duration: Some(FlowJobsDuration::new(uuids.len())), branch_chosen: None, branchall, id: status_module.id(), @@ -3192,6 +3308,7 @@ async fn push_next_flow_job( mut flow_jobs, status, mut flow_jobs_success, + mut flow_jobs_duration, .. }) => { let uuid = one_uuid?; @@ -3199,11 +3316,15 @@ async fn push_next_flow_job( if let Some(flow_jobs_success) = &mut flow_jobs_success { flow_jobs_success.push(None); } + if let Some(flow_jobs_duration) = &mut flow_jobs_duration { + flow_jobs_duration.push(&None); + } FlowStatusModule::InProgress { job: uuid, iterator: None, flow_jobs: Some(flow_jobs), flow_jobs_success, + flow_jobs_duration, branch_chosen: None, branchall: Some(status), id: status_module.id(), @@ -3220,6 +3341,7 @@ async fn push_next_flow_job( iterator: None, flow_jobs: None, flow_jobs_success: None, + flow_jobs_duration: None, branch_chosen: Some(branch), branchall: None, id: status_module.id(), @@ -3396,6 +3518,7 @@ struct ForloopNextIteration { itered: Vec>, flow_jobs: Vec, flow_jobs_success: Option>>, + flow_jobs_duration: Option, new_args: Iter, while_loop: bool, } @@ -3411,6 +3534,7 @@ struct NextBranch { status: BranchAllStatus, flow_jobs: Vec, flow_jobs_success: Option>>, + flow_jobs_duration: Option, } #[derive(Debug)] @@ -3662,13 +3786,18 @@ async fn compute_next_flow_transform( FlowModuleValue::WhileloopFlow { modules, modules_node, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg let is_simple = is_simple_modules(&modules, flow.failure_module.as_ref()); - let (flow_jobs, flow_jobs_success) = match status_module { + let (flow_jobs, flow_jobs_success, flow_jobs_duration) = match status_module { FlowStatusModule::InProgress { flow_jobs: Some(flow_jobs), flow_jobs_success, + flow_jobs_duration, .. - } => (flow_jobs.clone(), flow_jobs_success.clone()), - _ => (vec![], Some(vec![])), + } => ( + flow_jobs.clone(), + flow_jobs_success.clone(), + flow_jobs_duration.clone(), + ), + _ => (vec![], Some(vec![]), Some(FlowJobsDuration::new(0))), }; let next_loop_idx = flow_jobs.len(); next_loop_iteration( @@ -3679,6 +3808,7 @@ async fn compute_next_flow_transform( itered: vec![], flow_jobs: flow_jobs, flow_jobs_success: flow_jobs_success, + flow_jobs_duration: flow_jobs_duration, new_args: Iter { index: next_loop_idx as i32, value: windmill_common::worker::to_raw_value(&next_loop_idx), @@ -3877,72 +4007,78 @@ async fn compute_next_flow_transform( )) } FlowModuleValue::BranchAll { branches, parallel, .. } => { - let (branch_status, flow_jobs, flow_jobs_success) = match status_module { - FlowStatusModule::WaitingForPriorSteps { .. } - | FlowStatusModule::WaitingForEvents { .. } - | FlowStatusModule::WaitingForExecutor { .. } => { - if branches.is_empty() { - return Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: None }); - } else if parallel { - let len = branches.len(); - let payloads: Vec = branches - .into_iter() - .enumerate() - .filter_map(|(i, Branch { modules, modules_node, .. })| { - let Some(payload) = payload_from_modules( - modules, - modules_node, - flow.failure_module.as_ref(), - flow.same_worker, - || format!("{}-{i}", status.step), - || format!("{}/branchall-{}", flow_job.runnable_path(), i), - false, - ) else { - return None; - }; - Some(JobPayloadWithTag { - payload, - tag: None, - delete_after_use, - timeout: None, - on_behalf_of: None, - }) - }) - .collect::>(); - if payloads.is_empty() { + let (branch_status, flow_jobs, flow_jobs_success, flow_jobs_duration) = + match status_module { + FlowStatusModule::WaitingForPriorSteps { .. } + | FlowStatusModule::WaitingForEvents { .. } + | FlowStatusModule::WaitingForExecutor { .. } => { + if branches.is_empty() { return Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: None }); + } else if parallel { + let len = branches.len(); + let payloads: Vec = branches + .into_iter() + .enumerate() + .filter_map(|(i, Branch { modules, modules_node, .. })| { + let Some(payload) = payload_from_modules( + modules, + modules_node, + flow.failure_module.as_ref(), + flow.same_worker, + || format!("{}-{i}", status.step), + || format!("{}/branchall-{}", flow_job.runnable_path(), i), + false, + ) else { + return None; + }; + Some(JobPayloadWithTag { + payload, + tag: None, + delete_after_use, + timeout: None, + on_behalf_of: None, + }) + }) + .collect::>(); + if payloads.is_empty() { + return Ok(NextFlowTransform::EmptyInnerFlows { + branch_chosen: None, + }); + } + return Ok(NextFlowTransform::Continue( + ContinuePayload::ParallelJobs(payloads), + NextStatus::AllFlowJobs { + branchall: Some(BranchAllStatus { branch: 0, len }), + iterator: None, + simple_input_transforms: None, + }, + )); + } else { + ( + BranchAllStatus { branch: 0, len: branches.len() }, + vec![], + Some(vec![]), + Some(FlowJobsDuration::new(0)), + ) } - return Ok(NextFlowTransform::Continue( - ContinuePayload::ParallelJobs(payloads), - NextStatus::AllFlowJobs { - branchall: Some(BranchAllStatus { branch: 0, len }), - iterator: None, - simple_input_transforms: None, - }, - )); - } else { - ( - BranchAllStatus { branch: 0, len: branches.len() }, - vec![], - Some(vec![]), - ) } - } - FlowStatusModule::InProgress { - branchall: Some(BranchAllStatus { branch, len }), - flow_jobs: Some(flow_jobs), - flow_jobs_success, - .. - } if !parallel => ( - BranchAllStatus { branch: branch + 1, len: len.clone() }, - flow_jobs.clone(), - flow_jobs_success.clone(), - ), + FlowStatusModule::InProgress { + branchall: Some(BranchAllStatus { branch, len }), + flow_jobs: Some(flow_jobs), + flow_jobs_success, + flow_jobs_duration, + .. + } if !parallel => ( + BranchAllStatus { branch: branch + 1, len: len.clone() }, + flow_jobs.clone(), + flow_jobs_success.clone(), + flow_jobs_duration.clone(), + ), - _ => Err(Error::BadRequest(format!( - "Unrecognized module status for BranchAll {status_module:?}" - )))?, - }; + _ => Err(Error::BadRequest(format!( + "Unrecognized module status for BranchAll {status_module:?}" + )))?, + }; let Branch { modules, modules_node, .. } = branches .into_iter() @@ -3972,7 +4108,6 @@ async fn compute_next_flow_transform( branch_chosen: Some(BranchChosen::Default), }); }; - Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(JobPayloadWithTag { payload, @@ -3985,6 +4120,7 @@ async fn compute_next_flow_transform( status: branch_status, flow_jobs, flow_jobs_success, + flow_jobs_duration, }), )) } @@ -4130,6 +4266,7 @@ async fn next_forloop_status( itered, flow_jobs: vec![], flow_jobs_success: Some(vec![]), + flow_jobs_duration: Some(FlowJobsDuration::new(0)), new_args: iter, while_loop: false, }) @@ -4142,6 +4279,7 @@ async fn next_forloop_status( iterator: Some(FlowIterator { itered, index }), flow_jobs: Some(flow_jobs), flow_jobs_success, + flow_jobs_duration, .. } if !*parallel => { let itered_new = if itered.is_empty() { @@ -4192,6 +4330,7 @@ async fn next_forloop_status( itered: itered_new.clone(), flow_jobs: flow_jobs.clone(), flow_jobs_success: flow_jobs_success.clone(), + flow_jobs_duration: flow_jobs_duration.clone(), new_args: Iter { index: index as i32, value: next.to_owned() }, while_loop: false, }) diff --git a/cli/src/commands/sync/sync.ts b/cli/src/commands/sync/sync.ts index 4ab1af2252..bb641b5e0d 100644 --- a/cli/src/commands/sync/sync.ts +++ b/cli/src/commands/sync/sync.ts @@ -65,7 +65,10 @@ import { } from "../../utils/metadata.ts"; import { OpenFlow } from "../../../gen/types.gen.ts"; import { pushResource } from "../resource/resource.ts"; -import { newPathAssigner, PathAssigner } from "../../../windmill-utils-internal/src/path-utils/path-assigner.ts"; +import { + newPathAssigner, + PathAssigner, +} from "../../../windmill-utils-internal/src/path-utils/path-assigner.ts"; import { extractInlineScripts as extractInlineScriptsForFlows } from "../../../windmill-utils-internal/src/inline-scripts/extractor.ts"; // Merge CLI options with effective settings, preserving CLI flags as overrides @@ -158,7 +161,9 @@ async function addCodebaseDigestIfRelevant( try { parsed = yamlParseContent(path, content); } catch (error) { - log.error(`Failed to parse YAML content for codebase digest at path: ${path}`); + log.error( + `Failed to parse YAML content for codebase digest at path: ${path}` + ); throw error; } if (parsed && typeof parsed == "object") { @@ -258,7 +263,10 @@ export interface InlineScript { content: string; } -export function extractInlineScriptsForApps(rec: any, pathAssigner: PathAssigner): InlineScript[] { +export function extractInlineScriptsForApps( + rec: any, + pathAssigner: PathAssigner +): InlineScript[] { if (!rec) { return []; } @@ -349,10 +357,12 @@ function ZipFSElement( flow.value.modules, {}, SEP, - defaultTs, + defaultTs ); } catch (error) { - log.error(`Failed to extract inline scripts for flow at path: ${p}`); + log.error( + `Failed to extract inline scripts for flow at path: ${p}` + ); throw error; } for (const s of inlineScripts) { @@ -386,9 +396,14 @@ function ZipFSElement( } let inlineScripts; try { - inlineScripts = extractInlineScriptsForApps(app?.["value"], newPathAssigner(defaultTs)); + inlineScripts = extractInlineScriptsForApps( + app?.["value"], + newPathAssigner(defaultTs) + ); } catch (error) { - log.error(`Failed to extract inline scripts for app at path: ${p}`); + log.error( + `Failed to extract inline scripts for app at path: ${p}` + ); throw error; } for (const s of inlineScripts) { @@ -913,13 +928,17 @@ async function compareDynFSElement( try { parsedV = JSON.parse(v); } catch (error) { - log.error(`Failed to parse new JSON content for comparison at path: ${k}`); + log.error( + `Failed to parse new JSON content for comparison at path: ${k}` + ); throw error; } try { parsedM2 = JSON.parse(m2[k]); } catch (error) { - log.error(`Failed to parse existing JSON content for comparison at path: ${k}`); + log.error( + `Failed to parse existing JSON content for comparison at path: ${k}` + ); throw error; } if (deepEqual(parsedV, parsedM2)) { @@ -932,11 +951,11 @@ async function compareDynFSElement( continue; } if (!ignoreCodebaseChanges) { - if (before.codebase != undefined) { + if (before?.codebase != undefined) { delete before.codebase; m2[k] = yamlStringify(before, yamlOptions); } - if (after.codebase != undefined) { + if (after?.codebase != undefined) { if (before.codebase != after.codebase) { codebaseChanges[k] = after.codebase; } @@ -1241,7 +1260,6 @@ export async function pull( opts: GlobalOptions & SyncOptions & { repository?: string; promotion?: string } ) { - const originalCliOpts = { ...opts }; opts = await mergeConfigWithConfigFile(opts); @@ -1345,7 +1363,10 @@ export async function pull( ...(specificItems && isSpecificItem(change.path, specificItems) ? { branch_specific: true, - branch_specific_path: getBranchSpecificPath(change.path, specificItems) + branch_specific_path: getBranchSpecificPath( + change.path, + specificItems + ), } : {}), })), @@ -1380,7 +1401,10 @@ export async function pull( // Determine if this file should be written to a branch-specific path let targetPath = change.path; if (specificItems && isSpecificItem(change.path, specificItems)) { - const branchSpecificPath = getBranchSpecificPath(change.path, specificItems); + const branchSpecificPath = getBranchSpecificPath( + change.path, + specificItems + ); if (branchSpecificPath) { targetPath = branchSpecificPath; } @@ -1430,12 +1454,24 @@ export async function pull( } } if (exts.some((e) => change.path.endsWith(e))) { - log.info(`Editing script content of ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`); + log.info( + `Editing script content of ${targetPath}${ + targetPath !== change.path + ? colors.gray(` (branch-specific override for ${change.path})`) + : "" + }` + ); } else if ( change.path.endsWith(".yaml") || change.path.endsWith(".json") ) { - log.info(`Editing ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`); + log.info( + `Editing ${getTypeStrFromPath(change.path)} ${targetPath}${ + targetPath !== change.path + ? colors.gray(` (branch-specific override for ${change.path})`) + : "" + }` + ); } await Deno.writeTextFile(target, change.after); @@ -1447,10 +1483,22 @@ export async function pull( await ensureDir(path.dirname(target)); if (opts.stateful) { await ensureDir(path.dirname(stateTarget)); - log.info(`Adding ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`); + log.info( + `Adding ${getTypeStrFromPath(change.path)} ${targetPath}${ + targetPath !== change.path + ? colors.gray(` (branch-specific override for ${change.path})`) + : "" + }` + ); } await Deno.writeTextFile(target, change.content); - log.info(`Writing ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`); + log.info( + `Writing ${getTypeStrFromPath(change.path)} ${targetPath}${ + targetPath !== change.path + ? colors.gray(` (branch-specific override for ${change.path})`) + : "" + }` + ); if (opts.stateful) { await Deno.copyFile(target, stateTarget); } @@ -1528,7 +1576,10 @@ export async function pull( ...(specificItems && isSpecificItem(change.path, specificItems) ? { branch_specific: true, - branch_specific_path: getBranchSpecificPath(change.path, specificItems) + branch_specific_path: getBranchSpecificPath( + change.path, + specificItems + ), } : {}), })), @@ -1560,7 +1611,10 @@ function prettyChanges(changes: Change[], specificItems?: SpecificItemsConfig) { // Check if this will be written as a branch-specific file if (specificItems && isSpecificItem(change.path, specificItems)) { - const branchSpecificPath = getBranchSpecificPath(change.path, specificItems); + const branchSpecificPath = getBranchSpecificPath( + change.path, + specificItems + ); if (branchSpecificPath) { displayPath = branchSpecificPath; branchNote = " (branch-specific)"; @@ -1569,17 +1623,26 @@ function prettyChanges(changes: Change[], specificItems?: SpecificItemsConfig) { if (change.name === "added") { log.info( - colors.green(`+ ${getTypeStrFromPath(change.path)} ` + displayPath + colors.gray(branchNote)) + colors.green( + `+ ${getTypeStrFromPath(change.path)} ` + + displayPath + + colors.gray(branchNote) + ) ); } else if (change.name === "deleted") { log.info( - colors.red(`- ${getTypeStrFromPath(change.path)} ` + displayPath + colors.gray(branchNote)) + colors.red( + `- ${getTypeStrFromPath(change.path)} ` + + displayPath + + colors.gray(branchNote) + ) ); } else if (change.name === "edited") { log.info( colors.yellow( `~ ${getTypeStrFromPath(change.path)} ` + - displayPath + colors.gray(branchNote) + + displayPath + + colors.gray(branchNote) + (change.codebase ? ` (codebase changed)` : "") ) ); @@ -1794,7 +1857,10 @@ export async function push( ...(specificItems && isSpecificItem(change.path, specificItems) ? { branch_specific: true, - branch_specific_path: getBranchSpecificPath(change.path, specificItems) + branch_specific_path: getBranchSpecificPath( + change.path, + specificItems + ), } : {}), })), @@ -1938,7 +2004,10 @@ export async function push( const currentBranch = getCurrentGitBranch(); if (currentBranch && isBranchSpecificFile(resourceFilePath)) { - serverPath = fromBranchSpecificPath(resourceFilePath, currentBranch); + serverPath = fromBranchSpecificPath( + resourceFilePath, + currentBranch + ); } await pushResource( @@ -1960,7 +2029,10 @@ export async function push( // Check if this is a branch-specific item and get the original branch-specific path let originalBranchSpecificPath: string | undefined; if (specificItems && isSpecificItem(change.path, specificItems)) { - originalBranchSpecificPath = getBranchSpecificPath(change.path, specificItems); + originalBranchSpecificPath = getBranchSpecificPath( + change.path, + specificItems + ); } await pushObj( @@ -2010,7 +2082,10 @@ export async function push( // For branch-specific items, we read from branch-specific files but push to base server paths let localFilePath = change.path; if (specificItems && isSpecificItem(change.path, specificItems)) { - const branchSpecificPath = getBranchSpecificPath(change.path, specificItems); + const branchSpecificPath = getBranchSpecificPath( + change.path, + specificItems + ); if (branchSpecificPath) { localFilePath = branchSpecificPath; } @@ -2024,7 +2099,7 @@ export async function push( opts.plainSecrets ?? false, [], opts.message, - localFilePath // Pass the actual local file path + localFilePath // Pass the actual local file path ); if (stateTarget) { @@ -2216,7 +2291,10 @@ export async function push( ...(specificItems && isSpecificItem(change.path, specificItems) ? { branch_specific: true, - branch_specific_path: getBranchSpecificPath(change.path, specificItems) + branch_specific_path: getBranchSpecificPath( + change.path, + specificItems + ), } : {}), })), diff --git a/frontend/openapi-ts-error-1758271586180.log b/frontend/openapi-ts-error-1758271586180.log new file mode 100644 index 0000000000..4c0634a7c7 --- /dev/null +++ b/frontend/openapi-ts-error-1758271586180.log @@ -0,0 +1,28 @@ +Error parsing /home/rfiszel/windmill/backend/windmill-api/openapi.yaml: bad indentation of a mapping entry (8025:25) + + 8022 | description: job args + 8023 | content: + 8024 | application/json: + 8025 | schema: {}\ +--------------------------------^ + 8026 | + 8027 | /w/{workspace}/jobs/queue/get_scheduled_for_by_ids: +ParserError: Error parsing /home/rfiszel/windmill/backend/windmill-api/openapi.yaml: bad indentation of a mapping entry (8025:25) + + 8022 | description: job args + 8023 | content: + 8024 | application/json: + 8025 | schema: {}\ +--------------------------------^ + 8026 | + 8027 | /w/{workspace}/jobs/queue/get_scheduled_for_by_ids: + at Object.parse (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/parsers/yaml.js:44:23) + at getResult (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:116:22) + at runNextPlugin (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:64:32) + at /home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:55:9 + at new Promise () + at Object.run (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:54:12) + at parseFile (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/parse.js:130:38) + at parse (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/parse.js:56:30) + at async $RefParser.parse (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/index.js:115:28) + at async $RefParser.resolve (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/index.js:145:13) \ No newline at end of file diff --git a/frontend/src/lib/components/FlowLogViewer.svelte b/frontend/src/lib/components/FlowLogViewer.svelte index a22e0a4211..6374f69930 100644 --- a/frontend/src/lib/components/FlowLogViewer.svelte +++ b/frontend/src/lib/components/FlowLogViewer.svelte @@ -19,7 +19,7 @@ import { twMerge } from 'tailwind-merge' import FlowJobsMenu from './flows/map/FlowJobsMenu.svelte' import BarsStaggered from './icons/BarsStaggered.svelte' - import type { GlobalIterationBounds, GraphModuleState } from './graph/model' + import type { GraphModuleState } from './graph/model' import type { NavigationChain } from '$lib/keyboardChain' import { updateLinks } from '$lib/keyboardChain' import FlowLogRow from './FlowLogRow.svelte' @@ -62,8 +62,6 @@ timelineAvailableWidths: Record timelinelWidth: number showTimeline?: boolean - globalIterationBounds?: Record - loadPreviousIterations?: (key: string, amount: number) => void } let { @@ -92,9 +90,7 @@ timelineNow, timelineAvailableWidths = $bindable(), timelinelWidth, - showTimeline = true, - globalIterationBounds, - loadPreviousIterations + showTimeline = true }: Props = $props() function getJobLink(jobId: string | undefined): string { @@ -769,13 +765,8 @@ total={timelineTotal} min={timelineMin} items={moduleItems ?? []} - hasMoreIterations={globalIterationBounds?.[module.id] && - (globalIterationBounds[module.id].iteration_from ?? 0) > 0} now={timelineNow} {timelinelWidth} - loadPreviousIterations={() => { - loadPreviousIterations?.(module.id, 20) - }} onSelectIteration={(id) => { if ( module.value.type !== 'forloopflow' && @@ -856,8 +847,6 @@ bind:timelineAvailableWidths {timelinelWidth} {showTimeline} - {globalIterationBounds} - {loadPreviousIterations} /> {/each} diff --git a/frontend/src/lib/components/FlowLogViewerWrapper.svelte b/frontend/src/lib/components/FlowLogViewerWrapper.svelte index 274290ec1f..4549fc8304 100644 --- a/frontend/src/lib/components/FlowLogViewerWrapper.svelte +++ b/frontend/src/lib/components/FlowLogViewerWrapper.svelte @@ -1,6 +1,6 @@ { - timelineCompute?.updateInputs(flowModules, durationStatuses, flowDone) + timelineCompute?.updateInputs(flowModulesIds, durationStatuses, flowDone) }} /> - {#if items}
{/if}
{max ? displayDate(new Date(max), true) : ''}{#if !max && min}{#if now} - {msToSec(now - min, 3)}s + {msToSec(now - min, 1)}s {/if}{/if}
@@ -109,61 +118,87 @@ />
{/if} - {#each Object.values(flowModules) as k (k)} - {@const iterationFrom = globalIterationBounds[buildSubflowKey(k)]?.iteration_from ?? 0} -
- {#if iterationFrom > 0} -
- -
- {/if} - + {#each flowModules as { id: k, type: typ } (k)} + {@const subItems = items?.[k]?.filter((x) => x.created_at && x.started_at)} +
-
{k.startsWith('subflow:') ? k.substring(8) : k}
-
+
{k.startsWith('subflow:') ? k.substring(8) : k} + {#if localModuleStates[k]?.selectedForloop && (typ == 'forloopflow' || typ == 'whileloopflow')} + + + + {/if} +
+
+ {#if subItems?.length > 1} +
+ {subItems?.length} jobs +
+ {/if} {#if min && total} -
- {#each items?.[k] ?? [] as b} - {@const waitingLen = b?.created_at - ? b.started_at - ? b.started_at - b?.created_at - : b.duration_ms - ? 0 - : now - b?.created_at - : 0} -
- - {#if b.started_at} + subItems?.[index]?.id} + > + {#snippet item({ index, style })} + {@const b = subItems?.[index]} + {#if b?.created_at} + + {@const waitingLen = b?.created_at + ? b.started_at + ? b.started_at - b?.created_at + : b.duration_ms + ? 0 + : now - b?.created_at + : 0} +
- {/if} -
- {/each} -
+ {#if b.started_at} + + {/if} +
+ {:else} +
+
+ +
+
+ {/if} + {/snippet} + {/if}
diff --git a/frontend/src/lib/components/FlowTimelineBar.svelte b/frontend/src/lib/components/FlowTimelineBar.svelte index a78a9fa18a..12ad614049 100644 --- a/frontend/src/lib/components/FlowTimelineBar.svelte +++ b/frontend/src/lib/components/FlowTimelineBar.svelte @@ -21,8 +21,6 @@ showZoomButtons?: boolean onZoom?: () => void zoom?: 'in' | 'out' - hasMoreIterations?: boolean - loadPreviousIterations?: () => void onSelectIteration?: (id: string) => void idToIterationIndex?: (id: string) => number | undefined showIterations?: string[] @@ -39,8 +37,6 @@ showZoomButtons = false, onZoom, zoom = 'in', - hasMoreIterations, - loadPreviousIterations, onSelectIteration, idToIterationIndex, showIterations, @@ -112,26 +108,61 @@ return { left: leftPercent, width: widthPercent } } - function getOverlapOpacity(item: TimelineItem, allItems: TimelineItem[]): number { - if (!item.started_at) return 1 + // More efficient version using sweep line algorithm for computing all overlaps at once + function computeAllOverlaps(items: TimelineItem[]): Record { + const overlapCounts = new Map() - const itemEnd = item.duration_ms ? item.started_at + item.duration_ms : now - let overlapCount = 0 - - for (const otherItem of allItems) { - if (otherItem.id === item.id || !otherItem.started_at) continue - - const otherEnd = otherItem.duration_ms ? otherItem.started_at + otherItem.duration_ms : now - - // Check if time ranges overlap - if (item.started_at < otherEnd && otherItem.started_at < itemEnd) { - overlapCount++ - } + // Create events for start and end times + interface Event { + time: number + type: 'start' | 'end' + itemId: string } - // Base opacity of 1, reduce by 0.2 for each overlap, minimum 0.3 - return Math.max(0.3, 1 - overlapCount * 0.2) + const events: Event[] = [] + + for (const item of items) { + if (!item.started_at) continue + + const endTime = item.duration_ms ? item.started_at + item.duration_ms : now + events.push({ time: item.started_at, type: 'start', itemId: item.id }) + events.push({ time: endTime, type: 'end', itemId: item.id }) + overlapCounts.set(item.id, 0) + } + + // Sort events by time, with end events before start events at the same time + events.sort((a, b) => { + if (a.time !== b.time) return a.time - b.time + return a.type === 'end' ? -1 : 1 + }) + + // Sweep through events + const activeItems = new Set() + + for (const event of events) { + if (event.type === 'start') { + // Count current active items as overlaps for this item + overlapCounts.set(event.itemId, activeItems.size) + + // Update overlap counts for all currently active items + for (const activeId of activeItems) { + overlapCounts.set(activeId, overlapCounts.get(activeId)! + 1) + } + + activeItems.add(event.itemId) + } else { + activeItems.delete(event.itemId) + } + } + return Object.fromEntries(overlapCounts.entries()) } + + // At component level, compute once when items change + const allOverlaps = $derived(computeAllOverlaps(filteredItems)) + const maximumOverlaps = $derived(Math.max(...Object.values(allOverlaps))) + const opacity = $derived(Math.max(0.02, 1 / maximumOverlaps)) + // Then in your template, use: + // allOverlaps.get(item.id) ?? 0 {#if min && filteredItems.length > 0 && startItem?.started_at} @@ -155,24 +186,6 @@ {/if}
- {:else if hasMoreIterations} - - - {#snippet text()} - Load previous iterations - {/snippet} - {:else}
{/if} @@ -205,7 +218,8 @@ {#each filteredItems as item, i} {#if item.started_at} {@const position = calculateItemPosition(item)} - {@const opacity = getOverlapOpacity(item, filteredItems)} + { - scheduledFor = response }) + .then((response) => { + scheduledFor = response + }) + .catch((error) => { + console.error('Failed to fetch scheduled for:', error) + }) } catch (error) { console.error('Failed to fetch scheduled for:', error) } diff --git a/frontend/src/lib/components/flows/flowModuleNextId.ts b/frontend/src/lib/components/flows/flowModuleNextId.ts index 428ceccf47..1d0cd14ec6 100644 --- a/frontend/src/lib/components/flows/flowModuleNextId.ts +++ b/frontend/src/lib/components/flows/flowModuleNextId.ts @@ -6,6 +6,8 @@ import { charsToNumber, numberToChars } from './idUtils' // Computes the next available id export function nextId(flowState: FlowState, fullFlow: OpenFlow): string { const allIds = dfs(fullFlow.value.modules, (fm) => fm.id) + console.log('allIds', allIds) + const max = allIds.concat(Object.keys(flowState)).reduce((acc, key) => { if (key.length >= 4) { return acc diff --git a/frontend/src/lib/components/graph/model.ts b/frontend/src/lib/components/graph/model.ts index a97f603617..8ed90faeeb 100644 --- a/frontend/src/lib/components/graph/model.ts +++ b/frontend/src/lib/components/graph/model.ts @@ -27,11 +27,6 @@ export type DurationStatus = { byJob: Record } -export type GlobalIterationBounds = { - iteration_from?: number - iteration_total?: number -} - export type FlowStatusViewerContext = { flowStateStore?: FlowState retryStatus: StateStore> @@ -58,6 +53,10 @@ export type GraphModuleState = { selectedForloopIndex?: number selectedForLoopSetManually?: boolean flow_jobs_success?: (boolean | undefined)[] + flow_jobs_duration?: { + started_at?: (string | undefined)[] + duration_ms?: (number | undefined)[] + } flow_jobs?: string[] iteration_total?: number retries?: number diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 346ed62d1c..09feda8a6c 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -89,7 +89,7 @@ components: minimum: 0 maximum: 100 retry_if: - $ref: '#/components/schemas/RetryIf' + $ref: "#/components/schemas/RetryIf" RetryIf: type: object @@ -569,6 +569,17 @@ components: type: array items: type: boolean + flow_jobs_duration: + type: object + properties: + started_at: + type: array + items: + type: string + duration_ms: + type: array + items: + type: integer branch_chosen: type: object properties: From d36501488278072e2018df1fc93bfe676b446c6b Mon Sep 17 00:00:00 2001 From: hugocasa Date: Fri, 19 Sep 2025 19:33:05 +0200 Subject: [PATCH 3/4] build on macos with deno_core (#6642) --- backend/Cargo.lock | 1 + backend/Cargo.toml | 4 ++++ backend/windmill-worker/Cargo.toml | 2 ++ frontend/README_DEV.md | 23 ++++++++++++++--------- 4 files changed, 21 insertions(+), 9 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 4a03e346dc..593de8a8e5 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15796,6 +15796,7 @@ dependencies = [ "itertools 0.14.0", "jsonwebtoken 8.3.0", "lazy_static", + "libffi-sys", "libloading 0.8.8", "mappable-rc", "mime_guess", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index e7f56b29cc..ac9af8e0e1 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -68,6 +68,7 @@ jemalloc = ["windmill-common/jemalloc", "dep:tikv-jemallocator", "dep:tikv-jemal 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"] +deno_core_mac = ["deno_core", "windmill-worker/libffi_mac"] kafka = ["windmill-api/kafka"] nats = ["windmill-api/nats"] otel = ["windmill-common/otel", "windmill-worker/otel"] @@ -275,6 +276,9 @@ deno_runtime = { version = "0.198.0", features = ["transpile"] } deno_telemetry = "0.12.0" deno_error = "=0.5.5" +# only used with special deno_core_mac feature to prevent ffi issue on macos, requires libffi to be installed +libffi-sys = { version = "2.3.0", features = ["system"]} + google-cloud-pubsub = "0.30.0" google-cloud-googleapis = {version = "0.16.1", features = ["pubsub"]} # TODO: remove once deno fixes the issue on their end diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 19c8c0d31f..d249a78319 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -22,6 +22,7 @@ cloud = [] sqlx = [] deno_core = ["dep:deno_fetch", "dep:deno_webidl", "dep:deno_web", "dep:deno_net", "dep:deno_console", "dep:deno_url", "dep:deno_core", "dep:deno_ast", "dep:deno_tls", "dep:deno_permissions", "dep:deno_io", "dep:deno_runtime", "dep:deno_telemetry", "dep:deno_error", "dep:winapi"] +libffi_mac = ["dep:libffi-sys"] otel = ["windmill-common/otel", "dep:opentelemetry"] dind = ["dep:bollard"] php = ["dep:windmill-parser-php"] @@ -147,3 +148,4 @@ deno_io = { workspace = true, optional = true } deno_runtime = { workspace = true, optional = true } deno_telemetry = { workspace = true, optional = true } winapi = { workspace = true, optional = true } +libffi-sys = { workspace = true, optional = true } diff --git a/frontend/README_DEV.md b/frontend/README_DEV.md index c17ac944c8..c5ba2a0c15 100644 --- a/frontend/README_DEV.md +++ b/frontend/README_DEV.md @@ -139,15 +139,6 @@ If you develop wasm parser for new language you can also pass `--wasm-pkg > ~/.zshrc - source ~/.zshrc - ``` - In the root folder: ```bash @@ -166,6 +157,20 @@ In the frontend folder: REMOTE=http://127.0.0.1:8000 REMOTE_LSP=http://127.0.0.1:3001 npm run dev ``` +**Known issue on M1 Mac while running `cargo run`** + +- You may encounter `linking with cc failed` build time error. +- To solve this run: + ```bash + echo 'export RUSTFLAGS="-L/opt/homebrew/opt/libomp/lib"' >> ~/.zshrc + source ~/.zshrc + ``` + +**Known issue on M1 Mac while running `cargo run` with the `deno_core` feature** + +- You may encounter ``failed to run custom build command for `libffi-sys v2.3.0` `` build time error. +- To solve this use the `deno_core_mac` feature flag _instead_ of `deno_core`. You might need to install `libffi` (e.g. `brew install libffi`). + ### Formatting This project uses [prettier](https://prettier.io/docs/en/install.html) and From 70e9ae14a9541862aaa31b8435abb0185805d4b7 Mon Sep 17 00:00:00 2001 From: Alexander Petric Date: Fri, 19 Sep 2025 13:43:16 -0400 Subject: [PATCH 4/4] fix: teams api improvements (#6643) * init * adding cache + ui nits * improve caching * sqlx * ee repo ref * remove useless comments * ci errors * pr comments * ee repo ref --- ...1f7f387f5055c47f493271d26731336257384.json | 10 +- ...d9474b17887711128dbb2ef15d247d50686b0.json | 22 ---- ...4caecda6335eda5b2e97e5a7370361653ff48.json | 26 ---- ...524d1a5b196a7b77afcb02f46b84b22088bbf.json | 20 --- ...857d432dff44e343b8f0610208d42ff5afd14.json | 14 -- ...0cb549a34b96554ae1872355b90304f5dcb76.json | 4 +- backend/ee-repo-ref.txt | 2 +- backend/windmill-api/openapi.yaml | 42 +++--- backend/windmill-api/src/lib.rs | 3 + backend/windmill-api/src/teams_cache_oss.rs | 3 + backend/windmill-api/src/teams_oss.rs | 13 +- .../src/lib/components/ChannelSelector.svelte | 121 ++++++++++++++--- .../lib/components/ConnectionSection.svelte | 13 +- .../components/ErrorOrRecoveryHandler.svelte | 36 +++-- .../src/lib/components/InstanceSetting.svelte | 53 ++------ .../lib/components/InstanceSettings.svelte | 41 ++---- .../src/lib/components/TeamSelector.svelte | 124 ++++++++++++------ 17 files changed, 286 insertions(+), 261 deletions(-) delete mode 100644 backend/.sqlx/query-1ace9bdcde90fd2261fd64344a2d9474b17887711128dbb2ef15d247d50686b0.json delete mode 100644 backend/.sqlx/query-50c17c7848760aaf0f869acfc444caecda6335eda5b2e97e5a7370361653ff48.json delete mode 100644 backend/.sqlx/query-84d048ea323758842dc564c700b524d1a5b196a7b77afcb02f46b84b22088bbf.json delete mode 100644 backend/.sqlx/query-9c72b2962d0919353cfe5af710e857d432dff44e343b8f0610208d42ff5afd14.json create mode 100644 backend/windmill-api/src/teams_cache_oss.rs diff --git a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json index e7ed0aee65..d29a18c691 100644 --- a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json +++ b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json @@ -46,11 +46,11 @@ ] }, "nullable": [ - false, - false, - false, - false, - false, + true, + true, + true, + true, + true, true, true ] diff --git a/backend/.sqlx/query-1ace9bdcde90fd2261fd64344a2d9474b17887711128dbb2ef15d247d50686b0.json b/backend/.sqlx/query-1ace9bdcde90fd2261fd64344a2d9474b17887711128dbb2ef15d247d50686b0.json deleted file mode 100644 index 9a8ef973a4..0000000000 --- a/backend/.sqlx/query-1ace9bdcde90fd2261fd64344a2d9474b17887711128dbb2ef15d247d50686b0.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT teams_team_id FROM workspace_settings WHERE workspace_id = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "teams_team_id", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - true - ] - }, - "hash": "1ace9bdcde90fd2261fd64344a2d9474b17887711128dbb2ef15d247d50686b0" -} diff --git a/backend/.sqlx/query-50c17c7848760aaf0f869acfc444caecda6335eda5b2e97e5a7370361653ff48.json b/backend/.sqlx/query-50c17c7848760aaf0f869acfc444caecda6335eda5b2e97e5a7370361653ff48.json deleted file mode 100644 index f3dc153254..0000000000 --- a/backend/.sqlx/query-50c17c7848760aaf0f869acfc444caecda6335eda5b2e97e5a7370361653ff48.json +++ /dev/null @@ -1,26 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n WITH assigned_teams AS (\n SELECT teams_team_id\n FROM workspace_settings\n ),\n all_teams AS (\n SELECT jsonb_array_elements(CASE\n WHEN jsonb_typeof(value::jsonb) = 'array' THEN value::jsonb\n ELSE '[]'::jsonb\n END) AS team\n FROM global_settings\n WHERE name = 'teams'\n )\n SELECT team->>'team_name' AS team_name, team->>'team_internal_id' AS team_id\n FROM all_teams\n WHERE NOT EXISTS (\n SELECT 1\n FROM assigned_teams\n WHERE assigned_teams.teams_team_id = team->>'team_internal_id'\n )\n AND team->>'team_name' IS NOT NULL\n AND team->>'team_id' IS NOT NULL\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "team_name", - "type_info": "Text" - }, - { - "ordinal": 1, - "name": "team_id", - "type_info": "Text" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - null, - null - ] - }, - "hash": "50c17c7848760aaf0f869acfc444caecda6335eda5b2e97e5a7370361653ff48" -} diff --git a/backend/.sqlx/query-84d048ea323758842dc564c700b524d1a5b196a7b77afcb02f46b84b22088bbf.json b/backend/.sqlx/query-84d048ea323758842dc564c700b524d1a5b196a7b77afcb02f46b84b22088bbf.json deleted file mode 100644 index e4473b360c..0000000000 --- a/backend/.sqlx/query-84d048ea323758842dc564c700b524d1a5b196a7b77afcb02f46b84b22088bbf.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT value FROM global_settings WHERE name = 'teams'", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "value", - "type_info": "Jsonb" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - false - ] - }, - "hash": "84d048ea323758842dc564c700b524d1a5b196a7b77afcb02f46b84b22088bbf" -} diff --git a/backend/.sqlx/query-9c72b2962d0919353cfe5af710e857d432dff44e343b8f0610208d42ff5afd14.json b/backend/.sqlx/query-9c72b2962d0919353cfe5af710e857d432dff44e343b8f0610208d42ff5afd14.json deleted file mode 100644 index a2f8a647c5..0000000000 --- a/backend/.sqlx/query-9c72b2962d0919353cfe5af710e857d432dff44e343b8f0610208d42ff5afd14.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO global_settings (name, value)\n VALUES ('teams', $1)\n ON CONFLICT (name) DO UPDATE SET value = $1", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "9c72b2962d0919353cfe5af710e857d432dff44e343b8f0610208d42ff5afd14" -} diff --git a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json index 54e94cfb8f..99269c9851 100644 --- a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json +++ b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json @@ -18,8 +18,8 @@ "Left": [] }, "nullable": [ - false, - true + true, + false ] }, "hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76" diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 7d8849c19a..694b79ccac 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -daa1c26c268c778e55756f02a459b6c7628c9267 +1dc77acb47053b2dd84569d70884456de0188290 diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 40e3ab25e7..c0e3cf4000 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -2195,6 +2195,12 @@ paths: - workspace parameters: - $ref: "#/components/parameters/WorkspaceId" + - name: search + in: query + description: Search teams by name + required: false + schema: + type: string responses: "200": description: status @@ -2212,15 +2218,27 @@ paths: /w/{workspace}/workspaces/available_teams_channels: get: - summary: list available teams channels + summary: list available channels for a specific team operationId: listAvailableTeamsChannels tags: - workspace parameters: - $ref: "#/components/parameters/WorkspaceId" + - name: team_id + in: query + description: Microsoft Teams team ID + required: true + schema: + type: string + - name: search + in: query + description: Search channels by name + required: false + schema: + type: string responses: "200": - description: status + description: List of channels for the specified team content: application/json: schema: @@ -2232,10 +2250,7 @@ paths: type: string channel_id: type: string - service_url: - type: string - tenant_id: - type: string + /w/{workspace}/workspaces/connect_teams: post: @@ -3897,21 +3912,6 @@ paths: items: type: string - /teams/sync: - post: - operationId: syncTeams - summary: synchronize Microsoft Teams information (teams/channels) - tags: - - teams - responses: - "200": - description: Teams information successfully synchronized - content: - application/json: - schema: - type: array - items: - $ref: "#/components/schemas/TeamInfo" /teams/activities: post: diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 26e2a803c7..eec6574131 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -156,6 +156,9 @@ pub mod stripe_ee; mod stripe_oss; #[cfg(feature = "private")] pub mod teams_ee; +#[cfg(feature = "private")] +pub mod teams_cache_ee; +mod teams_cache_oss; mod teams_oss; mod token; mod tracing_init; diff --git a/backend/windmill-api/src/teams_cache_oss.rs b/backend/windmill-api/src/teams_cache_oss.rs new file mode 100644 index 0000000000..9501c16f00 --- /dev/null +++ b/backend/windmill-api/src/teams_cache_oss.rs @@ -0,0 +1,3 @@ +#[cfg(feature = "private")] +#[allow(unused)] +pub use crate::teams_cache_ee::*; diff --git a/backend/windmill-api/src/teams_oss.rs b/backend/windmill-api/src/teams_oss.rs index 95d4883690..037e556c99 100644 --- a/backend/windmill-api/src/teams_oss.rs +++ b/backend/windmill-api/src/teams_oss.rs @@ -23,6 +23,13 @@ pub async fn workspaces_list_available_teams_ids() -> Result )); } +#[cfg(not(feature = "private"))] +pub async fn workspaces_list_available_teams_channels() -> Result { + return Err(Error::BadRequest( + "Teams only available on enterprise".to_string(), + )); +} + #[cfg(not(feature = "private"))] pub async fn connect_teams() -> Result { return Err(Error::BadRequest( @@ -37,12 +44,6 @@ pub async fn run_teams_message_test_job() -> Result { )); } -#[cfg(not(feature = "private"))] -pub async fn workspaces_list_available_teams_channels() -> Result { - return Err(Error::BadRequest( - "Teams only available on enterprise".to_string(), - )); -} #[cfg(all(feature = "enterprise", not(feature = "private")))] pub fn teams_service() -> Router { diff --git a/frontend/src/lib/components/ChannelSelector.svelte b/frontend/src/lib/components/ChannelSelector.svelte index 30bf70e14b..fe5ee8f5c5 100644 --- a/frontend/src/lib/components/ChannelSelector.svelte +++ b/frontend/src/lib/components/ChannelSelector.svelte @@ -1,5 +1,8 @@
- = 1 || (searchFilterText.length === 0 && selectedChannel) ? displayChannels().filter(channel => channel.channel_id && channel.channel_name).map((channel) => ({ + label: channel.channel_name ?? 'Unknown Channel', + value: channel.channel_id ?? '' + })) : []} + placeholder={isFetching ? "Searching..." : (teamId ? "Search channels..." : "Select a team first")} + clearable + disabled={disabled || isFetching || !teamId} + bind:filterText={searchFilterText} + bind:value={ + () => selectedChannel?.channel_id, + (value) => { + selectedChannel = value ? displayChannels().find((channel) => channel.channel_id === value) : undefined + } + } + /> + {:else} + { - if (el.teams_channel) { - const team = teams.find((team) => team.team_name === el.teams_channel.team_name) || null - return { - teams_channel: { - team_id: team?.team_id, - team_name: team?.team_name, - channel_id: team?.channels.find( - (channel) => channel.channel_id === el.teams_channel.channel_id - )?.channel_id, - channel_name: team?.channels.find( - (channel) => channel.channel_id === el.teams_channel.channel_id - )?.channel_name - } - } - } - return el - }) } $values = nvalues @@ -141,12 +118,22 @@ setupSnowflakeUrls() } - // Remove empty or invalid teams_channel entries + // Remove empty or invalid entries for critical error channels $values.critical_error_channels = $values.critical_error_channels.filter((entry) => { - if (entry && typeof entry == 'object' && 'teams_channel' in entry) { + if (!entry || typeof entry !== 'object') return false + if ('teams_channel' in entry) { return isValidTeamsChannel(entry.teams_channel) } - return true + if ('slack_channel' in entry) { + return ( + typeof entry.slack_channel === 'string' && entry.slack_channel.trim() !== '' + ) + } + if ('email' in entry) { + return typeof entry.email === 'string' && entry.email.trim() !== '' + } + // Unknown shape + return false }) let shouldReloadPage = false diff --git a/frontend/src/lib/components/TeamSelector.svelte b/frontend/src/lib/components/TeamSelector.svelte index 899314bb31..d8d0fbc2e9 100644 --- a/frontend/src/lib/components/TeamSelector.svelte +++ b/frontend/src/lib/components/TeamSelector.svelte @@ -3,7 +3,7 @@ import { RefreshCcw } from 'lucide-svelte' import { WorkspaceService } from '$lib/gen' import Select from './select/Select.svelte' - import { onMount } from 'svelte' + import { debounce } from '$lib/utils' interface TeamItem { team_id: string @@ -12,7 +12,6 @@ interface Props { disabled?: boolean - placeholder?: string selectedTeam?: TeamItem | undefined containerClass?: string showRefreshButton?: boolean @@ -23,7 +22,6 @@ let { disabled = false, - placeholder = 'Select a team', selectedTeam = $bindable(), containerClass = 'w-64', showRefreshButton = true, @@ -33,68 +31,120 @@ }: Props = $props() let isFetching = $state(false) - onMount(() => { - if (!teams) loadTeams() + let searchResults = $state([]) + + // Only enable search mode if no teams are provided + const searchMode = !teams + + // Determine which teams to show: provided teams or search results + // In search mode, include the selected team if it exists + let displayTeams = $derived(() => { + const baseTeams = teams || searchResults; + if (searchMode && selectedTeam && !baseTeams.find(t => t.team_id === selectedTeam?.team_id)) { + return [selectedTeam, ...baseTeams]; + } + return baseTeams; }) - async function loadTeams() { + // Create separate filter text for search mode + let searchFilterText = $state('') + + // Debounced search function + const debouncedSearch = debounce(async (query: string) => { + await searchTeams(query) + }, 500) + + // Watch for search filter text changes (only in search mode) + $effect(() => { + if (searchMode) { + if (searchFilterText.length >= 1) { + debouncedSearch.debounced(searchFilterText) + } else if (searchFilterText.length === 0) { + searchResults = [] + } + } + }) + + async function searchTeams(query: string) { + if (!query) return + isFetching = true try { const response = (await WorkspaceService.listAvailableTeamsIds({ - workspace: $workspaceStore! + workspace: $workspaceStore!, + search: query })) as unknown as TeamItem[] - teams = response || [] + searchResults = response || [] isFetching = false - console.log('Teams loaded:', teams.length) - return teams + return searchResults } catch (error) { isFetching = false onError?.(error) - console.error('Error loading teams:', error) + console.error('Error searching teams:', error) + searchResults = [] return [] } } + + async function refreshSearch() { + if (searchMode && searchFilterText.length >= 2) { + await searchTeams(searchFilterText) + } + }
- = 1 || (searchFilterText.length === 0 && selectedTeam) ? displayTeams().map((team) => ({ + label: team.team_name, + value: team.team_id + })) : []} + placeholder={isFetching ? "Searching..." : "Search teams..."} + clearable + disabled={disabled || isFetching} + bind:filterText={searchFilterText} + bind:value={ + () => selectedTeam?.team_id, + (value) => { + selectedTeam = value ? displayTeams().find((team) => team.team_id === value) : undefined + } + } + /> + {:else} +