Result node message

* critical alert ui

* updating ui, backend logic

* revert

* type check fix npm

* checking out cli files from main

* moving alert icon

* adding sqlx mock data

* more sqlx changes

* feat(frontend): nodes from flow can be connected directly in expr input through a plug icon (#4652)

* Add flow prop picker

# Conflicts:
#	frontend/src/lib/components/propertyPicker/PropPicker.svelte

* fix unwanted copy

* cleaning

* Fix unset context

* move button and always display input

* fix unwanted proppicker display

* update

* update

* clean all

* clean all

---------

Co-authored-by: Ruben Fiszel <ruben@rubenfiszel.com>

* replace hide/show with toggle

* adding mutable setting and navigation to settings to configure channels

* merge fix

* ee non ee changees

* auto-acknowledge when muted

* pr comments

* fix bad log

* user inner modal component

* update unaknowledge alerts after acknowledging from modal

* aknowledge -> acknowledge

* format

* adding backend support for workspace critical alerts

* immediately check for alerts

* immediately check for alerts

* adding openapi changes

* simplify loading of superadmin/ee

* update modal logic

* frontend logic update

* sqlx prepare

* adding mute functionality for workspace critical alerts + show alerts button for instance and workspace settings

* npm / rust warnings & errors

* reverting non-mac cargo toml

* reverting non-mac cargo toml

* mute toggles in critical alert ui

* ui polish

* adding acknowledged_workspace column

* sqlx prepare

* make sure we wait for all stores to be loaded

* adding workspace mute logic to report_critical alert

* auto ack recovered alerts

* toggle workspace as superadmin and critical alerts in logs menu

* adding critical alert button if width < 786px

* refresh on change

* Move notification to logs (#4698)

* Move notification to logs

* removing unused button

---------

Co-authored-by: Alexander Petric <petric.al@gmail.com>

* tailwind typo

* auto ack on mute

* don't change deref

* sqlx prep

* avoid renaming db column, keep acknowledged instead of acknowledged_global

* hide critical alert menu when not ee

* if workspace muted, also acknowledge global when workspace set

* removing save button for mute setting and improve ux/responsiveness

* fix: deployment callbacks have a concurrency limit of 1 on same path

* sqlx prep

* ee-repo ref

* z-[9999] for modal

---------

Co-authored-by: Guilhem <guilhemlemouel@gmail.com>
Co-authored-by: Ruben Fiszel <ruben@rubenfiszel.com>
Co-authored-by: Ruben Fiszel <ruben@windmill.dev>

.github/workflows/build_windows_worker_.yml

@@ -0,0 +1,58 @@
+name: Build windows executable for this branch
+
+on:
+  workflow_dispatch:
+
+env:
+  CARGO_INCREMENTAL: 0
+  SQLX_OFFLINE: true
+  DISABLE_EMBEDDING: true
+  RUST_LOG: info
+
+jobs:
+  cargo_build_windows:
+    runs-on: windows-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Read EE repo commit hash
+        shell: pwsh
+        run: |
+          $ee_repo_ref = Get-Content .\backend\ee-repo-ref.txt
+          echo "ee_repo_ref=$ee_repo_ref" | Out-File -FilePath $env:GITHUB_ENV -Append
+
+      - name: Checkout windmill-ee-private repository
+        uses: actions/checkout@v4
+        with:
+          repository: windmill-labs/windmill-ee-private
+          path: ./windmill-ee-private
+          ref: ${{ env.ee_repo_ref }}
+          token: ${{ secrets.WINDMILL_EE_PRIVATE_ACCESS }}
+          fetch-depth: 0
+
+      - name: Substitute EE code
+        shell: bash
+        run: |
+          ./backend/substitute_ee_code.sh --copy --dir ./windmill-ee-private
+
+      - name: Cargo build windows
+        timeout-minutes: 90
+        run: |
+          vcpkg.exe install openssl-windows:x64-windows
+          vcpkg.exe install openssl:x64-windows-static
+          vcpkg.exe integrate install
+          $env:VCPKGRS_DYNAMIC=1
+          $env:OPENSSL_DIR="${Env:VCPKG_INSTALLATION_ROOT}\installed\x64-windows-static"
+          mkdir frontend/build && cd backend
+          New-Item -Path . -Name "windmill-api/openapi-deref.yaml" -ItemType "File" -Force
+          cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core
+
+      - name: Rename binary with corresponding architecture
+        run: |
+          Rename-Item -Path ".\backend\target\release\windmill.exe" -NewName "windmill-ee.exe"
+
+      - name: Upload artifact
+        uses: actions/upload-artifact@v3
+        with:
+          name: windmill-ee-binary
+          path: ./backend/target/release/windmill-ee.exe

.gitignore

@@ -6,4 +6,4 @@ frontend/src/routes/test.svelte
 CaddyfileRemoteMalo
 *.swp
 **/.idea/
-.direnv
+.direnv

CHANGELOG.md

@@ -1,5 +1,161 @@
 # Changelog
 
+## [1.424.0](https://github.com/windmill-labs/windmill/compare/v1.423.2...v1.424.0) (2024-11-14)
+
+
+### Features
+
+* allow setting password and login type from superadmin UI ([2a1bff3](https://github.com/windmill-labs/windmill/commit/2a1bff3160b65eadba399229b5f53950ec2c0d48))
+* **backend:** monitor minimal version of living workers ([#4704](https://github.com/windmill-labs/windmill/issues/4704)) ([50ff183](https://github.com/windmill-labs/windmill/commit/50ff183baeaa2a39c2b0188ee8dd6172b08ae801))
+* Support mistral anthropic for ai ([#4692](https://github.com/windmill-labs/windmill/issues/4692)) ([556b4a4](https://github.com/windmill-labs/windmill/commit/556b4a41a1d010bb8a1796f485d343c1e412d278))
+
+
+### Bug Fixes
+
+* add countCompletedJobs api ([2a78359](https://github.com/windmill-labs/windmill/commit/2a78359af7b8e4700634c2ad2745c1d14d2da4f7))
+* add queue_couts api ([10b6b1d](https://github.com/windmill-labs/windmill/commit/10b6b1dc04cb2b2d4ec5ceb26a5dbd2ac7bd1394))
+* autoscaling when count < min worker set to min_workers ([180bb82](https://github.com/windmill-labs/windmill/commit/180bb826436f9960def8c6cf3e9692714ffdf917))
+* deployment callbacks have a concurrency limit of 1 on same path ([c44fc4f](https://github.com/windmill-labs/windmill/commit/c44fc4f5ab8bcb5dffec08d5ed4ccd61feb1cb13))
+
+## [1.423.2](https://github.com/windmill-labs/windmill/compare/v1.423.1...v1.423.2) (2024-11-13)
+
+
+### Bug Fixes
+
+* fix intempestive expr type change in flow transform ([84bd06e](https://github.com/windmill-labs/windmill/commit/84bd06e90650e949c11e2b09447a7bad1ab60a95))
+
+## [1.423.1](https://github.com/windmill-labs/windmill/compare/v1.423.0...v1.423.1) (2024-11-12)
+
+
+### Bug Fixes
+
+* **autoscaling:** autoscaling thresholds to be >= and not > ([de00944](https://github.com/windmill-labs/windmill/commit/de00944f09699ff33a382f5a14f515ccf90b2454))
+
+## [1.423.0](https://github.com/windmill-labs/windmill/compare/v1.422.1...v1.423.0) (2024-11-12)
+
+
+### Features
+
+* s3 input available for public apps ([#4685](https://github.com/windmill-labs/windmill/issues/4685)) ([1671005](https://github.com/windmill-labs/windmill/commit/167100510032ab53cd609fb2c7629e67faceb093))
+
+
+### Bug Fixes
+
+* prevent underflow for autoscaling scalein ([5e9870a](https://github.com/windmill-labs/windmill/commit/5e9870a8a993032ec7d45c62e0023008da42c74a))
+* support multiple pip-extra-index-url with commas ([50861fc](https://github.com/windmill-labs/windmill/commit/50861fccc0cd86d4c76120c3306adf94f375330e))
+
+## [1.422.1](https://github.com/windmill-labs/windmill/compare/v1.422.0...v1.422.1) (2024-11-12)
+
+
+### Bug Fixes
+
+* fix password inputs ([e5e174a](https://github.com/windmill-labs/windmill/commit/e5e174ae9516f4c6b94ceb6e258b467f5c9a1f1a))
+
+## [1.422.0](https://github.com/windmill-labs/windmill/compare/v1.421.2...v1.422.0) (2024-11-11)
+
+
+### Features
+
+* expandable subflows in flows ([#4683](https://github.com/windmill-labs/windmill/issues/4683)) ([d44976f](https://github.com/windmill-labs/windmill/commit/d44976f35e45ade510d1ec220b5a1503e11f3db9))
+* **frontend:** critical alerts UI ([#4653](https://github.com/windmill-labs/windmill/issues/4653)) ([d9148ea](https://github.com/windmill-labs/windmill/commit/d9148eaa78680a93d81d71847a7df67e01f3c110))
+
+## [1.421.2](https://github.com/windmill-labs/windmill/compare/v1.421.1...v1.421.2) (2024-11-08)
+
+
+### Bug Fixes
+
+* **bash:** correctly propagate exit errors ([8bc9a02](https://github.com/windmill-labs/windmill/commit/8bc9a021a88391147c9170f56b5a0edddd55bc7d))
+
+## [1.421.1](https://github.com/windmill-labs/windmill/compare/v1.421.0...v1.421.1) (2024-11-08)
+
+
+### Bug Fixes
+
+* **python-client:** fix small break params of write_s3_file ([3ea12f1](https://github.com/windmill-labs/windmill/commit/3ea12f1821500e8b84549b892e3e1bb56a6ace4b))
+
+## [1.421.0](https://github.com/windmill-labs/windmill/compare/v1.420.1...v1.421.0) (2024-11-07)
+
+
+### Features
+
+* http custom routes for static assets ([#4666](https://github.com/windmill-labs/windmill/issues/4666)) ([dc8bd6d](https://github.com/windmill-labs/windmill/commit/dc8bd6d2b5d6f5cd8521f9034853175ec78d5639))
+
+
+### Bug Fixes
+
+* improve nested schema editor field change ([a3feca7](https://github.com/windmill-labs/windmill/commit/a3feca719799ea2bf08f2b49350e6b732a24abf4))
+
+## [1.420.1](https://github.com/windmill-labs/windmill/compare/v1.420.0...v1.420.1) (2024-11-07)
+
+
+### Bug Fixes
+
+* improve prop filtering on flow prop picker ([0501ad8](https://github.com/windmill-labs/windmill/commit/0501ad8a99f6f7d9fd26c996b754e8afe8f958b1))
+
+## [1.420.0](https://github.com/windmill-labs/windmill/compare/v1.419.0...v1.420.0) (2024-11-07)
+
+
+### Features
+
+* **frontend:** detect expr in flow input transform + filter right panel based on expr ([#4651](https://github.com/windmill-labs/windmill/issues/4651)) ([e9b7dca](https://github.com/windmill-labs/windmill/commit/e9b7dca20387e775fa50aaecd832890251582cf9))
+* **frontend:** nodes from flow can be connected directly in expr input through a plug icon ([#4652](https://github.com/windmill-labs/windmill/issues/4652)) ([ceaf56c](https://github.com/windmill-labs/windmill/commit/ceaf56c21ee2e548bc93859f9e0303e53b25b241))
+
+## [1.419.0](https://github.com/windmill-labs/windmill/compare/v1.418.0...v1.419.0) (2024-11-06)
+
+
+### Features
+
+* Add full-text search on windmill service logs ([#4576](https://github.com/windmill-labs/windmill/issues/4576)) ([77735d8](https://github.com/windmill-labs/windmill/commit/77735d859cfee9204f67a8e4f9885228d657a41d))
+* show path in flow script picker ([#4574](https://github.com/windmill-labs/windmill/issues/4574)) ([8e27392](https://github.com/windmill-labs/windmill/commit/8e27392afacbb725aaf9f9f892ab8a6171b59ce5))
+* websocket authentication ([#4635](https://github.com/windmill-labs/windmill/issues/4635)) ([4dda0fb](https://github.com/windmill-labs/windmill/commit/4dda0fb8cd8262ad3a2ab2b9d27e7043ac3bb891))
+
+
+### Bug Fixes
+
+* clarify error messages when job timeout or cancelled with more details ([771d740](https://github.com/windmill-labs/windmill/commit/771d740701902166f8b4e3f77aa9c5579237cb15))
+* **cli:** improve handling of deleted items on windows ([9a8dcc9](https://github.com/windmill-labs/windmill/commit/9a8dcc9a250caefa0b7c9523e1321599b7471c8b))
+* display logs in native mode when script fails ([#4655](https://github.com/windmill-labs/windmill/issues/4655)) ([7578ceb](https://github.com/windmill-labs/windmill/commit/7578cebaf92e729a26fd665e4b6f8357d34f59eb))
+* **frontend:** arg input json handling when the value is not of the same type as schema ([#4479](https://github.com/windmill-labs/windmill/issues/4479)) ([8d8156b](https://github.com/windmill-labs/windmill/commit/8d8156bd0773da3ddec81c46ad5fda114ecd3dda))
+* **frontend:** improve flow prop picker design ([323912c](https://github.com/windmill-labs/windmill/commit/323912c73c18d3bb6d136f2e6458389270364a0e))
+
+## [1.418.0](https://github.com/windmill-labs/windmill/compare/v1.417.3...v1.418.0) (2024-11-04)
+
+
+### Features
+
+* **frontend:** improve and simplify scheduled poll flows ([#4560](https://github.com/windmill-labs/windmill/issues/4560)) ([22ab51e](https://github.com/windmill-labs/windmill/commit/22ab51e9914376b47827f0be66708deef3a37767))
+* **oauth:** add snowflake oauth support ([#4622](https://github.com/windmill-labs/windmill/issues/4622)) ([693b7a4](https://github.com/windmill-labs/windmill/commit/693b7a4fd4da0e2b14cb6c8ea19537aa600225be))
+
+
+### Bug Fixes
+
+* **prometheus:** improve queue_count when tags have no more jobs ([09156b6](https://github.com/windmill-labs/windmill/commit/09156b65c2a0f0ac59fce1dede2b68b790323c49))
+
+## [1.417.3](https://github.com/windmill-labs/windmill/compare/v1.417.2...v1.417.3) (2024-11-04)
+
+
+### Bug Fixes
+
+* **cli:** improve pulling instance with folderPerInstance ([1e7909e](https://github.com/windmill-labs/windmill/commit/1e7909e95931e1e121cb3767bad093239254935d))
+
+## [1.417.2](https://github.com/windmill-labs/windmill/compare/v1.417.1...v1.417.2) (2024-11-04)
+
+
+### Bug Fixes
+
+* **cli:** improve pulling instance with folderPerInstance ([00748f4](https://github.com/windmill-labs/windmill/commit/00748f43feab4f28f0dd2682e3dc8ccc15a5f49f))
+
+## [1.417.1](https://github.com/windmill-labs/windmill/compare/v1.417.0...v1.417.1) (2024-11-04)
+
+
+### Bug Fixes
+
+* missing opts for pull and push instance configs ([#4630](https://github.com/windmill-labs/windmill/issues/4630)) ([e92a338](https://github.com/windmill-labs/windmill/commit/e92a338432d2732e6be74c3604b9fa95239e1ca3))
+
+## 1.417.0 (2024-11-03)
+
+*  feat(cli): instance yaml files inside the instance prefix folders (#4627)
+
 ## [1.416.2](https://github.com/windmill-labs/windmill/compare/v1.416.1...v1.416.2) (2024-11-02)
 
 

Dockerfile

@@ -175,9 +175,9 @@ RUN /usr/local/bin/python3 -m pip install pip-tools
 COPY --from=builder /frontend/build /static_frontend
 COPY --from=builder /windmill/target/release/windmill ${APP}/windmill
 
-COPY --from=denoland/deno:2.0.2 --chmod=755 /usr/bin/deno /usr/bin/deno
+COPY --from=denoland/deno:2.0.4 --chmod=755 /usr/bin/deno /usr/bin/deno
 
-COPY --from=oven/bun:1.1.32 /usr/local/bin/bun /usr/bin/bun
+COPY --from=oven/bun:1.1.34 /usr/local/bin/bun /usr/bin/bun
 
 COPY --from=php:8.3.7-cli /usr/local/bin/php /usr/bin/php
 COPY --from=composer:2.7.6 /usr/bin/composer /usr/bin/composer

README.md

@@ -355,6 +355,8 @@ you to have it being synced automatically everyday.
 See the [./frontend/README_DEV.md](./frontend/README_DEV.md) file for all
 running options.
 
+Using [Nix](./frontend/README_DEV.md#nix).
+
 ### only Frontend
 This will use the backend of <https://app.windmill.dev> but your own frontend
 with hot-code reloading. Note that you will need to use a username / password login due to CSRF checks using a different auth provider.

backend/.gitignore

@@ -2,7 +2,7 @@ target/
 .env
 oauth.json
 oauth2.json
-windmill-api/openapi-deref.yaml
 tracing.folded
 heaptrack*
 index/
+windmill-api/openapi-*.*

backend/.sqlx/query-00588a40dde5189ac1c61505f17acb0f4c244c60477427505bf5bd1b104d3bf9.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE alerts SET acknowledged_workspace = true, acknowledged = true WHERE workspace_id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "00588a40dde5189ac1c61505f17acb0f4c244c60477427505bf5bd1b104d3bf9"
+}

backend/.sqlx/query-00ce4ed3ca0eac7cb6283b047353a64b9e78c4beb423f04baef9a53fbf87e9f9.json

@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO alerts (alert_type, message) VALUES ('recovered_critical_error', $1)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "00ce4ed3ca0eac7cb6283b047353a64b9e78c4beb423f04baef9a53fbf87e9f9"
-}

backend/.sqlx/query-0230bd64d2b6719c3536a106e18a68c7b43b3a9a9efa92b5517132e9c0d8a25b.json

@@ -1,12 +1,12 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT openai_resource_path, code_completion_enabled FROM workspace_settings WHERE workspace_id = $1",
+  "query": "SELECT ai_resource, code_completion_enabled FROM workspace_settings WHERE workspace_id = $1",
   "describe": {
     "columns": [
       {
         "ordinal": 0,
-        "name": "openai_resource_path",
-        "type_info": "Varchar"
+        "name": "ai_resource",
+        "type_info": "Jsonb"
       },
       {
         "ordinal": 1,
@@ -24,5 +24,5 @@
       false
     ]
   },
-  "hash": "ceb97024ebf1a1c00ea1ed4952f66b229ca4622c537cc252d8ed52b4d24270ee"
+  "hash": "0230bd64d2b6719c3536a106e18a68c7b43b3a9a9efa92b5517132e9c0d8a25b"
 }

backend/.sqlx/query-0331a81262e2d3c1bcfaeb64617b11eb68ab4599d64e5e5af639a9ac5d791fd0.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT\n                -- slack_team_id, \n                -- slack_name, \n                -- slack_command_script, \n                -- CASE WHEN slack_email = 'missing@email.xyz' THEN NULL ELSE slack_email END AS slack_email,\n                auto_invite_domain IS NOT NULL AS \"auto_invite_enabled!\",\n                CASE WHEN auto_invite_operator IS TRUE THEN 'operator' ELSE 'developer' END AS \"auto_invite_as!\", \n                CASE WHEN auto_add IS TRUE THEN 'add' ELSE 'invite' END AS \"auto_invite_mode!\", \n                webhook, \n                deploy_to, \n                error_handler, \n                openai_resource_path, \n                code_completion_enabled, \n                error_handler_extra_args, \n                error_handler_muted_on_cancel, \n                large_file_storage, \n                git_sync,\n                default_app,\n                default_scripts,\n                workspace.name\n            FROM workspace_settings\n            LEFT JOIN workspace ON workspace.id = workspace_settings.workspace_id\n            WHERE workspace_id = $1",
+  "query": "SELECT\n                -- slack_team_id, \n                -- slack_name, \n                -- slack_command_script, \n                -- CASE WHEN slack_email = 'missing@email.xyz' THEN NULL ELSE slack_email END AS slack_email,\n                auto_invite_domain IS NOT NULL AS \"auto_invite_enabled!\",\n                CASE WHEN auto_invite_operator IS TRUE THEN 'operator' ELSE 'developer' END AS \"auto_invite_as!\", \n                CASE WHEN auto_add IS TRUE THEN 'add' ELSE 'invite' END AS \"auto_invite_mode!\", \n                webhook, \n                deploy_to, \n                error_handler, \n                ai_resource, \n                code_completion_enabled, \n                error_handler_extra_args, \n                error_handler_muted_on_cancel, \n                large_file_storage, \n                git_sync,\n                default_app,\n                default_scripts,\n                workspace.name\n            FROM workspace_settings\n            LEFT JOIN workspace ON workspace.id = workspace_settings.workspace_id\n            WHERE workspace_id = $1",
   "describe": {
     "columns": [
       {
@@ -35,8 +35,8 @@
       },
       {
         "ordinal": 6,
-        "name": "openai_resource_path",
-        "type_info": "Varchar"
+        "name": "ai_resource",
+        "type_info": "Jsonb"
       },
       {
         "ordinal": 7,
@@ -102,5 +102,5 @@
       false
     ]
   },
-  "hash": "188534f4b29f6461b1a6214d060f183c830b19a403ebb7b8be55a691675010c3"
+  "hash": "0331a81262e2d3c1bcfaeb64617b11eb68ab4599d64e5e5af639a9ac5d791fd0"
 }

backend/.sqlx/query-034583442e6f8ae38d6c4e4aac26f17c8d9d0e657f28276228fc90d3e22e1304.json

@@ -1,16 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE workspace_settings SET openai_resource_path = $1, code_completion_enabled = $2 WHERE workspace_id = $3",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Bool",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "034583442e6f8ae38d6c4e4aac26f17c8d9d0e657f28276228fc90d3e22e1304"
-}

backend/.sqlx/query-044e2b428ee6e2dd4543c87ad8835e239cf7567d18b8b3fa6608ea3a9d206ca7.json

@@ -0,0 +1,18 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO alerts (alert_type, message, acknowledged, acknowledged_workspace, workspace_id, resource)\n        VALUES ('critical_error', $1, $2, $3, $4, $5)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Bool",
+        "Bool",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "044e2b428ee6e2dd4543c87ad8835e239cf7567d18b8b3fa6608ea3a9d206ca7"
+}

backend/.sqlx/query-0b94bd4c98a11ca1b7e5e34dd1ee6fcb0b7a54ed4218fa3cf23cc929d009d50f.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, edited_by = $6, email = $7, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL\n            WHERE workspace_id = $8 AND path = $9",
+  "query": "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, initial_messages = $6, url_runnable_args = $7, edited_by = $8, email = $9, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL\n            WHERE workspace_id = $10 AND path = $11",
   "describe": {
     "columns": [],
     "parameters": {
@@ -10,6 +10,8 @@
         "Varchar",
         "Bool",
         "JsonbArray",
+        "JsonbArray",
+        "Jsonb",
         "Varchar",
         "Varchar",
         "Text",
@@ -18,5 +20,5 @@
     },
     "nullable": []
   },
-  "hash": "acbf74cf3302bfcf7615285070d3f8958932bb8a2dda715f1b9152ab44442780"
+  "hash": "0b94bd4c98a11ca1b7e5e34dd1ee6fcb0b7a54ed4218fa3cf23cc929d009d50f"
 }

backend/.sqlx/query-0ee63ef2dd5c88edba2a1f56d31f29876724922f148dad7af35b36efbf70207a.json

@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM healthchecks WHERE check_type = $1",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "0ee63ef2dd5c88edba2a1f56d31f29876724922f148dad7af35b36efbf70207a"
-}

backend/.sqlx/query-11b698f82a54aac68b3617047dfe2b18dd6da7d962118fee276af354218baac2.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\" FROM http_trigger",
+  "query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\", static_asset_config as \"static_asset_config: _\" FROM http_trigger",
   "describe": {
     "columns": [
       {
@@ -65,6 +65,11 @@
             }
           }
         }
+      },
+      {
+        "ordinal": 10,
+        "name": "static_asset_config: _",
+        "type_info": "Jsonb"
       }
     ],
     "parameters": {
@@ -80,8 +85,9 @@
       false,
       false,
       false,
-      false
+      false,
+      true
     ]
   },
-  "hash": "fe4f91ca7e179e58041a5c0b1a36015175e9301c01a99e8f2b82cb853349a183"
+  "hash": "11b698f82a54aac68b3617047dfe2b18dd6da7d962118fee276af354218baac2"
 }

backend/.sqlx/query-1610c79b8d238e3a35d795f34a2a0100b933b7f199d41cc2f554d57c811d55f3.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings SET ai_resource = $1, code_completion_enabled = $2 WHERE workspace_id = $3",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Jsonb",
+        "Bool",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "1610c79b8d238e3a35d795f34a2a0100b933b7f199d41cc2f554d57c811d55f3"
+}

backend/.sqlx/query-1730f39fd1793d45fbb41b21389c61296a3ff7489ae12f52a19f9543173ac597.json

@@ -65,8 +65,8 @@
       },
       {
         "ordinal": 12,
-        "name": "openai_resource_path",
-        "type_info": "Varchar"
+        "name": "ai_resource",
+        "type_info": "Jsonb"
       },
       {
         "ordinal": 13,
@@ -117,6 +117,11 @@
         "ordinal": 22,
         "name": "deploy_ui",
         "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 23,
+        "name": "mute_critical_alerts",
+        "type_info": "Bool"
       }
     ],
     "parameters": {
@@ -147,6 +152,7 @@
       true,
       false,
       true,
+      true,
       true
     ]
   },

backend/.sqlx/query-1e7ce0c140410ae799f9c0c5772e4be4506bfd238c88f1b1c3ddf39b29071446.json

@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT\n                flow_status AS \"flow_status!: Json<Box<RawValue>>\",\n                raw_flow->'modules'->(flow_status->'step')::int AS \"module: Json<Box<RawValue>>\"\n            FROM queue WHERE id = $1 AND workspace_id = $2 LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "flow_status!: Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 1,
+        "name": "module: Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      true,
+      null
+    ]
+  },
+  "hash": "1e7ce0c140410ae799f9c0c5772e4be4506bfd238c88f1b1c3ddf39b29071446"
+}

backend/.sqlx/query-286fa00c088df146c08c1934556f06a4243c66787ab8759a8045e60effd2fb77.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT value\n            FROM resource\n            WHERE path = $1 AND workspace_id = $2",
+  "query": "SELECT value\n                FROM resource\n                WHERE path = $1 AND workspace_id = $2",
   "describe": {
     "columns": [
       {
@@ -19,5 +19,5 @@
       true
     ]
   },
-  "hash": "acdaa5151f8f7f37bb8c8c5a7d146789887e47db9695fc26b1dfaedd735e1e60"
+  "hash": "286fa00c088df146c08c1934556f06a4243c66787ab8759a8045e60effd2fb77"
 }

backend/.sqlx/query-28987898c7e4b172d466bf08f33c2da733e71f8a253b0122c17b9e021919809e.json

@@ -0,0 +1,18 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO alerts (alert_type, message, acknowledged, acknowledged_workspace, workspace_id, resource)\n        VALUES ('recovered_critical_error', $1, $2, $3, $4, $5)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Bool",
+        "Bool",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "28987898c7e4b172d466bf08f33c2da733e71f8a253b0122c17b9e021919809e"
+}

backend/.sqlx/query-31dd23f6768052e486668172e15eba1a3f9b6a8e5678a7ee8e5456ff4405d6f9.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT raw_flow->'failure_module' != 'null'::jsonb\n            FROM completed_job\n            WHERE id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "31dd23f6768052e486668172e15eba1a3f9b6a8e5678a7ee8e5456ff4405d6f9"
+}

backend/.sqlx/query-333b484ffa030dee08e7b1161fcbc48af411377d2d9f58f92fc9d5eacdf0fba1.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "INSERT INTO http_trigger (workspace_id, path, route_path, route_path_key, script_path, is_flow, is_async, requires_auth, http_method, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now())",
+  "query": "INSERT INTO http_trigger (workspace_id, path, route_path, route_path_key, script_path, is_flow, is_async, requires_auth, http_method, static_asset_config, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now())",
   "describe": {
     "columns": [],
     "parameters": {
@@ -27,11 +27,12 @@
             }
           }
         },
+        "Jsonb",
         "Varchar",
         "Varchar"
       ]
     },
     "nullable": []
   },
-  "hash": "c229744534f17f7622c3dee21bb1e7292ff17e6dffe58e80e53bff8baade07c8"
+  "hash": "333b484ffa030dee08e7b1161fcbc48af411377d2d9f58f92fc9d5eacdf0fba1"
 }

backend/.sqlx/query-344b3a5d9683273a956b5156fed5ab9fdff1a7252198e4ae290d8c8f937ff177.json

@@ -0,0 +1,54 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, alert_type, message, created_at, COALESCE(acknowledged, false) AS acknowledged, workspace_id\n                 FROM alerts\n                 WHERE COALESCE(acknowledged, false) = $1\n                 ORDER BY created_at DESC\n                 LIMIT $2 OFFSET $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 1,
+        "name": "alert_type",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "message",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 4,
+        "name": "acknowledged",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 5,
+        "name": "workspace_id",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Bool",
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      null,
+      true
+    ]
+  },
+  "hash": "344b3a5d9683273a956b5156fed5ab9fdff1a7252198e4ae290d8c8f937ff177"
+}

backend/.sqlx/query-349396e8fdd96d45875110bc06767e6eb876792ec8f83e9b03c2fb46bb12e0b9.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE password SET login_type = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "349396e8fdd96d45875110bc06767e6eb876792ec8f83e9b03c2fb46bb12e0b9"
+}

backend/.sqlx/query-487d377e2df67fc3ea39d183ba9f99d45828d7c8e0ff10c5d74c454472e0493c.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "UPDATE http_trigger \n                SET route_path = $1, route_path_key = $2, script_path = $3, path = $4, is_flow = $5, http_method = $6, edited_by = $7, email = $8, is_async = $9, requires_auth = $10, edited_at = now() \n                WHERE workspace_id = $11 AND path = $12",
+  "query": "UPDATE http_trigger \n                SET route_path = $1, route_path_key = $2, script_path = $3, path = $4, is_flow = $5, http_method = $6, static_asset_config = $7, edited_by = $8, email = $9, is_async = $10, requires_auth = $11, edited_at = now() \n                WHERE workspace_id = $12 AND path = $13",
   "describe": {
     "columns": [],
     "parameters": {
@@ -24,6 +24,7 @@
             }
           }
         },
+        "Jsonb",
         "Varchar",
         "Varchar",
         "Bool",
@@ -34,5 +35,5 @@
     },
     "nullable": []
   },
-  "hash": "4d8640e84fccf1a0b799d8396a51e69345137e68d5096c70ba0a4332075d97ea"
+  "hash": "487d377e2df67fc3ea39d183ba9f99d45828d7c8e0ff10c5d74c454472e0493c"
 }

backend/.sqlx/query-4a3a8207627418ba7b7eabaccd7ec72884f2a1eacf0ab0e0c6ccbf56c654c14a.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE alerts SET acknowledged = true, acknowledged_workspace = true WHERE resource = $1 AND alert_type = 'critical_error'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "4a3a8207627418ba7b7eabaccd7ec72884f2a1eacf0ab0e0c6ccbf56c654c14a"
+}

backend/.sqlx/query-4c0067c2135a259aea5cc2db60f7375a9a33671be8ef406427d90f67a98c9c9f.json

@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO alerts (alert_type, message) VALUES ('critical_error', $1)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "4c0067c2135a259aea5cc2db60f7375a9a33671be8ef406427d90f67a98c9c9f"
-}

backend/.sqlx/query-4d30c5a2894d655741d167f5589f5816583f0bae78a0dd3270dedb377dfa944a.json

@@ -0,0 +1,53 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, alert_type, message, created_at, COALESCE(acknowledged, false) AS acknowledged, workspace_id\n                 FROM alerts\n                 ORDER BY created_at DESC\n                 LIMIT $1 OFFSET $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 1,
+        "name": "alert_type",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "message",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 4,
+        "name": "acknowledged",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 5,
+        "name": "workspace_id",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      null,
+      true
+    ]
+  },
+  "hash": "4d30c5a2894d655741d167f5589f5816583f0bae78a0dd3270dedb377dfa944a"
+}

backend/.sqlx/query-4e9668a46bad9e82baa51422946d373b18b6577198df7545c94bd19be3446775.json

@@ -1,108 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO websocket_trigger (workspace_id, path, url, script_path, is_flow, enabled, filters, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now()) RETURNING *",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "path",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 1,
-        "name": "url",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 2,
-        "name": "script_path",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 3,
-        "name": "is_flow",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 4,
-        "name": "workspace_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 5,
-        "name": "edited_by",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 6,
-        "name": "email",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 7,
-        "name": "edited_at",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 8,
-        "name": "extra_perms",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 9,
-        "name": "server_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 10,
-        "name": "last_server_ping",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 11,
-        "name": "error",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 12,
-        "name": "enabled",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 13,
-        "name": "filters",
-        "type_info": "JsonbArray"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "Bool",
-        "Bool",
-        "JsonbArray",
-        "Varchar",
-        "Varchar"
-      ]
-    },
-    "nullable": [
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      true,
-      true,
-      true,
-      false,
-      false
-    ]
-  },
-  "hash": "4e9668a46bad9e82baa51422946d373b18b6577198df7545c94bd19be3446775"
-}

backend/.sqlx/query-5303cb9dd5903aa4791ef8e5e5881a50a832e65c8c9632e2e12cd9c2747f2fc7.json

@@ -1,98 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT *\n            FROM websocket_trigger\n            WHERE enabled IS TRUE AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "path",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 1,
-        "name": "url",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 2,
-        "name": "script_path",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 3,
-        "name": "is_flow",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 4,
-        "name": "workspace_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 5,
-        "name": "edited_by",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 6,
-        "name": "email",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 7,
-        "name": "edited_at",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 8,
-        "name": "extra_perms",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 9,
-        "name": "server_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 10,
-        "name": "last_server_ping",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 11,
-        "name": "error",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 12,
-        "name": "enabled",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 13,
-        "name": "filters",
-        "type_info": "JsonbArray"
-      }
-    ],
-    "parameters": {
-      "Left": []
-    },
-    "nullable": [
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      true,
-      true,
-      true,
-      false,
-      false
-    ]
-  },
-  "hash": "5303cb9dd5903aa4791ef8e5e5881a50a832e65c8c9632e2e12cd9c2747f2fc7"
-}

backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json

@@ -65,8 +65,8 @@
       },
       {
         "ordinal": 12,
-        "name": "openai_resource_path",
-        "type_info": "Varchar"
+        "name": "ai_resource",
+        "type_info": "Jsonb"
       },
       {
         "ordinal": 13,
@@ -117,6 +117,11 @@
         "ordinal": 22,
         "name": "deploy_ui",
         "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 23,
+        "name": "mute_critical_alerts",
+        "type_info": "Bool"
       }
     ],
     "parameters": {
@@ -147,6 +152,7 @@
       true,
       false,
       true,
+      true,
       true
     ]
   },

backend/.sqlx/query-6268eabd561502a44e273d6391102278974623f7a7bcad6891d7abadf4d3ea03.json

@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE workspace_settings SET openai_resource_path = NULL, code_completion_enabled = $1 WHERE workspace_id = $2",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Bool",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "6268eabd561502a44e273d6391102278974623f7a7bcad6891d7abadf4d3ea03"
-}

backend/.sqlx/query-65da41c7ded54cdee8d33211561c068b72294cc99ff44ed0a13179df508ebc6a.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE alerts\n         SET\n           acknowledged = true,\n           acknowledged_workspace = CASE\n             WHEN $2::text IS NOT NULL AND workspace_id = $2 THEN true\n             ELSE acknowledged_workspace\n           END\n         WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int4",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "65da41c7ded54cdee8d33211561c068b72294cc99ff44ed0a13179df508ebc6a"
+}

backend/.sqlx/query-6940ddc5ee8d2a1048213d46f52d964b199604ba66dedbe9f89cea727d726a2d.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings SET ai_resource = NULL, code_completion_enabled = $1 WHERE workspace_id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Bool",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "6940ddc5ee8d2a1048213d46f52d964b199604ba66dedbe9f89cea727d726a2d"
+}

backend/.sqlx/query-7113d7cc72e44e4b7e01b69cc18cbe7b0399cf8ec0e9e6d2b05ceef589c432df.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "UPDATE http_trigger SET script_path = $1, path = $2, is_flow = $3, http_method = $4, edited_by = $5, email = $6, is_async = $7, requires_auth = $8, edited_at = now() \n                WHERE workspace_id = $9 AND path = $10",
+  "query": "UPDATE http_trigger SET script_path = $1, path = $2, is_flow = $3, http_method = $4, static_asset_config = $5, edited_by = $6, email = $7, is_async = $8, requires_auth = $9, edited_at = now() \n                WHERE workspace_id = $10 AND path = $11",
   "describe": {
     "columns": [],
     "parameters": {
@@ -22,6 +22,7 @@
             }
           }
         },
+        "Jsonb",
         "Varchar",
         "Varchar",
         "Bool",
@@ -32,5 +33,5 @@
     },
     "nullable": []
   },
-  "hash": "55d44f569f8ebfccddf975e1a330ef0dc286f4138efe923832371cdbac7157b0"
+  "hash": "7113d7cc72e44e4b7e01b69cc18cbe7b0399cf8ec0e9e6d2b05ceef589c432df"
 }

backend/.sqlx/query-777e7084edf9a5d14f299ff479ae43c8ead47a6a531d4b031a1342c9b2f16506.json

@@ -0,0 +1,54 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, alert_type, message, created_at, COALESCE(acknowledged_workspace, false) AS acknowledged, workspace_id\n                 FROM alerts\n                 WHERE workspace_id = $1\n                 ORDER BY created_at DESC\n                 LIMIT $2 OFFSET $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 1,
+        "name": "alert_type",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "message",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 4,
+        "name": "acknowledged",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 5,
+        "name": "workspace_id",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      null,
+      true
+    ]
+  },
+  "hash": "777e7084edf9a5d14f299ff479ae43c8ead47a6a531d4b031a1342c9b2f16506"
+}

backend/.sqlx/query-7c32176755c6ea2b6ae531860d436caae3fa256fc0803749ec5107632669adb3.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE alerts \n         SET\n           acknowledged = true,\n           acknowledged_workspace = CASE\n             WHEN $1::text IS NOT NULL THEN true\n             ELSE acknowledged_workspace\n           END\n         WHERE ($1::text IS NOT NULL AND workspace_id = $1)\n            OR ($1::text IS NULL)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "7c32176755c6ea2b6ae531860d436caae3fa256fc0803749ec5107632669adb3"
+}

backend/.sqlx/query-7c5a29de07cbe42326a15d4d7fd9714c20b767508e27368be6de38d7b18a3a4d.json

@@ -0,0 +1,55 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, alert_type, message, created_at, COALESCE(acknowledged_workspace, false) AS acknowledged, workspace_id\n                 FROM alerts\n                 WHERE workspace_id = $1 AND COALESCE(acknowledged_workspace, false) = $2\n                 ORDER BY created_at DESC\n                 LIMIT $3 OFFSET $4",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 1,
+        "name": "alert_type",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "message",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 4,
+        "name": "acknowledged",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 5,
+        "name": "workspace_id",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Bool",
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      null,
+      true
+    ]
+  },
+  "hash": "7c5a29de07cbe42326a15d4d7fd9714c20b767508e27368be6de38d7b18a3a4d"
+}

backend/.sqlx/query-a59fae29ebcc9aa53308b777ead3d2b652618ac454f139db738f4499cb4eb079.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE alerts SET acknowledged = true",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "a59fae29ebcc9aa53308b777ead3d2b652618ac454f139db738f4499cb4eb079"
+}

backend/.sqlx/query-a69ba7471d1a5faa145bebbd17d43e6dbe02e9e92ebbc31a50c99c3a04719284.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "WITH uuid_table as (\n            select gen_random_uuid() as uuid from generate_series(1, $11)\n        )\n        INSERT INTO queue \n            (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout)\n            (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14 FROM uuid_table) \n        RETURNING id",
+  "query": "WITH uuid_table as (\n            select gen_random_uuid() as uuid from generate_series(1, $11)\n        )\n        INSERT INTO queue \n            (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout, raw_code, raw_lock, raw_flow, flow_status)\n            (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14, $15, $16, $17, $18 FROM uuid_table) \n        RETURNING id",
   "describe": {
     "columns": [
       {
@@ -72,12 +72,16 @@
         "Int4",
         "Int4",
         "Int4",
-        "Int4"
+        "Int4",
+        "Text",
+        "Text",
+        "Jsonb",
+        "Jsonb"
       ]
     },
     "nullable": [
       false
     ]
   },
-  "hash": "0a686ca61444d7ad7484071727aa039a6ea6697e5a49a633b767c052aa3e0a18"
+  "hash": "a69ba7471d1a5faa145bebbd17d43e6dbe02e9e92ebbc31a50c99c3a04719284"
 }

backend/.sqlx/query-aa6907b7266ee437dfbd77a9bf6c047fda88f3e72a0d10323a5e4020cf857c42.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT raw_flow->'failure_module' != 'null'::jsonb\n            FROM queue\n            WHERE id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "aa6907b7266ee437dfbd77a9bf6c047fda88f3e72a0d10323a5e4020cf857c42"
+}

backend/.sqlx/query-ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT wm_version FROM worker_ping WHERE wm_version != $1 AND ping_at > now() - interval '5 minutes'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "wm_version",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6"
+}

backend/.sqlx/query-ae543dfa106fa6ad4e9bf45cda1110d4702a80f455c63af6fcbd7cf45bbc4f7a.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT flow_version.value AS \"value: sqlx::types::Json<Box<RawValue>>\" FROM flow \n                    LEFT JOIN flow_version\n                        ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n                    WHERE flow.path = $1 AND flow.workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "ae543dfa106fa6ad4e9bf45cda1110d4702a80f455c63af6fcbd7cf45bbc4f7a"
+}

backend/.sqlx/query-b5c839baab25c4dcdd503d380cf7a886242277cd50555f20b2e22e13942d2a3a.json

@@ -0,0 +1,65 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT\n                        hostname,\n                        mode::text,\n                        worker_group,\n                        log_ts,\n                        file_path,\n                        ok_lines,\n                        err_lines,\n                        json_fmt\n                    FROM log_file\n                    WHERE log_ts > $1\n                    ORDER BY log_ts ASC LIMIT $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "hostname",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "mode",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "worker_group",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "log_ts",
+        "type_info": "Timestamp"
+      },
+      {
+        "ordinal": 4,
+        "name": "file_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "ok_lines",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 6,
+        "name": "err_lines",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 7,
+        "name": "json_fmt",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Timestamp",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      null,
+      true,
+      false,
+      false,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "b5c839baab25c4dcdd503d380cf7a886242277cd50555f20b2e22e13942d2a3a"
+}

backend/.sqlx/query-c9930fcfe79541af570eace58ba7e15a0816a6b4fd036cf7b991a210654b2633.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\" FROM http_trigger WHERE workspace_id = $1",
+  "query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\", static_asset_config as \"static_asset_config: _\" FROM http_trigger WHERE workspace_id = $1",
   "describe": {
     "columns": [
       {
@@ -65,6 +65,11 @@
             }
           }
         }
+      },
+      {
+        "ordinal": 10,
+        "name": "static_asset_config: _",
+        "type_info": "Jsonb"
       }
     ],
     "parameters": {
@@ -82,8 +87,9 @@
       false,
       false,
       false,
-      false
+      false,
+      true
     ]
   },
-  "hash": "02f1a6eeb27067dc438459238e7b016f5ccf9e3fe0ffbe88471f15aad8f74441"
+  "hash": "c9930fcfe79541af570eace58ba7e15a0816a6b4fd036cf7b991a210654b2633"
 }

backend/.sqlx/query-cd33a9d63f4706a7e3b1e23cd0a4b2e3ecb30aae6510d1fcd08493b07c8b0952.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE websocket_trigger SET enabled = FALSE, error = $1, server_id = NULL, last_server_ping = NULL WHERE workspace_id = $2 AND path = $3",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "cd33a9d63f4706a7e3b1e23cd0a4b2e3ecb30aae6510d1fcd08493b07c8b0952"
+}

backend/.sqlx/query-ceda377fa534656ac12a7f41db77db1f054ec751855c8db7352b0e9c20b63ef8.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT ai_resource FROM workspace_settings WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "ai_resource",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "ceda377fa534656ac12a7f41db77db1f054ec751855c8db7352b0e9c20b63ef8"
+}

backend/.sqlx/query-dd967c5983fa0ff05e2b320ad0e0b5a152784826cb8fb4381c1ffe228cb7feb6.json

@@ -0,0 +1,64 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT\n                        hostname,\n                        mode::text,\n                        worker_group,\n                        log_ts,\n                        file_path,\n                        ok_lines,\n                        err_lines,\n                        json_fmt\n                    FROM log_file\n                    ORDER BY log_ts ASC LIMIT $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "hostname",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "mode",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "worker_group",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "log_ts",
+        "type_info": "Timestamp"
+      },
+      {
+        "ordinal": 4,
+        "name": "file_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "ok_lines",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 6,
+        "name": "err_lines",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 7,
+        "name": "json_fmt",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      null,
+      true,
+      false,
+      false,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "dd967c5983fa0ff05e2b320ad0e0b5a152784826cb8fb4381c1ffe228cb7feb6"
+}

backend/.sqlx/query-de1abe57b6aa61155f747a3bcb98359f70eade6654b5a884915f07f3ef3fe15e.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM queue WHERE id = $2",
+  "query": "SELECT raw_flow->'modules'->($1)::text->'value'->>'type' = 'flow' FROM queue WHERE id = $2 LIMIT 1",
   "describe": {
     "columns": [
       {
@@ -19,5 +19,5 @@
       null
     ]
   },
-  "hash": "3e539fef054ad31bc1736e27276087775a721a6ee7ae35b03fd4ce3563ea3838"
+  "hash": "de1abe57b6aa61155f747a3bcb98359f70eade6654b5a884915f07f3ef3fe15e"
 }

backend/.sqlx/query-e05dbe046e846c092a96b6b0a9d872c721169d418586be2d15cee6b929049098.json

@@ -1,12 +1,12 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT openai_resource_path FROM workspace_settings WHERE workspace_id = $1",
+  "query": "SELECT mute_critical_alerts FROM workspace_settings WHERE workspace_id = $1",
   "describe": {
     "columns": [
       {
         "ordinal": 0,
-        "name": "openai_resource_path",
-        "type_info": "Varchar"
+        "name": "mute_critical_alerts",
+        "type_info": "Bool"
       }
     ],
     "parameters": {
@@ -18,5 +18,5 @@
       true
     ]
   },
-  "hash": "f8b689ec3e09f14dc02ac9e9f98d252a14d3a2acfadcd9cd6ba27ca9799f4706"
+  "hash": "e05dbe046e846c092a96b6b0a9d872c721169d418586be2d15cee6b929049098"
 }

backend/.sqlx/query-f22168826350797e88153b65a5b1e906a7868f34c062f154b2ee4f24c57aa5c3.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings SET mute_critical_alerts = $1 WHERE workspace_id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Bool",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f22168826350797e88153b65a5b1e906a7868f34c062f154b2ee4f24c57aa5c3"
+}

backend/.sqlx/query-f2baee15e6d1fecd6d2d7b39fda1b50a15ec8bd349f1081af54da5dc2f5e3021.json

@@ -1,101 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT *\n          FROM websocket_trigger\n          WHERE workspace_id = $1 AND path = $2",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "path",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 1,
-        "name": "url",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 2,
-        "name": "script_path",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 3,
-        "name": "is_flow",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 4,
-        "name": "workspace_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 5,
-        "name": "edited_by",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 6,
-        "name": "email",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 7,
-        "name": "edited_at",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 8,
-        "name": "extra_perms",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 9,
-        "name": "server_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 10,
-        "name": "last_server_ping",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 11,
-        "name": "error",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 12,
-        "name": "enabled",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 13,
-        "name": "filters",
-        "type_info": "JsonbArray"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": [
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      false,
-      true,
-      true,
-      true,
-      false,
-      false
-    ]
-  },
-  "hash": "f2baee15e6d1fecd6d2d7b39fda1b50a15ec8bd349f1081af54da5dc2f5e3021"
-}

backend/.sqlx/query-f904702536c106b0e5da8facae119c6af887c49a29ae44b3a95350ff27fb1ccf.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, http_method as \"http_method: _\", edited_by, email, edited_at, extra_perms, is_async, requires_auth\n            FROM http_trigger\n            WHERE workspace_id = $1 AND path = $2",
+  "query": "SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, http_method as \"http_method: _\", edited_by, email, edited_at, extra_perms, is_async, requires_auth, static_asset_config as \"static_asset_config: _\"\n            FROM http_trigger\n            WHERE workspace_id = $1 AND path = $2",
   "describe": {
     "columns": [
       {
@@ -80,6 +80,11 @@
         "ordinal": 12,
         "name": "requires_auth",
         "type_info": "Bool"
+      },
+      {
+        "ordinal": 13,
+        "name": "static_asset_config: _",
+        "type_info": "Jsonb"
       }
     ],
     "parameters": {
@@ -101,8 +106,9 @@
       false,
       false,
       false,
-      false
+      false,
+      true
     ]
   },
-  "hash": "4fb95eae1c871241efe2ef79615ce03cba0e4a12aad3274e4829d98e38ca1491"
+  "hash": "f904702536c106b0e5da8facae119c6af887c49a29ae44b3a95350ff27fb1ccf"
 }

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "windmill"
-version = "1.416.2"
+version = "1.424.0"
 authors.workspace = true
 edition.workspace = true
 
@@ -29,7 +29,7 @@ members = [
 ]
 
 [workspace.package]
-version = "1.416.2"
+version = "1.424.0"
 authors = ["Ruben Fiszel <ruben@windmill.dev>"]
 edition = "2021"
 
@@ -154,7 +154,7 @@ tracing-appender = "^0"
 prometheus = { version = "^0", default-features = false }
 cookie = { version = "0.17.0" }
 phf = { version = "0.11", features = ["macros"] }
-rust-embed = "^6"
+rust-embed = { version = "^6", features = ["interpolate-folder-path"] }
 mime_guess = "^2"
 hex = "^0"
 sql-builder = "^3"
@@ -180,6 +180,7 @@ tokio-util = { version = "^0", features = ["io"] }
 json-pointer = "^0"
 itertools = "^0"
 regex = "^1"
+semver = "^1"
 
 deno_fetch = "0.195.0"
 deno_tls = "0.158.0"

backend/ee-repo-ref.txt

@@ -1 +1 @@
-f136a2f499e0fe7c10c54c79488851980d796eb2
+51dcbf93b0d127af9f33fa346cc63fcd2475d4fa

backend/migrations/20241030150214_ws_auth.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+ALTER TABLE websocket_trigger DROP COLUMN initial_messages, DROP COLUMN url_runnable_args;

backend/migrations/20241030150214_ws_auth.up.sql

@@ -0,0 +1,2 @@
+-- Add up migration script here
+ALTER TABLE websocket_trigger ADD COLUMN initial_messages JSONB[] DEFAULT '{}', ADD COLUMN url_runnable_args JSONB DEFAULT '{}';

backend/migrations/20241105133531_alter_openai_resource_path_column_in_workspace_settings.down.sql

@@ -0,0 +1,7 @@
+-- Add down migration script here
+ALTER TABLE workspace_settings
+ALTER COLUMN ai_resource TYPE text
+USING ai_resource->>'path';
+
+ALTER TABLE workspace_settings
+RENAME COLUMN ai_resource to openai_resource_path;

backend/migrations/20241105133531_alter_openai_resource_path_column_in_workspace_settings.up.sql

@@ -0,0 +1,7 @@
+-- Add up migration script here
+ALTER TABLE workspace_settings
+ALTER COLUMN openai_resource_path TYPE jsonb
+USING jsonb_build_object('provider', 'openai', 'path', openai_resource_path);
+
+ALTER TABLE workspace_settings
+RENAME COLUMN openai_resource_path TO ai_resource;

backend/migrations/20241105225151_alert_add_acknowledged.down.sql

@@ -0,0 +1 @@
+ALTER TABLE alerts DROP COLUMN acknowledged;

backend/migrations/20241105225151_alert_add_acknowledged.up.sql

@@ -0,0 +1,9 @@
+-- Step 1: Add the new column 'acknowledged' to the 'alerts' table
+ALTER TABLE alerts
+ADD COLUMN acknowledged BOOLEAN DEFAULT NULL;
+
+-- Step 2: Update all existing rows to set 'acknowledged' to true
+-- we don't want to pop up notifications to all users after migrations
+-- but only show new alerts from the point of the upgrade
+UPDATE alerts
+SET acknowledged = TRUE;

backend/migrations/20241107163553_http_route_static_asset.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+ALTER TABLE http_trigger DROP COLUMN static_asset_config;

backend/migrations/20241107163553_http_route_static_asset.up.sql

@@ -0,0 +1,2 @@
+-- Add up migration script here
+ALTER TABLE http_trigger ADD COLUMN static_asset_config JSONB;

backend/migrations/20241108201445_alert_add_workspace_id.down.sql

@@ -0,0 +1,4 @@
+ALTER TABLE alerts DROP COLUMN workspace_id;
+ALTER TABLE alerts DROP COLUMN acknowledged_workspace;
+ALTER TABLE alerts DROP COLUMN resource;
+ALTER TABLE workspace_settings DROP COLUMN mute_critical_alerts;

backend/migrations/20241108201445_alert_add_workspace_id.up.sql

@@ -0,0 +1,4 @@
+ALTER TABLE alerts ADD COLUMN workspace_id TEXT DEFAULT NULL;
+ALTER TABLE alerts ADD COLUMN acknowledged_workspace BOOL DEFAULT NULL;
+ALTER TABLE alerts ADD COLUMN resource TEXT DEFAULT NULL;
+ALTER TABLE workspace_settings ADD COLUMN mute_critical_alerts BOOL DEFAULT NULL;

backend/oauth_connect.json

@@ -171,5 +171,6 @@
       "user-library-modify",
       "user-library-read"
     ]
-  }
-}
+  },
+  "snowflake_oauth": {}
+}

backend/parsers/windmill-parser-py-imports/src/lib.rs

@@ -62,6 +62,7 @@ static PYTHON_IMPORTS_REPLACEMENT: phf::Map<&'static str, &'static str> = phf_ma
     "lokalise" => "python-lokalise-api",
     "msgraph" => "msgraph-sdk",
     "pythonjsonlogger" => "python-json-logger",
+    "socks" => "PySocks",
 };
 
 fn replace_import(x: String) -> String {

backend/src/main.rs

@@ -7,7 +7,6 @@
  */
 
 use anyhow::Context;
-use git_version::git_version;
 use monitor::{
     reload_timeout_wait_result_setting, send_current_log_file_to_object_store,
     send_logs_to_object_store,
@@ -31,18 +30,19 @@ use windmill_common::ee::{maybe_renew_license_key_on_start, LICENSE_KEY_ID, LICE
 
 use windmill_common::{
     global_settings::{
-        BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
-        CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING,
-        ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING,
-        EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING,
-        JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING,
-        OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
-        REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
-        SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
+        BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
+        CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
+        DEFAULT_TAGS_WORKSPACES_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING,
+        EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING,
+        JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING,
+        LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING,
+        REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
+        RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING,
+        TIMEOUT_WAIT_RESULT_SETTING,
     },
     scripts::ScriptLang,
     stats_ee::schedule_stats,
-    utils::{hostname, rd_string, Mode},
+    utils::{hostname, rd_string, Mode, GIT_VERSION},
     worker::{reload_custom_tags_setting, HUB_CACHE_DIR, TMP_DIR, WORKER_GROUP},
     DB, METRICS_ENABLED,
 };
@@ -67,16 +67,17 @@ use windmill_worker::{
     get_hub_script_content_and_requirements, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR,
     BUN_DEPSTAR_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM,
     GO_BIN_CACHE_DIR, GO_CACHE_DIR, LOCK_CACHE_DIR, PIP_CACHE_DIR, POWERSHELL_CACHE_DIR,
-    RUST_CACHE_DIR, TAR_PIP_CACHE_DIR, TMP_LOGS_DIR, UV_CACHE_DIR,
+    PY311_CACHE_DIR, RUST_CACHE_DIR, TAR_PIP_CACHE_DIR, TMP_LOGS_DIR, UV_CACHE_DIR,
 };
 
 use crate::monitor::{
     initial_load, load_keep_job_dir, load_metrics_debug_enabled, load_require_preexisting_user,
     load_tag_per_workspace_enabled, load_tag_per_workspace_workspaces, monitor_db, monitor_pool,
     reload_base_url_setting, reload_bunfig_install_scopes_setting,
-    reload_critical_error_channels_setting, reload_extra_pip_index_url_setting,
-    reload_hub_base_url_setting, reload_job_default_timeout_setting, reload_jwt_secret_setting,
-    reload_license_key, reload_npm_config_registry_setting, reload_pip_index_url_setting,
+    reload_critical_alert_mute_ui_setting, reload_critical_error_channels_setting,
+    reload_extra_pip_index_url_setting, reload_hub_base_url_setting,
+    reload_job_default_timeout_setting, reload_jwt_secret_setting, reload_license_key,
+    reload_npm_config_registry_setting, reload_pip_index_url_setting,
     reload_retention_period_setting, reload_scim_token_setting, reload_smtp_config,
     reload_worker_config,
 };
@@ -84,7 +85,6 @@ use crate::monitor::{
 #[cfg(feature = "parquet")]
 use crate::monitor::reload_s3_cache_setting;
 
-const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
 const DEFAULT_NUM_WORKERS: usize = 1;
 const DEFAULT_PORT: u16 = 8000;
 const DEFAULT_SERVER_BIND_ADDR: Ipv4Addr = Ipv4Addr::new(0, 0, 0, 0);
@@ -531,7 +531,7 @@ Windmill Community Edition {GIT_VERSION}
 
         #[cfg(feature = "tantivy")]
         let (index_reader, index_writer) = if should_index_jobs {
-            let (r, w) = windmill_indexer::indexer_ee::init_index(&db).await?;
+            let (r, w) = windmill_indexer::completed_runs_ee::init_index(&db).await?;
             (Some(r), Some(w))
         } else {
             (None, None)
@@ -543,26 +543,61 @@ Windmill Community Edition {GIT_VERSION}
             let index_writer2 = index_writer.clone();
             async {
                 if let Some(index_writer) = index_writer2 {
-                    windmill_indexer::indexer_ee::run_indexer(db.clone(), index_writer, indexer_rx)
-                        .await;
+                    windmill_indexer::completed_runs_ee::run_indexer(
+                        db.clone(),
+                        index_writer,
+                        indexer_rx,
+                    )
+                    .await;
+                }
+                Ok(())
+            }
+        };
+
+        #[cfg(all(feature = "tantivy", feature = "parquet"))]
+        let (log_index_reader, log_index_writer) = if should_index_jobs {
+            let (r, w) = windmill_indexer::service_logs_ee::init_index(&db).await?;
+            (Some(r), Some(w))
+        } else {
+            (None, None)
+        };
+
+        #[cfg(all(feature = "tantivy", feature = "parquet"))]
+        let log_indexer_f = {
+            let log_indexer_rx = killpill_rx.resubscribe();
+            let log_index_writer2 = log_index_writer.clone();
+            async {
+                if let Some(log_index_writer) = log_index_writer2 {
+                    windmill_indexer::service_logs_ee::run_indexer(
+                        db.clone(),
+                        log_index_writer,
+                        log_indexer_rx,
+                    )
+                    .await;
                 }
                 Ok(())
             }
         };
 
         #[cfg(not(feature = "tantivy"))]
-        let (index_reader, index_writer) = (None, None);
+        let index_reader = None;
 
         #[cfg(not(feature = "tantivy"))]
         let indexer_f = async { Ok(()) as anyhow::Result<()> };
 
+        #[cfg(not(all(feature = "tantivy", feature = "parquet")))]
+        let log_index_reader = None;
+
+        #[cfg(not(all(feature = "tantivy", feature = "parquet")))]
+        let log_indexer_f = async { Ok(()) as anyhow::Result<()> };
+
         let server_f = async {
             if !is_agent {
                 windmill_api::run_server(
                     db.clone(),
                     rsmq2,
                     index_reader,
-                    index_writer,
+                    log_index_reader,
                     addr,
                     server_killpill_rx,
                     base_internal_tx,
@@ -782,6 +817,12 @@ Windmill Community Edition {GIT_VERSION}
                                                         tracing::error!(error = %e, "Could not reload jwt secret setting");
                                                     }
                                                 },
+                                                CRITICAL_ALERT_MUTE_UI_SETTING => {
+                                                    tracing::info!("Critical alert UI setting changed");
+                                                    if let Err(e) = reload_critical_alert_mute_ui_setting(&db).await {
+                                                        tracing::error!(error = %e, "Could not reload critical alert UI setting");
+                                                    }
+                                                },
                                                 a @_ => {
                                                     tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a);
                                                 }
@@ -842,7 +883,8 @@ Windmill Community Edition {GIT_VERSION}
             monitor_f,
             server_f,
             metrics_f,
-            indexer_f
+            indexer_f,
+            log_indexer_f
         )?;
     } else {
         tracing::info!("Nothing to do, exiting.");
@@ -955,13 +997,14 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
     for x in [
         LOCK_CACHE_DIR,
         TMP_LOGS_DIR,
-        PIP_CACHE_DIR,
         UV_CACHE_DIR,
         TAR_PIP_CACHE_DIR,
         DENO_CACHE_DIR,
         DENO_CACHE_DIR_DEPS,
         DENO_CACHE_DIR_NPM,
         BUN_CACHE_DIR,
+        PY311_CACHE_DIR,
+        PIP_CACHE_DIR,
         BUN_DEPSTAR_CACHE_DIR,
         BUN_BUNDLE_CACHE_DIR,
         GO_CACHE_DIR,

backend/src/monitor.rs

@@ -34,12 +34,12 @@ use windmill_common::{
     error,
     flow_status::FlowStatusModule,
     global_settings::{
-        BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
-        DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING,
-        EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING,
-        HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING,
-        KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING,
-        PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
+        BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
+        CRITICAL_ERROR_CHANNELS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
+        DEFAULT_TAGS_WORKSPACES_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING,
+        EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING,
+        JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING,
+        OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
         REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
         SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
     },
@@ -51,11 +51,11 @@ use windmill_common::{
     utils::{now_from_db, rd_string, report_critical_error, Mode},
     worker::{
         load_worker_config, make_pull_query, make_suspended_pull_query, reload_custom_tags_setting,
-        DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, SMTP_CONFIG, WORKER_CONFIG,
-        WORKER_GROUP,
+        update_min_version, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, SMTP_CONFIG,
+        WORKER_CONFIG, WORKER_GROUP,
     },
-    BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS,
-    METRICS_DEBUG_ENABLED, METRICS_ENABLED,
+    BASE_URL, CRITICAL_ALERT_MUTE_UI_ENABLED, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL,
+    HUB_BASE_URL, JOB_RETENTION_SECS, METRICS_DEBUG_ENABLED, METRICS_ENABLED,
 };
 use windmill_queue::cancel_job;
 use windmill_worker::{
@@ -111,6 +111,9 @@ lazy_static::lazy_static! {
         "Number of jobs in the queue",
         &["tag"]
     ).unwrap();
+
+    static ref QUEUE_COUNT_TAGS: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(Vec::new()));
+
 }
 
 pub async fn initial_load(
@@ -128,6 +131,10 @@ pub async fn initial_load(
         tracing::error!("Error loading expose debug metrics: {e:#}");
     }
 
+    if let Err(e) = reload_critical_alert_mute_ui_setting(db).await {
+        tracing::error!("Error loading critical alert mute ui setting: {e:#}");
+    }
+
     if let Err(e) = load_tag_per_workspace_enabled(db).await {
         tracing::error!("Error loading default tag per workpsace: {e:#}");
     }
@@ -223,6 +230,24 @@ pub async fn load_tag_per_workspace_workspaces(db: &DB) -> error::Result<()> {
     Ok(())
 }
 
+pub async fn reload_critical_alert_mute_ui_setting(db: &DB) -> error::Result<()> {
+    if let Ok(Some(serde_json::Value::Bool(t))) =
+        load_value_from_global_settings(db, CRITICAL_ALERT_MUTE_UI_SETTING).await
+    {
+        CRITICAL_ALERT_MUTE_UI_ENABLED.store(t, Ordering::Relaxed);
+
+        if t {
+            if let Err(e) = sqlx::query!("UPDATE alerts SET acknowledged = true")
+                .execute(db)
+                .await
+            {
+                tracing::error!("Error updating alerts: {}", e.to_string());
+            }
+        }
+    }
+    Ok(())
+}
+
 pub async fn load_metrics_debug_enabled(db: &DB) -> error::Result<()> {
     let metrics_enabled = load_value_from_global_settings(db, EXPOSE_DEBUG_METRICS_SETTING).await;
     match metrics_enabled {
@@ -1062,6 +1087,10 @@ pub async fn monitor_db(
         }
     };
 
+    let update_min_worker_version_f = async {
+        update_min_version(db).await;
+    };
+
     join!(
         expired_items_f,
         zombie_jobs_f,
@@ -1070,6 +1099,7 @@ pub async fn monitor_db(
         worker_groups_alerts_f,
         jobs_waiting_alerts_f,
         apply_autoscaling_f,
+        update_min_worker_version_f,
     );
 }
 
@@ -1089,12 +1119,23 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
     if metrics_enabled || save_metrics {
         let queue_counts = windmill_common::queue::get_queue_counts(db).await;
 
+        if metrics_enabled {
+            for q in QUEUE_COUNT_TAGS.read().await.iter() {
+                if queue_counts.get(q).is_none() {
+                    (*QUEUE_COUNT).with_label_values(&[q]).set(0);
+                }
+            }
+        }
+
+        let mut tags_to_watch = vec![];
         for q in queue_counts {
             let count = q.1;
             let tag = q.0;
+
             if metrics_enabled {
                 let metric = (*QUEUE_COUNT).with_label_values(&[&tag]);
                 metric.set(count as i64);
+                tags_to_watch.push(tag.to_string());
             }
 
             // save queue_count and delay metrics per tag
@@ -1119,6 +1160,10 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
                 }
             }
         }
+        if metrics_enabled {
+            let mut w = QUEUE_COUNT_TAGS.write().await;
+            *w = tags_to_watch;
+        }
     }
 
     // clean queue metrics older than 14 days
@@ -1285,7 +1330,7 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                 ON CONFLICT (job_id) DO UPDATE SET logs = job_logs.logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE job_logs.job_id = $1", r.id)
                 .execute(db).await;
             tracing::error!(error_message);
-            report_critical_error(error_message, db.clone()).await;
+            report_critical_error(error_message, db.clone(), Some(&r.workspace_id), None).await;
         }
     }
 
@@ -1399,7 +1444,7 @@ async fn handle_zombie_flows(
                 flow.id, flow.workspace_id
             );
             tracing::error!(error_message);
-            report_critical_error(error_message, db.clone()).await;
+            report_critical_error(error_message, db.clone(), Some(&flow.workspace_id), None).await;
             // if the flow hasn't started and is a zombie, we can simply restart it
             sqlx::query!(
                 "UPDATE queue SET running = false, started_at = null WHERE id = $1 AND canceled = false",
@@ -1419,7 +1464,7 @@ async fn handle_zombie_flows(
                     format!("Flow {id} was cancelled because it")
                 }
             );
-            report_critical_error(reason.clone(), db.clone()).await;
+            report_critical_error(reason.clone(), db.clone(), Some(&flow.workspace_id), None).await;
             cancel_zombie_flow_job(db, flow, &rsmq, reason).await?;
         }
     }

backend/tests/worker.rs

@@ -1112,6 +1112,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
                         custom_concurrency_key: None,
                         concurrent_limit: None,
                         concurrency_time_window_s: None,
+                        is_trigger: None,
                     }
                     .into(),
                     stop_after_if: Default::default(),
@@ -1153,6 +1154,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
                                 custom_concurrency_key: None,
                                 concurrent_limit: None,
                                 concurrency_time_window_s: None,
+                                is_trigger: None,
                             }
                             .into(),
                             stop_after_if: Default::default(),
@@ -1276,6 +1278,8 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
                         custom_concurrency_key: None,
                         concurrent_limit: None,
                         concurrency_time_window_s: None,
+                        is_trigger: None,
+
                     }.into(),
                     stop_after_if: Default::default(),
                     stop_after_all_iters_if: Default::default(),
@@ -1327,6 +1331,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
                                     custom_concurrency_key: None,
                                     concurrent_limit: None,
                                     concurrency_time_window_s: None,
+                                    is_trigger: None,
                                 }.into(),
                                 stop_after_if: Default::default(),
                                 stop_after_all_iters_if: Default::default(),
@@ -1364,6 +1369,8 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
                                     custom_concurrency_key: None,
                                     concurrent_limit: None,
                                     concurrency_time_window_s: None,
+                                    is_trigger: None,
+
                                 }.into(),
                                 stop_after_if: Default::default(),
                                 stop_after_all_iters_if: Default::default(),
@@ -1424,6 +1431,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
                         custom_concurrency_key: None,
                         concurrent_limit: None,
                         concurrency_time_window_s: None,
+                        is_trigger: None,
                     }.into(),
                     stop_after_if: Default::default(),
                     stop_after_all_iters_if: Default::default(),

backend/windmill-api/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: "3.0.3"
 
 info:
-  version: 1.416.2
+  version: 1.424.0
   title: Windmill API
 
   contact:
@@ -190,7 +190,7 @@ paths:
               schema:
                 type: string
 
-  /w/{workspace}/users/{username}:
+  /w/{workspace}/users/get/{username}:
     get:
       summary: get user (require admin privilege)
       operationId: getUser
@@ -283,6 +283,71 @@ paths:
             text/plain:
               schema:
                 type: string
+                
+  /users/set_password_of/{user}:
+    post:
+      summary: set password for a specific user (require super admin)
+      operationId: setPasswordForUser
+      tags:
+        - user
+      parameters:
+        - name: user
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        description: set password
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                password:
+                  type: string
+              required:
+                - password
+      responses:
+        "200":
+          description: password set
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /users/set_login_type/{user}:
+    post:
+      summary: set login type for a specific user (require super admin)
+      operationId: setLoginTypeForUser
+      tags:
+        - user
+      parameters:
+        - name: user
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        description: set login type
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                login_type:
+                  type: string
+              required:
+                - login_type
+      responses:
+        "200":
+          description: login type set
+          content:
+            text/plain:
+              schema:
+                type: string
+
 
   /users/create:
     post:
@@ -724,6 +789,8 @@ paths:
                       type: string
                     tls_implicit:
                       type: boolean
+                    disable_tls:
+                      type: boolean
                   required:
                     - host
                     - username
@@ -731,6 +798,7 @@ paths:
                     - port
                     - from
                     - tls_implicit
+                    - disable_tls
               required:
                 - to
                 - smtp
@@ -770,6 +838,78 @@ paths:
               schema:
                 type: string
 
+  /settings/critical_alerts:
+    get:
+      summary: Get all critical alerts
+      operationId: getCriticalAlerts
+      tags:
+        - setting
+      parameters:
+        - in: query
+          name: page
+          schema:
+            type: integer
+            default: 1
+            description: The page number to retrieve (minimum value is 1)
+        - in: query
+          name: page_size
+          schema:
+            type: integer
+            default: 10
+            maximum: 100
+            description: Number of alerts per page (maximum is 100)
+        - in: query
+          name: acknowledged
+          schema:
+            type: boolean
+            nullable: true
+            description: Filter by acknowledgment status; true for acknowledged, false for unacknowledged, and omit for all alerts
+      responses:
+        "200":
+          description: Successfully retrieved all critical alerts
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/CriticalAlert'
+
+  /settings/critical_alerts/{id}/acknowledge:
+    post:
+      summary: Acknowledge a critical alert
+      operationId: acknowledgeCriticalAlert
+      tags:
+        - setting
+      parameters:
+        - in: path
+          name: id
+          required: true
+          schema:
+            type: integer
+          description: The ID of the critical alert to acknowledge
+      responses:
+        "200":
+          description: Successfully acknowledged the critical alert
+          content:
+            application/json:
+              schema:
+                type: string
+                example: "Critical alert acknowledged"
+
+  /settings/critical_alerts/acknowledge_all:
+    post:
+      summary: Acknowledge all unacknowledged critical alerts
+      operationId: acknowledgeAllCriticalAlerts
+      tags:
+        - setting
+      responses:
+        "200":
+          description: Successfully acknowledged all unacknowledged critical alerts.
+          content:
+            application/json:
+              schema:
+                type: string
+                example: "All unacknowledged critical alerts acknowledged"
 
   /settings/test_license_key:
     post:
@@ -1541,8 +1681,8 @@ paths:
                     type: string
                   deploy_to:
                     type: string
-                  openai_resource_path:
-                    type: string
+                  ai_resource:
+                    $ref: "#/components/schemas/AiResource"
                   code_completion_enabled:
                     type: boolean
                   error_handler:
@@ -1561,6 +1701,8 @@ paths:
                     type: string
                   default_scripts:
                     $ref: "#/components/schemas/WorkspaceDefaultScripts"
+                  mute_critical_alerts:
+                    type: boolean
                 required:
                   - code_completion_enabled
                   - automatic_billing
@@ -1826,8 +1968,8 @@ paths:
               required:
                 - code_completion_enabled
               properties:
-                openai_resource_path:
-                  type: string
+                ai_resource:
+                  $ref: "#/components/schemas/AiResource"
                 code_completion_enabled:
                   type: boolean
       responses:
@@ -1855,12 +1997,15 @@ paths:
               schema:
                 type: object
                 properties:
-                  exists_openai_resource_path:
+                  ai_provider:
+                    type: string
+                  exists_ai_resource:
                     type: boolean
                   code_completion_enabled:
                     type: boolean
                 required:
-                  - exists_openai_resource_path
+                  - ai_provider
+                  - exists_ai_resource
                   - code_completion_enabled
 
   /w/{workspace}/workspaces/edit_error_handler:
@@ -2574,6 +2719,112 @@ paths:
                 items:
                   $ref: "#/components/schemas/ContextualVariable"
 
+  /w/{workspace}/workspaces/critical_alerts:
+    get:
+      summary: Get all critical alerts for this workspace
+      operationId: workspaceGetCriticalAlerts
+      tags:
+        - setting
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - in: query
+          name: page
+          schema:
+            type: integer
+            default: 1
+            description: The page number to retrieve (minimum value is 1)
+        - in: query
+          name: page_size
+          schema:
+            type: integer
+            default: 10
+            maximum: 100
+            description: Number of alerts per page (maximum is 100)
+        - in: query
+          name: acknowledged
+          schema:
+            type: boolean
+            nullable: true
+            description: Filter by acknowledgment status; true for acknowledged, false for unacknowledged, and omit for all alerts
+      responses:
+        "200":
+          description: Successfully retrieved all critical alerts
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/CriticalAlert'
+
+  /w/{workspace}/workspaces/critical_alerts/{id}/acknowledge:
+    post:
+      summary: Acknowledge a critical alert for this workspace
+      operationId: workspaceAcknowledgeCriticalAlert
+      tags:
+        - setting
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - in: path
+          name: id
+          required: true
+          schema:
+            type: integer
+          description: The ID of the critical alert to acknowledge
+      responses:
+        "200":
+          description: Successfully acknowledged the critical alert
+          content:
+            application/json:
+              schema:
+                type: string
+                example: "Critical alert acknowledged"
+
+  /w/{workspace}/workspaces/critical_alerts/acknowledge_all:
+    post:
+      summary: Acknowledge all unacknowledged critical alerts for this workspace
+      operationId: workspaceAcknowledgeAllCriticalAlerts
+      tags:
+        - setting
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      responses:
+        "200":
+          description: Successfully acknowledged all unacknowledged critical alerts.
+          content:
+            application/json:
+              schema:
+                type: string
+                example: "All unacknowledged critical alerts acknowledged"
+
+  /w/{workspace}/workspaces/critical_alerts/mute:
+    post:
+      summary: Mute critical alert UI for this workspace
+      operationId: workspaceMuteCriticalAlertsUI
+      tags:
+        - setting
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      requestBody:
+        description: Boolean flag to mute critical alerts.
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                mute_critical_alerts:
+                  type: boolean
+                  description: Whether critical alerts should be muted.
+                  example: true
+      responses:
+        '200':
+          description: Successfully updated mute critical alert settings.
+          content:
+            application/json:
+              schema:
+                type: string
+                example: "Updated mute critical alert UI settings for workspace: workspace_id"
+
   /oauth/login_callback/{client_name}:
     post:
       security: []
@@ -5911,6 +6162,39 @@ paths:
                 required:
                   - database_length
 
+  /w/{workspace}/jobs/completed/count_jobs:
+    get:
+      summary: count number of completed jobs with filter
+      operationId: countCompletedJobs
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: completed_after_s_ago
+          in: query
+          schema:
+            type: integer
+        - name: success
+          in: query
+          schema:
+            type: boolean
+        - name: tags
+          in: query
+          schema:
+            type: string
+        - name: all_workspaces
+          in: query
+          schema:
+            type: boolean
+      responses:
+        "200":
+          description: Count of completed jobs
+          content:
+            application/json:
+              schema:
+                type: integer
+
+
   /w/{workspace}/jobs/queue/list_filtered_uuids:
     get:
       summary: get the ids of all jobs matching the given filters
@@ -8113,6 +8397,23 @@ paths:
                     - id
                     - values
 
+  /workers/queue_counts:
+    get:
+      summary: get counts of jobs waiting for an executor per tag
+      operationId: getCountsOfJobsWaitingPerTag
+      tags:
+        - worker
+      responses:
+        "200":
+          description: queue counts
+          content:
+            application/json:
+              schema: 
+                type: object
+                additionalProperties:
+                  type: integer
+
+              
   /configs/list_worker_groups:
     get:
       summary: list worker groups
@@ -9143,6 +9444,14 @@ paths:
           in: query
           schema:
             type: string
+        - name: content_type
+          in: query
+          schema:
+            type: string
+        - name: content_disposition
+          in: query
+          schema:
+            type: string
       requestBody:
         description: File content
         required: true
@@ -9546,6 +9855,109 @@ paths:
                     items:
                       $ref: "#/components/schemas/JobSearchHit"
 
+  /srch/index/search/service_logs:
+    get:
+      summary: Search through service logs with a string query
+      operationId: searchLogsIndex
+      tags:
+        - indexSearch
+      parameters:
+        - name: search_query
+          in: query
+          required: true
+          schema:
+            type: string
+        - name: mode
+          in: query
+          required: true
+          schema:
+            type: string
+        - name: worker_group
+          in: query
+          required: false
+          schema:
+            type: string
+        - name: hostname
+          in: query
+          required: true
+          schema:
+            type: string
+        - name: min_ts
+          in: query
+          required: false
+          schema:
+            type: string
+            format: date-time
+        - name: max_ts
+          in: query
+          required: false
+          schema:
+            type: string
+            format: date-time
+      responses:
+        "200":
+          description: search results
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  query_parse_errors:
+                    description: a list of the terms that couldn't be parsed (and thus ignored)
+                    type: array
+                    items:
+                      type: string
+                  hits:
+                    description: log files that matched the query
+                    type: array
+                    items:
+                      $ref: "#/components/schemas/LogSearchHit"
+
+  /srch/index/search/count_service_logs:
+    get:
+      summary: Search and count the log line hits on every provided host
+      operationId: countSearchLogsIndex
+      tags:
+        - indexSearch
+      parameters:
+        - name: search_query
+          in: query
+          required: true
+          schema:
+            type: string
+        - name: hosts
+          in: query
+          required: true
+          schema:
+            type: string
+        - name: min_ts
+          in: query
+          required: false
+          schema:
+            type: string
+            format: date-time
+        - name: max_ts
+          in: query
+          required: false
+          schema:
+            type: string
+            format: date-time
+      responses:
+        "200":
+          description: search results
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  query_parse_errors:
+                    description: a list of the terms that couldn't be parsed (and thus ignored)
+                    type: array
+                    items:
+                      type: string
+                  count_per_host:
+                    description: count of log lines that matched the query per hostname
+                    type: object
 
 components:
   securitySchemes:
@@ -9926,6 +10338,16 @@ components:
   schemas:
     $ref: "../../openflow.openapi.yaml#/components/schemas"
 
+    AiResource:
+      type: object
+      properties:
+        path:
+          type: string
+        provider:
+          type: string
+      required:
+        - path
+        - provider
     Script:
       type: object
       properties:
@@ -10744,7 +11166,7 @@ components:
             - "jobs.disapproval"
             - "jobs.delete"
             - "account.delete"
-            - "openai.request"
+            - "ai.request"
             - "resources.create"
             - "resources.update"
             - "resources.delete"
@@ -11357,6 +11779,17 @@ components:
           type: string
         route_path:
           type: string
+        static_asset_config:
+          type: object
+          properties:
+            s3:
+              type: string
+            storage:
+              type: string
+            filename:
+              type: string
+          required:
+            - s3
         is_flow:
           type: boolean
         extra_perms:
@@ -11403,6 +11836,17 @@ components:
           type: string
         route_path:
           type: string
+        static_asset_config:
+          type: object
+          properties:
+            s3:
+              type: string
+            storage:
+              type: string
+            filename:
+              type: string
+          required:
+            - s3
         is_flow:
           type: boolean
         http_method:
@@ -11436,6 +11880,17 @@ components:
           type: string
         route_path:
           type: string
+        static_asset_config:
+          type: object
+          properties:
+            s3:
+              type: string
+            storage:
+              type: string
+            filename:
+              type: string
+          required:
+            - s3
         is_flow:
           type: boolean
         http_method:
@@ -11522,6 +11977,12 @@ components:
             required:
               - key
               - value
+        initial_messages:
+          type: array
+          items:
+            $ref: "#/components/schemas/WebsocketTriggerInitialMessage"
+        url_runnable_args:
+          $ref: "#/components/schemas/ScriptArgs"
         
       required:
         - path
@@ -11535,6 +11996,8 @@ components:
         - workspace_id
         - enabled
         - filters
+        - initial_messages
+        - url_runnable_args
 
     NewWebsocketTrigger:
       type: object
@@ -11560,6 +12023,12 @@ components:
             required:
               - key
               - value
+        initial_messages:
+          type: array
+          items:
+            $ref: "#/components/schemas/WebsocketTriggerInitialMessage"
+        url_runnable_args:
+          $ref: "#/components/schemas/ScriptArgs"
         
       required:
         - path
@@ -11567,6 +12036,8 @@ components:
         - url
         - is_flow
         - filters
+        - initial_messages
+        - url_runnable_args
 
     EditWebsocketTrigger:
       type: object
@@ -11590,6 +12061,12 @@ components:
             required:
               - key
               - value
+        initial_messages:
+          type: array
+          items:
+            $ref: "#/components/schemas/WebsocketTriggerInitialMessage"
+        url_runnable_args:
+          $ref: "#/components/schemas/ScriptArgs"
               
       required:
         - path
@@ -11597,6 +12074,34 @@ components:
         - url
         - is_flow
         - filters
+        - initial_messages
+        - url_runnable_args
+
+    WebsocketTriggerInitialMessage:
+      anyOf:
+        - type: object
+          properties:
+            raw_message: 
+              type: string
+          required:
+            - raw_message
+        - type: object
+          properties:
+            runnable_result:
+              type: object
+              properties:
+                path: 
+                  type: string
+                args:
+                  $ref: "#/components/schemas/ScriptArgs"
+                is_flow:
+                  type: boolean
+              required:
+                - path
+                - args
+                - is_flow
+          required: 
+            - runnable_result
 
     Group:
       type: object
@@ -11917,6 +12422,10 @@ components:
           type: object
           additionalProperties:
             type: object
+        s3_inputs:
+          type: array
+          items:
+            type: object
         execution_mode:
           type: string
           enum: [viewer, publisher, anonymous]
@@ -12500,6 +13009,12 @@ components:
         dancer:
           type: string
 
+    LogSearchHit:
+      type: object
+      properties:
+        dancer:
+          type: string
+
     AutoscalingEvent:
       type: object
       properties:
@@ -12517,3 +13032,28 @@ components:
         applied_at:
           type: string
           format: date-time
+
+    CriticalAlert:
+      type: object
+      properties:
+        id:
+          type: integer
+          description: Unique identifier for the alert
+        alert_type:
+          type: string
+          description: Type of alert (e.g., critical_error)
+        message:
+          type: string
+          description: The message content of the alert
+        created_at:
+          type: string
+          format: date-time
+          description: Time when the alert was created
+        acknowledged:
+          type: boolean
+          nullable: true
+          description: Acknowledgment status of the alert, can be true, false, or null if not set
+        workspace_id:
+          type: string
+          nullable: true
+          description: Workspace id if the alert is in the scope of a workspace

backend/windmill-api/src/ai.rs

@@ -0,0 +1,507 @@
+use crate::{
+    db::{ApiAuthed, DB},
+    variables::decrypt,
+};
+use anthropic::AnthropicCache;
+use axum::{
+    body::Bytes,
+    extract::{Path, Query},
+    response::IntoResponse,
+    routing::post,
+    Extension, Router,
+};
+use lazy_static::lazy_static;
+use mistral::MistralCache;
+use openai::OpenaiCache;
+use quick_cache::sync::Cache;
+use reqwest::{Client, RequestBuilder};
+use serde::{Deserialize, Deserializer};
+use windmill_audit::audit_ee::audit_log;
+use windmill_audit::ActionKind;
+use windmill_common::error::{to_anyhow, Result};
+use windmill_common::variables::build_crypt;
+
+use windmill_common::error::Error;
+
+use serde_json::value::{RawValue, Value};
+use std::collections::HashMap;
+
+lazy_static::lazy_static! {
+    static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new()
+        .timeout(std::time::Duration::from_secs(60 * 5))
+        .user_agent("windmill/beta")
+        .build().unwrap();
+}
+
+trait AiRequest {
+    fn prepare_request(self, path: &str, body: Bytes) -> Result<RequestBuilder>;
+}
+
+mod openai {
+    use super::*;
+
+    use super::{get_variable_or_self, KeyCache};
+
+    const API_VERSION: &str = "2023-05-15";
+
+    #[derive(Deserialize, Debug)]
+    struct OpenaiResource {
+        api_key: String,
+        organization_id: Option<String>,
+    }
+
+    #[derive(Deserialize, Debug)]
+    struct OpenaiClientCredentialsOauthResource {
+        client_id: String,
+        client_secret: String,
+        token_url: String,
+        user: Option<String>,
+    }
+
+    #[derive(Deserialize, Debug)]
+    #[serde(untagged, rename_all = "snake_case")]
+    enum OpenaiConfig {
+        Resource(OpenaiResource),
+        ClientCredentialsOauthResource(OpenaiClientCredentialsOauthResource),
+    }
+
+    lazy_static::lazy_static! {
+        pub static ref OPENAI_AZURE_BASE_PATH: Option<String> = std::env::var("OPENAI_AZURE_BASE_PATH").ok();
+    }
+
+    #[derive(Deserialize, Debug)]
+    struct OpenaiCredentials {
+        access_token: String,
+    }
+
+    #[derive(Clone, Debug, Deserialize)]
+    pub struct OpenaiCache {
+        api_key: String,
+        organization_id: Option<String>,
+        azure_base_path: Option<String>,
+        user: Option<String>,
+    }
+
+    impl OpenaiCache {
+        pub fn new(
+            api_key: String,
+            organization_id: Option<String>,
+            azure_base_path: Option<String>,
+            user: Option<String>,
+        ) -> Self {
+            Self { api_key, organization_id, azure_base_path, user }
+        }
+    }
+
+    const BASE_URL: &str = "https://api.openai.com/v1";
+    impl AiRequest for OpenaiCache {
+        fn prepare_request(self, openai_path: &str, mut body: Bytes) -> Result<RequestBuilder> {
+            let OpenaiCache { api_key, azure_base_path, organization_id, user } = self;
+            if user.is_some() {
+                tracing::debug!("Adding user to request body");
+                let mut json_body: HashMap<String, Box<RawValue>> = serde_json::from_slice(&body)
+                    .map_err(|e| {
+                    Error::InternalErr(format!("Failed to parse request body: {}", e))
+                })?;
+
+                let user_json_string = serde_json::Value::String(user.unwrap()).to_string(); // makes sure to escape characters
+
+                json_body.insert(
+                    "user".to_string(),
+                    RawValue::from_string(user_json_string)
+                        .map_err(|e| Error::InternalErr(format!("Failed to parse user: {}", e)))?,
+                );
+
+                body = serde_json::to_vec(&json_body)
+                    .map_err(|e| {
+                        Error::InternalErr(format!("Failed to reserialize request body: {}", e))
+                    })?
+                    .into();
+            }
+
+            let base_url = if let Some(base_url) = azure_base_path {
+                base_url
+            } else {
+                BASE_URL.to_string()
+            };
+            let url = format!("{}/{}", base_url, openai_path);
+            let mut request = HTTP_CLIENT
+                .post(url)
+                .header("content-type", "application/json")
+                .body(body);
+
+            if base_url != BASE_URL {
+                request = request
+                    .header("api-key", api_key)
+                    .query(&[("api-version", API_VERSION)])
+            } else {
+                request = request.header("authorization", format!("Bearer {}", api_key))
+            }
+
+            if let Some(org_id) = organization_id {
+                request = request.header("OpenAI-Organization", org_id);
+            }
+
+            Ok(request)
+        }
+    }
+
+    async fn get_openai_key_using_credentials_flow(
+        mut resource: OpenaiClientCredentialsOauthResource,
+        db: &DB,
+        w_id: &str,
+    ) -> Result<String> {
+        resource.client_id = get_variable_or_self(resource.client_id, db, w_id).await?;
+        resource.client_secret = get_variable_or_self(resource.client_secret, db, w_id).await?;
+        resource.token_url = get_variable_or_self(resource.token_url, db, w_id).await?;
+        let mut params = HashMap::new();
+        params.insert("grant_type", "client_credentials");
+        let response = HTTP_CLIENT
+            .post(resource.token_url)
+            .form(&params)
+            .basic_auth(resource.client_id, Some(resource.client_secret))
+            .send()
+            .await
+            .map_err(|err| {
+                Error::InternalErr(format!(
+                    "Failed to get OpenAI credentials using credentials flow: {}",
+                    err
+                ))
+            })?;
+        let response = response.json::<OpenaiCredentials>().await.map_err(|err| {
+            Error::InternalErr(format!(
+                "Failed to parse OpenAI credentials from credentials flow: {}",
+                err
+            ))
+        })?;
+        Ok(response.access_token)
+    }
+
+    pub async fn get_cached_value(db: &DB, w_id: &str, resource: Value) -> Result<KeyCache> {
+        let config = serde_json::from_value(resource)
+            .map_err(|e| Error::InternalErr(format!("validating openai resource {e:#}")))?;
+
+        let mut user = None::<String>;
+        let mut resource = match config {
+            OpenaiConfig::Resource(resource) => {
+                tracing::debug!("Getting OpenAI key from static resource");
+                resource
+            }
+            OpenaiConfig::ClientCredentialsOauthResource(resource) => {
+                tracing::debug!("Getting OpenAI key with client credentials flow");
+                user = resource.user.clone();
+                let token = get_openai_key_using_credentials_flow(resource, db, w_id).await?;
+                OpenaiResource { api_key: token, organization_id: None }
+            }
+        };
+
+        resource.api_key = get_variable_or_self(resource.api_key, db, w_id).await?;
+
+        if let Some(organization_id) = resource.organization_id {
+            resource.organization_id = Some(get_variable_or_self(organization_id, db, w_id).await?);
+        }
+
+        if user.is_some() {
+            user = Some(get_variable_or_self(user.unwrap(), db, w_id).await?);
+        }
+
+        let azure_base_path = sqlx::query_scalar!(
+            "SELECT value
+            FROM global_settings
+            WHERE name = 'openai_azure_base_path'",
+        )
+        .fetch_optional(db)
+        .await?;
+
+        let azure_base_path = if let Some(azure_base_path) = azure_base_path {
+            Some(
+                serde_json::from_value::<String>(azure_base_path).map_err(|e| {
+                    Error::InternalErr(format!("validating openai azure base path {e:#}"))
+                })?,
+            )
+        } else {
+            OPENAI_AZURE_BASE_PATH.clone()
+        };
+
+        let workspace_cache = OpenaiCache::new(
+            resource.api_key.clone(),
+            resource.organization_id.clone(),
+            azure_base_path.clone(),
+            user.clone(),
+        );
+        Ok(KeyCache::Openai(workspace_cache))
+    }
+}
+
+mod anthropic {
+
+    use super::*;
+
+    #[derive(Clone, Deserialize, Debug)]
+    pub struct AnthropicCache {
+        #[serde(rename = "apiKey")]
+        pub api_key: String,
+    }
+
+    const API_VERSION: &str = "2023-06-01";
+
+    impl AnthropicCache {
+        pub fn new(api_key: String) -> Self {
+            Self { api_key }
+        }
+    }
+
+    const BASE_URL: &str = "https://api.anthropic.com";
+    impl AiRequest for AnthropicCache {
+        fn prepare_request(self, anthropic_path: &str, body: Bytes) -> Result<RequestBuilder> {
+            let AnthropicCache { api_key } = self;
+            let url = format!("{}/{}", BASE_URL, anthropic_path);
+            let request = HTTP_CLIENT
+                .post(url)
+                .header("x-api-key", api_key)
+                .header("anthropic-version", API_VERSION)
+                .header("content-type", "application/json")
+                .body(body);
+            Ok(request)
+        }
+    }
+
+    pub async fn get_cached_value(db: &DB, w_id: &str, resource: Value) -> Result<KeyCache> {
+        let mut resource: AnthropicCache = serde_json::from_value(resource)
+            .map_err(|e| Error::InternalErr(format!("validating anthropic resource {e:#}")))?;
+        resource.api_key = get_variable_or_self(resource.api_key, db, w_id).await?;
+        let workspace_cache = AnthropicCache::new(resource.api_key);
+        Ok(KeyCache::Anthropic(workspace_cache))
+    }
+}
+
+mod mistral {
+    use super::*;
+    #[derive(Deserialize, Clone, Debug)]
+    pub struct MistralCache {
+        #[serde(rename = "apiKey")]
+        pub api_key: String,
+    }
+
+    impl MistralCache {
+        pub fn new(api_key: String) -> Self {
+            Self { api_key }
+        }
+    }
+
+    const BASE_URL: &str = "https://api.mistral.ai";
+    impl AiRequest for MistralCache {
+        fn prepare_request(self, mistral_path: &str, body: Bytes) -> Result<RequestBuilder> {
+            let MistralCache { api_key } = self;
+
+            let url = format!("{}/{}", BASE_URL, mistral_path);
+            let request = HTTP_CLIENT
+                .post(url)
+                .header("content-type", "application/json")
+                .header("Accept", "application/json")
+                .header("authorization", format!("Bearer {}", api_key))
+                .body(body);
+            Ok(request)
+        }
+    }
+
+    pub async fn get_cached_value(db: &DB, w_id: &str, resource: Value) -> Result<KeyCache> {
+        let mut resource: MistralCache = serde_json::from_value(resource)
+            .map_err(|e| Error::InternalErr(format!("validating mistral resource {e:#}")))?;
+        resource.api_key = get_variable_or_self(resource.api_key, db, w_id).await?;
+
+        let workspace_cache = MistralCache::new(resource.api_key);
+        Ok(KeyCache::Mistral(workspace_cache))
+    }
+}
+
+#[derive(Clone, Debug)]
+pub enum KeyCache {
+    Openai(OpenaiCache),
+    Anthropic(AnthropicCache),
+    Mistral(MistralCache),
+}
+
+#[derive(Clone, Debug)]
+pub struct AiCache {
+    pub path: String,
+    pub cached_key: KeyCache,
+    pub expires_at: std::time::Instant,
+}
+
+impl AiCache {
+    pub fn new(path: String, cached_key: KeyCache) -> Self {
+        Self {
+            path,
+            cached_key,
+            expires_at: std::time::Instant::now() + std::time::Duration::from_secs(60),
+        }
+    }
+    fn is_expired(&self) -> bool {
+        self.expires_at < std::time::Instant::now()
+    }
+}
+
+lazy_static! {
+    pub static ref AI_KEY_CACHE: Cache<String, AiCache> = Cache::new(500);
+}
+
+struct Variable {
+    value: String,
+    is_secret: bool,
+}
+
+#[derive(Deserialize, Debug)]
+struct ProxyQueryParams {
+    no_cache: Option<bool>,
+}
+
+async fn get_variable_or_self(path: String, db: &DB, w_id: &str) -> Result<String> {
+    if !path.starts_with("$var:") {
+        return Ok(path);
+    }
+    let path = path.strip_prefix("$var:").unwrap().to_string();
+    let mut variable = sqlx::query_as!(
+        Variable,
+        "SELECT value, is_secret
+        FROM variable
+        WHERE path = $1 AND workspace_id = $2",
+        &path,
+        &w_id
+    )
+    .fetch_one(db)
+    .await?;
+    if variable.is_secret {
+        let mc = build_crypt(db, w_id).await?;
+        variable.value = decrypt(&mc, variable.value)?;
+    }
+    Ok(variable.value)
+}
+
+#[derive(Deserialize, Debug)]
+pub struct AiResource {
+    pub path: String,
+    #[serde(deserialize_with = "check_if_valid_ai_provider")]
+    pub provider: String,
+}
+
+fn check_if_valid_ai_provider<'de, D>(provider: D) -> std::result::Result<String, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let provider = String::deserialize(provider)?;
+    match provider.as_str() {
+        "anthropic" | "openai" | "mistral" => Ok(provider),
+        _ => Err(serde::de::Error::custom(
+            "Only the following Ai providers are supported: openai, anthropic and mistral"
+                .to_string(),
+        )),
+    }
+}
+
+pub fn workspaced_service() -> Router {
+    let router = Router::new().route("/proxy/*ai", post(proxy));
+
+    router
+}
+
+async fn proxy(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path((w_id, ai_path)): Path<(String, String)>,
+    Query(query_params): Query<ProxyQueryParams>,
+    body: Bytes,
+) -> impl IntoResponse {
+    let workspace_cache = AI_KEY_CACHE.get(&w_id);
+    let ai_cache = match workspace_cache {
+        Some(cache) if !cache.is_expired() && !query_params.no_cache.unwrap_or(false) => {
+            cache.cached_key
+        }
+        _ => {
+            let ai_resource = sqlx::query_scalar!(
+                "SELECT ai_resource FROM workspace_settings WHERE workspace_id = $1",
+                &w_id
+            )
+            .fetch_one(&db)
+            .await?;
+
+            if ai_resource.is_none() {
+                return Err(Error::InternalErr("AI resource not configured".to_string()));
+            }
+
+            let ai_resource = serde_json::from_value::<AiResource>(ai_resource.unwrap())
+                .map_err(|e| Error::BadRequest(e.to_string()))?;
+            let ai_resource_path = ai_resource.path;
+
+            let resource = sqlx::query_scalar!(
+                "SELECT value
+                FROM resource
+                WHERE path = $1 AND workspace_id = $2",
+                &ai_resource_path,
+                &w_id
+            )
+            .fetch_optional(&db)
+            .await?
+            .ok_or_else(|| {
+                Error::InternalErr(format!(
+                    "Could not find the {} resource at path {ai_resource_path}, update the resource path in the workspace settings", ai_resource.provider
+                ))
+            })?;
+
+            if resource.is_none() {
+                return Err(Error::InternalErr(format!(
+                    "{} resource missing value",
+                    ai_resource.provider
+                )));
+            }
+
+            let resource = resource.unwrap();
+
+            let ai_cache = match ai_resource.provider.as_str() {
+                "openai" => openai::get_cached_value(&db, &w_id, resource).await,
+                "anthropic" => anthropic::get_cached_value(&db, &w_id, resource).await,
+                "mistral" => mistral::get_cached_value(&db, &w_id, resource).await,
+                provider => {
+                    return Err(Error::BadRequest(format!("{} is not supported", provider)))
+                }
+            };
+            let ai_cache = ai_cache?;
+            AI_KEY_CACHE.insert(
+                w_id.clone(),
+                AiCache::new(ai_resource_path, ai_cache.clone()),
+            );
+            ai_cache
+        }
+    };
+    let (path, request) = match ai_cache {
+        KeyCache::Openai(cached) => ("openai_path", cached.prepare_request(&ai_path, body)),
+        KeyCache::Anthropic(cached) => ("anthropic_path", cached.prepare_request(&ai_path, body)),
+        KeyCache::Mistral(cached) => ("mistral_path", cached.prepare_request(&ai_path, body)),
+    };
+
+    let response = request?.send().await.map_err(to_anyhow)?;
+
+    let mut tx = db.begin().await?;
+
+    audit_log(
+        &mut *tx,
+        &authed,
+        "ai.request",
+        ActionKind::Execute,
+        &w_id,
+        Some(&authed.email),
+        Some([(path, &format!("{:?}", ai_path)[..])].into()),
+    )
+    .await?;
+    tx.commit().await?;
+
+    if response.error_for_status_ref().is_err() {
+        let err_msg = response.text().await.unwrap_or("".to_string());
+        return Err(Error::AiError(err_msg));
+    }
+
+    let status_code = response.status();
+    let headers = response.headers().clone();
+    let stream = response.bytes_stream();
+    Ok((status_code, headers, axum::body::Body::from_stream(stream)))
+}

backend/windmill-api/src/apps.rs

@@ -7,6 +7,12 @@ use std::collections::HashMap;
  * Please see the included NOTICE for copyright information and
  * LICENSE-AGPL for a copy of the license.
  */
+
+#[cfg(feature = "parquet")]
+use crate::{job_helpers_ee::{
+    get_random_file_name, get_s3_resource, get_workspace_s3_resource, upload_file_internal,
+    UploadFileResponse,
+}, users::fetch_api_authed_from_permissioned_as};
 use crate::{
     db::{ApiAuthed, DB},
     resources::get_resource_value_interpolated_internal,
@@ -23,7 +29,13 @@ use axum::{
     Router,
 };
 use hyper::StatusCode;
+#[cfg(feature = "parquet")]
+use itertools::Itertools;
 use magic_crypt::MagicCryptTrait;
+#[cfg(feature = "parquet")]
+use object_store::{Attribute, Attributes};
+#[cfg(feature = "parquet")]
+use regex::Regex;
 use serde::{Deserialize, Serialize};
 use serde_json::{json, value::RawValue};
 use sha2::{Digest, Sha256};
@@ -32,6 +44,8 @@ use sqlx::{types::Uuid, FromRow};
 use std::str;
 use windmill_audit::audit_ee::audit_log;
 use windmill_audit::ActionKind;
+#[cfg(feature = "parquet")]
+use windmill_common::s3_helpers::build_object_store_client;
 use windmill_common::{
     apps::ListAppQuery,
     db::UserDB,
@@ -69,6 +83,7 @@ pub fn workspaced_service() -> Router {
 pub fn unauthed_service() -> Router {
     Router::new()
         .route("/execute_component/*path", post(execute_component))
+        .route("/upload_s3_file/*path", post(upload_s3_file_from_app))
         .route("/public_app/:secret", get(get_public_app_by_secret))
         .route("/public_resource/*path", get(get_public_resource))
 }
@@ -179,6 +194,14 @@ pub struct PolicyTriggerableInputs {
     allow_user_resources: AllowUserResources,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct S3Input {
+    allowed_resources: Vec<String>,
+    allow_user_resources: bool,
+    allow_workspace_resource: bool,
+    file_key_regex: String,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Policy {
     pub on_behalf_of: Option<String>,
@@ -192,6 +215,7 @@ pub struct Policy {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub triggerables_v2: Option<HashMap<String, PolicyTriggerableInputs>>,
     pub execution_mode: ExecutionMode,
+    pub s3_inputs: Option<Vec<S3Input>>,
 }
 
 #[derive(Deserialize)]
@@ -432,9 +456,7 @@ async fn get_latest_version(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
-
 ) -> JsonResult<Option<AppHistory>> {
-
     let mut tx = user_db.begin(&authed).await?;
     let row = sqlx::query!(
         "SELECT a.id as app_id, av.id as version_id, dm.deployment_msg as deployment_msg
@@ -457,7 +479,6 @@ async fn get_latest_version(
     } else {
         return Ok(Json(None));
     }
-
 }
 
 async fn update_app_history(
@@ -1067,6 +1088,49 @@ fn digest(code: &str) -> String {
     format!("rawscript/{:x}", result)
 }
 
+async fn get_on_behalf_details_from_policy_and_authed(
+    policy: &Policy,
+    opt_authed: &Option<ApiAuthed>,
+) -> Result<(String, String, String)> {
+    let (username, permissioned_as, email) = match policy.execution_mode {
+        ExecutionMode::Anonymous => {
+            let username = opt_authed
+                .as_ref()
+                .map(|a| a.username.clone())
+                .unwrap_or_else(|| "anonymous".to_string());
+            let (permissioned_as, email) = get_on_behalf_of(&policy)?;
+            (username, permissioned_as, email)
+        }
+        ExecutionMode::Publisher => {
+            let username = opt_authed
+                .as_ref()
+                .map(|a| a.username.clone())
+                .ok_or_else(|| {
+                    Error::BadRequest(
+                        "publisher execution mode requires authentication".to_string(),
+                    )
+                })?;
+            let (permissioned_as, email) = get_on_behalf_of(&policy)?;
+            (username, permissioned_as, email)
+        }
+        ExecutionMode::Viewer => {
+            let (username, email) = opt_authed
+                .as_ref()
+                .map(|a| (a.username.clone(), a.email.clone()))
+                .ok_or_else(|| {
+                    Error::BadRequest("Required to be authed in viewer mode".to_string())
+                })?;
+            (
+                username.clone(),
+                username_to_permissioned_as(&username),
+                email,
+            )
+        }
+    };
+
+    Ok((username, permissioned_as, email))
+}
+
 async fn execute_component(
     OptAuthed(opt_authed): OptAuthed,
     Extension(db): Extension<DB>,
@@ -1129,6 +1193,7 @@ async fn execute_component(
                 triggerables_v2: Some(hm),
                 on_behalf_of: None,
                 on_behalf_of_email: None,
+                s3_inputs: None,
             }
         }
         _ => {
@@ -1146,41 +1211,8 @@ async fn execute_component(
         }
     };
 
-    let (username, permissioned_as, email) = match policy.execution_mode {
-        ExecutionMode::Anonymous => {
-            let username = opt_authed
-                .as_ref()
-                .map(|a| a.username.clone())
-                .unwrap_or_else(|| "anonymous".to_string());
-            let (permissioned_as, email) = get_on_behalf_of(&policy)?;
-            (username, permissioned_as, email)
-        }
-        ExecutionMode::Publisher => {
-            let username = opt_authed
-                .as_ref()
-                .map(|a| a.username.clone())
-                .ok_or_else(|| {
-                    Error::BadRequest(
-                        "publisher execution mode requires authentication".to_string(),
-                    )
-                })?;
-            let (permissioned_as, email) = get_on_behalf_of(&policy)?;
-            (username, permissioned_as, email)
-        }
-        ExecutionMode::Viewer => {
-            let (username, email) = opt_authed
-                .as_ref()
-                .map(|a| (a.username.clone(), a.email.clone()))
-                .ok_or_else(|| {
-                    Error::BadRequest("Required to be authed in viewer mode".to_string())
-                })?;
-            (
-                username.clone(),
-                username_to_permissioned_as(&username),
-                email,
-            )
-        }
-    };
+    let (username, permissioned_as, email) =
+        get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?;
 
     let (job_payload, (args, job_id), tag) = match payload {
         ExecuteApp { args, component, raw_code: Some(raw_code), path: None, .. } => {
@@ -1249,6 +1281,248 @@ async fn execute_component(
     Ok(uuid.to_string())
 }
 
+#[cfg(not(feature = "parquet"))]
+async fn upload_s3_file_from_app() -> Result<()> {
+    return Err(Error::BadRequest(
+        "This endpoint requires the parquet feature to be enabled".to_string(),
+    ));
+}
+
+
+#[cfg(feature = "parquet")]
+#[derive(Debug, Deserialize, Clone)]
+struct UploadFileToS3Query {
+    file_key: Option<String>,
+    file_extension: Option<String>,
+    s3_resource_path: Option<String>,
+    content_type: Option<String>,
+    content_disposition: Option<String>,
+    force_viewer_file_key_regex: Option<String>,
+    force_viewer_allow_user_resources: Option<bool>,
+    force_viewer_allow_workspace_resource: Option<bool>,
+    force_viewer_allowed_resources: Option<String>,
+}
+
+#[cfg(feature = "parquet")]
+async fn upload_s3_file_from_app(
+    OptAuthed(opt_authed): OptAuthed,
+    Extension(db): Extension<DB>,
+    Path((w_id, path)): Path<(String, StripPath)>,
+    Query(query): Query<UploadFileToS3Query>,
+    request: axum::extract::Request,
+) -> JsonResult<UploadFileResponse> {
+    let policy = if let Some(file_key_regex) = query.force_viewer_file_key_regex {
+        Some(Policy {
+            execution_mode: ExecutionMode::Viewer,
+            triggerables: None,
+            triggerables_v2: None,
+            on_behalf_of: None,
+            on_behalf_of_email: None,
+            s3_inputs: Some(vec![S3Input {
+                file_key_regex: file_key_regex,
+                allow_user_resources: query.force_viewer_allow_user_resources.unwrap_or(false),
+                allow_workspace_resource: query
+                    .force_viewer_allow_workspace_resource
+                    .unwrap_or(false),
+                allowed_resources: query
+                    .force_viewer_allowed_resources
+                    .map(|s| s.split(',').map(|s| s.to_string()).collect())
+                    .unwrap_or_default(),
+            }]),
+        })
+    } else {
+        let policy_o = sqlx::query_scalar!(
+            "SELECT policy from app WHERE path = $1 AND workspace_id = $2",
+            &path.0,
+            &w_id
+        )
+        .fetch_optional(&db)
+        .await?;
+
+        policy_o
+            .map(|p| serde_json::from_value::<Policy>(p).map_err(to_anyhow))
+            .transpose()?
+    };
+
+    let user_db = UserDB::new(db.clone());
+
+    let (s3_resource_opt, file_key) = if policy.as_ref().is_some_and(|p| p.s3_inputs.is_some()) {
+        let policy = policy.unwrap();
+        let s3_inputs = policy.s3_inputs.as_ref().unwrap();
+
+        let (username, permissioned_as, email) =
+            get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?;
+
+        let on_behalf_authed =
+            fetch_api_authed_from_permissioned_as(permissioned_as, email, &w_id, &db, username)
+                .await?;
+
+        if let Some(file_key) = query.file_key {
+            // file key is provided => requires workspace, user or list policy and must match the regex
+            let matching_s3_inputs = if let Some(ref s3_resource_path) = query.s3_resource_path {
+                s3_inputs
+                    .iter()
+                    .filter(|s3_input| {
+                        s3_input.allowed_resources.contains(s3_resource_path)
+                            || s3_input.allow_user_resources
+                    })
+                    .sorted_by_key(|i| i.allow_user_resources) // consider user resources last
+                    .collect::<Vec<_>>()
+            } else {
+                s3_inputs
+                    .iter()
+                    .filter(|s3_input| s3_input.allow_workspace_resource)
+                    .collect::<Vec<_>>()
+            };
+
+            let matched_input = matching_s3_inputs.iter().find(|s3_input| {
+                match Regex::new(&s3_input.file_key_regex) {
+                    Ok(re) => re.is_match(&file_key),
+                    Err(e) => {
+                        tracing::error!("Error compiling regex: {}", e);
+                        false
+                    }
+                }
+            });
+
+            if let Some(matched_input) = matched_input {
+                if let Some(ref s3_resource_path) = query.s3_resource_path {
+                    if matched_input.allow_user_resources {
+                        if let Some(authed) = opt_authed {
+                            (
+                                Some(
+                                    get_s3_resource(
+                                        &authed,
+                                        &db,
+                                        Some(user_db),
+                                        "",
+                                        &w_id,
+                                        s3_resource_path,
+                                        None,
+                                        None,
+                                    )
+                                    .await?,
+                                ),
+                                file_key,
+                            )
+                        } else {
+                            return Err(Error::BadRequest(
+                                "User resources are not allowed without being logged in"
+                                    .to_string(),
+                            ));
+                        }
+                    } else {
+                        (
+                            Some(
+                                get_s3_resource(
+                                    &on_behalf_authed,
+                                    &db,
+                                    Some(user_db),
+                                    "",
+                                    &w_id,
+                                    s3_resource_path,
+                                    None,
+                                    None,
+                                )
+                                .await?,
+                            ),
+                            file_key,
+                        )
+                    }
+                } else {
+                    let (_, s3_resource_opt) =
+                        get_workspace_s3_resource(&on_behalf_authed, &db, None, "", &w_id, None)
+                            .await?;
+                    (s3_resource_opt, file_key)
+                }
+            } else {
+                return Err(Error::BadRequest(
+                    "No matching s3 resource found for the given file key".to_string(),
+                ));
+            }
+        } else {
+            // no file key => requires unnamed upload policy => allow workspace resource and file_key_regex is empty
+            let has_unnamed_policy = s3_inputs.iter().any(|s3_input| {
+                s3_input.allow_workspace_resource && s3_input.file_key_regex.is_empty()
+            });
+
+            if !has_unnamed_policy {
+                return Err(Error::BadRequest(
+                    "no policy found for unnamed s3 file uplooad".to_string(),
+                ));
+            }
+
+            // for now, we place all files into `windmill_uploads` folder with a random name
+            // TODO: make the folder configurable via the workspace settings
+            let file_key = get_random_file_name(query.file_extension);
+
+            let (_, s3_resource_opt) =
+                get_workspace_s3_resource(&on_behalf_authed, &db, None, "", &w_id, None).await?;
+
+            (s3_resource_opt, file_key)
+        }
+    } else {
+        // backward compatibility (no policy)
+        // if no policy but logged in, use the user's auth to get the s3 resource
+        if let Some(authed) = opt_authed {
+            let file_key = query
+                .file_key
+                .unwrap_or_else(|| get_random_file_name(query.file_extension));
+
+            if let Some(ref s3_resource_path) = query.s3_resource_path {
+                (
+                    Some(
+                        get_s3_resource(
+                            &authed,
+                            &db,
+                            Some(user_db),
+                            "",
+                            &w_id,
+                            s3_resource_path,
+                            None,
+                            None,
+                        )
+                        .await?,
+                    ),
+                    file_key,
+                )
+            } else {
+                let (_, s3_resource) =
+                    get_workspace_s3_resource(&authed, &db, None, "", &w_id, None).await?;
+
+                (s3_resource, file_key)
+            }
+        } else {
+            return Err(Error::BadRequest("Missing s3 policy".to_string()));
+        }
+    };
+
+    let s3_resource = s3_resource_opt.ok_or(Error::InternalErr(
+        "No files storage resource defined at the workspace level".to_string(),
+    ))?;
+    let s3_client = build_object_store_client(&s3_resource).await?;
+
+    let options = Attributes::from_iter(vec![
+        (
+            Attribute::ContentType,
+            query.content_type.unwrap_or_else(|| {
+                mime_guess::from_path(&file_key)
+                    .first_or_octet_stream()
+                    .to_string()
+            }),
+        ),
+        (
+            Attribute::ContentDisposition,
+            query.content_disposition.unwrap_or("inline".to_string()),
+        ),
+    ])
+    .into();
+
+    upload_file_internal(s3_client, &file_key, request, options).await?;
+
+    return Ok(Json(UploadFileResponse { file_key }));
+}
+
 fn get_on_behalf_of(policy: &Policy) -> Result<(String, String)> {
     let permissioned_as = policy
         .on_behalf_of

backend/windmill-api/src/flows.rs

@@ -1209,6 +1209,7 @@ mod tests {
                         .into(),
                         hash: None,
                         tag_override: None,
+                        is_trigger: None,
                     }),
                     stop_after_if: None,
                     stop_after_all_iters_if: None,
@@ -1236,6 +1237,7 @@ mod tests {
                         custom_concurrency_key: None,
                         concurrent_limit: None,
                         concurrency_time_window_s: None,
+                        is_trigger: None,
                     }),
                     stop_after_if: Some(StopAfterIf {
                         expr: "foo = 'bar'".to_string(),
@@ -1290,6 +1292,7 @@ mod tests {
                     input_transforms: HashMap::new(),
                     hash: None,
                     tag_override: None,
+                    is_trigger: None,
                 }
                 .into(),
                 stop_after_if: Some(StopAfterIf {

backend/windmill-api/src/http_triggers.rs

@@ -1,24 +1,5 @@
-use axum::{
-    extract::{Path, Query},
-    response::IntoResponse,
-    routing::{delete, get, post},
-    Extension, Json, Router,
-};
-use http::{HeaderMap, StatusCode};
-use serde::{Deserialize, Serialize};
-use sql_builder::{bind::Bind, SqlBuilder};
-use sqlx::prelude::FromRow;
-use std::collections::HashMap;
-use tower_http::cors::CorsLayer;
-use windmill_audit::{audit_ee::audit_log, ActionKind};
-use windmill_common::{
-    db::UserDB,
-    error::{self, JsonResult},
-    utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
-    worker::{to_raw_value, CLOUD_HOSTED},
-};
-use windmill_queue::PushArgsOwned;
-
+#[cfg(feature = "parquet")]
+use crate::job_helpers_ee::get_workspace_s3_resource;
 use crate::{
     db::{ApiAuthed, DB},
     jobs::{
@@ -27,6 +8,31 @@ use crate::{
     },
     users::{fetch_api_authed, OptAuthed},
 };
+use axum::{
+    extract::{Path, Query},
+    response::IntoResponse,
+    routing::{delete, get, post},
+    Extension, Json, Router,
+};
+#[cfg(feature = "parquet")]
+use http::header::IF_NONE_MATCH;
+use http::{HeaderMap, StatusCode};
+use serde::{Deserialize, Serialize};
+use sql_builder::{bind::Bind, SqlBuilder};
+use sqlx::prelude::FromRow;
+use std::collections::HashMap;
+use tower_http::cors::CorsLayer;
+use windmill_audit::{audit_ee::audit_log, ActionKind};
+#[cfg(feature = "parquet")]
+use windmill_common::s3_helpers::build_object_store_client;
+use windmill_common::{
+    db::UserDB,
+    error::{self, JsonResult},
+    s3_helpers::S3Object,
+    utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
+    worker::{to_raw_value, CLOUD_HOSTED},
+};
+use windmill_queue::PushArgsOwned;
 
 lazy_static::lazy_static! {
     static ref ROUTE_PATH_KEY_RE: regex::Regex = regex::Regex::new(r"/:\w+").unwrap();
@@ -99,6 +105,7 @@ struct NewTrigger {
     is_async: bool,
     requires_auth: bool,
     http_method: HttpMethod,
+    static_asset_config: Option<sqlx::types::Json<S3Object>>,
 }
 
 #[derive(FromRow, Serialize)]
@@ -116,6 +123,7 @@ struct Trigger {
     is_async: bool,
     requires_auth: bool,
     http_method: HttpMethod,
+    static_asset_config: Option<sqlx::types::Json<S3Object>>,
 }
 
 #[derive(Deserialize)]
@@ -127,6 +135,7 @@ struct EditTrigger {
     is_async: bool,
     requires_auth: bool,
     http_method: HttpMethod,
+    static_asset_config: Option<sqlx::types::Json<S3Object>>,
 }
 
 #[derive(Deserialize)]
@@ -182,7 +191,7 @@ async fn get_trigger(
     let path = path.to_path();
     let trigger = sqlx::query_as!(
         Trigger,
-        r#"SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, http_method as "http_method: _", edited_by, email, edited_at, extra_perms, is_async, requires_auth
+        r#"SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, http_method as "http_method: _", edited_by, email, edited_at, extra_perms, is_async, requires_auth, static_asset_config as "static_asset_config: _"
             FROM http_trigger
             WHERE workspace_id = $1 AND path = $2"#,
         w_id,
@@ -209,7 +218,7 @@ async fn create_trigger(
 
     let mut tx = user_db.begin(&authed).await?;
     sqlx::query!(
-        "INSERT INTO http_trigger (workspace_id, path, route_path, route_path_key, script_path, is_flow, is_async, requires_auth, http_method, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now())",
+        "INSERT INTO http_trigger (workspace_id, path, route_path, route_path_key, script_path, is_flow, is_async, requires_auth, http_method, static_asset_config, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now())",
         w_id,
         ct.path,
         ct.route_path,
@@ -218,7 +227,8 @@ async fn create_trigger(
         ct.is_flow,
         ct.is_async,
         ct.requires_auth,
-        ct.http_method as HttpMethod,
+        ct.http_method as _,
+        ct.static_asset_config as _,
         &authed.username,
         &authed.email
     )
@@ -261,14 +271,15 @@ async fn update_trigger(
 
         sqlx::query!(
             "UPDATE http_trigger 
-                SET route_path = $1, route_path_key = $2, script_path = $3, path = $4, is_flow = $5, http_method = $6, edited_by = $7, email = $8, is_async = $9, requires_auth = $10, edited_at = now() 
-                WHERE workspace_id = $11 AND path = $12",
+                SET route_path = $1, route_path_key = $2, script_path = $3, path = $4, is_flow = $5, http_method = $6, static_asset_config = $7, edited_by = $8, email = $9, is_async = $10, requires_auth = $11, edited_at = now() 
+                WHERE workspace_id = $12 AND path = $13",
             ct.route_path,
             &route_path_key,
             ct.script_path,
             ct.path,
             ct.is_flow,
-            ct.http_method as HttpMethod,
+            ct.http_method as _,
+            ct.static_asset_config as _,
             &authed.username,
             &authed.email,
             ct.is_async,
@@ -279,12 +290,13 @@ async fn update_trigger(
         .execute(&mut *tx).await?;
     } else {
         sqlx::query!(
-            "UPDATE http_trigger SET script_path = $1, path = $2, is_flow = $3, http_method = $4, edited_by = $5, email = $6, is_async = $7, requires_auth = $8, edited_at = now() 
-                WHERE workspace_id = $9 AND path = $10",
+            "UPDATE http_trigger SET script_path = $1, path = $2, is_flow = $3, http_method = $4, static_asset_config = $5, edited_by = $6, email = $7, is_async = $8, requires_auth = $9, edited_at = now() 
+                WHERE workspace_id = $10 AND path = $11",
             ct.script_path,
             ct.path,
             ct.is_flow,
-            ct.http_method as HttpMethod,
+            ct.http_method as _,
+            ct.static_asset_config as _,
             &authed.username,
             &authed.email,
             ct.is_async,
@@ -405,6 +417,7 @@ struct TriggerRoute {
     edited_by: String,
     email: String,
     http_method: HttpMethod,
+    static_asset_config: Option<sqlx::types::Json<S3Object>>,
 }
 
 async fn get_http_route_trigger(
@@ -421,7 +434,7 @@ async fn get_http_route_trigger(
         let route_path = StripPath(splitted.collect::<Vec<_>>().join("/"));
         let triggers = sqlx::query_as!(
             TriggerRoute,
-            r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _" FROM http_trigger WHERE workspace_id = $1"#,
+            r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _", static_asset_config as "static_asset_config: _" FROM http_trigger WHERE workspace_id = $1"#,
             w_id
         )
         .fetch_all(db)
@@ -430,7 +443,7 @@ async fn get_http_route_trigger(
     } else {
         let triggers = sqlx::query_as!(
             TriggerRoute,
-            r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _" FROM http_trigger"#,
+            r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _", static_asset_config as "static_asset_config: _" FROM http_trigger"#,
         )
         .fetch_all(db)
         .await?;
@@ -518,6 +531,90 @@ async fn route_job(
             Ok(trigger) => trigger,
             Err(e) => return e.into_response(),
         };
+
+    #[cfg(not(feature = "parquet"))]
+    if trigger.static_asset_config.is_some() {
+        return error::Error::InternalErr(
+            "Static asset configuration is not supported in this build".to_string(),
+        )
+        .into_response();
+    }
+
+    #[cfg(feature = "parquet")]
+    if let Some(sqlx::types::Json(config)) = trigger.static_asset_config {
+        let build_static_response_f = async {
+            let (_, s3_resource_opt) = get_workspace_s3_resource(
+                &authed,
+                &db,
+                None,
+                &"NO_TOKEN".to_string(), // no token is provided in this case
+                &trigger.workspace_id,
+                config.storage,
+            )
+            .await?;
+            let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
+                "No files storage resource defined at the workspace level".to_string(),
+            ))?;
+            let s3_client = build_object_store_client(&s3_resource).await?;
+            let path = object_store::path::Path::from(config.s3);
+            let s3_object = s3_client.get(&path).await.map_err(|err| {
+                tracing::warn!("Error retrieving file from S3: {:?}", err);
+                error::Error::InternalErr(format!("Error retrieving file: {}", err.to_string()))
+            })?;
+            let mut response_headers = http::HeaderMap::new();
+            if let Some(ref e_tag) = s3_object.meta.e_tag {
+                if let Some(if_none_match) = headers.get(IF_NONE_MATCH) {
+                    if if_none_match == e_tag {
+                        return Ok::<_, error::Error>((
+                            StatusCode::NOT_MODIFIED,
+                            response_headers,
+                            axum::body::Body::empty(),
+                        ));
+                    }
+                }
+                if let Ok(e_tag) = e_tag.parse() {
+                    response_headers.insert("etag", e_tag);
+                }
+            }
+            response_headers.insert(
+                "content-type",
+                s3_object
+                    .attributes
+                    .get(&object_store::Attribute::ContentType)
+                    .map(|s| s.parse().ok())
+                    .flatten()
+                    .unwrap_or("application/octet-stream".parse().unwrap()),
+            );
+            response_headers.insert(
+                "content-disposition",
+                config.filename.as_ref().map_or_else(
+                    || {
+                        s3_object
+                            .attributes
+                            .get(&object_store::Attribute::ContentDisposition)
+                            .map(|s| s.parse().ok())
+                            .flatten()
+                            .unwrap_or("inline".parse().unwrap())
+                    },
+                    |filename| {
+                        format!("inline; filename=\"{}\"", filename)
+                            .parse()
+                            .unwrap_or("inline".parse().unwrap())
+                    },
+                ),
+            );
+
+            let body_stream = axum::body::Body::from_stream(s3_object.into_stream());
+            Ok::<_, error::Error>((StatusCode::OK, response_headers, body_stream))
+        };
+        match build_static_response_f.await {
+            Ok((status, headers, body_stream)) => {
+                return (status, headers, body_stream).into_response()
+            }
+            Err(e) => return e.into_response(),
+        }
+    }
+
     let headers = headers
         .iter()
         .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))

backend/windmill-api/src/indexer_ee.rs

@@ -3,3 +3,7 @@ use axum::Router;
 pub fn workspaced_service() -> Router {
     Router::new()
 }
+
+pub fn global_service() -> Router {
+    Router::new()
+}

backend/windmill-api/src/job_helpers_ee.rs

@@ -1,5 +1,61 @@
 use axum::Router;
+use serde::Serialize;
+use uuid::Uuid;
+use windmill_common::s3_helpers::StorageResourceType;
+
+#[cfg(feature = "parquet")]
+use crate::db::{ApiAuthed, DB};
+#[cfg(feature = "parquet")]
+use object_store::{ObjectStore, PutMultipartOpts};
+#[cfg(feature = "parquet")]
+use std::sync::Arc;
+use windmill_common::error;
+#[cfg(feature = "parquet")]
+use windmill_common::{db::UserDB, s3_helpers::ObjectStoreResource};
+#[derive(Serialize)]
+pub struct UploadFileResponse {
+    pub file_key: String,
+}
 
 pub fn workspaced_service() -> Router {
     Router::new()
 }
+
+#[cfg(feature = "parquet")]
+pub async fn get_workspace_s3_resource<'c>(
+    _authed: &ApiAuthed,
+    _db: &DB,
+    _user_db: Option<UserDB>,
+    _token: &str,
+    _w_id: &str,
+    _storage: Option<String>,
+) -> windmill_common::error::Result<(Option<bool>, Option<ObjectStoreResource>)> {
+    // implementation is not open source
+    Ok((None, None))
+}
+
+pub fn get_random_file_name(_file_extension: Option<String>) -> String {
+    todo!()
+}
+
+pub async fn get_s3_resource<'c>(
+    _authed: &ApiAuthed,
+    _db: &DB,
+    _user_db: Option<UserDB>,
+    _token: &str,
+    _w_id: &str,
+    _resource_path: &str,
+    _resource_type: Option<StorageResourceType>,
+    _job_id: Option<Uuid>,
+) -> error::Result<ObjectStoreResource> {
+    todo!()
+}
+
+pub async fn upload_file_internal(
+    _s3_client: Arc<dyn ObjectStore>,
+    _file_key: &str,
+    _request: axum::extract::Request,
+    _options: PutMultipartOpts,
+) -> error::Result<()> {
+    todo!()
+}

backend/windmill-api/src/jobs.rs

@@ -12,6 +12,7 @@ use quick_cache::sync::Cache;
 use serde_json::value::RawValue;
 use sqlx::Pool;
 use std::collections::HashMap;
+use std::ops::{Deref, DerefMut};
 #[cfg(feature = "prometheus")]
 use std::sync::atomic::Ordering;
 use tokio::io::AsyncReadExt;
@@ -65,7 +66,7 @@ use windmill_common::{
     db::UserDB,
     error::{self, to_anyhow, Error},
     flow_status::{Approval, FlowStatus, FlowStatusModule},
-    flows::FlowValue,
+    flows::{add_virtual_items_if_necessary, FlowValue},
     jobs::{script_path_to_payload, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode},
     oauth2::HmacSha256,
     scripts::{ScriptHash, ScriptLang},
@@ -205,6 +206,7 @@ pub fn workspaced_service() -> Router {
         .route("/queue/list_filtered_uuids", get(list_filtered_uuids))
         .route("/queue/cancel_selection", post(cancel_selection))
         .route("/completed/count", get(count_completed_jobs))
+        .route("/completed/count_jobs", get(count_completed_jobs_detail))
         .route(
             "/completed/list",
             get(list_completed_jobs).layer(cors.clone()),
@@ -600,7 +602,7 @@ async fn get_flow_job_debug_info(
     Extension(db): Extension<DB>,
     Path((w_id, id)): Path<(String, Uuid)>,
 ) -> error::Result<Response> {
-    let job = get_queued_job(&id, w_id.as_str(), &db).await?;
+    let job = get_queued_job_ex(&db, &w_id, id, false, None).await?;
     if let Some(job) = job {
         let is_flow = job.is_flow();
         if job.is_flow_step || !is_flow {
@@ -734,6 +736,67 @@ fn generate_get_job_query(no_logs: bool, table: &str) -> String {
     {join}
     WHERE id = $1 AND {table}.workspace_id = $2");
 }
+pub async fn get_queued_job_ex(
+    db: &DB,
+    workspace_id: &str,
+    job_id: Uuid,
+    no_logs: bool,
+    // first optional is if authed need to be checked, second is the opt_authed itself
+    opt_authed: Option<&Option<ApiAuthed>>,
+) -> error::Result<Option<JobExtended<QueuedJob>>> {
+    let query = if no_logs { &*GET_QUEUED_JOB_QUERY_NO_LOGS } else { &*GET_QUEUED_JOB_QUERY };
+    let job = sqlx::query_as::<_, JobExtended<QueuedJob>>(query)
+        .bind(job_id)
+        .bind(workspace_id)
+        .fetch_optional(db)
+        .await?;
+
+    if let Some(job) = job.as_ref() {
+        if opt_authed.is_some_and(|x| x.is_none()) && job.created_by != "anonymous" {
+            return Err(Error::BadRequest(
+                "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
+            ));
+        }
+    }
+
+    Ok(job)
+}
+pub async fn get_completed_job_ex(
+    db: &DB,
+    workspace_id: &str,
+    job_id: Uuid,
+    no_logs: bool,
+    // first optional is if authed need to be checked, second is the opt_authed itself
+    opt_authed: Option<&Option<ApiAuthed>>,
+) -> error::Result<Option<JobExtended<CompletedJob>>> {
+    let query = if no_logs { &*GET_COMPLETED_JOB_QUERY_NO_LOGS } else { &*GET_COMPLETED_JOB_QUERY };
+    let cjob = sqlx::query_as::<_, JobExtended<CompletedJob>>(query)
+        .bind(job_id)
+        .bind(workspace_id)
+        .fetch_optional(db)
+        .await?;
+
+    if let Some(job) = cjob.as_ref() {
+        if opt_authed.is_some_and(|x| x.is_none()) && job.created_by != "anonymous" {
+            return Err(Error::BadRequest(
+                "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
+            ));
+        }
+    }
+
+    if let Some(mut cjob) = cjob {
+        let CompletedJobWithFormattedResult { mut cj, result } = format_completed_job_result(cjob.inner);
+        cj.result = match result {
+            Some(FormattedResult::RawValue(rv)) => rv,
+            Some(FormattedResult::Vec(v)) => Some(to_raw_value(&v)),
+            None => None,
+        }.map(sqlx::types::Json);
+        cjob.inner = cj;
+        return Ok(Some(cjob));
+    }
+
+    Ok(cjob)
+}
 pub async fn get_job_internal(
     db: &DB,
     workspace_id: &str,
@@ -742,48 +805,18 @@ pub async fn get_job_internal(
     // first optional is if authed need to be checked, second is the opt_authed itself
     opt_authed: Option<&Option<ApiAuthed>>,
 ) -> error::Result<Job> {
-    let cjob_maybe = sqlx::query_as::<_, CompletedJob>(if no_logs {
-        &*GET_COMPLETED_JOB_QUERY_NO_LOGS
-    } else {
-        &*GET_COMPLETED_JOB_QUERY
-    })
-    .bind(job_id)
-    .bind(workspace_id)
-    .fetch_optional(db)
-    .await?
-    .map(Job::CompletedJob);
-
-    if let Some(cjob) = cjob_maybe {
-        Ok(match cjob {
-            Job::CompletedJob(cjob) => {
-                if opt_authed.is_some_and(|x| x.is_none()) && cjob.created_by != "anonymous" {
-                    return Err(Error::BadRequest(
-                        "As a non logged in user, you can only see jobs ran by anonymous users"
-                            .to_string(),
-                    ));
-                }
-                Job::CompletedJobWithFormattedResult(format_completed_job_result(cjob))
-            }
-            cjob => cjob,
-        })
-    } else {
-        let job_o = sqlx::query_as::<_, QueuedJob>(if no_logs {
-            &*GET_QUEUED_JOB_QUERY_NO_LOGS
-        } else {
-            &*GET_QUEUED_JOB_QUERY
-        })
-        .bind(job_id)
-        .bind(workspace_id)
-        .fetch_optional(db)
+    let cjob = get_completed_job_ex(db, workspace_id, job_id, no_logs, opt_authed.clone())
         .await?
-        .map(Job::QueuedJob);
-        let job: Job = not_found_if_none(job_o, "Job", job_id.to_string())?;
-        if opt_authed.is_some_and(|x| x.is_none()) && job.created_by() != "anonymous" {
-            return Err(Error::BadRequest(
-                "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
-            ));
+        .map(Job::CompletedJob);
+
+    match cjob {
+        Some(cjob) => Ok(cjob),
+        None => {
+            let job_maybe = get_queued_job_ex(db, workspace_id, job_id, no_logs, opt_authed)
+                .await?
+                .map(Job::QueuedJob);
+            not_found_if_none(job_maybe, "Job", job_id.to_string())
         }
-        Ok(job)
     }
 }
 
@@ -1560,6 +1593,50 @@ async fn count_queue_jobs(
     ))
 }
 
+#[derive(Deserialize)]
+pub struct CountCompletedJobsQuery {
+    completed_after_s_ago: Option<i64>,
+    success: Option<bool>,
+    tags: Option<String>,
+    all_workspaces: Option<bool>,
+}
+
+async fn count_completed_jobs_detail(
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    Query(query): Query<CountCompletedJobsQuery>,
+) -> error::JsonResult<i64> {
+    let mut sqlb = SqlBuilder::select_from("completed_job");
+    sqlb
+        .field("COUNT(*) as count");
+
+    if !query.all_workspaces.unwrap_or(false) {
+        sqlb.and_where_eq("workspace_id", "?".bind(&w_id));
+    }
+
+    if let Some(after_s_ago) = query.completed_after_s_ago {
+        let after = Utc::now() - chrono::Duration::seconds(after_s_ago);
+        sqlb.and_where_gt("started_at + duration_ms / 1000 * interval '1 second'", "?".bind(&after.to_rfc3339()));
+    }
+
+    if let Some(success) = query.success {
+        sqlb.and_where_eq("success", "?".bind(&success));
+    }
+
+    if let Some(tags) = query.tags {
+        sqlb.and_where_in("tag", &tags.split(",").map(|t| format!("'{}'", t)).collect::<Vec<_>>());
+    }
+
+    let sql = sqlb.sql()?;
+    let stats = sqlx::query_scalar::<_, i64>(&sql)
+        .fetch_one(&db)
+        .await?;
+
+    Ok(Json(stats))
+}
+
+
+
 async fn count_completed_jobs(
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
@@ -1735,7 +1812,6 @@ async fn resume_suspended_job_internal(
     let trigger_email = match &parent_flow {
         Job::CompletedJob(job) => &job.email,
         Job::QueuedJob(job) => &job.email,
-        Job::CompletedJobWithFormattedResult(job) => &job.cj.email,
     };
     conditionally_require_authed_user(authed.clone(), flow_status, trigger_email)?;
 
@@ -2010,7 +2086,6 @@ pub async fn get_suspended_job_flow(
     let trigger_email = match &flow {
         Job::CompletedJob(job) => &job.email,
         Job::QueuedJob(job) => &job.email,
-        Job::CompletedJobWithFormattedResult(job) => &job.cj.email,
     };
     conditionally_require_authed_user(authed.clone(), flow_status.clone(), trigger_email)?;
 
@@ -2225,13 +2300,45 @@ pub async fn get_resume_urls(
     Ok(Json(res))
 }
 
+#[derive(sqlx::FromRow, Debug, Serialize)]
+pub struct JobExtended<T> {
+    #[sqlx(flatten)]
+    #[serde(flatten)]
+    inner: T,
+
+    #[sqlx(skip)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub self_wait_time_ms: Option<i64>,
+    #[sqlx(skip)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub aggregate_wait_time_ms: Option<i64>,
+}
+
+impl<T> JobExtended<T> {
+    pub fn new(self_wait_time_ms: Option<i64>, aggregate_wait_time_ms: Option<i64>, inner: T) -> Self {
+        Self { inner, self_wait_time_ms, aggregate_wait_time_ms }
+    }
+}
+
+impl<T> Deref for JobExtended<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl<T> DerefMut for JobExtended<T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.inner
+    }
+}
+
 #[derive(Serialize, Debug)]
 #[serde(tag = "type")]
 pub enum Job {
-    QueuedJob(QueuedJob),
-    CompletedJob(CompletedJob),
-    #[serde(rename = "CompletedJob")]
-    CompletedJobWithFormattedResult(CompletedJobWithFormattedResult),
+    QueuedJob(JobExtended<QueuedJob>),
+    CompletedJob(JobExtended<CompletedJob>),
 }
 
 impl Job {
@@ -2239,27 +2346,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => &job.created_by,
             Job::CompletedJob(job) => &job.created_by,
-            Job::CompletedJobWithFormattedResult(job) => &job.cj.created_by,
-        }
-    }
-    pub fn raw_flow(&self) -> Option<FlowValue> {
-        match self {
-            Job::QueuedJob(job) => job
-                .raw_flow
-                .as_ref()
-                .map(|rf| serde_json::from_str(rf.0.get()).ok())
-                .flatten(),
-            Job::CompletedJob(job) => job
-                .raw_flow
-                .as_ref()
-                .map(|rf| serde_json::from_str(rf.0.get()).ok())
-                .flatten(),
-            Job::CompletedJobWithFormattedResult(job) => job
-                .cj
-                .raw_flow
-                .as_ref()
-                .map(|rf| serde_json::from_str(rf.0.get()).ok())
-                .flatten(),
         }
     }
 
@@ -2279,13 +2365,6 @@ impl Job {
                     job.logs = Some(logs.to_string());
                 }
             }
-            Job::CompletedJobWithFormattedResult(job) => {
-                if let Some(ref mut l) = job.cj.logs {
-                    l.push_str(logs);
-                } else {
-                    job.cj.logs = Some(logs.to_string());
-                }
-            }
         }
     }
 
@@ -2293,7 +2372,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.logs.as_ref().map(|l| l.len()),
             Job::CompletedJob(job) => job.logs.as_ref().map(|l| l.len()),
-            Job::CompletedJobWithFormattedResult(job) => job.cj.logs.as_ref().map(|l| l.len()),
         }
     }
 
@@ -2301,7 +2379,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.logs.clone(),
             Job::CompletedJob(job) => job.logs.clone(),
-            Job::CompletedJobWithFormattedResult(job) => job.cj.logs.clone(),
         }
     }
     pub fn flow_status(&self) -> Option<FlowStatus> {
@@ -2316,19 +2393,12 @@ impl Job {
                 .as_ref()
                 .map(|rf| serde_json::from_str(rf.0.get()).ok())
                 .flatten(),
-            Job::CompletedJobWithFormattedResult(job) => job
-                .cj
-                .flow_status
-                .as_ref()
-                .map(|rf| serde_json::from_str(rf.0.get()).ok())
-                .flatten(),
         }
     }
     pub fn is_flow_step(&self) -> bool {
         match self {
             Job::QueuedJob(job) => job.is_flow_step,
             Job::CompletedJob(job) => job.is_flow_step,
-            Job::CompletedJobWithFormattedResult(job) => job.cj.is_flow_step,
         }
     }
 
@@ -2343,7 +2413,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => &job.job_kind,
             Job::CompletedJob(job) => &job.job_kind,
-            Job::CompletedJobWithFormattedResult(job) => &job.cj.job_kind,
         }
     }
 
@@ -2351,7 +2420,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.id,
             Job::CompletedJob(job) => job.id,
-            Job::CompletedJobWithFormattedResult(job) => job.cj.id,
         }
     }
 
@@ -2359,7 +2427,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => &job.workspace_id,
             Job::CompletedJob(job) => &job.workspace_id,
-            Job::CompletedJobWithFormattedResult(job) => &job.cj.workspace_id,
         }
     }
 
@@ -2367,7 +2434,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.script_path.as_ref(),
             Job::CompletedJob(job) => job.script_path.as_ref(),
-            Job::CompletedJobWithFormattedResult(job) => job.cj.script_path.as_ref(),
         }
         .map(String::as_str)
         .unwrap_or("tmp/main")
@@ -2377,7 +2443,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.args.as_ref(),
             Job::CompletedJob(job) => job.args.as_ref(),
-            Job::CompletedJobWithFormattedResult(job) => job.cj.args.as_ref(),
         }
     }
 
@@ -2426,10 +2491,6 @@ impl Job {
                 job.self_wait_time_ms = self_wait_time;
                 job.aggregate_wait_time_ms = aggregate_wait_time;
             }
-            Job::CompletedJobWithFormattedResult(job) => {
-                job.cj.self_wait_time_ms = self_wait_time;
-                job.cj.aggregate_wait_time_ms = aggregate_wait_time;
-            }
         }
         Ok(())
     }
@@ -2557,86 +2618,82 @@ impl UnifiedJob {
 impl<'a> From<UnifiedJob> for Job {
     fn from(uj: UnifiedJob) -> Self {
         match uj.typ.as_ref() {
-            "CompletedJob" => Job::CompletedJob(CompletedJob {
-                workspace_id: uj.workspace_id,
-                id: uj.id,
-                parent_job: uj.parent_job,
-                created_by: uj.created_by,
-                created_at: uj.created_at,
-                started_at: uj.started_at.unwrap_or(uj.created_at),
-                duration_ms: uj.duration_ms.unwrap(),
-                success: uj.success.unwrap(),
-                script_hash: uj.script_hash,
-                script_path: uj.script_path,
-                args: None,
-                result: None,
-                logs: None,
-                flow_status: None,
-                deleted: uj.deleted,
-                canceled: uj.canceled,
-                canceled_by: uj.canceled_by,
-                raw_code: None,
-                canceled_reason: None,
-                job_kind: uj.job_kind,
-                schedule_path: uj.schedule_path,
-                permissioned_as: uj.permissioned_as,
-                raw_flow: None,
-                is_flow_step: uj.is_flow_step,
-                language: uj.language,
-                is_skipped: uj.is_skipped,
-                email: uj.email,
-                visible_to_owner: uj.visible_to_owner,
-                mem_peak: uj.mem_peak,
-                tag: uj.tag,
-                priority: uj.priority,
-                labels: uj.labels,
-                self_wait_time_ms: uj.self_wait_time_ms,
-                aggregate_wait_time_ms: uj.aggregate_wait_time_ms,
-            }),
-            "QueuedJob" => Job::QueuedJob(QueuedJob {
-                workspace_id: uj.workspace_id,
-                id: uj.id,
-                parent_job: uj.parent_job,
-                created_by: uj.created_by,
-                created_at: uj.created_at,
-                started_at: uj.started_at,
-                script_hash: uj.script_hash,
-                script_path: uj.script_path,
-                args: None,
-                running: uj.running.unwrap(),
-                scheduled_for: uj.scheduled_for.unwrap(),
-                logs: None,
-                flow_status: None,
-                raw_code: None,
-                raw_lock: None,
-                canceled: uj.canceled,
-                canceled_by: uj.canceled_by,
-                canceled_reason: None,
-                last_ping: None,
-                job_kind: uj.job_kind,
-                schedule_path: uj.schedule_path,
-                permissioned_as: uj.permissioned_as,
-                raw_flow: None,
-                is_flow_step: uj.is_flow_step,
-                language: uj.language,
-                same_worker: false,
-                pre_run_error: None,
-                email: uj.email,
-                visible_to_owner: uj.visible_to_owner,
-                suspend: uj.suspend,
-                mem_peak: uj.mem_peak,
-                root_job: None,
-                leaf_jobs: None,
-                tag: uj.tag,
-                concurrent_limit: uj.concurrent_limit,
-                concurrency_time_window_s: uj.concurrency_time_window_s,
-                timeout: None,
-                flow_step_id: None,
-                cache_ttl: None,
-                priority: uj.priority,
-                self_wait_time_ms: uj.self_wait_time_ms,
-                aggregate_wait_time_ms: uj.aggregate_wait_time_ms,
-            }),
+            "CompletedJob" => Job::CompletedJob(JobExtended::new(uj.self_wait_time_ms, uj.aggregate_wait_time_ms, CompletedJob {
+                    workspace_id: uj.workspace_id,
+                    id: uj.id,
+                    parent_job: uj.parent_job,
+                    created_by: uj.created_by,
+                    created_at: uj.created_at,
+                    started_at: uj.started_at.unwrap_or(uj.created_at),
+                    duration_ms: uj.duration_ms.unwrap(),
+                    success: uj.success.unwrap(),
+                    script_hash: uj.script_hash,
+                    script_path: uj.script_path,
+                    args: None,
+                    result: None,
+                    logs: None,
+                    flow_status: None,
+                    deleted: uj.deleted,
+                    canceled: uj.canceled,
+                    canceled_by: uj.canceled_by,
+                    raw_code: None,
+                    canceled_reason: None,
+                    job_kind: uj.job_kind,
+                    schedule_path: uj.schedule_path,
+                    permissioned_as: uj.permissioned_as,
+                    raw_flow: None,
+                    is_flow_step: uj.is_flow_step,
+                    language: uj.language,
+                    is_skipped: uj.is_skipped,
+                    email: uj.email,
+                    visible_to_owner: uj.visible_to_owner,
+                    mem_peak: uj.mem_peak,
+                    tag: uj.tag,
+                    priority: uj.priority,
+                    labels: uj.labels,
+            })),
+            "QueuedJob" => Job::QueuedJob(JobExtended::new(uj.self_wait_time_ms, uj.aggregate_wait_time_ms, QueuedJob {
+                    workspace_id: uj.workspace_id,
+                    id: uj.id,
+                    parent_job: uj.parent_job,
+                    created_by: uj.created_by,
+                    created_at: uj.created_at,
+                    started_at: uj.started_at,
+                    script_hash: uj.script_hash,
+                    script_path: uj.script_path,
+                    args: None,
+                    running: uj.running.unwrap(),
+                    scheduled_for: uj.scheduled_for.unwrap(),
+                    logs: None,
+                    flow_status: None,
+                    raw_code: None,
+                    raw_lock: None,
+                    canceled: uj.canceled,
+                    canceled_by: uj.canceled_by,
+                    canceled_reason: None,
+                    last_ping: None,
+                    job_kind: uj.job_kind,
+                    schedule_path: uj.schedule_path,
+                    permissioned_as: uj.permissioned_as,
+                    raw_flow: None,
+                    is_flow_step: uj.is_flow_step,
+                    language: uj.language,
+                    same_worker: false,
+                    pre_run_error: None,
+                    email: uj.email,
+                    visible_to_owner: uj.visible_to_owner,
+                    suspend: uj.suspend,
+                    mem_peak: uj.mem_peak,
+                    root_job: None,
+                    leaf_jobs: None,
+                    tag: uj.tag,
+                    concurrent_limit: uj.concurrent_limit,
+                    concurrency_time_window_s: uj.concurrency_time_window_s,
+                    timeout: None,
+                    flow_step_id: None,
+                    cache_ttl: None,
+                    priority: uj.priority,
+            })),
             t => panic!("job type {} not valid", t),
         }
     }
@@ -3079,7 +3136,7 @@ pub async fn run_workflow_as_code(
         i += 1;
     }
 
-    let job = get_queued_job(&job_id, &w_id, &db).await?;
+    let job = get_queued_job_ex(&db, &w_id, job_id, true, None).await?;
 
     if *CLOUD_HOSTED {
         tracing::info!("workflow_as_code_tracing id {i} ");
@@ -3087,6 +3144,7 @@ pub async fn run_workflow_as_code(
     }
 
     let job = not_found_if_none(job, "Queued Job", &job_id.to_string())?;
+    let JobExtended { inner: job, .. } = job;
     let (job_payload, tag, _delete_after_use, timeout) = match job.job_kind {
         JobKind::Preview => (
             JobPayload::Code(RawCode {
@@ -4275,18 +4333,26 @@ async fn run_flow_dependencies_job(
     wait_result
 }
 
+#[derive(Deserialize)]
+struct BatchRawScript {
+    content: String,
+    language: Option<ScriptLang>,
+    lock: Option<String>,
+}
+
 #[derive(Deserialize)]
 struct BatchInfo {
     kind: String,
     flow_value: Option<FlowValue>,
     path: Option<String>,
+    rawscript: Option<BatchRawScript>,
 }
 
 #[tracing::instrument(level = "trace", skip_all)]
 async fn add_batch_jobs(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
-    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
+    Extension(_rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path((w_id, n)): Path<(String, i32)>,
     Json(batch_info): Json<BatchInfo>,
 ) -> error::JsonResult<Vec<Uuid>> {
@@ -4302,6 +4368,10 @@ async fn add_batch_jobs(
         concurrent_limit,
         concurrent_time_window_s,
         timeout,
+        raw_code,
+        raw_lock,
+        raw_flow,
+        flow_status
     ) = match batch_info.kind.as_str() {
         "script" => {
             if let Some(path) = batch_info.path {
@@ -4329,6 +4399,10 @@ async fn add_batch_jobs(
                     concurrent_limit,
                     concurrency_time_window_s,
                     timeout,
+                    None,
+                    None,
+                    None,
+                    None,
                 )
             } else {
                 Err(anyhow::anyhow!(
@@ -4336,61 +4410,72 @@ async fn add_batch_jobs(
                 ))?
             }
         }
-        "flow" => {
-            let mut uuids: Vec<Uuid> = Vec::new();
-            let payload = if let Some(ref fv) = batch_info.flow_value {
-                JobPayload::RawFlow { value: fv.clone(), path: None, restarted_from: None }
-            } else {
-                if let Some(path) = batch_info.path.as_ref() {
-                    JobPayload::Flow {
-                        path: path.to_string(),
-                        dedicated_worker: None,
-                        apply_preprocessor: false,
-                    }
-                } else {
-                    Err(anyhow::anyhow!(
-                        "Path is required if no value is not provided"
-                    ))?
-                }
-            };
-            let mut tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);
-            for _ in 0..n {
-                let ehm = HashMap::new();
-                let (uuid, ntx) = push(
-                    &db,
-                    tx,
-                    &w_id,
-                    payload.clone(),
-                    PushArgs::from(&ehm),
-                    authed.display_username(),
-                    &authed.email,
-                    username_to_permissioned_as(&authed.username),
+        "rawscript" => {
+            if let Some(rawscript) = batch_info.rawscript {
+                (
+                    None,
+                    None,
+                    JobKind::Preview,
+                    rawscript.language,
                     None,
                     None,
                     None,
                     None,
                     None,
-                    false,
-                    false,
-                    None,
-                    true,
+                    Some(rawscript.content),
+                    rawscript.lock,
                     None,
                     None,
-                    None,
-                    None,
-                    Some(&authed.clone().into()),
                 )
-                .await?;
-                tx = PushIsolationLevel::Transaction(ntx);
-                uuids.push(uuid);
+            } else {
+                Err(anyhow::anyhow!(
+                    "rawscript is required for `rawscript` kind"
+                ))?
             }
-            match tx {
-                PushIsolationLevel::Transaction(tx) => {
-                    tx.commit().await?;
-                }
-                _ => (),
-            }
-            return Ok(Json(uuids));
+        }
+        "flow" => {
+            let (mut value, job_kind, path) = if let Some(value) = batch_info.flow_value {
+                (value, JobKind::FlowPreview, None)
+            } else if let Some(path) = batch_info.path {
+                let value_json = sqlx::query!(
+                    "SELECT flow_version.value AS \"value: sqlx::types::Json<Box<RawValue>>\" FROM flow 
+                    LEFT JOIN flow_version
+                        ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
+                    WHERE flow.path = $1 AND flow.workspace_id = $2",
+                    &path, &w_id
+                )
+                .fetch_optional(&db)
+                .await?
+                .ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", path)))?;
+                let value =
+                    serde_json::from_str::<FlowValue>(value_json.value.get()).map_err(|err| {
+                        Error::InternalErr(format!(
+                            "could not convert json to flow for {path}: {err:?}"
+                        ))
+                    })?;
+                (value, JobKind::Flow, Some(path))
+            } else {
+                Err(anyhow::anyhow!(
+                    "Path is required if no value is not provided"
+                ))?
+            };
+            add_virtual_items_if_necessary(&mut value.modules);
+            let flow_status = FlowStatus::new(&value);
+            (
+                None,                             // script_hash
+                path,                             // script_path
+                job_kind,                         // job_kind
+                None,                             // language
+                None,                             // dedicated_worker
+                value.concurrency_key.clone(),    // custom_concurrency_key
+                value.concurrent_limit.clone(),   // concurrent_limit
+                value.concurrency_time_window_s,  // concurrency_time_window_s
+                None,                             // timeout
+                None,                             // raw_code
+                None,                             // raw_lock
+                Some(value),                      // raw_flow
+                Some(flow_status),                // flow_status
+            )
         }
         "noop" => (
             None,
@@ -4402,6 +4487,10 @@ async fn add_batch_jobs(
             None,
             None,
             None,
+            None,
+            None,
+            None,
+            None,
         ),
         _ => {
             return Err(error::Error::BadRequest(format!(
@@ -4428,8 +4517,8 @@ async fn add_batch_jobs(
             select gen_random_uuid() as uuid from generate_series(1, $11)
         )
         INSERT INTO queue 
-            (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout)
-            (SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14 FROM uuid_table) 
+            (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout, raw_code, raw_lock, raw_flow, flow_status)
+            (SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14, $15, $16, $17, $18 FROM uuid_table) 
         RETURNING id"#,
             hash.map(|h| h.0),
             path,
@@ -4444,7 +4533,11 @@ async fn add_batch_jobs(
             n,
             concurrent_limit,
             concurrent_time_window_s,
-            timeout
+            timeout,
+            raw_code,
+            raw_lock,
+            raw_flow.map(sqlx::types::Json) as Option<sqlx::types::Json<FlowValue>>,
+            flow_status.map(sqlx::types::Json) as Option<sqlx::types::Json<FlowStatus>>
         )
         .fetch_all(&db)
         .await?;
@@ -5023,24 +5116,10 @@ async fn get_completed_job<'a>(
     Extension(db): Extension<DB>,
     Path((w_id, id)): Path<(String, Uuid)>,
 ) -> error::Result<Response> {
-    let job_o = sqlx::query_as::<_, CompletedJob>("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, 
-    CASE WHEN args is null or pg_column_size(args) < 90000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 90000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind,
-    schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped,
-    raw_lock, email, visible_to_owner, mem_peak, tag, priority, result->'wm_labels' as labels FROM completed_job WHERE id = $1 AND workspace_id = $2")
-        .bind(id)
-        .bind(&w_id)
-        .fetch_optional(&db)
+    let job_o = get_completed_job_ex(&db, &w_id, id, false, Some(&opt_authed))
         .await?;
 
     let cj = not_found_if_none(job_o, "Completed Job", id.to_string())?;
-
-    if opt_authed.is_none() && cj.created_by != "anonymous" {
-        return Err(Error::BadRequest(
-            "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
-        ));
-    }
-    let cj = format_completed_job_result(cj);
-
     let response = Json(cj).into_response();
     // let extra_log = query_scalar!(
     //     "SELECT substr(logs, $1) as logs FROM large_logs WHERE workspace_id = $2 AND job_id = $3",

backend/windmill-api/src/lib.rs

@@ -25,7 +25,6 @@ use argon2::Argon2;
 use axum::extract::DefaultBodyLimit;
 use axum::{middleware::from_extractor, routing::get, Extension, Router};
 use db::DB;
-use git_version::git_version;
 use http::HeaderValue;
 use reqwest::Client;
 use std::collections::HashMap;
@@ -40,7 +39,7 @@ use tower_http::{
 };
 use windmill_common::db::UserDB;
 use windmill_common::worker::{ALL_TAGS, CLOUD_HOSTED};
-use windmill_common::{BASE_URL, INSTANCE_NAME};
+use windmill_common::{BASE_URL, INSTANCE_NAME, utils::GIT_VERSION};
 
 use crate::scim_ee::has_scim_token;
 use windmill_common::error::AppError;
@@ -63,13 +62,14 @@ mod http_triggers;
 mod indexer_ee;
 mod inputs;
 mod integration;
+mod ai;
+
 #[cfg(feature = "parquet")]
 mod job_helpers_ee;
 pub mod job_metrics;
 pub mod jobs;
 pub mod oauth2_ee;
 mod oidc_ee;
-mod openai;
 mod raw_apps;
 mod resources;
 mod saml_ee;
@@ -93,11 +93,9 @@ mod workers;
 mod workspaces;
 mod workspaces_ee;
 
-pub const GIT_VERSION: &str =
-    git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
-
 pub const DEFAULT_BODY_LIMIT: usize = 2097152 * 100; // 200MB
 
+
 lazy_static::lazy_static! {
 
     pub static ref REQUEST_SIZE_LIMIT: Arc<RwLock<usize>> = Arc::new(RwLock::new(DEFAULT_BODY_LIMIT));
@@ -154,18 +152,18 @@ pub async fn add_webhook_allowed_origin(
 type IndexReader = ();
 
 #[cfg(not(feature = "tantivy"))]
-type IndexWriter = ();
+type ServiceLogIndexReader = ();
 
 #[cfg(feature = "tantivy")]
-type IndexReader = windmill_indexer::indexer_ee::IndexReader;
+type IndexReader = windmill_indexer::completed_runs_ee::IndexReader;
 #[cfg(feature = "tantivy")]
-type IndexWriter = windmill_indexer::indexer_ee::IndexWriter;
+type ServiceLogIndexReader = windmill_indexer::service_logs_ee::ServiceLogIndexReader;
 
 pub async fn run_server(
     db: DB,
     rsmq: Option<rsmq_async::MultiplexedRsmq>,
-    index_reader: Option<IndexReader>,
-    index_writer: Option<IndexWriter>,
+    job_index_reader: Option<IndexReader>,
+    log_index_reader: Option<ServiceLogIndexReader>,
     addr: SocketAddr,
     mut rx: tokio::sync::broadcast::Receiver<()>,
     port_tx: tokio::sync::oneshot::Sender<String>,
@@ -205,8 +203,9 @@ pub async fn run_server(
         .layer(Extension(rsmq.clone()))
         .layer(Extension(user_db.clone()))
         .layer(Extension(auth_cache.clone()))
-        .layer(Extension(index_reader))
-        .layer(Extension(index_writer))
+        .layer(Extension(job_index_reader))
+        .layer(Extension(log_index_reader))
+        // .layer(Extension(index_writer))
         .layer(CookieManagerLayer::new())
         .layer(Extension(WebhookShared::new(rx.resubscribe(), db.clone())))
         .layer(DefaultBodyLimit::max(
@@ -281,7 +280,7 @@ pub async fn run_server(
                         .nest("/job_helpers", job_helpers_service)
                         .nest("/jobs", jobs::workspaced_service())
                         .nest("/oauth", oauth2_ee::workspaced_service())
-                        .nest("/openai", openai::workspaced_service())
+                        .nest("/ai", ai::workspaced_service())
                         .nest("/raw_apps", raw_apps::workspaced_service())
                         .nest("/resources", resources::workspaced_service())
                         .nest("/schedules", schedule::workspaced_service())
@@ -322,6 +321,10 @@ pub async fn run_server(
                     "/srch/w/:workspace_id/index",
                     indexer_ee::workspaced_service(),
                 )
+                .nest(
+                    "/srch/index",
+                    indexer_ee::global_service(),
+                )
                 .nest("/oidc", oidc_ee::global_service())
                 .nest(
                     "/saml",

backend/windmill-api/src/openai.rs

@@ -1,344 +0,0 @@
-use std::collections::HashMap;
-
-use crate::db::{ApiAuthed, DB};
-
-use axum::{
-    body::Bytes,
-    extract::{Extension, Path, Query},
-    response::IntoResponse,
-    routing::post,
-    Router,
-};
-use quick_cache::sync::Cache;
-use reqwest::Client;
-use serde_json::value::RawValue;
-use windmill_audit::audit_ee::audit_log;
-use windmill_audit::ActionKind;
-use windmill_common::{
-    error::{to_anyhow, Error},
-    variables::build_crypt,
-};
-
-use crate::variables::decrypt;
-use serde::Deserialize;
-
-lazy_static::lazy_static! {
-    static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new()
-        .timeout(std::time::Duration::from_secs(60 * 5))
-        .user_agent("windmill/beta")
-        .build().unwrap();
-}
-
-pub fn workspaced_service() -> Router {
-    let router = Router::new().route("/proxy/*openai_path", post(proxy));
-
-    router
-}
-
-#[derive(Deserialize)]
-struct OpenaiResource {
-    api_key: String,
-    organization_id: Option<String>,
-}
-
-#[derive(Deserialize)]
-struct OpenaiClientCredentialsOauthResource {
-    client_id: String,
-    client_secret: String,
-    token_url: String,
-    user: Option<String>,
-}
-
-#[derive(Deserialize)]
-#[serde(untagged)]
-enum OpenaiConfig {
-    Resource(OpenaiResource),
-    ClientCredentialsOauthResource(OpenaiClientCredentialsOauthResource),
-}
-
-struct Variable {
-    value: String,
-    is_secret: bool,
-}
-async fn get_variable_or_self(path: String, db: &DB, w_id: &String) -> Result<String, Error> {
-    if !path.starts_with("$var:") {
-        return Ok(path);
-    }
-    let path = path.strip_prefix("$var:").unwrap().to_string();
-    let mut variable = sqlx::query_as!(
-        Variable,
-        "SELECT value, is_secret
-        FROM variable
-        WHERE path = $1 AND workspace_id = $2",
-        &path,
-        &w_id
-    )
-    .fetch_one(db)
-    .await?;
-    if variable.is_secret {
-        let mc = build_crypt(&db, &w_id).await?;
-        variable.value = decrypt(&mc, variable.value)?;
-    }
-    Ok(variable.value)
-}
-
-lazy_static::lazy_static! {
-    pub static ref OPENAI_AZURE_BASE_PATH: Option<String> = std::env::var("OPENAI_AZURE_BASE_PATH").ok();
-    static ref OPENAI_KEY_CACHE: Cache<String, OpenaiKeyCache> = Cache::new(500);
-}
-
-#[derive(Deserialize)]
-struct OpenaiCredentials {
-    access_token: String,
-}
-async fn get_openai_key_using_credentials_flow(
-    mut resource: OpenaiClientCredentialsOauthResource,
-    db: &DB,
-    w_id: &String,
-) -> Result<String, Error> {
-    resource.client_id = get_variable_or_self(resource.client_id, &db, &w_id).await?;
-    resource.client_secret = get_variable_or_self(resource.client_secret, &db, &w_id).await?;
-    resource.token_url = get_variable_or_self(resource.token_url, &db, &w_id).await?;
-    let mut params = HashMap::new();
-    params.insert("grant_type", "client_credentials");
-    let response = HTTP_CLIENT
-        .post(resource.token_url)
-        .form(&params)
-        .basic_auth(resource.client_id, Some(resource.client_secret))
-        .send()
-        .await
-        .map_err(|err| {
-            Error::InternalErr(format!(
-                "Failed to get OpenAI credentials using credentials flow: {}",
-                err
-            ))
-        })?;
-    let response = response.json::<OpenaiCredentials>().await.map_err(|err| {
-        Error::InternalErr(format!(
-            "Failed to parse OpenAI credentials from credentials flow: {}",
-            err
-        ))
-    })?;
-    Ok(response.access_token)
-}
-
-#[derive(Clone)]
-struct OpenaiKeyCache {
-    api_key: String,
-    organization_id: Option<String>,
-    azure_base_path: Option<String>,
-    user: Option<String>,
-    expires_at: std::time::Instant,
-}
-
-impl OpenaiKeyCache {
-    fn new(
-        api_key: String,
-        organization_id: Option<String>,
-        azure_base_path: Option<String>,
-        expires_at: std::time::Instant,
-        user: Option<String>,
-    ) -> Self {
-        Self { api_key, organization_id, azure_base_path, expires_at, user }
-    }
-    fn is_expired(&self) -> bool {
-        self.expires_at < std::time::Instant::now()
-    }
-}
-
-#[derive(Deserialize)]
-struct ProxyQueryParams {
-    no_cache: Option<bool>,
-}
-async fn proxy(
-    authed: ApiAuthed,
-    Extension(db): Extension<DB>,
-    Path((w_id, openai_path)): Path<(String, String)>,
-    Query(query_params): Query<ProxyQueryParams>,
-    mut body: Bytes,
-) -> impl IntoResponse {
-    let workspace_cache = OPENAI_KEY_CACHE.get(&w_id);
-    let (api_key, organization_id, azure_base_path, user) = if query_params
-        .no_cache
-        .unwrap_or(false)
-        || workspace_cache.is_none()
-        || workspace_cache.clone().unwrap().is_expired()
-    {
-        let openai_resource_path = sqlx::query_scalar!(
-            "SELECT openai_resource_path FROM workspace_settings WHERE workspace_id = $1",
-            &w_id
-        )
-        .fetch_one(&db)
-        .await?;
-
-        if openai_resource_path.is_none() {
-            return Err(Error::InternalErr(
-                "OpenAI resource not configured".to_string(),
-            ));
-        }
-
-        let openai_resource_path = openai_resource_path.unwrap();
-
-        let resource = sqlx::query_scalar!(
-            "SELECT value
-            FROM resource
-            WHERE path = $1 AND workspace_id = $2",
-            &openai_resource_path,
-            &w_id
-        )
-        .fetch_optional(&db)
-        .await?
-        .ok_or_else(|| {
-            Error::InternalErr(format!(
-                "Could not find the OpenAI resource at path {openai_resource_path}, update the resource path in the workspace settings"
-            ))
-        })?;
-
-        if resource.is_none() {
-            return Err(Error::InternalErr(
-                "OpenAI resource missing value".to_string(),
-            ));
-        }
-
-        let config: OpenaiConfig = serde_json::from_value(resource.unwrap())
-            .map_err(|e| Error::InternalErr(format!("validating openai resource {e:#}")))?;
-
-        let mut user = None::<String>;
-        let mut resource = match config {
-            OpenaiConfig::Resource(resource) => {
-                tracing::debug!("Getting OpenAI key from static resource");
-                resource
-            }
-            OpenaiConfig::ClientCredentialsOauthResource(resource) => {
-                tracing::debug!("Getting OpenAI key with client credentials flow");
-                user = resource.user.clone();
-                let token = get_openai_key_using_credentials_flow(resource, &db, &w_id).await?;
-                OpenaiResource { api_key: token, organization_id: None }
-            }
-        };
-
-        resource.api_key = get_variable_or_self(resource.api_key, &db, &w_id).await?;
-
-        if resource.organization_id.is_some() {
-            resource.organization_id =
-                Some(get_variable_or_self(resource.organization_id.unwrap(), &db, &w_id).await?);
-        }
-
-        if user.is_some() {
-            user = Some(get_variable_or_self(user.unwrap(), &db, &w_id).await?);
-        }
-
-        let expires_at = std::time::Instant::now() + std::time::Duration::from_secs(60);
-
-        let azure_base_path = sqlx::query_scalar!(
-            "SELECT value
-            FROM global_settings
-            WHERE name = 'openai_azure_base_path'",
-        )
-        .fetch_optional(&db)
-        .await?;
-
-        let azure_base_path = if let Some(azure_base_path) = azure_base_path {
-            Some(
-                serde_json::from_value::<String>(azure_base_path).map_err(|e| {
-                    Error::InternalErr(format!("validating openai azure base path {e:#}"))
-                })?,
-            )
-        } else {
-            OPENAI_AZURE_BASE_PATH.clone()
-        };
-
-        let workspace_cache = OpenaiKeyCache::new(
-            resource.api_key.clone(),
-            resource.organization_id.clone(),
-            azure_base_path.clone(),
-            expires_at,
-            user.clone(),
-        );
-        OPENAI_KEY_CACHE.insert(w_id.clone(), workspace_cache);
-        (
-            resource.api_key,
-            resource.organization_id,
-            azure_base_path,
-            user,
-        )
-    } else {
-        tracing::debug!("Using cached OpenAI key");
-        let workspace_cache = workspace_cache.unwrap();
-        (
-            workspace_cache.api_key.clone(),
-            workspace_cache.organization_id.clone(),
-            workspace_cache.azure_base_path.clone(),
-            workspace_cache.user.clone(),
-        )
-    };
-
-    if user.is_some() {
-        tracing::debug!("Adding user to request body");
-        let mut json_body: HashMap<String, Box<RawValue>> = serde_json::from_slice(&body)
-            .map_err(|e| Error::InternalErr(format!("Failed to parse request body: {}", e)))?;
-
-        let user_json_string = serde_json::Value::String(user.unwrap()).to_string(); // makes sure to escape characters
-
-        json_body.insert(
-            "user".to_string(),
-            RawValue::from_string(user_json_string)
-                .map_err(|e| Error::InternalErr(format!("Failed to parse user: {}", e)))?,
-        );
-
-        body = serde_json::to_vec(&json_body)
-            .map_err(|e| Error::InternalErr(format!("Failed to reserialize request body: {}", e)))?
-            .into();
-    }
-
-    let base_url = if let Some(base_url) = azure_base_path {
-        base_url
-    } else {
-        "https://api.openai.com/v1".to_string()
-    };
-
-    let url = format!("{}/{}", base_url, openai_path);
-    let mut request = HTTP_CLIENT
-        .post(url)
-        .header("content-type", "application/json")
-        .body(body);
-
-    if base_url != "https://api.openai.com/v1" {
-        request = request
-            .header("api-key", api_key)
-            .query(&[("api-version", "2023-05-15")])
-    } else {
-        request = request.header("authorization", format!("Bearer {}", api_key))
-    }
-
-    if let Some(org_id) = organization_id {
-        request = request.header("OpenAI-Organization", org_id);
-    }
-
-    let response = request.send().await.map_err(to_anyhow)?;
-
-    let mut tx = db.begin().await?;
-    audit_log(
-        &mut *tx,
-        &authed,
-        "openai.request",
-        ActionKind::Execute,
-        &w_id,
-        Some(&authed.email),
-        Some([("openai_path", &format!("{:?}", openai_path)[..])].into()),
-    )
-    .await?;
-    tx.commit().await?;
-
-    if response.error_for_status_ref().is_err() {
-        return Err(Error::OpenAIError(
-            response.text().await.unwrap_or("".to_string()),
-        ));
-    }
-
-    let status_code = response.status();
-    let headers = response.headers().clone();
-    let stream = response.bytes_stream();
-
-    Ok((status_code, headers, axum::body::Body::from_stream(stream)))
-}

backend/windmill-api/src/resources.rs

@@ -300,7 +300,7 @@ async fn get_resource(
     .await?;
     tx.commit().await?;
     if resource_o.is_none() {
-        explain_resource_perm_error(&path, &w_id, &db).await?;
+        explain_resource_perm_error(&path, &w_id, &db, &authed).await?;
     }
     let resource = not_found_if_none(resource_o, "Resource", path)?;
     Ok(Json(resource))
@@ -343,7 +343,7 @@ async fn get_resource_value(
 
     tx.commit().await?;
     if value_o.is_none() {
-        explain_resource_perm_error(&path, &w_id, &db).await?;
+        explain_resource_perm_error(&path, &w_id, &db, &authed).await?;
     }
 
     let value = not_found_if_none(value_o, "Resource", path)?;
@@ -354,6 +354,7 @@ async fn explain_resource_perm_error(
     path: &str,
     w_id: &str,
     db: &sqlx::Pool<Postgres>,
+    authed: &ApiAuthed,
 ) -> windmill_common::error::Result<()> {
     let extra_perms = sqlx::query_scalar!(
         "SELECT extra_perms from resource WHERE path = $1 AND workspace_id = $2",
@@ -378,12 +379,12 @@ async fn explain_resource_perm_error(
         .fetch_optional(db)
         .await?;
         return Err(Error::NotAuthorized(format!(
-            "Resource exists but you don't have access to it:\nresource perms: {}\nfolder perms: {}",
+            "Resource exists but you don't have access to it:\nresource perms: {}\nfolder perms: {}\nauthed as: {authed:?}",
             serde_json::to_string_pretty(&extra_perms).unwrap_or_default(), serde_json::to_string_pretty(&folder_extra_perms).unwrap_or_default()
         )));
     } else {
         return Err(Error::NotAuthorized(format!(
-            "Resource exists but you don't have access to it:\nresource perms: {}",
+            "Resource exists but you don't have access to it:\nresource perms: {}\nauthed as: {authed:?}",
             serde_json::to_string_pretty(&extra_perms).unwrap_or_default()
         )));
     }
@@ -457,7 +458,7 @@ pub async fn get_resource_value_interpolated_internal(
     .await?;
     tx.commit().await?;
     if value_o.is_none() {
-        explain_resource_perm_error(path, workspace, db).await?;
+        explain_resource_perm_error(path, workspace, db, &authed).await?;
     }
 
     let value = not_found_if_none(value_o, "Resource", path)?;

backend/windmill-api/src/settings.rs

@@ -58,7 +58,10 @@ pub fn global_service() -> Router {
         )
         .route("/renew_license_key", post(renew_license_key))
         .route("/customer_portal", post(create_customer_portal_session))
-        .route("/test_critical_channels", post(test_critical_channels));
+        .route("/test_critical_channels", post(test_critical_channels))
+        .route("/critical_alerts", get(get_critical_alerts))
+        .route("/critical_alerts/:id/acknowledge", post(acknowledge_critical_alert))
+        .route("/critical_alerts/acknowledge_all", post(acknowledge_all_critical_alerts));
 
     #[cfg(feature = "parquet")]
     {
@@ -430,3 +433,50 @@ pub async fn test_critical_channels(
 pub async fn test_critical_channels() -> Result<String> {
     Ok("Critical channels require EE".to_string())
 }
+
+#[cfg(feature = "enterprise")]
+pub async fn get_critical_alerts(
+    Extension(db): Extension<DB>,
+    authed: ApiAuthed,
+    Query(params): Query<crate::utils::AlertQueryParams>,
+) -> JsonResult<Vec<crate::utils::CriticalAlert>> {
+    require_super_admin(&db, &authed.email).await?;
+
+    crate::utils::get_critical_alerts(db, params, None).await
+}
+
+#[cfg(not(feature = "enterprise"))]
+pub async fn get_critical_alerts() -> error::Error {
+    error::Error::NotFound("Critical Alerts require EE".to_string())
+}
+
+#[cfg(feature = "enterprise")]
+pub async fn acknowledge_critical_alert(
+    Extension(db): Extension<DB>,
+    authed: ApiAuthed,
+    Path(id): Path<i32>,
+) -> error::Result<String> {
+    require_super_admin(&db, &authed.email).await?;
+    
+    crate::utils::acknowledge_critical_alert(db, None, id).await
+}
+
+#[cfg(not(feature = "enterprise"))]
+pub async fn acknowledge_critical_alert() -> error::Error {
+    error::Error::NotFound("Critical Alerts require EE".to_string())
+}
+
+#[cfg(feature = "enterprise")]
+pub async fn acknowledge_all_critical_alerts(
+    Extension(db): Extension<DB>,
+    authed: ApiAuthed,
+) -> error::Result<String> {
+    require_super_admin(&db, &authed.email).await?;
+
+    crate::utils::acknowledge_all_critical_alerts(db, None).await
+}
+
+#[cfg(not(feature = "enterprise"))]
+pub async fn acknowledge_all_critical_alerts() -> error::Error {
+    error::Error::NotFound("Critical Alerts require EE".to_string())
+}

backend/windmill-api/src/static_assets.rs

@@ -23,7 +23,7 @@ pub async fn static_handler(OriginalUri(original_uri): OriginalUri) -> StaticFil
 }
 
 #[derive(RustEmbed)]
-#[folder = "../../frontend/build/"]
+#[folder = "${FRONTEND_BUILD_DIR:-../../frontend/build/}"]
 struct Asset;
 pub struct StaticFile(Uri);
 

backend/windmill-api/src/users.rs

@@ -90,6 +90,9 @@ pub fn global_service() -> Router {
         .route("/accept_invite", post(accept_invite))
         .route("/list_as_super_admin", get(list_users_as_super_admin))
         .route("/setpassword", post(set_password))
+        .route("/set_password_of/:user", post(set_password_of_user))
+        .route("/set_login_type/:user", post(set_login_type))
+
         .route("/create", post(create_user))
         .route("/update/:user", post(update_user))
         .route("/delete/:user", delete(delete_user))
@@ -129,7 +132,8 @@ fn username_override_from_label(label: Option<String>) -> Option<String> {
         Some(label)
             if label.starts_with("webhook-")
                 || label.starts_with("http-")
-                || label.starts_with("email-") =>
+                || label.starts_with("email-")
+                || label.starts_with("ws-") =>
         {
             Some(label)
         }
@@ -745,10 +749,20 @@ pub async fn fetch_api_authed(
     username_override: String,
 ) -> error::Result<ApiAuthed> {
     let permissioned_as = username_to_permissioned_as(username.as_str());
+    fetch_api_authed_from_permissioned_as(permissioned_as, email, w_id, db, username_override).await
+}
+
+pub async fn fetch_api_authed_from_permissioned_as(
+    permissioned_as: String,
+    email: String,
+    w_id: &str,
+    db: &DB,
+    username_override: String,
+) -> error::Result<ApiAuthed> {
     let authed =
         fetch_authed_from_permissioned_as(permissioned_as, email.clone(), w_id, db).await?;
     Ok(ApiAuthed {
-        username: username,
+        username: authed.username,
         email: email,
         is_admin: authed.is_admin,
         is_operator: authed.is_operator,
@@ -855,6 +869,12 @@ pub struct EditPassword {
     pub password: String,
 }
 
+#[derive(Deserialize)]
+pub struct EditLoginType {
+    pub login_type: String,
+}
+
+
 #[derive(FromRow, Serialize)]
 pub struct TruncatedToken {
     pub label: Option<String>,
@@ -2017,7 +2037,52 @@ async fn set_password(
     authed: ApiAuthed,
     Json(ep): Json<EditPassword>,
 ) -> Result<String> {
-    crate::users_ee::set_password(db, argon2, authed, ep).await
+    let email = authed.email.clone();
+    crate::users_ee::set_password(db, argon2, authed, &email, ep).await
+}
+
+async fn set_password_of_user(
+    Extension(db): Extension<DB>,
+    Extension(argon2): Extension<Arc<Argon2<'_>>>,
+    Path(email): Path<String>,
+    authed: ApiAuthed,
+    Json(ep): Json<EditPassword>,
+) -> Result<String> {
+    require_super_admin(&db, &authed.email).await?;
+    crate::users_ee::set_password(db, argon2, authed, &email, ep).await
+}
+
+async fn set_login_type(
+    Extension(db): Extension<DB>,
+    Path(email): Path<String>,
+    authed: ApiAuthed,
+    Json(et): Json<EditLoginType>,
+) -> Result<String> {
+    require_super_admin(&db, &authed.email).await?;
+    let mut tx = db.begin().await?;
+
+    sqlx::query!(
+        "UPDATE password SET login_type = $1 WHERE email = $2",
+        et.login_type,
+        email
+    )
+    .execute(&mut *tx)
+    .await?;
+
+    audit_log(
+        &mut *tx,
+        &authed,
+        "users.set_login_type",
+        ActionKind::Update,
+        "global",
+        Some(&email),
+        None,
+    )
+    .await?;
+
+    tx.commit().await?;
+    Ok(format!("login type of {} updated to {}", email, et.login_type))
+    
 }
 
 async fn login(

backend/windmill-api/src/users_ee.rs

@@ -27,6 +27,7 @@ pub async fn set_password(
     _db: DB,
     _argon2: Arc<Argon2<'_>>,
     _authed: ApiAuthed,
+    _user_email: &str,
     _ep: EditPassword,
 ) -> Result<String> {
     Err(Error::InternalErr(

backend/windmill-api/src/utils.rs

@@ -16,6 +16,12 @@ use windmill_common::{
     DB,
 };
 
+#[cfg(feature = "enterprise")]
+use windmill_common::error::JsonResult;
+
+#[cfg(feature = "enterprise")]
+use axum::Json;
+
 #[derive(Deserialize)]
 pub struct WithStarredInfoQuery {
     pub with_starred_info: Option<bool>,
@@ -162,3 +168,159 @@ pub fn content_plain(body: Body) -> Response {
         .body(body)
         .unwrap()
 }
+
+use serde::Serialize;
+
+#[derive(Serialize)]
+pub struct CriticalAlert {
+    id: i32,
+    alert_type: String,
+    message: String,
+    created_at: chrono::DateTime<chrono::Utc>,
+    acknowledged: Option<bool>,
+    workspace_id: Option<String>,
+}
+
+#[cfg(feature = "enterprise")]
+#[derive(Deserialize, Debug)]
+pub struct AlertQueryParams {
+    pub page: Option<i32>,
+    pub page_size: Option<i32>,
+    pub acknowledged: Option<bool>,
+}
+
+#[cfg(feature = "enterprise")]
+pub async fn get_critical_alerts(
+    db: DB,
+    params: AlertQueryParams,
+    workspace_id: Option<String>,
+) -> JsonResult<Vec<CriticalAlert>> {
+    let page = params.page.unwrap_or(1).max(1);
+    let page_size = params.page_size.unwrap_or(10).min(100) as i64;
+    let offset = ((page - 1) * page_size as i32) as i64;
+
+    let alerts = if let Some(workspace_id) = workspace_id {
+        // `workspace_id` is provided => workspace admin
+        if params.acknowledged.is_none() {
+            // Case: return all rows where `workspace_id` matches
+            sqlx::query_as!(
+                CriticalAlert,
+                "SELECT id, alert_type, message, created_at, COALESCE(acknowledged_workspace, false) AS acknowledged, workspace_id
+                 FROM alerts
+                 WHERE workspace_id = $1
+                 ORDER BY created_at DESC
+                 LIMIT $2 OFFSET $3",
+                workspace_id,
+                page_size,
+                offset
+            )
+            .fetch_all(&db)
+            .await?
+        } else {
+            // Case: return rows where `acknowledged_workspace` matches `params.acknowledged`
+            sqlx::query_as!(
+                CriticalAlert,
+                "SELECT id, alert_type, message, created_at, COALESCE(acknowledged_workspace, false) AS acknowledged, workspace_id
+                 FROM alerts
+                 WHERE workspace_id = $1 AND COALESCE(acknowledged_workspace, false) = $2
+                 ORDER BY created_at DESC
+                 LIMIT $3 OFFSET $4",
+                workspace_id,
+                params.acknowledged,
+                page_size,
+                offset
+            )
+            .fetch_all(&db)
+            .await?
+        }
+    } else {
+        // `workspace_id` is not provided => superadmin
+        if params.acknowledged.is_none() {
+            // Case: Return all rows unfiltered with global acknowledged as acknowledged
+            sqlx::query_as!(
+                CriticalAlert,
+                "SELECT id, alert_type, message, created_at, COALESCE(acknowledged, false) AS acknowledged, workspace_id
+                 FROM alerts
+                 ORDER BY created_at DESC
+                 LIMIT $1 OFFSET $2",
+                page_size,
+                offset
+            )
+            .fetch_all(&db)
+            .await?
+        } else {
+            // Case: Return rows where global acknowledged matches params.acknowledged
+            sqlx::query_as!(
+                CriticalAlert,
+                "SELECT id, alert_type, message, created_at, COALESCE(acknowledged, false) AS acknowledged, workspace_id
+                 FROM alerts
+                 WHERE COALESCE(acknowledged, false) = $1
+                 ORDER BY created_at DESC
+                 LIMIT $2 OFFSET $3",
+                params.acknowledged,
+                page_size,
+                offset
+            )
+            .fetch_all(&db)
+            .await?
+        }
+    };
+
+    Ok(Json(alerts))
+}
+
+#[cfg(feature = "enterprise")]
+pub async fn acknowledge_critical_alert(
+    db: DB,
+    workspace_id: Option<String>,
+    id: i32,
+) -> error::Result<String> {
+    sqlx::query!(
+        "UPDATE alerts
+         SET
+           acknowledged = true,
+           acknowledged_workspace = CASE
+             WHEN $2::text IS NOT NULL AND workspace_id = $2 THEN true
+             ELSE acknowledged_workspace
+           END
+         WHERE id = $1",
+        id,
+        workspace_id
+    )
+    .execute(&db)
+    .await?;
+
+    tracing::info!(
+        "Acknowledged critical alert with id: {}{}",
+        id,
+        workspace_id.map_or_else(|| "".to_string(), |w| format!(" for workspace_id: {}", w))
+    );
+    Ok("Critical alert acknowledged".to_string())
+}
+
+#[cfg(feature = "enterprise")]
+pub async fn acknowledge_all_critical_alerts(
+    db: DB,
+    workspace_id: Option<String>,
+) -> error::Result<String> {
+    sqlx::query!(
+        "UPDATE alerts 
+         SET
+           acknowledged = true,
+           acknowledged_workspace = CASE
+             WHEN $1::text IS NOT NULL THEN true
+             ELSE acknowledged_workspace
+           END
+         WHERE ($1::text IS NOT NULL AND workspace_id = $1)
+            OR ($1::text IS NULL)",
+        workspace_id
+    )
+    .execute(&db)
+    .await?;
+
+    tracing::info!(
+        "Acknowledged all unacknowledged critical alerts{}",
+        workspace_id.map_or_else(|| "".to_string(), |w| format!(" for workspace_id: {}", w))
+    );
+    Ok("All unacknowledged critical alerts acknowledged".to_string())
+}

backend/windmill-api/src/websocket_triggers.rs

@@ -1,9 +1,10 @@
+use anyhow::Context;
 use axum::{
     extract::{Path, Query},
     routing::{delete, get, post},
     Extension, Json, Router,
 };
-use futures::StreamExt;
+use futures::{stream::SplitSink, SinkExt, StreamExt};
 use http::StatusCode;
 use itertools::Itertools;
 use rand::seq::SliceRandom;
@@ -11,16 +12,20 @@ use serde::{
     de::{self, MapAccess, Visitor},
     Deserialize, Deserializer, Serialize,
 };
-use serde_json::Value;
+use serde_json::{value::RawValue, Value};
 use sql_builder::{bind::Bind, SqlBuilder};
 use sqlx::prelude::FromRow;
 use std::{collections::HashMap, fmt};
-use tokio_tungstenite::connect_async;
+use tokio::net::TcpStream;
+use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
+use uuid::Uuid;
 use windmill_audit::{audit_ee::audit_log, ActionKind};
 use windmill_common::{
     db::UserDB,
-    error::{self, JsonResult},
-    utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
+    error::{self, to_anyhow, JsonResult},
+    utils::{
+        not_found_if_none, paginate, report_critical_error, require_admin, Pagination, StripPath,
+    },
     worker::{to_raw_value, CLOUD_HOSTED},
     INSTANCE_NAME,
 };
@@ -28,9 +33,7 @@ use windmill_queue::PushArgsOwned;
 
 use crate::{
     db::{ApiAuthed, DB},
-    jobs::{
-        run_wait_result_flow_by_path_internal, run_wait_result_script_by_path_internal, RunJobQuery,
-    },
+    jobs::{run_flow_by_path_inner, run_script_by_path_inner, RunJobQuery},
     users::fetch_api_authed,
 };
 
@@ -52,7 +55,29 @@ struct NewWebsocketTrigger {
     script_path: String,
     is_flow: bool,
     enabled: Option<bool>,
-    filters: Vec<serde_json::Value>,
+    filters: Vec<Box<RawValue>>,
+    initial_messages: Vec<Box<RawValue>>,
+    url_runnable_args: Box<RawValue>,
+}
+
+#[derive(Deserialize)]
+struct JsonFilter {
+    key: String,
+    value: serde_json::Value,
+}
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum Filter {
+    JsonFilter(JsonFilter),
+}
+
+#[derive(Deserialize)]
+enum InitialMessage {
+    #[serde(rename = "raw_message")]
+    RawMessage(String),
+    #[serde(rename = "runnable_result")]
+    RunnableResult { path: String, args: Box<RawValue>, is_flow: bool },
 }
 
 #[derive(FromRow, Serialize, Clone)]
@@ -70,7 +95,9 @@ pub struct WebsocketTrigger {
     extra_perms: serde_json::Value,
     error: Option<String>,
     enabled: bool,
-    filters: Vec<serde_json::Value>,
+    filters: Vec<sqlx::types::Json<Box<RawValue>>>,
+    initial_messages: Vec<sqlx::types::Json<Box<RawValue>>>,
+    url_runnable_args: sqlx::types::Json<Box<RawValue>>,
 }
 
 #[derive(Deserialize)]
@@ -79,7 +106,9 @@ struct EditWebsocketTrigger {
     url: String,
     script_path: String,
     is_flow: bool,
-    filters: Vec<serde_json::Value>,
+    filters: Vec<Box<RawValue>>,
+    initial_messages: Vec<Box<RawValue>>,
+    url_runnable_args: Box<RawValue>,
 }
 
 #[derive(Deserialize)]
@@ -133,14 +162,13 @@ async fn get_websocket_trigger(
 ) -> error::JsonResult<WebsocketTrigger> {
     let mut tx = user_db.begin(&authed).await?;
     let path = path.to_path();
-    let trigger = sqlx::query_as!(
-        WebsocketTrigger,
+    let trigger = sqlx::query_as::<_, WebsocketTrigger>(
         r#"SELECT *
           FROM websocket_trigger
           WHERE workspace_id = $1 AND path = $2"#,
-        w_id,
-        path,
     )
+    .bind(w_id)
+    .bind(path)
     .fetch_optional(&mut *tx)
     .await?;
     tx.commit().await?;
@@ -164,19 +192,27 @@ async fn create_websocket_trigger(
     require_admin(authed.is_admin, &authed.username)?;
 
     let mut tx = user_db.begin(&authed).await?;
-    sqlx::query_as!(
-        WebsocketTrigger,
-      "INSERT INTO websocket_trigger (workspace_id, path, url, script_path, is_flow, enabled, filters, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now()) RETURNING *",
-      w_id,
-      ct.path,
-      ct.url,
-      ct.script_path,
-      ct.is_flow,
-      ct.enabled.unwrap_or(true),
-      &ct.filters,
-      &authed.username,
-      &authed.email
+
+    let filters = ct.filters.into_iter().map(sqlx::types::Json).collect_vec();
+    let initial_messages = ct
+        .initial_messages
+        .into_iter()
+        .map(sqlx::types::Json)
+        .collect_vec();
+    sqlx::query_as::<_, WebsocketTrigger>(
+      "INSERT INTO websocket_trigger (workspace_id, path, url, script_path, is_flow, enabled, filters, initial_messages, url_runnable_args, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now()) RETURNING *",
     )
+    .bind(&w_id)
+    .bind(&ct.path)
+    .bind(ct.url)
+    .bind(ct.script_path)
+    .bind(ct.is_flow)
+    .bind(ct.enabled.unwrap_or(true))
+    .bind(filters.as_slice())
+    .bind(initial_messages.as_slice())
+    .bind(sqlx::types::Json(ct.url_runnable_args))
+    .bind(&authed.username)
+    .bind(&authed.email)
     .fetch_one(&mut *tx).await?;
 
     audit_log(
@@ -204,15 +240,24 @@ async fn update_websocket_trigger(
     let path = path.to_path();
     let mut tx = user_db.begin(&authed).await?;
 
+    let filters = ct.filters.into_iter().map(sqlx::types::Json).collect_vec();
+    let initial_messages = ct
+        .initial_messages
+        .into_iter()
+        .map(sqlx::types::Json)
+        .collect_vec();
+
     // important to update server_id, last_server_ping and error to NULL to stop current websocket listener
     sqlx::query!(
-        "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, edited_by = $6, email = $7, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL
-            WHERE workspace_id = $8 AND path = $9",
+        "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, initial_messages = $6, url_runnable_args = $7, edited_by = $8, email = $9, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL
+            WHERE workspace_id = $10 AND path = $11",
         ct.url,
         ct.script_path,
         ct.path,
         ct.is_flow,
-        &ct.filters,
+        filters.as_slice() as &[sqlx::types::Json<Box<RawValue>>],
+        initial_messages.as_slice() as &[sqlx::types::Json<Box<RawValue>>],
+        sqlx::types::Json(ct.url_runnable_args) as sqlx::types::Json<Box<RawValue>>,
         &authed.username,
         &authed.email,
         w_id,
@@ -335,8 +380,7 @@ async fn listen_to_unlistened_websockets(
     rsmq: &Option<rsmq_async::MultiplexedRsmq>,
     killpill_rx: &tokio::sync::broadcast::Receiver<()>,
 ) -> () {
-    match sqlx::query_as!(
-        WebsocketTrigger,
+    match sqlx::query_as::<_, WebsocketTrigger>(
         r#"SELECT *
             FROM websocket_trigger
             WHERE enabled IS TRUE AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')"#
@@ -422,8 +466,6 @@ impl<'de, 'a> Visitor<'de> for SupersetVisitor<'a> {
             if key == self.key {
                 // Deserialize the value for the key and check if it's a superset
                 let json_value: Value = map.next_value()?;
-                tracing::info!("json_value: {:?}", json_value);
-                tracing::info!("value_to_check: {:?}", self.value_to_check);
                 return Ok(is_superset(&json_value, self.value_to_check));
             } else {
                 // Skip the value if it's not the one we're interested in
@@ -470,149 +512,431 @@ where
     deserializer.deserialize_map(SupersetVisitor { key, value_to_check })
 }
 
+async fn wait_runnable_result(
+    path: String,
+    is_flow: bool,
+    args: &Box<RawValue>,
+    ws_trigger: &WebsocketTrigger,
+    username_override: String,
+    db: &DB,
+    rsmq: Option<rsmq_async::MultiplexedRsmq>,
+) -> error::Result<String> {
+    let user_db = UserDB::new(db.clone());
+    let authed = fetch_api_authed(
+        ws_trigger.edited_by.clone(),
+        ws_trigger.email.clone(),
+        &ws_trigger.workspace_id,
+        &db,
+        username_override,
+    )
+    .await?;
+
+    let args = serde_json::from_str::<Option<HashMap<String, Box<RawValue>>>>(args.get())
+        .map_err(|e| error::Error::BadRequest(format!("invalid json: {}", e)))?
+        .unwrap_or_else(HashMap::new);
+
+    let label_prefix = Some(format!("ws-{}-", ws_trigger.path));
+    let (_, job_id) = if is_flow {
+        run_flow_by_path_inner(
+            authed,
+            db.clone(),
+            user_db,
+            rsmq.clone(),
+            ws_trigger.workspace_id.clone(),
+            StripPath(path.clone()),
+            RunJobQuery::default(),
+            PushArgsOwned { args, extra: None },
+            label_prefix,
+        )
+        .await?
+    } else {
+        run_script_by_path_inner(
+            authed,
+            db.clone(),
+            user_db,
+            rsmq.clone(),
+            ws_trigger.workspace_id.clone(),
+            StripPath(path.clone()),
+            RunJobQuery::default(),
+            PushArgsOwned { args, extra: None },
+            label_prefix,
+        )
+        .await?
+    };
+
+    let start_time = tokio::time::Instant::now();
+
+    loop {
+        if start_time.elapsed() > tokio::time::Duration::from_secs(300) {
+            return Err(anyhow::anyhow!(
+                "Timed out after 5m waiting for runnable {path} (is_flow: {is_flow}) to complete",
+            )
+            .into());
+        }
+
+        #[derive(sqlx::FromRow)]
+        struct RawResult {
+            result: Option<sqlx::types::Json<Box<RawValue>>>,
+            success: bool,
+        }
+
+        let result = sqlx::query_as::<_, RawResult>(
+            "SELECT result, success FROM completed_job WHERE id = $1 AND workspace_id = $2",
+        )
+        .bind(Uuid::parse_str(&job_id).unwrap())
+        .bind(&ws_trigger.workspace_id)
+        .fetch_optional(db)
+        .await;
+
+        match result {
+            Ok(Some(r)) => {
+                if !r.success {
+                    return Err(anyhow::anyhow!(
+                        "Runnable {path} (is_flow: {is_flow}) failed: {:?}",
+                        r.result
+                    )
+                    .into());
+                } else {
+                    return Ok(r.result.map(|r| r.get().to_owned()).unwrap_or_default());
+                }
+            }
+            Ok(None) => {
+                // not yet done, wait for 5s and check again
+                tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+            }
+            Err(err) => {
+                return Err(anyhow::anyhow!(
+                    "Error fetching job result for runnable {path} (is_flow: {is_flow}): {err}",
+                )
+                .into());
+            }
+        }
+    }
+}
+
+async fn send_initial_messages(
+    ws_trigger: &WebsocketTrigger,
+    mut writer: SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>,
+    db: &DB,
+    rsmq: Option<rsmq_async::MultiplexedRsmq>,
+) -> error::Result<()> {
+    let initial_messages: Vec<InitialMessage> = ws_trigger
+        .initial_messages
+        .iter()
+        .filter_map(|m| serde_json::from_str(m.get()).ok())
+        .collect_vec();
+
+    for start_message in initial_messages {
+        match start_message {
+            InitialMessage::RawMessage(msg) => {
+                let msg = if msg.starts_with("\"") && msg.ends_with("\"") {
+                    msg[1..msg.len() - 1].to_string()
+                } else {
+                    msg
+                };
+                tracing::info!(
+                    "Sending raw message initial message to websocket {}: {}",
+                    ws_trigger.url,
+                    msg
+                );
+                writer
+                    .send(tokio_tungstenite::tungstenite::Message::Text(msg))
+                    .await
+                    .map_err(to_anyhow)
+                    .with_context(|| "failed to send raw message")?;
+            }
+            InitialMessage::RunnableResult { path, is_flow, args } => {
+                tracing::info!(
+                    "Running runnable {path} (is_flow: {is_flow}) for initial message to websocket {}",
+                    ws_trigger.url,
+                );
+
+                let result = wait_runnable_result(
+                    path.clone(),
+                    is_flow,
+                    &args,
+                    ws_trigger,
+                    "init".to_string(),
+                    db,
+                    rsmq.clone(),
+                )
+                .await?;
+
+                tracing::info!(
+                    "Sending runnable {path} (is_flow: {is_flow}) result to websocket {}",
+                    ws_trigger.url
+                );
+
+                let result = if result.starts_with("\"") && result.ends_with("\"") {
+                    result[1..result.len() - 1].to_string()
+                } else {
+                    result
+                };
+
+                writer
+                    .send(tokio_tungstenite::tungstenite::Message::Text(result))
+                    .await
+                    .map_err(to_anyhow)
+                    .with_context(|| {
+                        format!("Failed to send runnable {path} (is_flow: {is_flow}) result")
+                    })?;
+            }
+        }
+    }
+
+    Ok(())
+}
+
+async fn get_url_from_runnable(
+    path: &str,
+    is_flow: bool,
+    ws_trigger: &WebsocketTrigger,
+    db: &DB,
+    rsmq: Option<rsmq_async::MultiplexedRsmq>,
+) -> error::Result<String> {
+    tracing::info!("Running runnable {path} (is_flow: {is_flow}) to get websocket URL",);
+
+    let result = wait_runnable_result(
+        path.to_string(),
+        is_flow,
+        &ws_trigger.url_runnable_args.0,
+        ws_trigger,
+        "url".to_string(),
+        db,
+        rsmq,
+    )
+    .await?;
+
+    if result.starts_with("\"") && result.ends_with("\"") {
+        Ok(result[1..result.len() - 1].to_string())
+    } else {
+        Err(anyhow::anyhow!("Runnable {path} (is_flow: {is_flow}) did not return a string").into())
+    }
+}
+
+async fn update_ping(db: &DB, ws_trigger: &WebsocketTrigger, error: Option<&str>) -> Option<()> {
+    match sqlx::query_scalar!(
+        "UPDATE websocket_trigger SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND server_id = $4 AND enabled IS TRUE RETURNING 1",
+        error,
+        ws_trigger.workspace_id,
+        ws_trigger.path,
+        *INSTANCE_NAME
+    ).fetch_optional(db).await {
+        Ok(updated) => {
+            if updated.flatten().is_none() {
+                tracing::info!("Websocket {} changed, disabled, or deleted, stopping...", ws_trigger.url); 
+                return None;
+            }
+        },
+        Err(err) => {
+            tracing::warn!("Error updating ping of websocket {}: {:?}", ws_trigger.url, err);
+        }
+    };
+
+    Some(())
+}
+
+async fn loop_ping(db: &DB, ws_trigger: &WebsocketTrigger, error: Option<&str>) -> () {
+    loop {
+        if let None = update_ping(db, ws_trigger, error).await {
+            return;
+        }
+        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+    }
+}
+
+async fn disable_with_error(db: &DB, ws_trigger: &WebsocketTrigger, error: String) {
+    match sqlx::query!(
+        "UPDATE websocket_trigger SET enabled = FALSE, error = $1, server_id = NULL, last_server_ping = NULL WHERE workspace_id = $2 AND path = $3",
+        error,
+        ws_trigger.workspace_id,
+        ws_trigger.path,
+    )
+    .execute(db).await {
+        Ok(_) => {
+            report_critical_error(format!("Disabling websocket {} because of error: {}", ws_trigger.url, error), db.clone(), Some(&ws_trigger.workspace_id), None).await;
+        },
+        Err(disable_err) => {
+            report_critical_error(
+                format!("Could not disable websocket {} with err {}, disabling because of error {}", ws_trigger.path, disable_err, error), 
+                db.clone(),
+                Some(&ws_trigger.workspace_id),
+                None,
+            ).await;
+        }
+    }
+}
+
 async fn listen_to_websocket(
     ws_trigger: WebsocketTrigger,
     db: DB,
     rsmq: Option<rsmq_async::MultiplexedRsmq>,
     mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
 ) -> () {
-    async fn update_ping(db: DB, ws_trigger: &WebsocketTrigger, error: Option<&str>) -> Option<()> {
-        match sqlx::query_scalar!(
-            "UPDATE websocket_trigger SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND server_id = $4 AND enabled IS TRUE RETURNING 1",
-            error,
-            ws_trigger.workspace_id,
-            ws_trigger.path,
-            *INSTANCE_NAME
-        ).fetch_optional(&db).await {
-            Ok(updated) => {
-                if updated.flatten().is_none() {
-                    tracing::info!("Websocket {} changed, disabled, or deleted, stopping...", ws_trigger.url); 
-                    return None;
-                }
-            },
-            Err(err) => {
-                tracing::warn!("Error updating ping of websocket {}: {:?}", ws_trigger.url, err);
-            }
-        };
-
-        Some(())
-    }
+    update_ping(&db, &ws_trigger, Some("Connecting...")).await;
 
     let url = ws_trigger.url.as_str();
 
-    #[derive(Deserialize)]
-    struct JsonFilter {
-        key: String,
-        value: serde_json::Value,
-    }
-
-    #[derive(Deserialize)]
-    #[serde(untagged)]
-    enum Filter {
-        JsonFilter(JsonFilter),
-    }
     let filters: Vec<Filter> = ws_trigger
         .filters
         .iter()
-        .filter_map(|m| serde_json::from_value(m.clone()).ok())
+        .filter_map(|m| serde_json::from_str(m.get()).ok())
         .collect_vec();
 
     loop {
+        let connect_url = if url.starts_with("$") {
+            if url.starts_with("$flow:") || url.starts_with("$script:") {
+                let path = url.splitn(2, ':').nth(1).unwrap();
+                tokio::select! {
+                    biased;
+                    _ = killpill_rx.recv() => {
+                        return;
+                    },
+                    _ = loop_ping(&db, &ws_trigger, Some(
+                        "Waiting on runnable to return websocket URL..."
+                    )) => {
+                        return;
+                    },
+                    url_result = get_url_from_runnable(path, url.starts_with("$flow:"), &ws_trigger, &db, rsmq.clone()) => match url_result {
+                        Ok(url) => url,
+                        Err(err) => {
+                            disable_with_error(
+                                &db,
+                                &ws_trigger,
+                                format!(
+                                    "Error getting websocket URL from runnable after 5 tries: {:?}",
+                                    err
+                                ),
+                            )
+                            .await;
+                            return;
+                        }
+                    },
+                }
+            } else {
+                disable_with_error(
+                    &db,
+                    &ws_trigger,
+                    format!("Invalid websocket runnable path: {}", url),
+                )
+                .await;
+                return;
+            }
+        } else {
+            url.to_string()
+        };
+
         tokio::select! {
             biased;
             _ = killpill_rx.recv() => {
                 return;
             },
-            connection = connect_async(url) => {
+            _ = loop_ping(&db, &ws_trigger, Some("Connecting...")) => {
+                return;
+            },
+            connection = connect_async(connect_url) => {
                 match connection {
                     Ok((ws_stream, _)) => {
                         tracing::info!("Listening to websocket {}", url);
-                        if let None = update_ping(db.clone(), &ws_trigger, None).await {
+                        if let None = update_ping(&db, &ws_trigger, None).await {
                             return;
                         }
+                        let (writer, mut reader) = ws_stream.split();
                         let mut last_ping = tokio::time::Instant::now();
-                        let (_, mut read) = ws_stream.split();
-                        loop {
-                            tokio::select! {
-                                biased;
-                                _ = killpill_rx.recv() => {
-                                    return;
+
+                        tokio::select! {
+                            biased;
+                            _ = killpill_rx.recv() => {
+                                return;
+                            }
+                            _ = async {
+                                if let Err(err) = send_initial_messages(&ws_trigger, writer, &db, rsmq.clone()).await {
+                                    disable_with_error(&db, &ws_trigger, format!("Error sending initial messages: {:?}", err)).await;
+                                } else {
+                                    // if initial messages sent successfully, wait forever
+                                    futures::future::pending::<()>().await;
                                 }
-                                msg = read.next() => {
-                                    if let Some(msg) = msg {
-                                        if last_ping.elapsed() > tokio::time::Duration::from_secs(5) {
-                                            if let None = update_ping(db.clone(), &ws_trigger, None).await {
+                            } => {
+                                // was disabled => exit
+                                return;
+                            },
+                            _ = async {
+                                loop {
+                                    tokio::select! {
+                                        biased;
+                                        msg = reader.next() => {
+                                            if let Some(msg) = msg {
+                                                if last_ping.elapsed() > tokio::time::Duration::from_secs(5) {
+                                                    if let None = update_ping(&db, &ws_trigger, None).await {
+                                                        return;
+                                                    }
+                                                    last_ping = tokio::time::Instant::now();
+                                                }
+                                                match msg {
+                                                    Ok(msg) => {
+                                                        match msg {
+                                                            tokio_tungstenite::tungstenite::Message::Text(text) => {
+                                                                let mut should_handle = true;
+                                                                for filter in &filters {
+                                                                    match filter {
+                                                                        Filter::JsonFilter(JsonFilter { key, value }) => {
+                                                                            let mut deserializer = serde_json::Deserializer::from_str(text.as_str());
+                                                                            should_handle = match is_value_superset(&mut deserializer, key, &value) {
+                                                                                Ok(filter_match) => {
+                                                                                    filter_match
+                                                                                },
+                                                                                Err(err) => {
+                                                                                    tracing::warn!("Error deserializing filter for websocket {}: {:?}", url, err);
+                                                                                    false
+                                                                                }
+                                                                            };
+                                                                        }
+                                                                    }
+                                                                    if !should_handle {
+                                                                        break;
+                                                                    }
+                                                                }
+                                                                if should_handle {
+                                                                    if let Err(err) = run_job(&db, rsmq.clone(), &ws_trigger, text).await {
+                                                                        report_critical_error(format!("Failed to trigger job from websocket {}: {:?}", ws_trigger.url, err), db.clone(), Some(&ws_trigger.workspace_id), None).await;
+                                                                    };
+                                                                }
+                                                            },
+                                                            _ => {}
+                                                        }
+                                                    },
+                                                    Err(err) => {
+                                                        tracing::error!("Error reading from websocket {}: {:?}", url, err);
+                                                    }
+                                                }
+                                            } else {
+                                                tracing::error!("Websocket {} closed", url);
+                                                if let None =
+                                                    update_ping(&db, &ws_trigger, Some("Websocket closed")).await
+                                                {
+                                                    return;
+                                                }
+                                                tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+                                                break;
+                                            }
+                                        },
+                                        _ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => {
+                                            if let None = update_ping(&db, &ws_trigger, None).await {
                                                 return;
                                             }
                                             last_ping = tokio::time::Instant::now();
-                                        }
-                                        match msg {
-                                            Ok(msg) => {
-                                                match msg {
-                                                    tokio_tungstenite::tungstenite::Message::Text(text) => {
-                                                        let mut should_handle = true;
-                                                        for filter in &filters {
-                                                            match filter {
-                                                                Filter::JsonFilter(JsonFilter { key, value }) => {
-                                                                    let mut deserializer = serde_json::Deserializer::from_str(text.as_str());
-                                                                    should_handle = match is_value_superset(&mut deserializer, key, &value) {
-                                                                        Ok(filter_match) => {
-                                                                            filter_match
-                                                                        },
-                                                                        Err(err) => {
-                                                                            tracing::warn!("Error deserializing filter for websocket {}: {:?}", url, err);
-                                                                            false
-                                                                        }
-                                                                    };
-                                                                }
-                                                            }
-                                                            if !should_handle {
-                                                                break;
-                                                            }
-                                                        }
-                                                        if should_handle {
-                                                            let db_ = db.clone();
-                                                            let rsmq_ = rsmq.clone();
-                                                            let ws_trigger_ = ws_trigger.clone();
-                                                            tokio::spawn(async move {
-                                                                let url = ws_trigger_.url.clone();
-                                                                if let Err(err) = run_job(db_, rsmq_, ws_trigger_, text).await {
-                                                                    tracing::error!("Error running job on websocket {}: {:?}", url, err);
-                                                                };
-                                                            });
-                                                        }
-                                                    },
-                                                    _ => {}
-                                                }
-                                            },
-                                            Err(err) => {
-                                                tracing::error!("Error reading from websocket {}: {:?}", url, err);
-                                            }
-                                        }
-                                    } else {
-                                        tracing::error!("Websocket {} closed", url);
-                                        if let None =
-                                            update_ping(db.clone(), &ws_trigger, Some("Websocket closed")).await
-                                        {
-                                            return;
-                                        }
-                                        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
-                                        break;
+                                        },
                                     }
-                                },
-                                _ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => {
-                                    if let None = update_ping(db.clone(), &ws_trigger, None).await {
-                                        return;
-                                    }
-                                    last_ping = tokio::time::Instant::now();
-                                },
+                                }
+                            } => {
+                                return;
                             }
-                        }
+                        };
                     }
                     Err(err) => {
                         tracing::error!("Error connecting to websocket {}: {:?}", url, err);
                         if let None =
-                            update_ping(db.clone(), &ws_trigger, Some(err.to_string().as_str())).await
+                            update_ping(&db, &ws_trigger, Some(err.to_string().as_str())).await
                         {
                             return;
                         }
@@ -625,16 +949,18 @@ async fn listen_to_websocket(
 }
 
 async fn run_job(
-    db: DB,
+    db: &DB,
     rsmq: Option<rsmq_async::MultiplexedRsmq>,
-    trigger: WebsocketTrigger,
+    trigger: &WebsocketTrigger,
     msg: String,
 ) -> anyhow::Result<()> {
     let args = PushArgsOwned {
         args: HashMap::from([("msg".to_string(), to_raw_value(&msg))]),
         extra: Some(HashMap::from([(
             "wm_trigger".to_string(),
-            to_raw_value(&serde_json::json!({"kind": "websocket"})),
+            to_raw_value(
+                &serde_json::json!({"kind": "websocket", "websocket": { "url": trigger.url }}),
+            ),
         )])),
     };
     let label_prefix = Some(format!("ws-{}-", trigger.path));
@@ -643,7 +969,7 @@ async fn run_job(
         trigger.edited_by.clone(),
         trigger.email.clone(),
         &trigger.workspace_id,
-        &db,
+        db,
         "anonymous".to_string(),
     )
     .await?;
@@ -653,27 +979,27 @@ async fn run_job(
     let run_query = RunJobQuery::default();
 
     if trigger.is_flow {
-        run_wait_result_flow_by_path_internal(
-            db,
-            run_query,
-            StripPath(trigger.script_path.to_owned()),
+        run_flow_by_path_inner(
             authed,
-            rsmq,
+            db.clone(),
             user_db,
-            args,
+            rsmq,
             trigger.workspace_id.clone(),
+            StripPath(trigger.script_path.to_owned()),
+            run_query,
+            args,
             label_prefix,
         )
         .await?;
     } else {
-        run_wait_result_script_by_path_internal(
-            db,
-            run_query,
-            StripPath(trigger.script_path.to_owned()),
+        run_script_by_path_inner(
             authed,
-            rsmq,
+            db.clone(),
             user_db,
+            rsmq,
             trigger.workspace_id.clone(),
+            StripPath(trigger.script_path.to_owned()),
+            run_query,
             args,
             label_prefix,
         )

backend/windmill-api/src/workers.rs

@@ -36,6 +36,7 @@ pub fn global_service() -> Router {
         )
         .route("/get_default_tags", get(get_default_tags))
         .route("/queue_metrics", get(get_queue_metrics))
+        .route("/queue_counts", get(get_queue_counts))
 }
 
 #[derive(FromRow, Serialize, Deserialize)]
@@ -176,3 +177,12 @@ async fn get_queue_metrics(
 
     Ok(Json(queue_metrics))
 }
+
+async fn get_queue_counts(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+) -> JsonResult<std::collections::HashMap<String, u32>> {
+    require_super_admin(&db, &authed.email).await?;
+    let queue_counts = windmill_common::queue::get_queue_counts(&db).await;
+    Ok(Json(queue_counts))
+}

backend/windmill-api/src/workspaces.rs

@@ -8,6 +8,7 @@
 
 use std::collections::HashMap;
 
+use crate::ai::{AiResource, AI_KEY_CACHE};
 use crate::db::ApiAuthed;
 use crate::users_ee::send_email_if_possible;
 use crate::utils::get_instance_username_or_create_pending;
@@ -118,7 +119,11 @@ pub fn workspaced_service() -> Router {
         .route("/change_workspace_name", post(change_workspace_name))
         .route("/change_workspace_id", post(change_workspace_id))
         .route("/usage", get(get_usage))
-        .route("/used_triggers", get(get_used_triggers));
+        .route("/used_triggers", get(get_used_triggers))
+        .route("/critical_alerts", get(get_critical_alerts))
+        .route("/critical_alerts/:id/acknowledge", post(acknowledge_critical_alert))
+        .route("/critical_alerts/acknowledge_all", post(acknowledge_all_critical_alerts))
+        .route("/critical_alerts/mute", post(mute_critical_alerts));
 
     #[cfg(feature = "stripe")]
     {
@@ -168,7 +173,7 @@ pub struct WorkspaceSettings {
     pub plan: Option<String>,
     pub webhook: Option<String>,
     pub deploy_to: Option<String>,
-    pub openai_resource_path: Option<String>,
+    pub ai_resource: Option<serde_json::Value>,
     pub code_completion_enabled: bool,
     pub error_handler: Option<String>,
     pub error_handler_extra_args: Option<serde_json::Value>,
@@ -179,6 +184,7 @@ pub struct WorkspaceSettings {
     pub default_app: Option<String>,
     pub automatic_billing: bool,
     pub default_scripts: Option<serde_json::Value>,
+    pub mute_critical_alerts: Option<bool>,
 }
 
 #[derive(FromRow, Serialize, Debug)]
@@ -234,7 +240,7 @@ struct EditWebhook {
 
 #[derive(Deserialize)]
 struct EditCopilotConfig {
-    openai_resource_path: Option<String>,
+    ai_resource: Option<serde_json::Value>,
     code_completion_enabled: bool,
 }
 
@@ -645,23 +651,33 @@ async fn edit_copilot_config(
 
     let mut tx = db.begin().await?;
 
-    if let Some(openai_resource_path) = &eo.openai_resource_path {
+    if let Some(ai_resource) = &eo.ai_resource {
+        let path = serde_json::from_value::<AiResource>(ai_resource.clone())
+            .map_err(|e| Error::BadRequest(e.to_string()))?
+            .path;
         sqlx::query!(
-            "UPDATE workspace_settings SET openai_resource_path = $1, code_completion_enabled = $2 WHERE workspace_id = $3",
-            openai_resource_path,
+            "UPDATE workspace_settings SET ai_resource = $1, code_completion_enabled = $2 WHERE workspace_id = $3",
+            ai_resource,
             eo.code_completion_enabled,
             &w_id
         )
         .execute(&mut *tx)
         .await?;
+
+        if let Some(cached) = AI_KEY_CACHE.get(&w_id) {
+            if cached.path != path {
+                AI_KEY_CACHE.remove(&w_id);
+            }
+        }
     } else {
         sqlx::query!(
-            "UPDATE workspace_settings SET openai_resource_path = NULL, code_completion_enabled = $1 WHERE workspace_id = $2",
+            "UPDATE workspace_settings SET ai_resource = NULL, code_completion_enabled = $1 WHERE workspace_id = $2",
             eo.code_completion_enabled,
             &w_id,
         )
         .execute(&mut *tx)
         .await?;
+        AI_KEY_CACHE.remove(&w_id);
     }
     audit_log(
         &mut *tx,
@@ -672,10 +688,7 @@ async fn edit_copilot_config(
         Some(&authed.email),
         Some(
             [
-                (
-                    "openai_resource_path",
-                    &format!("{:?}", eo.openai_resource_path)[..],
-                ),
+                ("ai_resource", &format!("{:?}", eo.ai_resource)[..]),
                 (
                     "code_completion_enabled",
                     &format!("{:?}", eo.code_completion_enabled)[..],
@@ -692,7 +705,8 @@ async fn edit_copilot_config(
 
 #[derive(Serialize)]
 struct CopilotInfo {
-    pub exists_openai_resource_path: bool,
+    pub ai_provider: String,
+    pub exists_ai_resource: bool,
     pub code_completion_enabled: bool,
 }
 async fn get_copilot_info(
@@ -701,16 +715,32 @@ async fn get_copilot_info(
 ) -> JsonResult<CopilotInfo> {
     let mut tx = db.begin().await?;
     let record = sqlx::query!(
-        "SELECT openai_resource_path, code_completion_enabled FROM workspace_settings WHERE workspace_id = $1",
+        "SELECT ai_resource, code_completion_enabled FROM workspace_settings WHERE workspace_id = $1",
         &w_id
     )
     .fetch_one(&mut *tx)
     .await
-    .map_err(|e| Error::InternalErr(format!("getting openai_resource_path and code_completion_enabled: {e:#}")))?;
+    .map_err(|e| Error::InternalErr(format!("getting ai_resource and code_completion_enabled: {e:#}")))?;
     tx.commit().await?;
 
+    let (ai_provider, exists_ai_resource) = if let Some(ai_resource) = record.ai_resource {
+        let ai_resource = serde_json::from_value::<AiResource>(ai_resource);
+        let exist = ai_resource.is_ok();
+        (
+            if exist {
+                ai_resource.unwrap().provider
+            } else {
+                "".to_string()
+            },
+            exist,
+        )
+    } else {
+        ("".to_string(), false)
+    };
+
     Ok(Json(CopilotInfo {
-        exists_openai_resource_path: record.openai_resource_path.is_some(),
+        ai_provider,
+        exists_ai_resource,
         code_completion_enabled: record.code_completion_enabled,
     }))
 }
@@ -2223,7 +2253,7 @@ struct SimplifiedSettings {
     error_handler: Option<String>,
     error_handler_extra_args: Option<Value>,
     error_handler_muted_on_cancel: bool,
-    openai_resource_path: Option<String>,
+    ai_resource: Option<serde_json::Value>,
     code_completion_enabled: bool,
     large_file_storage: Option<Value>,
     git_sync: Option<Value>,
@@ -2588,7 +2618,7 @@ async fn tarball_workspace(
                 webhook, 
                 deploy_to, 
                 error_handler, 
-                openai_resource_path, 
+                ai_resource, 
                 code_completion_enabled, 
                 error_handler_extra_args, 
                 error_handler_muted_on_cancel, 
@@ -3050,3 +3080,93 @@ async fn get_usage(Extension(db): Extension<DB>, Path(w_id): Path<String>) -> Re
     .unwrap_or(0);
     Ok(usage.to_string())
 }
+
+#[cfg(feature = "enterprise")]
+pub async fn get_critical_alerts(
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    authed: ApiAuthed,
+    Query(params): Query<crate::utils::AlertQueryParams>,
+) -> JsonResult<Vec<crate::utils::CriticalAlert>> {
+    require_admin(authed.is_admin, &authed.username)?;
+
+    crate::utils::get_critical_alerts(db, params, Some(w_id)).await
+}
+
+#[cfg(not(feature = "enterprise"))]
+pub async fn get_critical_alerts() -> Error {
+    Error::NotFound("Critical Alerts require EE".to_string())
+}
+
+#[cfg(feature = "enterprise")]
+pub async fn acknowledge_critical_alert(
+    Extension(db): Extension<DB>,
+    Path((w_id, id)): Path<(String, i32)>,
+    authed: ApiAuthed,
+) -> Result<String> {
+    require_admin(authed.is_admin, &authed.username)?;
+    crate::utils::acknowledge_critical_alert(db, Some(w_id), id).await 
+}
+
+#[cfg(not(feature = "enterprise"))]
+pub async fn acknowledge_critical_alert() -> Error {
+    Error::NotFound("Critical Alerts require EE".to_string())
+}
+
+#[cfg(feature = "enterprise")]
+pub async fn acknowledge_all_critical_alerts(
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    authed: ApiAuthed,
+) -> Result<String> {
+    require_admin(authed.is_admin, &authed.username)?;
+    crate::utils::acknowledge_all_critical_alerts(db, Some(w_id)).await
+}
+
+#[cfg(not(feature = "enterprise"))]
+pub async fn acknowledge_all_critical_alerts() -> Error {
+    Error::NotFound("Critical Alerts require EE".to_string())
+}
+
+
+#[cfg(feature = "enterprise")]
+#[derive(Deserialize)]
+pub struct MuteCriticalAlertRequest {
+    pub mute_critical_alerts: Option<bool>,
+}
+
+#[cfg(feature = "enterprise")]
+async fn mute_critical_alerts(
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    ApiAuthed { is_admin, username, .. }: ApiAuthed,
+    Json(m_r): Json<MuteCriticalAlertRequest>,
+) -> Result<String> {
+    require_admin(is_admin, &username)?;
+
+    let mute_alerts = m_r.mute_critical_alerts.unwrap_or(false);
+
+    if mute_alerts {
+        sqlx::query!(
+            "UPDATE alerts SET acknowledged_workspace = true, acknowledged = true WHERE workspace_id = $1",
+            &w_id
+        )
+    .execute(&db)
+    .await?;
+    }
+
+    sqlx::query!(
+        "UPDATE workspace_settings SET mute_critical_alerts = $1 WHERE workspace_id = $2",
+        mute_alerts,
+        &w_id
+    )
+    .execute(&db)
+    .await?;
+
+    Ok(format!("Updated mute criticital alert ui settings for workspace: {}", &w_id))
+}
+
+#[cfg(not(feature = "enterprise"))]
+pub async fn mute_critical_alerts() -> Error {
+    Error::NotFound("Critical Alerts require EE".to_string())
+}

backend/windmill-common/Cargo.toml

@@ -58,6 +58,7 @@ async-stream.workspace = true
 const_format.workspace = true
 crc.workspace = true
 windmill-macros.workspace = true
+semver.workspace = true
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemalloc-ctl = { optional = true, workspace = true }

backend/windmill-common/src/error.rs

@@ -59,7 +59,7 @@ pub enum Error {
     #[error("Error: {0:#?}")]
     JsonErr(serde_json::Value),
     #[error("{0}")]
-    OpenAIError(String),
+    AiError(String),
     #[error("{0}")]
     AlreadyCompleted(String),
 }
@@ -90,7 +90,7 @@ impl IntoResponse for Error {
             Self::RequireAdmin(_) => axum::http::StatusCode::FORBIDDEN,
             Self::SqlErr(_)
             | Self::BadRequest(_)
-            | Self::OpenAIError(_)
+            | Self::AiError(_)
             | Self::QuotaExceeded(_) => axum::http::StatusCode::BAD_REQUEST,
             _ => axum::http::StatusCode::INTERNAL_SERVER_ERROR,
         };

backend/windmill-common/src/flows.rs

@@ -16,6 +16,7 @@ use rand::Rng;
 use serde::{Deserialize, Serialize, Serializer};
 
 use crate::{
+    error::Error,
     more_serde::{default_empty_string, default_id, default_null, default_true, is_default},
     scripts::{Schema, ScriptHash, ScriptLang},
 };
@@ -426,6 +427,8 @@ pub enum FlowModuleValue {
         hash: Option<ScriptHash>,
         #[serde(skip_serializing_if = "Option::is_none")]
         tag_override: Option<String>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        is_trigger: Option<bool>,
     },
     Flow {
         #[serde(default)]
@@ -474,6 +477,8 @@ pub enum FlowModuleValue {
         concurrent_limit: Option<i32>,
         #[serde(skip_serializing_if = "Option::is_none")]
         concurrency_time_window_s: Option<i32>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        is_trigger: Option<bool>,
     },
     Identity,
 }
@@ -505,6 +510,7 @@ struct UntaggedFlowModuleValue {
     custom_concurrency_key: Option<String>,
     concurrent_limit: Option<i32>,
     concurrency_time_window_s: Option<i32>,
+    is_trigger: Option<bool>,
 }
 
 impl<'de> Deserialize<'de> for FlowModuleValue {
@@ -522,6 +528,7 @@ impl<'de> Deserialize<'de> for FlowModuleValue {
                     .ok_or_else(|| serde::de::Error::missing_field("path"))?,
                 hash: untagged.hash,
                 tag_override: untagged.tag_override,
+                is_trigger: untagged.is_trigger,
             }),
             "flow" => Ok(FlowModuleValue::Flow {
                 input_transforms: untagged.input_transforms.unwrap_or_default(),
@@ -574,6 +581,7 @@ impl<'de> Deserialize<'de> for FlowModuleValue {
                 custom_concurrency_key: untagged.custom_concurrency_key,
                 concurrent_limit: untagged.concurrent_limit,
                 concurrency_time_window_s: untagged.concurrency_time_window_s,
+                is_trigger: untagged.is_trigger,
             }),
             "identity" => Ok(FlowModuleValue::Identity),
             other => Err(serde::de::Error::unknown_variant(
@@ -644,3 +652,29 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
         });
     }
 }
+
+pub async fn has_failure_module<'c>(flow: sqlx::types::Uuid, db: &sqlx::Pool<sqlx::Postgres>, completed: bool) -> Result<bool, Error> {
+    if completed {
+        sqlx::query_scalar!(
+            "SELECT raw_flow->'failure_module' != 'null'::jsonb
+            FROM completed_job
+            WHERE id = $1",
+            flow
+        )
+    } else {
+        sqlx::query_scalar!(
+            "SELECT raw_flow->'failure_module' != 'null'::jsonb
+            FROM queue
+            WHERE id = $1",
+            flow
+        )
+    }
+    .fetch_one(db)
+    .await
+    .map_err(|e| {
+        Error::InternalErr(format!(
+            "error during retrieval of has_failure_module: {e:#}"
+        ))
+    })
+    .map(|v| v.unwrap_or(false))
+}

backend/windmill-common/src/global_settings.rs

@@ -29,6 +29,7 @@ pub const AUTOMATE_USERNAME_CREATION_SETTING: &str = "automate_username_creation
 pub const HUB_BASE_URL_SETTING: &str = "hub_base_url";
 pub const HUB_ACCESSIBLE_URL_SETTING: &str = "hub_accessible_url";
 pub const CRITICAL_ERROR_CHANNELS_SETTING: &str = "critical_error_channels";
+pub const CRITICAL_ALERT_MUTE_UI_SETTING: &str = "critical_alert_mute_ui";
 pub const DEV_INSTANCE_SETTING: &str = "dev_instance";
 pub const JWT_SECRET_SETTING: &str = "jwt_secret";
 pub const EMAIL_DOMAIN_SETTING: &str = "email_domain";

backend/windmill-common/src/jobs.rs

@@ -108,13 +108,6 @@ pub struct QueuedJob {
     pub cache_ttl: Option<i32>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub priority: Option<i16>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[sqlx(skip)]
-    pub self_wait_time_ms: Option<i64>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[sqlx(skip)]
-    pub aggregate_wait_time_ms: Option<i64>,
 }
 
 impl QueuedJob {
@@ -198,8 +191,6 @@ impl Default for QueuedJob {
             flow_step_id: None,
             cache_ttl: None,
             priority: None,
-            self_wait_time_ms: None,
-            aggregate_wait_time_ms: None,
         }
     }
 }
@@ -253,13 +244,6 @@ pub struct CompletedJob {
     pub priority: Option<i16>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub labels: Option<serde_json::Value>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[sqlx(skip)]
-    pub self_wait_time_ms: Option<i64>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[sqlx(skip)]
-    pub aggregate_wait_time_ms: Option<i64>,
 }
 
 impl CompletedJob {

backend/windmill-common/src/lib.rs

@@ -86,6 +86,8 @@ lazy_static::lazy_static! {
     pub static ref METRICS_ENABLED: AtomicBool = AtomicBool::new(std::env::var("METRICS_PORT").is_ok() || std::env::var("METRICS_ADDR").is_ok());
     pub static ref METRICS_DEBUG_ENABLED: AtomicBool = AtomicBool::new(false);
 
+    pub static ref CRITICAL_ALERT_MUTE_UI_ENABLED: AtomicBool = AtomicBool::new(false);
+
     pub static ref BASE_URL: Arc<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
     pub static ref IS_READY: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
 

backend/windmill-common/src/s3_helpers.rs

@@ -109,10 +109,13 @@ pub struct S3AwsOidcResource {
     pub audience: Option<String>,
 }
 
-#[derive(Deserialize, Clone)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct S3Object {
     pub s3: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub storage: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub filename: Option<String>,
 }
 
 #[cfg(feature = "parquet")]