Compare commits

..

5 Commits

Author SHA1 Message Date
pyranota
d39b870b2a add windows flags 2025-04-11 15:55:07 +02:00
pyranota
f107936d44 fix compilation error 2025-04-11 15:34:58 +02:00
pyranota
9221a63edb Merge branch 'main' into rust-inc-comp 2025-04-11 15:12:50 +02:00
pyranota
99fcacb2e6 update dockerfiles 2025-04-11 15:10:53 +02:00
pyranota
cb49ae5458 feat: rust incremental compilation 2025-04-11 13:55:22 +02:00
456 changed files with 51469 additions and 21286 deletions

2
.env
View File

@@ -10,4 +10,4 @@ WM_IMAGE=ghcr.io/windmill-labs/windmill:main
# To rotate logs, set the following variables:
#LOG_MAX_SIZE=10m
#LOG_MAX_FILE=3
#LOG_MAX_FILE=3

View File

@@ -24,4 +24,4 @@ sed -i -e "/^wmill_pg =/s/= .*/= \">=$VERSION\"/" ${root_dirpath}/lsp/Pipfile
sed -i -zE "s/name = \"windmill\"\nversion = \"[^\"]*\"\\n(.*)/name = \"windmill\"\nversion = \"$VERSION\"\\n\\1/" ${root_dirpath}/backend/Cargo.lock
cd ${root_dirpath}/frontend && npm i --package-lock-only --ignore-scripts
cd ${root_dirpath}/frontend && npm i --package-lock-only

View File

@@ -64,7 +64,7 @@ jobs:
platforms: linux/amd64
push: true
build-args: |
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,license,otel,http_trigger,zip,oauth2,kafka,sqs_trigger,nats,postgres_trigger,gcp_trigger,mqtt_trigger,websocket,smtp,static_frontend,all_languages,deno_core,mcp
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,license,otel,http_trigger,zip,oauth2,kafka,sqs_trigger,nats,postgres_trigger,gcp_trigger,mqtt_trigger,websocket,smtp,static_frontend,all_languages
secrets: |
rh_username=${{ secrets.RH_USERNAME }}
rh_password=${{ secrets.RH_PASSWORD }}
@@ -81,7 +81,7 @@ jobs:
platforms: linux/arm64
push: true
build-args: |
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,license,otel,http_trigger,zip,oauth2,kafka,sqs_trigger,nats,postgres_trigger,gcp_trigger,mqtt_trigger,websocket,smtp,static_frontend,all_languages,deno_core,mcp
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,license,otel,http_trigger,zip,oauth2,kafka,sqs_trigger,nats,postgres_trigger,gcp_trigger,mqtt_trigger,websocket,smtp,static_frontend,all_languages
secrets: |
rh_username=${{ secrets.RH_USERNAME }}
rh_password=${{ secrets.RH_PASSWORD }}
@@ -111,7 +111,8 @@ jobs:
- uses: actions/upload-artifact@v4
with:
name: RHEL9-amd64 build
path: ${{ steps.extract-ee-amd64.outputs.destination
path:
${{ steps.extract-ee-amd64.outputs.destination
}}/windmill-ee-amd64-rhel9
# - uses: actions/upload-artifact@v4

View File

@@ -51,7 +51,7 @@ jobs:
$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,license,http_trigger,zip,oauth2,kafka,nats,sqs_trigger,postgres_trigger,gcp_trigger,mqtt_trigger,websocket,smtp,static_frontend,all_languages,mcp
cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,license,http_trigger,zip,oauth2,kafka,nats,sqs_trigger,postgres_trigger,gcp_trigger,mqtt_trigger,websocket,smtp,static_frontend,all_languages
- name: Rename binary with corresponding architecture
run: |
Rename-Item -Path ".\backend\target\release\windmill.exe" -NewName "windmill-ee.exe"

View File

@@ -67,7 +67,7 @@ jobs:
platforms: linux/amd64,linux/arm64
push: true
build-args: |
features=embedding,parquet,openidconnect,license,http_trigger,zip,oauth2,postgres_trigger,mqtt_trigger,websocket,smtp,static_frontend,all_languages,deno_core,mcp
features=embedding,parquet,openidconnect,license,http_trigger,zip,oauth2,postgres_trigger,mqtt_trigger,websocket,smtp,static_frontend,all_languages
tags: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:dev
${{ steps.meta-public.outputs.tags }}

View File

@@ -1,8 +1,10 @@
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.event_name != 'pull_request' && github.event_name !=
IMAGE_NAME:
${{ github.event_name != 'pull_request' && github.event_name !=
'workflow_dispatch' && github.repository || 'windmill-labs/windmill-test' }}
DEV_SHA: ${{ github.event_name != 'pull_request' && github.event_name !=
DEV_SHA:
${{ github.event_name != 'pull_request' && github.event_name !=
'workflow_dispatch' && 'dev' || github.event.inputs.tag || github.sha }}
name: Build windmill:main
on:
@@ -38,7 +40,8 @@ permissions: write-all
jobs:
build:
runs-on: ubicloud
if: (github.event_name != 'workflow_dispatch') || (github.event.inputs &&
if:
(github.event_name != 'workflow_dispatch') || (github.event.inputs &&
!github.event.inputs.ee)
steps:
- uses: actions/checkout@v4
@@ -92,7 +95,7 @@ jobs:
platforms: linux/amd64,linux/arm64
push: true
build-args: |
features=embedding,parquet,openidconnect,jemalloc,license,http_trigger,zip,oauth2,dind,postgres_trigger,mqtt_trigger,websocket,smtp,static_frontend,agent_worker_server,all_languages,deno_core,mcp
features=embedding,parquet,openidconnect,jemalloc,license,http_trigger,zip,oauth2,dind,postgres_trigger,mqtt_trigger,websocket,smtp,static_frontend,agent_worker_server,all_languages
tags: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ env.DEV_SHA }}
${{ steps.meta-public.outputs.tags }}
@@ -154,7 +157,7 @@ jobs:
platforms: linux/amd64,linux/arm64
push: true
build-args: |
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,agent_worker_server,tantivy,license,http_trigger,zip,oauth2,kafka,sqs_trigger,nats,otel,dind,postgres_trigger,mqtt_trigger,gcp_trigger,websocket,smtp,static_frontend,all_languages,deno_core,mcp
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,agent_worker_server,tantivy,license,http_trigger,zip,oauth2,kafka,sqs_trigger,nats,otel,dind,postgres_trigger,mqtt_trigger,gcp_trigger,websocket,smtp,static_frontend,all_languages
tags: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee:${{ env.DEV_SHA }}
${{ steps.meta-ee-public.outputs.tags }}
@@ -312,7 +315,7 @@ jobs:
needs: [run_integration_test, build]
if:
github.event_name != 'pull_request' && (github.ref == 'refs/heads/main' ||
startsWith(github.ref, 'refs/tags/v')) && (github.event_name != 'workflow_dispatch')
startsWith(github.ref, 'refs/tags/v')) && (github.event_name != 'workflow_dispatch')
steps:
- uses: actions/checkout@v4
with:
@@ -352,7 +355,7 @@ jobs:
verify_ee_image_vulnerabilities:
runs-on: ubicloud
needs: [tag_latest_ee]
if: startsWith(github.ref, 'refs/tags/v') && (github.event_name != 'workflow_dispatch')
if: startsWith(github.ref, 'refs/tags/v') && (github.event_name != 'workflow_dispatch')
steps:
- name: Checkout code
uses: actions/checkout@v4
@@ -394,7 +397,8 @@ jobs:
build_ee_nsjail:
needs: [build_ee]
runs-on: ubicloud
if: (github.event_name != 'pull_request') && ((github.event_name != 'workflow_dispatch') || (github.event.inputs.ee || github.event.inputs.nsjail))
if:
(github.event_name != 'pull_request') && ((github.event_name != 'workflow_dispatch') || (github.event.inputs.ee || github.event.inputs.nsjail))
steps:
- uses: actions/checkout@v4
@@ -433,7 +437,7 @@ jobs:
run: |
sed -i 's|FROM ghcr.io/windmill-labs/windmill-ee:dev|FROM ghcr.io/${{ env.IMAGE_NAME }}-ee:${{ env.DEV_SHA }}|' ./docker/DockerfileNsjail
cat ./docker/DockerfileNsjail | grep "FROM"
- name: Build and push publicly ee
uses: depot/build-push-action@v1
with:
@@ -447,10 +451,12 @@ jobs:
${{ steps.meta-ee-public.outputs.labels }}
org.opencontainers.image.licenses=Windmill-Enterprise-License
publish_ecr_s3:
needs: [build_ee_nsjail]
runs-on: ubicloud-standard-2-arm
if: (github.event_name != 'pull_request') && (github.event_name !=
if:
(github.event_name != 'pull_request') && (github.event_name !=
'workflow_dispatch')
env:
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}

View File

@@ -53,7 +53,7 @@ jobs:
$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,license,http_trigger,zip,oauth2,kafka,sqs_trigger,nats,postgres_trigger,mqtt_trigger,gcp_trigger,websocket,smtp,static_frontend,all_languages,mcp
cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,license,http_trigger,zip,oauth2,kafka,sqs_trigger,nats,postgres_trigger,mqtt_trigger,gcp_trigger,websocket,smtp,static_frontend,all_languages
- name: Rename binary with corresponding architecture
run: |
Rename-Item -Path ".\backend\target\release\windmill.exe" -NewName "windmill-ee.exe"

View File

@@ -1,149 +1,5 @@
# Changelog
## [1.485.3](https://github.com/windmill-labs/windmill/compare/v1.485.2...v1.485.3) (2025-04-29)
### Bug Fixes
* improve performance of background cleanup monitoring operations ([18dced3](https://github.com/windmill-labs/windmill/commit/18dced3c748cd5305f0934b26e50d69899563723))
## [1.485.2](https://github.com/windmill-labs/windmill/compare/v1.485.1...v1.485.2) (2025-04-29)
### Bug Fixes
* improve agent workers for deployed scripts ([60018aa](https://github.com/windmill-labs/windmill/commit/60018aadf62cecadf111e019d3600513a89810f1))
* make `#(extra_)requirements:` work better with pins ([#5680](https://github.com/windmill-labs/windmill/issues/5680)) ([1ab4160](https://github.com/windmill-labs/windmill/commit/1ab41603f4fd1526d0c944396ef250b184aed1f4))
* **python:** handle better relative imports with requirements or extra_requirements ([f662cf5](https://github.com/windmill-labs/windmill/commit/f662cf5d75beed8fd114ba171cbe0fa8e4b2773f))
## [1.485.1](https://github.com/windmill-labs/windmill/compare/v1.485.0...v1.485.1) (2025-04-28)
### Bug Fixes
* improve mcp mode api ([cf77ff0](https://github.com/windmill-labs/windmill/commit/cf77ff088b8382b861113120589de58f7cf241d0))
* MCP handle long names + invalid char in prop key + fix for not found resource type ([#5668](https://github.com/windmill-labs/windmill/issues/5668)) ([eadae95](https://github.com/windmill-labs/windmill/commit/eadae95a42d679bf8792bdefd8b9d19dbcbc4b57))
* skip_flow_update for dependency tracking table ([#5670](https://github.com/windmill-labs/windmill/issues/5670)) ([35b69da](https://github.com/windmill-labs/windmill/commit/35b69da25c5bd17deff5a54b635e9150cb865cc0))
## [1.485.0](https://github.com/windmill-labs/windmill/compare/v1.484.0...v1.485.0) (2025-04-28)
### Features
* add universal search to object viewer ([7254743](https://github.com/windmill-labs/windmill/commit/72547437fead0a071fceac27dae8628cdcae6a3e))
### Bug Fixes
* add svelte 5 boundaries to app components to contain errors ([1b16918](https://github.com/windmill-labs/windmill/commit/1b1691837a7e6b88afbacf7d88c14ca5e475b493))
* Fix object handling on some MCP clients + better frontend for MCP ([#5663](https://github.com/windmill-labs/windmill/issues/5663)) ([12c3202](https://github.com/windmill-labs/windmill/commit/12c32026e5879a65fc0f1cc9f2481087c4b95111))
## [1.484.0](https://github.com/windmill-labs/windmill/compare/v1.483.2...v1.484.0) (2025-04-26)
### Features
* Add MCP endpoints ([#5639](https://github.com/windmill-labs/windmill/issues/5639)) ([a34ac4f](https://github.com/windmill-labs/windmill/commit/a34ac4fa24c2a5482e45724e76316d57f64f7040))
* Add MCP only mode ([#5661](https://github.com/windmill-labs/windmill/issues/5661)) ([1625524](https://github.com/windmill-labs/windmill/commit/162552431138d68002c7060cad4ae31f1ec4c69c))
* Ansible improvements (vault, roles and git repos) ([#5655](https://github.com/windmill-labs/windmill/issues/5655)) ([fdd1642](https://github.com/windmill-labs/windmill/commit/fdd1642ce10866da1d8d373bda44f050e2e0f403))
### Bug Fixes
* check for valid teams_channel config when saving critical alerts settings ([#5660](https://github.com/windmill-labs/windmill/issues/5660)) ([dc5c8d8](https://github.com/windmill-labs/windmill/commit/dc5c8d8c5f8577b7ded3da1d684cdb735fa7a936))
* Fix CI for MCP + optimization ([#5657](https://github.com/windmill-labs/windmill/issues/5657)) ([b199a77](https://github.com/windmill-labs/windmill/commit/b199a77d486c5bfd086ca73a58a14bb747e386b5))
* fix token creation after mcp mode change to make it non workspace specific ([2b5dfcf](https://github.com/windmill-labs/windmill/commit/2b5dfcfb251471dcc39b04c54e25008d617cc34f))
* improve full-scaleout of autoscaling event logging ([8435eb3](https://github.com/windmill-labs/windmill/commit/8435eb3adff8429a73db88b12204f7cf8f14d3d2))
* improve skip failure on parallel branchall ([a7b2b51](https://github.com/windmill-labs/windmill/commit/a7b2b51444d757964560de3a89024b1c9b0fefe9))
## [1.483.2](https://github.com/windmill-labs/windmill/compare/v1.483.1...v1.483.2) (2025-04-23)
### Bug Fixes
* batch reruns query missing workspace_id check in subquery ([#5652](https://github.com/windmill-labs/windmill/issues/5652)) ([444a6ab](https://github.com/windmill-labs/windmill/commit/444a6abad670114c52e44f3606bf6fefc5d3fd98))
* **frontend:** fix validity check ([#5654](https://github.com/windmill-labs/windmill/issues/5654)) ([c41c1eb](https://github.com/windmill-labs/windmill/commit/c41c1eb587bf22364f1202310a0c64b6040ab968))
* improve MySQL datetime parser timezone handling (WIN-1155) ([#5645](https://github.com/windmill-labs/windmill/issues/5645)) ([5bca8f6](https://github.com/windmill-labs/windmill/commit/5bca8f60e970cc67839edb5dc491685f36cf0499))
* track relative imports in python and ts even if lockfile is provided ([e316dbd](https://github.com/windmill-labs/windmill/commit/e316dbd9bdd5c59e9aaba6a4472bb7d832834e84))
## [1.483.1](https://github.com/windmill-labs/windmill/compare/v1.483.0...v1.483.1) (2025-04-19)
### Bug Fixes
* pin libxml to 0.3.3 ([e5595e4](https://github.com/windmill-labs/windmill/commit/e5595e41b5c87704d814eff95bc00b82195728ba))
## [1.483.0](https://github.com/windmill-labs/windmill/compare/v1.482.1...v1.483.0) (2025-04-19)
### Features
* handle different aws auth resource type ([#5637](https://github.com/windmill-labs/windmill/issues/5637)) ([5b123b0](https://github.com/windmill-labs/windmill/commit/5b123b01a1318208450789b5bcade447a0b331c7))
* oidc support for sqs trigger ([#5614](https://github.com/windmill-labs/windmill/issues/5614)) ([34b307b](https://github.com/windmill-labs/windmill/commit/34b307b2be1f6cf92a81694325f4c333bdd7b055))
### Bug Fixes
* fix click outside popover fullscreen ([#5631](https://github.com/windmill-labs/windmill/issues/5631)) ([0811457](https://github.com/windmill-labs/windmill/commit/081145726a5d4ab81510a7637126a036823a1565))
* improve flow editor step switch performance ([58fa4c8](https://github.com/windmill-labs/windmill/commit/58fa4c80062a5704bbd13ddda1b2f00c7c9e40dd))
* linter in early stop doesn't include flow_input ([#5638](https://github.com/windmill-labs/windmill/issues/5638)) ([6a9bdfd](https://github.com/windmill-labs/windmill/commit/6a9bdfd3bd52ff802b6a71c3ae9504bfe7d0421f))
* output picker output opening doesn't change id ([#5641](https://github.com/windmill-labs/windmill/issues/5641)) ([64c72b6](https://github.com/windmill-labs/windmill/commit/64c72b6fce669e47f04dc620750840857cbe66cf))
## [1.482.1](https://github.com/windmill-labs/windmill/compare/v1.482.0...v1.482.1) (2025-04-16)
### Bug Fixes
* flow editor workspace script test use actual workspace script hash ([24e893b](https://github.com/windmill-labs/windmill/commit/24e893b8c50fafdb41f4b6e1777cb34aceafc466))
* **frontend:** postgres remove selectedTable ([#5386](https://github.com/windmill-labs/windmill/issues/5386)) ([bd7c6a2](https://github.com/windmill-labs/windmill/commit/bd7c6a2a46047de5fe89753decdfdf1f4851ee3f))
* **openapi:** fix openapi def of batch re-run jobs ([cb8731e](https://github.com/windmill-labs/windmill/commit/cb8731e7e37fb6cd052f5dae6fdce46e6ca2409c))
* show workspace color if superadmin and not in workspace + change workspace name when switching workspace ([#5625](https://github.com/windmill-labs/windmill/issues/5625)) ([cc4384f](https://github.com/windmill-labs/windmill/commit/cc4384f48cc89f883237a2082d854d69a7b5dc56))
## [1.482.0](https://github.com/windmill-labs/windmill/compare/v1.481.0...v1.482.0) (2025-04-15)
### Features
* add diff toggle to flow inline scripts ([#5550](https://github.com/windmill-labs/windmill/issues/5550)) ([b3ecde3](https://github.com/windmill-labs/windmill/commit/b3ecde3316252bcd7323de98149786349019ba7e))
* add gcp trigger ([#5501](https://github.com/windmill-labs/windmill/issues/5501)) ([6339775](https://github.com/windmill-labs/windmill/commit/63397754046eed41d32e28d4698db37b4c9b9710))
* add wildcards filter for worker/label/tags ([62f14d1](https://github.com/windmill-labs/windmill/commit/62f14d1cb95e3f1c7de85c46e1c6bb092247656c))
* add windmill context to autocomplete ([#5548](https://github.com/windmill-labs/windmill/issues/5548)) ([b47c151](https://github.com/windmill-labs/windmill/commit/b47c15165f93ca68a58f81cf2b86fc9467155482))
* agent workers v2 using http ([#5588](https://github.com/windmill-labs/windmill/issues/5588)) ([63fa499](https://github.com/windmill-labs/windmill/commit/63fa4990153f33434b49269922f7803d04e407cd))
* Batch re-run ([#5553](https://github.com/windmill-labs/windmill/issues/5553)) ([26b5ea5](https://github.com/windmill-labs/windmill/commit/26b5ea5023a100c57d077910c99a5e5703edf1c1))
* **frontend:** app editor code input component (monaco) ([#5566](https://github.com/windmill-labs/windmill/issues/5566)) ([177e16b](https://github.com/windmill-labs/windmill/commit/177e16bb18eed0d1c454b967aaa59547f61e8d26))
* handle sending selected lines to ai context ([#5527](https://github.com/windmill-labs/windmill/issues/5527)) ([5abdc3e](https://github.com/windmill-labs/windmill/commit/5abdc3e4403b5c604309bd99a24d7a2847a17b9b))
* Implement sending diff to ai ([#5510](https://github.com/windmill-labs/windmill/issues/5510)) ([e118d2c](https://github.com/windmill-labs/windmill/commit/e118d2cd5f9c641884a76229802a5228ef41f1a5))
* make azure a standalone AI provider ([#5558](https://github.com/windmill-labs/windmill/issues/5558)) ([2c5e58c](https://github.com/windmill-labs/windmill/commit/2c5e58cf1ab9225d516540b38d9e4dde482a3a7f))
* migrate to svelte5 + vite6 ([#4813](https://github.com/windmill-labs/windmill/issues/4813)) ([3c99b3f](https://github.com/windmill-labs/windmill/commit/3c99b3fdc7b78b1cdc7d8fb21d999296695f7889))
* **postgres-trigger:** postgres trigger fix circular dependencies and add remove associate resource ([#5606](https://github.com/windmill-labs/windmill/issues/5606)) ([1daeb2f](https://github.com/windmill-labs/windmill/commit/1daeb2f48f3026621b3ffc58e10f048d5911906c))
* **python:** per import requirement pin ([#5520](https://github.com/windmill-labs/windmill/issues/5520)) ([0b6d017](https://github.com/windmill-labs/windmill/commit/0b6d017fedc31e790a76cf29a1adaaf2a72acc61))
* signed s3 objects ([#5593](https://github.com/windmill-labs/windmill/issues/5593)) ([b9e8796](https://github.com/windmill-labs/windmill/commit/b9e879618bc223ce17effde8bb4c5d1df2ad6df5))
### Bug Fixes
* add support for ${} syntax without default in bash ([#5594](https://github.com/windmill-labs/windmill/issues/5594)) ([3950cfd](https://github.com/windmill-labs/windmill/commit/3950cfd7e3297d7f8ec56430d6462f6b67ecd3c2))
* app editor svelte 5 fixes ([#5570](https://github.com/windmill-labs/windmill/issues/5570)) ([b926076](https://github.com/windmill-labs/windmill/commit/b9260769883348ecd5aeb5684f527a8bf0073928))
* binding not working in nested array script arg ([#5585](https://github.com/windmill-labs/windmill/issues/5585)) ([f5d46d5](https://github.com/windmill-labs/windmill/commit/f5d46d5751bc875b7f4da1db06be40571ac55ab8))
* **cli:** properly handle enabled/disabled updates of schedules ([2629458](https://github.com/windmill-labs/windmill/commit/26294584d6c2ca02bbc4fc5f28cb8df6a5fb3790))
* **cli:** wmill-locks improvement ([8d062c4](https://github.com/windmill-labs/windmill/commit/8d062c47ecd9e84a81140d5c59814da9217dd434))
* Dynamic select does not work with tag //native ([#5576](https://github.com/windmill-labs/windmill/issues/5576)) ([1f3e7d9](https://github.com/windmill-labs/windmill/commit/1f3e7d9029051832db6ab1755b3cad38176a9e96)), closes [#5490](https://github.com/windmill-labs/windmill/issues/5490)
* fix list jobs by tag ([0c3cb37](https://github.com/windmill-labs/windmill/commit/0c3cb3700a3fb9b69e396487bd7491dbbd8861c0))
* flow editor svelte 5 issues ([#5567](https://github.com/windmill-labs/windmill/issues/5567)) ([4f6be6e](https://github.com/windmill-labs/windmill/commit/4f6be6ed340e26bf1ed95398a9dc9f1eb41b33dd))
* freeze when clicking script history diff button ([#5581](https://github.com/windmill-labs/windmill/issues/5581)) ([07094b6](https://github.com/windmill-labs/windmill/commit/07094b6aa21f10688b138d2a81d4fd5833f003fc))
* **frontend:** app builder - force json configuration in rich result ([#5565](https://github.com/windmill-labs/windmill/issues/5565)) ([6fae3a5](https://github.com/windmill-labs/windmill/commit/6fae3a566be06dae88ece8ec23f5723cd8f3f2b9))
* **frontend:** load all step jobs ([#5617](https://github.com/windmill-labs/windmill/issues/5617)) ([16bed59](https://github.com/windmill-labs/windmill/commit/16bed593dfd0b735a92d0928df5091547b98ae79))
* **frontend:** prevent deploy popover to show if deploy dropdown is open ([#5542](https://github.com/windmill-labs/windmill/issues/5542)) ([c2180c6](https://github.com/windmill-labs/windmill/commit/c2180c6eb34e14fe2292ff40aa6a99c627698d5e))
* **frontend:** proper each block binding + better app settings reactivity ([#5568](https://github.com/windmill-labs/windmill/issues/5568)) ([4c71af8](https://github.com/windmill-labs/windmill/commit/4c71af8a74627d0ba76917e0dac0ac9e5e984cca))
* improve app image picker UX ([#5589](https://github.com/windmill-labs/windmill/issues/5589)) ([f497a4b](https://github.com/windmill-labs/windmill/commit/f497a4bfae8d1bff097e0c2c9df8381a531dfeb9))
* legacy script gen model selection ([#5574](https://github.com/windmill-labs/windmill/issues/5574)) ([3507925](https://github.com/windmill-labs/windmill/commit/3507925624a43804a3be463b6f7913cea5821384))
* mssql ca_cert deserializing ([#5587](https://github.com/windmill-labs/windmill/issues/5587)) ([b4f8c88](https://github.com/windmill-labs/windmill/commit/b4f8c88c19bd4f844c3ecb53ececc340ee326b0e))
* number input in app multiselect yields NOT_NUMBER ([#5616](https://github.com/windmill-labs/windmill/issues/5616)) ([4aae6ab](https://github.com/windmill-labs/windmill/commit/4aae6ab634280adc1de9abd890100b7c12c89158))
* prevent invalid returned ai completion object errors ([#5564](https://github.com/windmill-labs/windmill/issues/5564)) ([9276c71](https://github.com/windmill-labs/windmill/commit/9276c717a21aaee3241845a9cc00d3fb6bce9eb9))
* Remaining svelte 5 bugs ([#5563](https://github.com/windmill-labs/windmill/issues/5563)) ([6e9ec63](https://github.com/windmill-labs/windmill/commit/6e9ec6323c265a747ef8696865297e6d47abb016))
* tenant id to never be undefined on teams ([#5572](https://github.com/windmill-labs/windmill/issues/5572)) ([102b58a](https://github.com/windmill-labs/windmill/commit/102b58a5f40dde22f15700d4b6c11eb7f3fbf4bb))
* validate saved module before passing to flow module editor ([#5580](https://github.com/windmill-labs/windmill/issues/5580)) ([2eb1a16](https://github.com/windmill-labs/windmill/commit/2eb1a161d15627b440195b65eec54998561f4ef6))
## [1.481.0](https://github.com/windmill-labs/windmill/compare/v1.480.1...v1.481.0) (2025-04-02)

View File

@@ -12,7 +12,7 @@
bind {$ADDRESS}
reverse_proxy /ws/* http://lsp:3001
# reverse_proxy /ws_mp/* http://multiplayer:3002
# reverse_proxy /api/srch/* http://windmill_indexer:8002
# reverse_proxy /api/srch/* http://windmill_indexer:8001
reverse_proxy /* http://windmill_server:8000
# tls /certs/cert.pem /certs/key.pem
}

View File

@@ -25,7 +25,6 @@ FROM node:20-alpine as frontend
# install dependencies
WORKDIR /frontend
COPY ./frontend/package.json ./frontend/package-lock.json ./
COPY ./frontend/scripts/ ./scripts/
RUN npm ci
# Copy all local files into the image.

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n gcp_resource_path,\n subscription_id,\n topic_id,\n workspace_id,\n delivery_type AS \"delivery_type: _\",\n delivery_config AS \"delivery_config: _\",\n subscription_mode AS \"subscription_mode: _\",\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM \n gcp_trigger\n WHERE \n workspace_id = $1\n ",
"query": "\n SELECT\n gcp_resource_path,\n subscription_id,\n topic_id,\n workspace_id,\n delivery_type AS \"delivery_type: _\",\n delivery_config AS \"delivery_config: _\",\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM \n gcp_trigger\n WHERE \n workspace_id = $1\n ",
"describe": {
"columns": [
{
@@ -45,71 +45,56 @@
},
{
"ordinal": 6,
"name": "subscription_mode: _",
"type_info": {
"Custom": {
"name": "gcp_subscription_mode",
"kind": {
"Enum": [
"create_update",
"existing"
]
}
}
}
},
{
"ordinal": 7,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 8,
"ordinal": 7,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 9,
"ordinal": 8,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"ordinal": 9,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 11,
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 12,
"ordinal": 11,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"ordinal": 12,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 14,
"ordinal": 13,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 15,
"ordinal": 14,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 16,
"ordinal": 15,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 17,
"ordinal": 16,
"name": "enabled",
"type_info": "Bool"
}
@@ -132,7 +117,6 @@
false,
false,
false,
false,
true,
true,
false,
@@ -140,5 +124,5 @@
false
]
},
"hash": "c2925a28212265bd9ac8e2d498b3588fc040c7a29ae6d9bab1f05af0b535e2ac"
"hash": "0102308ffa1c0dbfba54d29246535bb81146a4cfae0ec408435570e0813a3bef"
}

View File

@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.pgroups', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "122090a0f89e5248a0a0f199ebd24582fdb302883aebd2da187ac0084e767ea3"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n gcp_resource_path,\n subscription_id,\n topic_id,\n workspace_id,\n delivery_type AS \"delivery_type: _\",\n delivery_config AS \"delivery_config: _\",\n subscription_mode AS \"subscription_mode: _\",\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM\n gcp_trigger\n WHERE\n delivery_type != 'push'::DELIVERY_MODE AND\n enabled IS TRUE\n AND (last_server_ping IS NULL OR\n last_server_ping < now() - interval '15 seconds'\n )\n ",
"query": "\n SELECT\n gcp_resource_path,\n subscription_id,\n topic_id,\n workspace_id,\n delivery_type AS \"delivery_type: _\",\n delivery_config AS \"delivery_config: _\",\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM\n gcp_trigger\n WHERE\n delivery_type != 'push'::DELIVERY_MODE AND\n enabled IS TRUE\n AND (last_server_ping IS NULL OR\n last_server_ping < now() - interval '15 seconds'\n )\n ",
"describe": {
"columns": [
{
@@ -45,71 +45,56 @@
},
{
"ordinal": 6,
"name": "subscription_mode: _",
"type_info": {
"Custom": {
"name": "gcp_subscription_mode",
"kind": {
"Enum": [
"create_update",
"existing"
]
}
}
}
},
{
"ordinal": 7,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 8,
"ordinal": 7,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 9,
"ordinal": 8,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"ordinal": 9,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 11,
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 12,
"ordinal": 11,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"ordinal": 12,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 14,
"ordinal": 13,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 15,
"ordinal": 14,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 16,
"ordinal": 15,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 17,
"ordinal": 16,
"name": "enabled",
"type_info": "Bool"
}
@@ -130,7 +115,6 @@
false,
false,
false,
false,
true,
true,
false,
@@ -138,5 +122,5 @@
false
]
},
"hash": "2dcff8b16da75740c362015b2293f578a21813f038d75184a5034c37d4daf36e"
"hash": "15fbe481789a7817bf37415fb935f9ed537fd7d3b266d928af4d5d3dd8bb5c18"
}

View File

@@ -1,27 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_session_context($1, $2, $3, $4, $5, $6)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_session_context",
"type_info": "Void"
}
],
"parameters": {
"Left": [
"Bool",
"Text",
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "18ddbb9eb4c0ec03e6170f8c70634546cddaf1029618ed14015fd7b0a7017441"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n workspace_id,\n path,\n url,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled,\n filters AS \"filters: _\",\n initial_messages AS \"initial_messages: _\",\n url_runnable_args AS \"url_runnable_args: _\",\n can_return_message\n FROM \n websocket_trigger\n WHERE \n workspace_id = $1\n ",
"query": "SELECT workspace_id, path, url, script_path, is_flow, edited_by, email, edited_at, server_id, last_server_ping, extra_perms, error, enabled, filters as \"filters: _\", initial_messages as \"initial_messages: _\", url_runnable_args as \"url_runnable_args: _\", can_return_message FROM websocket_trigger\n WHERE workspace_id = $1",
"describe": {
"columns": [
{
@@ -114,5 +114,5 @@
false
]
},
"hash": "8eabeee5527da4aad3f46ad185015262b4304667449f3e1a71efd7477a39f7fc"
"hash": "1aa8ead10f8d994f6685d266fcbd409b0fff43111d9600e64b2348401ed8929d"
}

View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT jsonb_build_object(\n 'kind', jb.kind,\n 'script_path', jb.runnable_path,\n 'latest_schema', COALESCE(\n (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.workspace_id = $1 AND s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),\n (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.workspace_id = $1 AND flow.path = jb.runnable_path AND jb.kind = 'flow')\n ),\n 'schemas', ARRAY(\n SELECT jsonb_build_object(\n 'script_hash', LPAD(TO_HEX(COALESCE(s.hash, f.id)), 16, '0'),\n 'job_ids', ARRAY_AGG(DISTINCT j.id),\n 'schema', (ARRAY_AGG(COALESCE(s.schema, f.schema)))[1]\n ) FROM v2_job j\n LEFT JOIN script s ON s.hash = j.runnable_id AND j.kind = 'script'\n LEFT JOIN flow_version f ON f.id = j.runnable_id AND j.kind = 'flow'\n WHERE j.id = ANY(ARRAY_AGG(jb.id))\n GROUP BY COALESCE(s.hash, f.id)\n )\n ) FROM v2_job jb\n WHERE (jb.kind = 'flow' OR jb.kind = 'script')\n AND jb.workspace_id = $1 AND jb.id = ANY($2)\n GROUP BY jb.kind, jb.runnable_path",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "jsonb_build_object",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"UuidArray"
]
},
"nullable": [
null
]
},
"hash": "26761fbd7953416eb391de47b1694e0f4ab2bb96a6d838f1b1fdce4b58a8f5d4"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "WITH inserted_job AS (\n INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $21, $22, $23, $24, $25, $26,\n CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)\n ),\n inserted_runtime AS (\n INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)\n ),\n inserted_job_perms AS (\n INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n values ($1, $32, $33, $34, $35, $36, $37, $2) \n ON CONFLICT (job_id) DO UPDATE SET email = $32, username = $33, is_admin = $34, is_operator = $35, folders = $36, groups = $37, workspace_id = $2\n )\n INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 THEN now() END, $30, $31)",
"query": "INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $21, $22, $23, $24, $25, $26,\n CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)",
"describe": {
"columns": [],
"parameters": {
@@ -86,20 +86,10 @@
"Varchar",
"Int4",
"Int2",
"Bool",
"Bool",
"Timestamptz",
"Varchar",
"Int2",
"Varchar",
"Varchar",
"Bool",
"Bool",
"JsonbArray",
"TextArray"
"Bool"
]
},
"nullable": []
},
"hash": "cccdcb7fe7968eadfc04d8957a8e98b2f2d92a6d7f687a9dd5a70edb3d5a63e6"
"hash": "29624682d687790dd199c4af759132d79fdb2982de111cb5fd43e3d9ecd0f15e"
}

View File

@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.folders_read', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "29fbc3a8c35845a997cd548ed417b9cc3c82d815d99b3d435adcfbb5a9246124"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO app_version\n (app_id, value, created_by, raw_app)\n VALUES ($1, $2::text::json, $3, $4) RETURNING id",
"query": "INSERT INTO app_version\n (app_id, value, created_by)\n VALUES ($1, $2::text::json, $3) RETURNING id",
"describe": {
"columns": [
{
@@ -13,13 +13,12 @@
"Left": [
"Int8",
"Text",
"Varchar",
"Bool"
"Varchar"
]
},
"nullable": [
false
]
},
"hash": "56264b88a9f428e79c6e531a0011e0dbed634f3b12dedaa5fee0362d670887f6"
"hash": "2a3992b5e9abcfbb032d10e142d98efa969dae26a7242eb7ac12593ed5421ef3"
}

View File

@@ -1,95 +1,80 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n aws_auth_resource_type AS \"aws_auth_resource_type: _\",\n aws_resource_path,\n message_attributes,\n queue_url,\n workspace_id,\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM \n sqs_trigger\n WHERE \n workspace_id = $1\n ",
"query": "SELECT * FROM sqs_trigger\n WHERE workspace_id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "aws_auth_resource_type: _",
"type_info": {
"Custom": {
"name": "aws_auth_resource_type",
"kind": {
"Enum": [
"oidc",
"credentials"
]
}
}
}
},
{
"ordinal": 1,
"name": "aws_resource_path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "message_attributes",
"type_info": "TextArray"
},
{
"ordinal": 3,
"name": "queue_url",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 6,
"ordinal": 1,
"name": "queue_url",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "aws_resource_path",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "message_attributes",
"type_info": "TextArray"
},
{
"ordinal": 4,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"ordinal": 5,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 8,
"ordinal": 6,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 9,
"ordinal": 8,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 10,
"ordinal": 9,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 11,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 12,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"ordinal": 10,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 14,
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 15,
"ordinal": 12,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 14,
"name": "enabled",
"type_info": "Bool"
}
@@ -100,6 +85,7 @@
]
},
"nullable": [
false,
false,
false,
true,
@@ -109,8 +95,6 @@
false,
false,
false,
false,
false,
true,
true,
true,
@@ -118,5 +102,5 @@
false
]
},
"hash": "a7df493316f632fc636e4c3c90bef4c98035b5dc808bb8379251c0b35d945ba0"
"hash": "2b6c13191484b0c664f35e2c811082b00c44fd4a7e98a11b714133674a7b6da7"
}

View File

@@ -1,95 +1,80 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n aws_auth_resource_type AS \"aws_auth_resource_type: _\",\n aws_resource_path,\n message_attributes,\n queue_url,\n workspace_id,\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM \n sqs_trigger\n WHERE \n workspace_id = $1 AND \n path = $2\n ",
"query": "\n SELECT\n aws_resource_path,\n message_attributes,\n queue_url,\n workspace_id,\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM \n sqs_trigger\n WHERE \n workspace_id = $1 AND \n path = $2\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "aws_auth_resource_type: _",
"type_info": {
"Custom": {
"name": "aws_auth_resource_type",
"kind": {
"Enum": [
"oidc",
"credentials"
]
}
}
}
},
{
"ordinal": 1,
"name": "aws_resource_path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"ordinal": 1,
"name": "message_attributes",
"type_info": "TextArray"
},
{
"ordinal": 3,
"ordinal": 2,
"name": "queue_url",
"type_info": "Varchar"
},
{
"ordinal": 4,
"ordinal": 3,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 5,
"ordinal": 4,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 6,
"ordinal": 5,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"ordinal": 6,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 8,
"ordinal": 7,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 9,
"ordinal": 8,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 10,
"ordinal": 9,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 11,
"ordinal": 10,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 12,
"ordinal": 11,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"ordinal": 12,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 14,
"ordinal": 13,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 15,
"ordinal": 14,
"name": "enabled",
"type_info": "Bool"
}
@@ -101,7 +86,6 @@
]
},
"nullable": [
false,
false,
true,
false,
@@ -119,5 +103,5 @@
false
]
},
"hash": "5237f9d3f57f4e799968a4a9630c984775d23c7a4dc1a7ad37528a46ab89d6bd"
"hash": "2ef82fad8a6ccdc66228cfbce5393de351653ab9ac171fa0eea447c905440867"
}

View File

@@ -0,0 +1,27 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority)\n VALUES ($1, $2, $3, COALESCE($4, now()), CASE WHEN $3 THEN now() END, $5, $6) RETURNING id AS \"id!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Varchar",
"Uuid",
"Bool",
"Timestamptz",
"Varchar",
"Int2"
]
},
"nullable": [
false
]
},
"hash": "31df83e9eb6078e93ec5fe4168306caccb849db9e0f71d86da655b01c6a3e8d0"
}

View File

@@ -0,0 +1,24 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS (\n SELECT 1 FROM v2_as_completed_job \n WHERE workspace_id = $2 \n AND (job_kind = 'appscript' OR job_kind = 'preview')\n AND created_by = 'anonymous' \n AND started_at > now() - interval '3 hours'\n AND script_path LIKE $3 || '/%' \n AND result @> ('{\"s3\":\"' || $1 || '\"}')::jsonb \n )",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "43aa468aac174529a74e6108af55a383f8a20e98b8c502929f4dc5041a55e72f"
}

View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT content FROM script WHERE path = $1 AND workspace_id = $2\n AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND\n workspace_id = $2)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "content",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "443bd83bcea1d37c79cb080095343c98104529879f991c49585cd181e34aa827"
}

View File

@@ -1,12 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM metrics WHERE created_at < NOW() - INTERVAL '180 day'",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "45dde0dc2da12fe46b5975ed53b520fc4160c008c3d29fe7d99e609c0b6e3e6d"
}

View File

@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH worker_ids AS (SELECT unnest($1::text[]) as worker)\n SELECT worker_ids.worker FROM worker_ids\n LEFT JOIN worker_ping ON worker_ids.worker = worker_ping.worker\n WHERE worker_ping.worker IS NULL OR worker_ping.ping_at < now() - ('60 seconds')::interval",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "worker",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"TextArray"
]
},
"nullable": [
null
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
}

View File

@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.groups', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "6199e8be5cb13db71108e555ea20f0b76dc38476670f9fc0667b057d2766d42e"
}

View File

@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.user', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "6c90d5ea2a09b47b81fdab465062c25f8768b220e53bf469550a3a3697ab756a"
}

View File

@@ -1,34 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO sqs_trigger (\n aws_auth_resource_type,\n aws_resource_path,\n queue_url,\n message_attributes,\n workspace_id, \n path, \n script_path, \n is_flow, \n email, \n enabled, \n edited_by\n ) \n VALUES (\n $1, \n $2, \n $3, \n $4, \n $5, \n $6, \n $7,\n $8,\n $9,\n $10,\n $11\n )",
"describe": {
"columns": [],
"parameters": {
"Left": [
{
"Custom": {
"name": "aws_auth_resource_type",
"kind": {
"Enum": [
"oidc",
"credentials"
]
}
}
},
"Varchar",
"Varchar",
"TextArray",
"Varchar",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Bool",
"Varchar"
]
},
"nullable": []
},
"hash": "776909d3452aaf9e0ed41a5ec314b7bfc5bc8b6a76c98bf4daf7d715d18c52e6"
}

View File

@@ -0,0 +1,21 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n values ($1, $2, $3, $4, $5, $6, $7, $8) \n ON CONFLICT (job_id) DO UPDATE SET email = $2, username = $3, is_admin = $4, is_operator = $5, folders = $6, groups = $7, workspace_id = $8",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Varchar",
"Varchar",
"Bool",
"Bool",
"JsonbArray",
"TextArray",
"Varchar"
]
},
"nullable": []
},
"hash": "7d78efab0a588f56a13a7b5251f0a72f5d341b053218e7aec83a834cf7ccc98f"
}

View File

@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n active_pid \n FROM \n pg_replication_slots \n WHERE \n slot_name = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "active_pid",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Name"
]
},
"nullable": [
true
]
},
"hash": "7e64ba7e2362cc19d2aed9f34c9879983922e96a9baab7c1a2b09ed2b1c261e2"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO app_version\n (app_id, value, created_by, raw_app)\n VALUES ($1, $2::text::json, $3, $4) RETURNING id",
"query": "INSERT INTO app_version\n (app_id, value, created_by)\n VALUES ($1, $2::text::json, $3) RETURNING id",
"describe": {
"columns": [
{
@@ -13,13 +13,12 @@
"Left": [
"Int8",
"Text",
"Varchar",
"Bool"
"Varchar"
]
},
"nullable": [
false
]
},
"hash": "a9f6e5720f748418ae52cf0e31778b9bbef26008803f2dd2a043bdb091b2ff20"
"hash": "83d18f4e1cda2c1867168551d855d5626e766b5469c6976c986bdafc8b9407c5"
}

View File

@@ -1,34 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n sqs_trigger \n SET \n aws_auth_resource_type = $1,\n aws_resource_path = $2,\n queue_url = $3,\n message_attributes = $4, \n is_flow = $5, \n edited_by = $6, \n email = $7,\n script_path = $8,\n path = $9,\n edited_at = now(), \n error = NULL,\n server_id = NULL\n WHERE \n workspace_id = $10 AND \n path = $11\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
{
"Custom": {
"name": "aws_auth_resource_type",
"kind": {
"Enum": [
"oidc",
"credentials"
]
}
}
},
"Varchar",
"Varchar",
"TextArray",
"Bool",
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "8490024b96aa689d3c1bc5cbf94fcd8c5491732818404fee2397f00cccee0ad3"
}

View File

@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.folders_write', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "9897705b7c265333559df757a725711a40924ec551a7589a64ee96f8aa7f6a21"
}

View File

@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO sqs_trigger (\n aws_resource_path,\n queue_url,\n message_attributes,\n workspace_id, \n path, \n script_path, \n is_flow, \n email, \n enabled, \n edited_by\n ) \n VALUES (\n $1, \n $2, \n $3, \n $4, \n $5, \n $6, \n $7,\n $8,\n $9,\n $10\n )",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"TextArray",
"Varchar",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Bool",
"Varchar"
]
},
"nullable": []
},
"hash": "9b9bc21023f10a0b4bb45879c7da8e119bffd2982b97fc900358506a0a14bbb8"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "WITH to_update AS (\n SELECT q.id, q.workspace_id, r.ping, COALESCE(zjc.counter, 0) as counter\n FROM v2_job_queue q\n JOIN v2_job j ON j.id = q.id\n JOIN v2_job_runtime r ON r.id = j.id\n LEFT JOIN zombie_job_counter zjc ON zjc.job_id = q.id\n WHERE ping < now() - ($1 || ' seconds')::interval\n AND running = true\n AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow')\n AND same_worker = false\n AND (zjc.counter IS NULL OR zjc.counter <= $2)\n FOR UPDATE of q SKIP LOCKED\n ),\n zombie_jobs AS (\n UPDATE v2_job_queue q\n SET running = false, started_at = null\n FROM to_update tu\n WHERE q.id = tu.id AND (tu.counter IS NULL OR tu.counter < $2)\n RETURNING q.id, q.workspace_id, ping, tu.counter\n ),\n update_ping AS (\n UPDATE v2_job_runtime r\n SET ping = null\n FROM zombie_jobs zj\n WHERE r.id = zj.id\n ),\n increment_counter AS (\n INSERT INTO zombie_job_counter (job_id, counter)\n SELECT id, 1 FROM to_update WHERE counter < $2\n ON CONFLICT (job_id) DO UPDATE\n SET counter = zombie_job_counter.counter + 1\n ),\n update_concurrency AS (\n UPDATE concurrency_counter cc\n SET job_uuids = job_uuids - zj.id::text\n FROM zombie_jobs zj\n INNER JOIN concurrency_key ck ON ck.job_id = zj.id\n WHERE cc.concurrency_id = ck.key\n )\n SELECT id AS \"id!\", workspace_id AS \"workspace_id!\", ping, counter + 1 AS counter FROM to_update",
"query": "WITH to_update AS (\n SELECT q.id, q.workspace_id, r.ping, COALESCE(zjc.counter, 0) as counter\n FROM v2_job_queue q\n JOIN v2_job j ON j.id = q.id\n JOIN v2_job_runtime r ON r.id = j.id\n LEFT JOIN zombie_job_counter zjc ON zjc.job_id = q.id\n WHERE ping < now() - ($1 || ' seconds')::interval\n AND running = true\n AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow')\n AND same_worker = false\n AND (zjc.counter IS NULL OR zjc.counter <= $2)\n FOR UPDATE of q SKIP LOCKED\n ),\n zombie_jobs AS (\n UPDATE v2_job_queue q\n SET running = false, started_at = null\n FROM to_update tu\n WHERE q.id = tu.id AND (tu.counter IS NULL OR tu.counter < $2)\n RETURNING q.id, q.workspace_id, ping, tu.counter\n ),\n update_ping AS (\n UPDATE v2_job_runtime r\n SET ping = null\n FROM zombie_jobs zj\n WHERE r.id = zj.id\n ),\n increment_counter AS (\n INSERT INTO zombie_job_counter (job_id, counter)\n SELECT id, 1 FROM to_update WHERE counter < $2\n ON CONFLICT (job_id) DO UPDATE \n SET counter = zombie_job_counter.counter + 1\n ),\n update_concurrency AS (\n UPDATE concurrency_counter cc\n SET job_uuids = job_uuids - zj.id::text\n FROM zombie_jobs zj\n INNER JOIN concurrency_key ck ON ck.job_id = zj.id\n WHERE cc.concurrency_id = ck.key\n )\n SELECT id AS \"id!\", workspace_id AS \"workspace_id!\", ping, counter + 1 AS counter FROM to_update",
"describe": {
"columns": [
{
@@ -37,5 +37,5 @@
null
]
},
"hash": "daf9674838fb3e3653a356c7434c719616a614d77e726433737e5f5d9bd60134"
"hash": "abfc6b530565f125bd0b8ac99cd67fd22f14f8fce89e59890ba21e87fe323af5"
}

View File

@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT content FROM script WHERE path = $1 AND workspace_id = $2\n AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND\n workspace_id = $2)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "content",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "ac01e45d3335015f53f3d63fe159e631efb65c3d326b6b6ae8361a2116bff145"
}

View File

@@ -1,85 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n j.id,\n j.kind AS \"kind: _\",\n COALESCE(s.path, f.path) AS \"script_path!\",\n COALESCE(s.hash, f.id) AS \"script_hash!: _\",\n COALESCE(jc.started_at, jq.scheduled_for, make_date(1970, 1, 1)) AS \"scheduled_for!: _\",\n args AS input,\n COALESCE(s.schema, f.schema) AS \"schema: _\"\n FROM v2_job j\n LEFT JOIN script s ON j.runnable_id = s.hash AND j.kind = 'script'\n LEFT JOIN flow_version f ON j.runnable_id = f.id AND j.runnable_path = f.path AND j.kind = 'flow'\n LEFT JOIN v2_job_completed jc ON jc.id = j.id\n LEFT JOIN v2_job_queue jq ON jq.id = j.id\n WHERE j.id = ANY($1)\n AND j.workspace_id = $2\n AND COALESCE(s.hash, f.id) IS NOT NULL\n AND COALESCE(s.path, f.path) IS NOT NULL",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "kind: _",
"type_info": {
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode",
"appscript"
]
}
}
}
},
{
"ordinal": 2,
"name": "script_path!",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "script_hash!: _",
"type_info": "Int8"
},
{
"ordinal": 4,
"name": "scheduled_for!: _",
"type_info": "Timestamptz"
},
{
"ordinal": 5,
"name": "input",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "schema: _",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"UuidArray",
"Text"
]
},
"nullable": [
false,
false,
null,
null,
null,
true,
null
]
},
"hash": "ad88831c9021b79c9037a925c2b53be3a93dbe2a67bdc4d4342c84f439b5eaf9"
}

View File

@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": []
},
"hash": "ca8bf3dda133556ee7051d5436a5d76ae2223d59e9f321a8b6d0c27adc09f741"
}

View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE(\n (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),\n (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = jb.runnable_path AND jb.kind = 'flow')\n ) FROM v2_job jb\n WHERE jb.id = $1 AND jb.workspace_id = $2\n GROUP BY jb.kind, jb.runnable_path",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "coalesce",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
null
]
},
"hash": "d1fdfabd4f3bbe93cf9667e8e76bd8f537e68fabceb094fe535693fbd146eff2"
}

View File

@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "WITH worker_ids AS (SELECT unnest($1::text[]) as worker) \n SELECT worker_ids.worker FROM worker_ids \n LEFT JOIN worker_ping ON worker_ids.worker = worker_ping.worker \n WHERE worker_ping.worker IS NULL OR worker_ping.ping_at < now() - ('60 seconds')::interval",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "worker",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"TextArray"
]
},
"nullable": [
true
]
},
"hash": "ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927"
}

View File

@@ -1,95 +1,80 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n aws_auth_resource_type AS \"aws_auth_resource_type: _\",\n queue_url,\n aws_resource_path,\n message_attributes,\n workspace_id,\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM\n sqs_trigger\n WHERE\n enabled IS TRUE\n AND (last_server_ping IS NULL OR\n last_server_ping < now() - interval '15 seconds'\n )\n ",
"query": "\n SELECT\n queue_url,\n aws_resource_path,\n message_attributes,\n workspace_id,\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM\n sqs_trigger\n WHERE\n enabled IS TRUE\n AND (last_server_ping IS NULL OR\n last_server_ping < now() - interval '15 seconds'\n )\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "aws_auth_resource_type: _",
"type_info": {
"Custom": {
"name": "aws_auth_resource_type",
"kind": {
"Enum": [
"oidc",
"credentials"
]
}
}
}
},
{
"ordinal": 1,
"name": "queue_url",
"type_info": "Varchar"
},
{
"ordinal": 2,
"ordinal": 1,
"name": "aws_resource_path",
"type_info": "Varchar"
},
{
"ordinal": 3,
"ordinal": 2,
"name": "message_attributes",
"type_info": "TextArray"
},
{
"ordinal": 4,
"ordinal": 3,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 5,
"ordinal": 4,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 6,
"ordinal": 5,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"ordinal": 6,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 8,
"ordinal": 7,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 9,
"ordinal": 8,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 10,
"ordinal": 9,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 11,
"ordinal": 10,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 12,
"ordinal": 11,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"ordinal": 12,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 14,
"ordinal": 13,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 15,
"ordinal": 14,
"name": "enabled",
"type_info": "Bool"
}
@@ -98,7 +83,6 @@
"Left": []
},
"nullable": [
false,
false,
false,
true,
@@ -116,5 +100,5 @@
false
]
},
"hash": "bc0bedddcafad216c30c5061eef3f4ed8573cda89f655d6490d22f182e3f2f36"
"hash": "e6adaebcade2e25be800e8b888b23c94caae2421f3cae7c06d6346bd6de1d94a"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT worker, array_agg(v2_job_queue.id) as ids FROM v2_job_queue LEFT JOIN v2_job ON v2_job_queue.id = v2_job.id LEFT JOIN v2_job_runtime ON v2_job_queue.id = v2_job_runtime.id WHERE v2_job_queue.created_at < now() - ('60 seconds')::interval\n AND running = true AND ping IS NULL AND same_worker = true AND worker IS NOT NULL GROUP BY worker",
"query": "SELECT worker, array_agg(v2_job_queue.id) as ids FROM v2_job_queue LEFT JOIN v2_job ON v2_job_queue.id = v2_job.id LEFT JOIN v2_job_runtime ON v2_job_queue.id = v2_job_runtime.id WHERE v2_job_queue.created_at < now() - ('60 seconds')::interval \n AND running = true AND ping IS NULL AND same_worker = true AND worker IS NOT NULL GROUP BY worker",
"describe": {
"columns": [
{
@@ -22,5 +22,5 @@
null
]
},
"hash": "3e261911cc4c5289da49865f54350613f9e651540a279bd7d75e5e7d79f676a8"
"hash": "ebd25047d14bec8457465ba6ed9b6b15b8f2303157b4c6e80ac9e3d84a57d8b1"
}

View File

@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS (\n SELECT 1 FROM v2_as_completed_job\n WHERE workspace_id = $2\n AND (job_kind = 'appscript' OR job_kind = 'preview')\n AND created_by = 'anonymous'\n AND started_at > now() - interval '3 hours'\n AND script_path LIKE $3 || '/%'\n AND result @> ('{\"s3\":\"' || $1 || '\"}')::jsonb\n )",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "ec9980f80a8bfa4b09225035e8a1f78d7d61fbf83dda9b39bb22e0f9584d221b"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n gcp_resource_path,\n subscription_id,\n topic_id,\n workspace_id,\n delivery_type AS \"delivery_type: _\",\n delivery_config AS \"delivery_config: _\",\n subscription_mode AS \"subscription_mode: _\",\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM \n gcp_trigger\n WHERE \n workspace_id = $1 AND \n path = $2\n ",
"query": "\n SELECT\n gcp_resource_path,\n subscription_id,\n topic_id,\n workspace_id,\n delivery_type AS \"delivery_type: _\",\n delivery_config AS \"delivery_config: _\",\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled\n FROM \n gcp_trigger\n WHERE \n workspace_id = $1 AND \n path = $2\n ",
"describe": {
"columns": [
{
@@ -45,71 +45,56 @@
},
{
"ordinal": 6,
"name": "subscription_mode: _",
"type_info": {
"Custom": {
"name": "gcp_subscription_mode",
"kind": {
"Enum": [
"create_update",
"existing"
]
}
}
}
},
{
"ordinal": 7,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 8,
"ordinal": 7,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 9,
"ordinal": 8,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"ordinal": 9,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 11,
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 12,
"ordinal": 11,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"ordinal": 12,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 14,
"ordinal": 13,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 15,
"ordinal": 14,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 16,
"ordinal": 15,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 17,
"ordinal": 16,
"name": "enabled",
"type_info": "Bool"
}
@@ -133,7 +118,6 @@
false,
false,
false,
false,
true,
true,
false,
@@ -141,5 +125,5 @@
false
]
},
"hash": "1312b7fd622cc814a406c85dbbff61f003c29185c267642cfd898075ebda855d"
"hash": "f4f6336fc671b00bed7835124892f7a4d3bbe673f7c48153819dab385a5cb357"
}

View File

@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n sqs_trigger \n SET \n aws_resource_path = $1,\n queue_url = $2,\n message_attributes = $3, \n is_flow = $4, \n edited_by = $5, \n email = $6,\n script_path = $7,\n path = $8,\n edited_at = now(), \n error = NULL,\n server_id = NULL\n WHERE \n workspace_id = $9 AND \n path = $10\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"TextArray",
"Bool",
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "f68d23841e0e31cdf8633aaf0f32777e04e4965682e5b37afbe84194b756d5f5"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n DELETE\n FROM parallel_monitor_lock\n WHERE last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval\n RETURNING parent_flow_id, job_id, last_ping, (SELECT workspace_id FROM v2_job_queue q\n WHERE q.id = parent_flow_id AND q.running = true AND q.canceled_by IS NULL\n ) AS workspace_id\n ",
"query": "\n DELETE\n FROM parallel_monitor_lock\n WHERE last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval \n RETURNING parent_flow_id, job_id, last_ping, (SELECT workspace_id FROM v2_job_queue q\n WHERE q.id = parent_flow_id AND q.running = true AND q.canceled_by IS NULL\n ) AS workspace_id\n ",
"describe": {
"columns": [
{
@@ -36,5 +36,5 @@
null
]
},
"hash": "00c4a602aa6a50f2f922851ce63b5216e915c7649698687a00d47da55c70349f"
"hash": "f822df86bf511fe0d3044b69525a6ff2524167877929886adca9b0fa3d3fee45"
}

View File

@@ -1,21 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8",
"describe": {
"columns": [],
"parameters": {
"Left": [
"UuidArray",
"Varchar",
"Varchar",
"Bool",
"Bool",
"JsonbArray",
"TextArray",
"Varchar"
]
},
"nullable": []
},
"hash": "fe7221651a982861dede4116bc71fe2dce615ff76a53f72cb5386dc17e4e07aa"
}

View File

@@ -11,5 +11,5 @@
"remote.autoForwardPorts": true,
"conventionalCommits.scopes": [
"restructring triggers, decoding trigger message on work"
]
],
}

845
backend/Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -1,6 +1,6 @@
[package]
name = "windmill"
version = "1.485.3"
version = "1.481.0"
authors.workspace = true
edition.workspace = true
@@ -32,7 +32,7 @@ members = [
]
[workspace.package]
version = "1.485.3"
version = "1.481.0"
authors = ["Ruben Fiszel <ruben@windmill.dev>"]
edition = "2021"
@@ -64,7 +64,6 @@ cloud = ["windmill-queue/cloud", "windmill-worker/cloud", "windmill-common/cloud
jemalloc = ["windmill-common/jemalloc", "dep:tikv-jemallocator", "dep:tikv-jemalloc-sys", "dep:tikv-jemalloc-ctl"]
tantivy = ["dep:windmill-indexer", "windmill-api/tantivy", "windmill-indexer/enterprise", "windmill-indexer/parquet", "windmill-common/tantivy", "enterprise", "parquet"]
sqlx = ["windmill-worker/sqlx"]
deno_core = ["windmill-worker/deno_core", "windmill-api/deno_core", "dep:deno_core", "dep:v8"]
kafka = ["windmill-api/kafka"]
nats = ["windmill-api/nats"]
otel = ["windmill-common/otel", "windmill-worker/otel"]
@@ -72,9 +71,8 @@ dind = ["windmill-worker/dind"]
websocket = ["windmill-api/websocket"]
http_trigger = ["windmill-api/http_trigger"]
postgres_trigger = ["windmill-api/postgres_trigger"]
mcp = ["windmill-api/mcp"]
mqtt_trigger = ["windmill-api/mqtt_trigger"]
sqs_trigger = ["windmill-api/sqs_trigger", "windmill-common/aws_auth", "windmill-api/openidconnect"]
sqs_trigger = ["windmill-api/sqs_trigger"]
gcp_trigger = ["windmill-api/gcp_trigger"]
smtp = ["windmill-api/smtp", "windmill-common/smtp"]
license = ["windmill-api/license"]
@@ -84,6 +82,7 @@ static_frontend = ["windmill-api/static_frontend"]
scoped_cache = ["windmill-common/scoped_cache"]
# Languages
python = ["windmill-worker/python"]
deno_core = ["windmill-worker/deno_core", "dep:deno_core", "dep:v8"]
rust = ["windmill-worker/rust"]
mysql = ["windmill-worker/mysql"]
oracledb = ["windmill-worker/oracledb"]
@@ -99,7 +98,6 @@ all_languages = [ "python", "deno_core", "rust", "mysql", "oracledb", "mssql", "
[dependencies]
anyhow.workspace = true
tokio.workspace = true
tokio-stream.workspace = true
dotenv.workspace = true
windmill-queue.workspace = true
windmill-common = { workspace = true, default-features = false }
@@ -183,7 +181,6 @@ axum = { version = "^0.7", features = ["multipart"] }
headers = "^0"
hyper = { version = "^1", features = ["full"] }
tokio = { version = "^1.42.0", features = ["full", "tracing", "time"] }
tokio-stream = { version = "0.1.17" }
tower = "^0"
tower-http = { version = "^0.6", features = ["trace", "cors"] }
tower-cookies = "^0.10"
@@ -305,7 +302,6 @@ postgres-native-tls = "^0"
native-tls = "^0"
# samael will break compilation on MacOS. Use this fork instead to make it work
# samael = { git="https://github.com/njaremko/samael", rev="464d015e3ae393e4b5dd00b4d6baa1b617de0dd6", features = ["xmlsec"] }
libxml = { version = "=0.3.3" }
samael = { version="0.0.14", features = ["xmlsec"] }
gcp_auth = "0.9.0"
rust_decimal = { version = "^1", features = ["db-postgres", "serde-float"]}
@@ -369,8 +365,6 @@ tantivy = "0.22.0"
backon = "1.3.0"
flume = { version = "0.11.1", features = ["async"] }
# Macro-related
proc-macro2 = "1.0"
pulldown-cmark = "0.9"

View File

@@ -1 +1 @@
96a2129f585a1bc6567ef08bd27aaa4aca70543f
85c37983ffb8f622458425c182613206625c6cee

View File

@@ -1,2 +0,0 @@
-- Add down migration script here
ALTER TABLE app_version DROP COLUMN IF EXISTS raw_app;

View File

@@ -1,2 +0,0 @@
-- Add up migration script here
ALTER TABLE app_version ADD COLUMN IF NOT EXISTS raw_app BOOLEAN NOT NULL DEFAULT FALSE;

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1,22 +0,0 @@
-- Add up migration script here
CREATE OR REPLACE FUNCTION set_session_context(
admin BOOLEAN,
username TEXT,
groups TEXT,
pgroups TEXT,
folders_read TEXT,
folders_write TEXT
) RETURNS void AS $$
BEGIN
IF admin THEN
SET LOCAL ROLE windmill_admin;
ELSE
SET LOCAL ROLE windmill_user;
END IF;
PERFORM set_config('session.user', username, true);
PERFORM set_config('session.groups', groups, true);
PERFORM set_config('session.pgroups', pgroups, true);
PERFORM set_config('session.folders_read', folders_read, true);
PERFORM set_config('session.folders_write', folders_write, true);
END;
$$ LANGUAGE plpgsql;

View File

@@ -1,3 +0,0 @@
-- Add down migration script here
ALTER TABLE sqs_trigger DROP COLUMN aws_auth_resource_type;
DROP TYPE IF EXISTS AWS_AUTH_RESOURCE_TYPE;

View File

@@ -1,4 +0,0 @@
-- Add up migration script here
CREATE TYPE AWS_AUTH_RESOURCE_TYPE AS ENUM ('oidc', 'credentials');
ALTER TABLE sqs_trigger
ADD COLUMN aws_auth_resource_type AWS_AUTH_RESOURCE_TYPE DEFAULT 'credentials'::AWS_AUTH_RESOURCE_TYPE NOT NULL;

View File

@@ -1,3 +0,0 @@
-- Add down migration script here
ALTER TABLE gcp_trigger DROP COLUMN subscription_mode;
DROP TYPE GCP_SUBSCRIPTION_MODE;

View File

@@ -1,3 +0,0 @@
-- Add up migration script here
CREATE TYPE GCP_SUBSCRIPTION_MODE AS ENUM ('create_update', 'existing');
ALTER TABLE gcp_trigger ADD COLUMN subscription_mode GCP_SUBSCRIPTION_MODE NOT NULL DEFAULT 'create_update'::GCP_SUBSCRIPTION_MODE;

View File

@@ -1,3 +0,0 @@
-- Add down migration script here
ALTER TABLE websocket_trigger
ALTER COLUMN url TYPE VARCHAR(255);

View File

@@ -1,3 +0,0 @@
-- Add up migration script here
ALTER TABLE websocket_trigger
ALTER COLUMN url TYPE VARCHAR(1000);

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1,2 +0,0 @@
-- Add up migration script here
ALTER TYPE log_mode ADD VALUE 'mcp';

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1,2 +0,0 @@
-- Add up migration script here
CREATE INDEX IF NOT EXISTS idx_metrics_id_created_at ON public.metrics (id, created_at DESC) WHERE id LIKE 'queue_%';

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1,2 +0,0 @@
-- Add up migration script here
CREATE INDEX IF NOT EXISTS job_stats_id ON job_stats (job_id);

View File

@@ -11,7 +11,6 @@ mod mapping;
use async_recursion::async_recursion;
use itertools::Itertools;
use lazy_static::lazy_static;
use std::collections::HashMap;
use mapping::{FULL_IMPORTS_MAP, SHORT_IMPORTS_MAP};
#[cfg(not(target_arch = "wasm32"))]
@@ -21,7 +20,6 @@ use regex_lite::Regex;
use rustpython_parser::{
ast::{Stmt, StmtImport, StmtImportFrom, Suite},
text_size::TextRange,
Parse,
};
use sqlx::{Pool, Postgres};
@@ -43,11 +41,9 @@ fn replace_full_import(x: &str) -> Option<String> {
lazy_static! {
static ref RE: Regex = Regex::new(r"^\#\s?(\S+)\s*$").unwrap();
static ref PIN_RE: Regex = Regex::new(r"(?:\s*#\s*(pin|repin):\s*)(\S*)").unwrap();
static ref PKG_RE: Regex = Regex::new(r"^([^!=<>]+)(?:[!=<>]|$)").unwrap();
}
fn process_import(module: Option<String>, path: &str, level: usize) -> Vec<NImport> {
fn process_import(module: Option<String>, path: &str, level: usize) -> Vec<String> {
if level > 0 {
let mut imports = vec![];
let splitted_path = path.split("/");
@@ -56,18 +52,17 @@ fn process_import(module: Option<String>, path: &str, level: usize) -> Vec<NImpo
.take(splitted_path.count() - level)
.join("/");
if let Some(m) = module {
imports.push(NImport::Relative(format!("{base}/{}", m.replace(".", "/"))));
imports.push(format!("relative:{base}/{}", m.replace(".", "/")));
} else {
imports.push(NImport::Relative(format!("{base}")));
imports.push(format!("relative:{base}"));
}
imports
} else if let Some(module) = module {
let imprt = module.split('.').next().unwrap_or("").replace("_", "-");
if imprt == "u" || imprt == "f" {
vec![NImport::Relative(module.replace(".", "/"))]
vec![format!("relative:{}", module.replace(".", "/"))]
} else {
let pkg = replace_full_import(&module).unwrap_or(replace_import(imprt));
vec![NImport::Auto { key: if module == pkg { None } else { Some(module) }, pkg }]
vec![replace_full_import(&module).unwrap_or(replace_import(imprt))]
}
} else {
vec![]
@@ -78,68 +73,17 @@ pub fn parse_relative_imports(code: &str, path: &str) -> error::Result<Vec<Strin
let nimports = parse_code_for_imports(code, path)?;
return Ok(nimports
.into_iter()
.filter_map(|x| match x {
NImport::Relative(path) => Some(path),
_ => None,
.filter_map(|x| {
if x.starts_with("relative:") {
Some(x.replace("relative:", ""))
} else {
None
}
})
.collect());
}
#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
enum NImport {
// Order matters! First we want to resolve all repins
// manually repinned requirement
// e.g.:
// import pandas # repin: pandas==x.y.z
Repin {
pin: ImportPin,
key: String,
},
// manually pinned requirements
// e.g.:
// import pandas # pin: pandas>=x.y.z
// import pandas # pin: pandas<=x.y.z
//
// NOTE: It is possible for multiple pins exist on same import
// That's why we store vector of pins
Pin {
pins: Vec<ImportPin>,
key: String,
},
// Automatically inferred requirement
// e.g.:
// import pandas
Auto {
// Take `x.y.z` for example
// x is going to be the `root`
// and x.y.z is `full`
//
// `full` will be None if it is equal to root
//
// We will use `root` as a requirement name and pass to `uv pip compile` if it was not replaced with any pin
pkg: String,
// However we still need full, since all pins pin against full import names
key: Option<String>,
},
// Relative imports
Relative(String),
}
#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
enum NImportResolved {
Repin { pin: ImportPin, key: String },
Pin { pins: Vec<ImportPin>, key: String },
Auto { pkg: String, key: Option<String> },
}
#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
struct ImportPin {
pkg: String,
path: String,
}
fn parse_code_for_imports(code: &str, path: &str) -> error::Result<Vec<NImport>> {
fn parse_code_for_imports(code: &str, path: &str) -> error::Result<Vec<String>> {
let mut code = code.split(DEF_MAIN).next().unwrap_or("").to_string();
// remove main function decorator from end of file if it exists
@@ -160,54 +104,19 @@ fn parse_code_for_imports(code: &str, path: &str) -> error::Result<Vec<NImport>>
let ast = Suite::parse(&code, "main.py").map_err(|e| {
error::Error::ExecutionErr(format!("Error parsing code for imports: {}", e.to_string()))
})?;
let find_pin = |range: TextRange, key: String| {
let hs = code
.chars()
.skip(range.end().to_usize())
.take_while(|e| *e != '\n')
.collect::<String>();
if hs.trim_start().is_empty() {
return None;
}
PIN_RE.captures(&hs).and_then(|x| {
x.get(1).zip(x.get(2)).and_then(|(ty_m, pkg_m)| {
let pkg = pkg_m.as_str().to_owned();
if ty_m.as_str() == "pin" {
Some(vec![NImport::Pin {
pins: vec![ImportPin { pkg, path: path.to_owned() }],
key,
}])
} else if ty_m.as_str() == "repin" {
Some(vec![NImport::Repin {
pin: ImportPin { pkg, path: path.to_owned() },
key,
}])
} else {
None
}
})
})
};
let mut nimports: Vec<NImport> = ast
let nimports: Vec<String> = ast
.into_iter()
.filter_map(|x| match x {
Stmt::Import(StmtImport { names, range }) => names
.get(0)
.and_then(|al| find_pin(range, al.name.to_string()))
.or(Some(
names
.into_iter()
.map(|x| {
let name = x.name.to_string();
process_import(Some(name), path, 0)
})
.flatten()
.collect::<Vec<NImport>>(),
)),
Stmt::Import(StmtImport { names, .. }) => Some(
names
.into_iter()
.map(|x| {
let name = x.name.to_string();
process_import(Some(name), path, 0)
})
.flatten()
.collect::<Vec<String>>(),
),
Stmt::ImportFrom(StmtImportFrom { level: Some(i), module, .. }) if i.to_u32() > 0 => {
Some(process_import(
module.map(|x| x.to_string()),
@@ -215,25 +124,15 @@ fn parse_code_for_imports(code: &str, path: &str) -> error::Result<Vec<NImport>>
i.to_usize(),
))
}
Stmt::ImportFrom(StmtImportFrom { level: _, module, range, .. }) => find_pin(
range,
module.clone().map(|x| x.to_string()).unwrap_or_default(),
)
.or(Some(process_import(module.map(|x| x.to_string()), path, 0))),
Stmt::ImportFrom(StmtImportFrom { level: _, module, .. }) => {
Some(process_import(module.map(|x| x.to_string()), path, 0))
}
_ => None,
})
.flatten()
.filter(|x| {
if let NImport::Auto { ref pkg, .. } = x {
!STDIMPORTS.contains(&(*pkg).as_str())
} else {
true
}
})
.filter(|x| !STDIMPORTS.contains(&x.as_str()))
.unique()
.collect();
nimports.sort();
return Ok(nimports);
}
@@ -244,9 +143,8 @@ pub async fn parse_python_imports(
db: &Pool<Postgres>,
already_visited: &mut Vec<String>,
annotated_pyv_numeric: &mut Option<u32>,
) -> error::Result<(Vec<String>, Option<String>)> {
let mut compile_error_hint: Option<String> = None;
let mut imports = parse_python_imports_inner(
) -> error::Result<Vec<String>> {
parse_python_imports_inner(
code,
w_id,
path,
@@ -255,46 +153,7 @@ pub async fn parse_python_imports(
annotated_pyv_numeric,
&mut annotated_pyv_numeric.and_then(|_| Some(path.to_owned())),
)
.await?
.into_values()
.map(|nimport| match nimport {
NImportResolved::Pin { pins, .. } => pins.into_iter().map(|p| {
if let Some(hint) = &mut compile_error_hint{
hint.push_str(&format!("\n - pin to {} in {}", p.pkg, p.path));
} else {
compile_error_hint = Some("\n\nMultiple pins can cause problems during lockfile resolution.\nMake sure you checked every pin for conflicts:\n".into())
};
Ok(p.pkg)
}).collect_vec(),
NImportResolved::Repin { pin: ImportPin { pkg, .. }, .. } => vec![Ok(pkg)],
NImportResolved::Auto { pkg, key } => vec![
if let Some(key) = key {
Ok(format!("{pkg} # (mapped from {key})"))
} else {
Ok(pkg)
}
],
})
.flatten()
.collect::<error::Result<Vec<String>>>()?
.into_iter()
.unique()
.collect_vec();
imports.sort();
compile_error_hint
.as_mut()
.map(|e| e.push_str("\n\nNOTE: You can also `repin` to override all pins"));
Ok((imports, compile_error_hint))
}
fn extract_pkg_name(requirement: &str) -> String {
PKG_RE
.captures(requirement)
.map(|x| x.get(1).map(|m| m.as_str().to_string()).unwrap_or_default())
.unwrap_or_default()
.await
}
#[async_recursion]
@@ -306,7 +165,7 @@ async fn parse_python_imports_inner(
already_visited: &mut Vec<String>,
annotated_pyv_numeric: &mut Option<u32>,
path_where_annotated_pyv: &mut Option<String>,
) -> error::Result<HashMap<String, NImportResolved>> {
) -> error::Result<Vec<String>> {
let PythonAnnotations { py310, py311, py312, py313, .. } = PythonAnnotations::parse(&code);
// we pass only if there is none or only one annotation
@@ -335,6 +194,7 @@ async fn parse_python_imports_inner(
} else {
*annotated_pyv_numeric = Some(numeric);
}
*path_where_annotated_pyv = Some(path.to_owned());
}
Ok(())
@@ -349,216 +209,74 @@ async fn parse_python_imports_inner(
.lines()
.find_position(|x| x.starts_with("#requirements:") || x.starts_with("# requirements:"));
if let Some((pos, _)) = find_requirements {
let mut requirements = HashMap::new();
code.lines()
let lines = code
.lines()
.skip(pos + 1)
.map_while(|x| {
RE.captures(x).and_then(|x| {
x.get(1).map(|m| {
let requirement = m.as_str().to_string();
let key = extract_pkg_name(&requirement);
requirements.insert(
key.clone(),
NImportResolved::Pin {
pins: vec![ImportPin {
pkg: requirement.clone(),
path: Default::default(),
}],
key,
},
);
})
})
RE.captures(x)
.map(|x| x.get(1).unwrap().as_str().to_string())
})
.collect_vec();
Ok(requirements)
.collect();
Ok(lines)
} else {
let find_extra_requirements = code.lines().find_position(|x| {
x.starts_with("#extra_requirements:") || x.starts_with("# extra_requirements:")
});
let mut imports: HashMap<String, NImportResolved> = HashMap::new();
let mut imports: Vec<String> = vec![];
if let Some((pos, _)) = find_extra_requirements {
code.lines()
let lines: Vec<String> = code
.lines()
.skip(pos + 1)
.map_while(|x| {
RE.captures(x).and_then(|x| {
x.get(1).map(|m| {
let requirement = m.as_str().to_string();
let key = extract_pkg_name(&requirement);
imports.insert(
key.clone(),
NImportResolved::Pin {
pins: vec![ImportPin {
pkg: requirement,
path: Default::default(),
}],
key,
},
);
})
})
RE.captures(x)
.map(|x| x.get(1).unwrap().as_str().to_string())
})
.collect_vec();
.collect();
imports.extend(lines);
}
// Will get unsorted vector of imports found in current script
let mut nimports = parse_code_for_imports(code, path)?;
let nimports = parse_code_for_imports(code, path)?;
for n in nimports.iter() {
let nested = if n.starts_with("relative:") {
let rpath = n.replace("relative:", "");
let code = sqlx::query_scalar!(
r#"
SELECT content FROM script WHERE path = $1 AND workspace_id = $2
AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND
workspace_id = $2)
"#,
&rpath,
w_id
)
.fetch_optional(db)
.await?
.unwrap_or_else(|| "".to_string());
// It is important to note, that sorting is important and will always result in this pattern:
// 1. All Repins go first
// 2. All Pins go second
// 3. All Auto go third
// 4. All relative imports go the last
//
// This way we make sure all repins are resolved before (re)pins inside imported relative scripts.
nimports.sort();
for n in nimports.into_iter() {
let mut nested = match n {
NImport::Relative(rpath) => {
let code = sqlx::query_scalar!(
r#"
SELECT content FROM script WHERE path = $1 AND workspace_id = $2
AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND
workspace_id = $2)
"#,
if already_visited.contains(&rpath) {
vec![]
} else {
already_visited.push(rpath.clone());
parse_python_imports_inner(
&code,
w_id,
&rpath,
w_id
db,
already_visited,
annotated_pyv_numeric,
path_where_annotated_pyv,
)
.fetch_optional(db)
.await?
.unwrap_or_else(|| "".to_string());
if already_visited.contains(&rpath) {
vec![]
} else {
already_visited.push(rpath.clone());
// Because the algo goes depth first, this function will never return relative import
// This why we can safely assume later, that there is no relative imports
parse_python_imports_inner(
&code,
w_id,
&rpath,
db,
already_visited,
annotated_pyv_numeric,
path_where_annotated_pyv,
)
.await?
.into_values()
.collect_vec()
}
}
NImport::Repin { pin, key } => vec![NImportResolved::Repin { pin, key }],
NImport::Pin { pins, key } => vec![NImportResolved::Pin { pins, key }],
NImport::Auto { pkg, key } => vec![NImportResolved::Auto { pkg, key }],
} else {
vec![n.to_string()]
};
// Nested should also be sorted for the same reason
nested.sort();
// At this point there should be no NImport::Relative in `nested`
for imp in nested {
let key = match imp.clone() {
NImportResolved::Pin { key, .. } => key,
NImportResolved::Repin { key, .. } => key,
NImportResolved::Auto { key, pkg } => key.unwrap_or(pkg),
};
// Handled cases:
//
// 1.
// Error: Imported windmill scripts have different pins
//
// auto
// ├── pin:2
// └── pin:1
//
// Fix 1:
//
// auto
// ├── pin:1
// └── pin:1
//
// Fix 2:
//
// repin:1
// ├── pin:2
// └── pin:1
//
// 2.
// Error: Imported windmill scripts have different pins
//
// pin:2
// └── pin:1
//
// Fix 1:
//
// auto
// └── pin:1
//
// Fix 2:
//
// repin:2
// └── pin:1
//
// 3. repins allowed to be repinned again
//
// repin:2
// └── repin:1
//
match imp.clone() {
NImportResolved::Repin { .. } => {
if let Some(existing_import) = imports.get(&key) {
match existing_import {
// replace
p if matches!(
p,
NImportResolved::Pin { .. } | NImportResolved::Auto { .. }
) =>
{
imports.insert(key, imp);
}
// do nothing (older repins have greater precedence)
NImportResolved::Repin { .. } => {}
// Should not be possible
_ => {
return Err(anyhow::anyhow!(
"Internal error: cannot resolve requirement pins",
)
.into());
}
}
} else {
imports.insert(key, imp.clone());
}
}
NImportResolved::Pin { pins: new_pins, .. } => {
if let Some(existing_import) = imports.get_mut(&key) {
match existing_import {
// Check if pin is the same version, if same, do nothing, if not error
NImportResolved::Pin { pins: existing_pins, .. } => {
existing_pins.extend(new_pins)
}
// do nothing
NImportResolved::Repin { .. } => {}
// Replace with new pin
NImportResolved::Auto { .. } => {
imports.insert(key, imp);
}
}
} else {
imports.insert(key, imp.clone());
}
}
NImportResolved::Auto { .. } => {
if !imports.contains_key(&key) {
imports.insert(key, imp);
}
}
if !imports.contains(&imp) {
imports.push(imp);
}
}
}
imports.sort();
Ok(imports)
}
}

View File

@@ -325,7 +325,6 @@ pub static FULL_IMPORTS_MAP: PyMap = phf_map! {
"azure.mgmt.nspkg" => "azure-mgmt-nspkg",
"azure.keyvault.secrets" => "azure-keyvault-secrets",
"azure.storage.blob" => "azure-storage-blob",
"azure.storage.filedatalake" => "azure-storage-file-datalake",
// Add new entry here ^
};
@@ -377,6 +376,5 @@ pub static SHORT_IMPORTS_MAP: PyMap = phf_map! {
"socks" => "PySocks",
"taiga" => "python-taiga",
"docx" => "python-docx",
"vt" => "vt-py",
// Add new entry here ^
};

View File

@@ -19,7 +19,7 @@ def main():
";
let mut already_visited = vec![];
let (r, ..) = parse_python_imports(
let r = parse_python_imports(
code,
"test-workspace",
"f/foo/bar",
@@ -29,15 +29,7 @@ def main():
)
.await?;
// println!("{}", serde_json::to_string(&r)?);
assert_eq!(
r,
vec![
"matplotlib # (mapped from matplotlib.pyplot)",
"wmill",
"zanzibar # (mapped from zanzibar.estonie)"
]
);
assert_eq!(r, vec!["matplotlib", "wmill", "zanzibar"]);
Ok(())
}
@@ -60,7 +52,7 @@ def main():
";
let mut already_visited = vec![];
let (r, ..) = parse_python_imports(
let r = parse_python_imports(
code,
"test-workspace",
"f/foo/bar",
@@ -91,7 +83,7 @@ def main():
";
let mut already_visited = vec![];
let (r, ..) = parse_python_imports(
let r = parse_python_imports(
code,
"test-workspace",
"f/foo/bar",

View File

@@ -49,12 +49,8 @@ impl Visit for ImportsFinder {
pub fn parse_expr_for_imports(code: &str) -> anyhow::Result<Vec<String>> {
let cm: Lrc<SourceMap> = Default::default();
let fm = cm.new_source_file(FileName::Custom("main.d.ts".into()).into(), code.into());
let mut tss = TsSyntax::default();
tss.disallow_ambiguous_jsx_like;
tss.tsx = true;
tss.no_early_errors = true;
let lexer = Lexer::new(
Syntax::Typescript(tss),
Syntax::Typescript(TsSyntax::default()),
// EsVersion defaults to es5
Default::default(),
StringInput::from(&*fm),

View File

@@ -1,5 +1,3 @@
use std::collections::HashMap;
use anyhow::anyhow;
use serde_json::json;
use windmill_parser::{Arg, MainArgSignature, ObjectProperty, Typ};
@@ -210,52 +208,15 @@ pub struct AnsibleInventory {
resource_type: Option<String>,
pub pinned_resource: Option<String>,
}
#[derive(Debug, Clone)]
pub struct GitRepo {
pub url: String,
pub commit: Option<String>,
pub branch: Option<String>,
pub target_path: String,
}
#[derive(Debug, Clone)]
pub struct AnsibleRequirements {
pub python_reqs: Vec<String>,
pub roles_and_collections: Option<String>,
pub collections: Option<String>,
pub file_resources: Vec<FileResource>,
pub inventories: Vec<AnsibleInventory>,
pub vars: Vec<(String, String)>,
pub resources: Vec<(String, String)>,
pub options: AnsiblePlaybookOptions,
pub vault_password: Option<String>,
pub vault_id: Vec<String>,
pub git_repos: Vec<GitRepo>,
pub git_ssh_identity: Vec<String>,
}
impl Default for AnsibleRequirements {
fn default() -> Self {
Self {
python_reqs: vec![],
roles_and_collections: None,
file_resources: vec![],
inventories: vec![],
vars: vec![],
resources: vec![],
options: AnsiblePlaybookOptions {
verbosity: None,
forks: None,
timeout: None,
flush_cache: None,
force_handlers: None,
},
vault_password: None,
vault_id: vec![],
git_repos: vec![],
git_ssh_identity: vec![],
}
}
}
fn parse_inventories(inventory_yaml: &Yaml) -> anyhow::Result<Vec<AnsibleInventory>> {
@@ -314,7 +275,22 @@ pub fn parse_ansible_reqs(
return Ok((logs, None, inner_content.to_string()));
}
let mut ret = AnsibleRequirements::default();
let opts = AnsiblePlaybookOptions {
verbosity: None,
forks: None,
timeout: None,
flush_cache: None,
force_handlers: None,
};
let mut ret = AnsibleRequirements {
python_reqs: vec![],
collections: None,
file_resources: vec![],
inventories: vec![],
vars: vec![],
resources: vec![],
options: opts,
};
if let Yaml::Hash(doc) = &docs[0] {
for (key, value) in doc {
@@ -327,7 +303,7 @@ pub fn parse_ansible_reqs(
let mut out_str = String::new();
let mut emitter = YamlEmitter::new(&mut out_str);
emitter.dump(galaxy_requirements)?;
ret.roles_and_collections = Some(out_str);
ret.collections = Some(out_str);
}
if let Some(Yaml::Array(py_reqs)) =
deps.get(&Yaml::String("python".to_string()))
@@ -369,60 +345,11 @@ pub fn parse_ansible_reqs(
Yaml::String(key) if key == "inventory" => {
ret.inventories = parse_inventories(value)?;
}
Yaml::String(key) if key == "vault_password" => {
let Yaml::String(filename) = value else {
return Err(anyhow!(
"Vault Password File expects a String containing the file name"
));
};
ret.vault_password = Some(filename.to_string());
}
Yaml::String(key) if key == "vault_id" => {
let Yaml::Array(filenames) = value else {
return Err(anyhow!("Vault ID field expects an array of strings in the format: `label@filename`"));
};
for f in filenames {
let Yaml::String(filename) = f else {
return Err(anyhow!("The elements of the vault_id field should be strings in the format: `label@filename`"));
};
ret.vault_id.push(filename.to_string());
}
}
Yaml::String(key) if key == "options" => {
if let Yaml::Array(opts) = &value {
ret.options = parse_ansible_options(opts);
}
}
Yaml::String(key) if key == "git_repos" => {
let Yaml::Array(repos) = &value else {
return Err(anyhow!("git_repos field expects an array of repos"));
};
for r in repos {
ret.git_repos.push(
parse_git_repo(r)
.map_err(|e| anyhow!("Failed to parse git repo: {e}"))?,
);
}
}
Yaml::String(key) if key == "git_ssh_identity" => {
let Yaml::Array(indentities) = &value else {
return Err(anyhow!(
"git_ssh_identity expects an array of windmill variables (or secrets) containing ssh IDs"
));
};
for r in indentities {
let Yaml::String(file_name) = r else {
return Err(anyhow!(
"Git ssh identity file must be a string path to a Windmill variable/secret"
));
};
ret.git_ssh_identity.push(file_name.clone());
}
}
Yaml::String(key) => logs.push_str(&format!("\nUnknown field `{}`. Ignoring", key)),
_ => (),
}
@@ -437,38 +364,6 @@ pub fn parse_ansible_reqs(
Ok((logs, Some(ret), out_str))
}
fn parse_git_repo(r: &Yaml) -> anyhow::Result<GitRepo> {
let Yaml::Hash(repo) = r else {
return Err(anyhow!("Should be a Map"));
};
let url = repo
.get(&Yaml::String("url".to_string()))
.and_then(|v| v.as_str())
.map(|s| s.to_string())
.ok_or(anyhow!("Expected `url` field"))?;
let target_path = repo
.get(&Yaml::String("target".to_string()))
.and_then(|v| v.as_str())
.map(|s| s.to_string())
.ok_or(anyhow!(
"Expected `target` field (target directory for cloning the repo)"
))?;
let branch = repo
.get(&Yaml::String("branch".to_string()))
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let commit = repo
.get(&Yaml::String("commit".to_string()))
.and_then(|v| v.as_str())
.map(|s| s.to_string());
Ok(GitRepo { url, commit, branch, target_path })
}
fn parse_ansible_options(opts: &Vec<Yaml>) -> AnsiblePlaybookOptions {
let mut ret = AnsiblePlaybookOptions {
verbosity: None,
@@ -634,78 +529,3 @@ fn yaml_to_json(yaml: &Yaml) -> serde_json::Value {
_ => serde_json::Value::Null,
}
}
fn update_versions(
section: &str,
yaml: &mut Yaml,
versions: &HashMap<String, String>,
) -> anyhow::Result<String> {
let mut logs = String::new();
let Yaml::Hash(ref mut m) = yaml else {
return Err(anyhow!("{section} dependency should be a map"));
};
if let Some(Yaml::Array(elements)) = m.get_mut(&Yaml::String(section.to_string())) {
for el in elements {
let Yaml::Hash(ref mut h) = el else {
return Err(anyhow!("{section} dependency element should be a map"));
};
if let Some(name) = h
.get(&Yaml::String("name".to_string()))
.and_then(|n| n.as_str())
{
if let Some(version) = versions.get(name) {
h.insert(
Yaml::String("version".to_string()),
Yaml::String(version.to_string()),
);
} else {
logs.push_str(&format!("WARNING: {section} dependency `{name}` has no locked version, using the latest or system installed version.\n"));
}
} else {
return Err(anyhow!(
"{section} dependency element: missing or invalid `name` field"
));
}
}
}
Ok(logs)
}
pub fn add_versions_to_requirements_yaml(
input: &str,
role_versions: &HashMap<String, String>,
collection_versions: &HashMap<String, String>,
) -> anyhow::Result<(String,String)> {
let mut docs =
YamlLoader::load_from_str(input).map_err(|e| anyhow!("YAML parse error: {}", e))?;
let doc = &mut docs[0];
let mut logs = String::new();
logs.push_str(
&update_versions("roles", doc, role_versions)
.map_err(|e| anyhow!("Error updating role versions: {e}"))?,
);
logs.push_str(
&update_versions("collections", doc, collection_versions)
.map_err(|e| anyhow!("Error updating collection versions: {e}"))?,
);
if !logs.is_empty() {
logs.push_str("WARNING: You might want to try adding manual versions for these, otherwise there could be breaking changes on deployed scripts\n");
}
let mut out_str = String::new();
{
let mut emitter = YamlEmitter::new(&mut out_str);
emitter
.dump(doc)
.map_err(|e| anyhow!("YAML emit error: {}", e))?;
}
Ok((out_str, logs))
}

View File

@@ -1,33 +0,0 @@
import json
import matplotlib.pyplot as plt
# Path to the profiling JSON file
# file_path = "/tmp/windmill/profiling_main.json"
file_path = "/tmp/profiling.json"
# Load the JSON data
with open(file_path, "r") as f:
data = json.load(f)
# Extract timings for "pre pull->post pull"
pre_post_pull_timings = [
timing / 1000000.0 for entry in data["timings"]
for step, timing in entry["timings"]
# if step == "pre pull->post pull"
if step == "->job pulled from DB"
]
# Plotting the distribution
plt.figure(figsize=(10, 6))
# plt.hist(pre_post_pull_timings, bins=10, edgecolor='black')
plt.scatter(range(len(pre_post_pull_timings)), pre_post_pull_timings,
alpha=1.0, # Transparency level
s=40) # Size of the dots`)
plt.title("Distribution of 'pre pull->post pull' timings")
# plt.xlabel("Time (ms)")
# plt.ylabel("Frequency")
plt.xlabel("Sample Index")
plt.ylabel("Time (ms)")
plt.grid(True)
plt.tight_layout()
plt.show()

View File

@@ -265,8 +265,6 @@ async fn windmill_main() -> anyhow::Result<()> {
if mode == Mode::Standalone {
println!("Running in standalone mode");
} else if mode == Mode::MCP {
println!("Running in MCP mode");
}
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
@@ -299,7 +297,7 @@ async fn windmill_main() -> anyhow::Result<()> {
}
#[allow(unused_mut)]
let mut num_workers = if mode == Mode::Server || mode == Mode::Indexer || mode == Mode::MCP {
let mut num_workers = if mode == Mode::Server || mode == Mode::Indexer {
0
} else {
std::env::var("NUM_WORKERS")
@@ -321,9 +319,8 @@ async fn windmill_main() -> anyhow::Result<()> {
&& (mode == Mode::Server || mode == Mode::Standalone);
let indexer_mode = mode == Mode::Indexer;
let mcp_mode = mode == Mode::MCP;
let server_bind_address: IpAddr = if server_mode || indexer_mode || mcp_mode {
let server_bind_address: IpAddr = if server_mode || indexer_mode {
std::env::var("SERVER_BIND_ADDR")
.ok()
.and_then(|x| x.parse().ok())
@@ -332,16 +329,26 @@ async fn windmill_main() -> anyhow::Result<()> {
IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))
};
let (conn, first_suffix) = if mode == Mode::Agent {
tracing::info!(
"Creating http client for cluster using base internal url {}",
std::env::var("BASE_INTERNAL_URL").unwrap_or_default()
);
let mut first_worker_suffix = None;
let mut worker_names = vec![];
for _ in 0..num_workers {
let suffix = windmill_common::utils::worker_suffix(&hostname, &rd_string(5));
(
Connection::Http(build_agent_http_client(&suffix)),
Some(suffix),
)
worker_names.push(windmill_common::utils::worker_name_with_suffix(
mode == Mode::Agent,
WORKER_GROUP.as_str(),
&suffix,
));
if first_worker_suffix.is_none() {
first_worker_suffix = Some(suffix);
}
}
let conn = if mode == Mode::Agent {
let worker_suffix = first_worker_suffix.unwrap_or_else(|| {
panic!("there must be at least one worker in agent mode");
});
Connection::Http(build_agent_http_client(&worker_suffix))
} else {
println!("Connecting to database...");
@@ -359,7 +366,7 @@ async fn windmill_main() -> anyhow::Result<()> {
load_otel(&db).await;
tracing::info!("Database connected");
(Connection::Sql(db), None)
Connection::Sql(db)
};
let environment = load_base_url(&conn)
@@ -383,7 +390,7 @@ async fn windmill_main() -> anyhow::Result<()> {
.is_some_and(|x| x == "1" || x == "true");
if let Some(db) = conn.as_sql() {
if !is_agent && !indexer_mode && !mcp_mode {
if !is_agent && !indexer_mode {
let skip_migration = std::env::var("SKIP_MIGRATION")
.map(|val| val == "true")
.unwrap_or(false);
@@ -402,7 +409,6 @@ async fn windmill_main() -> anyhow::Result<()> {
let conn = if mode == Mode::Agent {
conn
} else {
// This time we use a pool of connections
let db = windmill_common::connect_db(server_mode, indexer_mode, worker_mode).await?;
Connection::Sql(db)
};
@@ -433,6 +439,16 @@ Windmill Community Edition {GIT_VERSION}
display_config(&ENV_SETTINGS);
if let Err(e) = reload_base_url_setting(&conn).await {
tracing::error!("Error loading base url: {:?}", e)
}
if let Some(db) = conn.as_sql() {
if let Err(e) = reload_critical_error_channels_setting(&db).await {
tracing::error!("Could loading critical error emails setting: {:?}", e);
}
}
#[cfg(feature = "enterprise")]
{
// load the license key and check if it's valid
@@ -446,7 +462,7 @@ Windmill Community Edition {GIT_VERSION}
if !valid_key && !server_mode {
tracing::error!("Invalid license key, workers require a valid license key");
}
if server_mode || mcp_mode {
if server_mode {
if let Some(db) = conn.as_sql() {
// only force renewal if invalid but not empty (= expired)
let renewed_now = maybe_renew_license_key_on_start(
@@ -466,10 +482,10 @@ Windmill Community Edition {GIT_VERSION}
}
}
if server_mode || worker_mode || indexer_mode || mcp_mode {
if server_mode || worker_mode || indexer_mode {
let port_var = std::env::var("PORT").ok().and_then(|x| x.parse().ok());
let port = if server_mode || indexer_mode || mcp_mode {
let port = if server_mode || indexer_mode {
port_var.unwrap_or(DEFAULT_PORT as u16)
} else {
port_var.unwrap_or(0)
@@ -650,7 +666,6 @@ Windmill Community Edition {GIT_VERSION}
server_killpill_rx,
base_internal_tx,
server_mode,
mcp_mode,
base_internal_url.clone(),
)
.await?;
@@ -671,34 +686,14 @@ Windmill Community Edition {GIT_VERSION}
if !killpill_rx.try_recv().is_ok() {
let base_internal_url = base_internal_rx.await?;
if worker_mode {
let mut workers = vec![];
for i in 0..num_workers {
let suffix: String = if i == 0 && first_suffix.as_ref().is_some() {
first_suffix.as_ref().unwrap().clone()
} else {
windmill_common::utils::worker_suffix(&hostname, &rd_string(5))
};
let worker_conn = WorkerConn {
conn: if i == 0 || mode != Mode::Agent {
conn.clone()
} else {
Connection::Http(build_agent_http_client(&suffix))
},
worker_name: windmill_common::utils::worker_name_with_suffix(
mode == Mode::Agent,
WORKER_GROUP.as_str(),
&suffix,
),
};
workers.push(worker_conn);
}
run_workers(
conn.clone(),
rx,
killpill_tx.clone(),
num_workers,
base_internal_url.clone(),
hostname.clone(),
&workers,
&worker_names,
)
.await?;
tracing::info!("All workers exited.");
@@ -1053,19 +1048,15 @@ Windmill Community Edition {GIT_VERSION}
}
}
if mcp_mode {
futures::try_join!(shutdown_signal, workers_f, server_f)?;
} else {
futures::try_join!(
shutdown_signal,
workers_f,
monitor_f,
server_f,
metrics_f,
indexer_f,
log_indexer_f
)?;
}
futures::try_join!(
shutdown_signal,
workers_f,
monitor_f,
server_f,
metrics_f,
indexer_f,
log_indexer_f
)?;
} else {
tracing::info!("Nothing to do, exiting.");
}
@@ -1143,20 +1134,16 @@ fn display_config(envs: &[&str]) {
)
}
pub struct WorkerConn {
conn: Connection,
worker_name: String,
}
pub async fn run_workers(
db: Connection,
mut rx: tokio::sync::broadcast::Receiver<()>,
tx: KillpillSender,
num_workers: i32,
base_internal_url: String,
hostname: String,
workers: &[WorkerConn],
worker_names: &[String],
) -> anyhow::Result<()> {
let mut killpill_rxs = vec![];
let num_workers = workers.len();
for _ in 0..num_workers {
killpill_rxs.push(rx.resubscribe());
}
@@ -1215,9 +1202,8 @@ pub async fn run_workers(
*windmill_worker::SLEEP_QUEUE
);
for i in 1..(num_workers + 1) {
let wk_conf = &workers[i as usize - 1];
let conn1 = wk_conf.conn.clone();
let worker_name = wk_conf.worker_name.clone();
let db1 = db.clone();
let worker_name = worker_names[i as usize - 1].clone();
let ip = ip.clone();
let rx = killpill_rxs.pop().unwrap();
let tx = tx.clone();
@@ -1230,7 +1216,7 @@ pub async fn run_workers(
}
let f = windmill_worker::run_worker(
&conn1,
&db1,
&hostname,
worker_name,
i as u64,

View File

@@ -34,12 +34,8 @@ use windmill_common::ee::{jobs_waiting_alerts, worker_groups_alerts};
#[cfg(feature = "oauth2")]
use windmill_common::global_settings::OAUTH_SETTING;
use windmill_common::{
agent_workers::DECODED_AGENT_TOKEN,
auth::create_token_for_owner,
ee::CriticalErrorChannel,
error,
flow_status::{FlowStatus, FlowStatusModule},
global_settings::{
utils::empty_string_as_none,
agent_workers::DECODED_AGENT_TOKEN, auth::create_token_for_owner, ee::CriticalErrorChannel, error, flow_status::{FlowStatus, FlowStatusModule}, global_settings::{
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,
@@ -49,32 +45,13 @@ use windmill_common::{
NUGET_CONFIG_SETTING, OTEL_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,
},
indexer::load_indexer_config,
jobs::QueuedJob,
jwt::JWT_SECRET,
oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH,
server::load_smtp_config,
tracing_init::JSON_FMT,
users::truncate_token,
utils::empty_string_as_none,
utils::{now_from_db, rd_string, report_critical_error, Mode},
worker::{
load_worker_config, reload_custom_tags_setting, store_pull_query,
store_suspended_pull_query, update_min_version, Connection, DEFAULT_TAGS_PER_WORKSPACE,
DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, SCRIPT_TOKEN_EXPIRY, SMTP_CONFIG, TMP_DIR,
WORKER_CONFIG, WORKER_GROUP,
},
KillpillSender, 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,
MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED,
SERVICE_LOG_RETENTION_SECS,
}, indexer::load_indexer_config, jobs::QueuedJob, jwt::JWT_SECRET, oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, server::load_smtp_config, tracing_init::JSON_FMT, users::truncate_token, utils::{now_from_db, rd_string, report_critical_error, Mode}, worker::{
load_worker_config, reload_custom_tags_setting, store_pull_query, store_suspended_pull_query, update_min_version, Connection, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, SCRIPT_TOKEN_EXPIRY, SMTP_CONFIG, TMP_DIR, WORKER_CONFIG, WORKER_GROUP
}, KillpillSender, 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, MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED, SERVICE_LOG_RETENTION_SECS,
};
use windmill_queue::{cancel_job, MiniPulledJob, SameWorkerPayload};
use windmill_worker::{
handle_job_error, AuthedClient, JobCompletedSender, SameWorkerSender, BUNFIG_INSTALL_SCOPES,
INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, MAVEN_REPOS, NO_DEFAULT_MAVEN,
NPM_CONFIG_REGISTRY, NUGET_CONFIG, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL,
handle_job_error, AuthedClient, JobCompletedSender, SameWorkerSender, BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, MAVEN_REPOS, NO_DEFAULT_MAVEN, NPM_CONFIG_REGISTRY, NUGET_CONFIG, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL
};
#[cfg(feature = "parquet")]
@@ -158,6 +135,7 @@ pub async fn initial_load(
}
}
if let Err(e) = load_metrics_enabled(conn).await {
tracing::error!("Error loading expose metrics: {e:#}");
}
@@ -194,13 +172,11 @@ pub async fn initial_load(
}
Connection::Http(_) => {
// TODO: reload worker config from http
WORKER_CONFIG.write().await.worker_tags = DECODED_AGENT_TOKEN
.as_ref()
.map(|x| x.tags.clone())
.unwrap_or_default();
WORKER_CONFIG.write().await.worker_tags = DECODED_AGENT_TOKEN.as_ref().map(|x| x.tags.clone()).unwrap_or_default();
}
}
}
if let Err(e) = reload_hub_base_url_setting(conn, server_mode).await {
tracing::error!("Error reloading hub base url: {:?}", e)
@@ -214,6 +190,7 @@ pub async fn initial_load(
if let Err(e) = reload_custom_tags_setting(db).await {
tracing::error!("Error reloading custom tags: {:?}", e)
}
}
#[cfg(feature = "parquet")]
@@ -248,8 +225,7 @@ pub async fn initial_load(
}
pub async fn load_metrics_enabled(conn: &Connection) -> error::Result<()> {
let metrics_enabled =
load_value_from_global_settings_with_conn(conn, EXPOSE_METRICS_SETTING, true).await;
let metrics_enabled = load_value_from_global_settings_with_conn(conn, EXPOSE_METRICS_SETTING, true).await;
match metrics_enabled {
Ok(Some(serde_json::Value::Bool(t))) => METRICS_ENABLED.store(t, Ordering::Relaxed),
_ => (),
@@ -369,13 +345,13 @@ pub async fn reload_critical_alert_mute_ui_setting(conn: &Connection) -> error::
load_value_from_global_settings_with_conn(conn, CRITICAL_ALERT_MUTE_UI_SETTING, true).await
{
CRITICAL_ALERT_MUTE_UI_ENABLED.store(t, Ordering::Relaxed);
}
Ok(())
}
pub async fn load_metrics_debug_enabled(conn: &Connection) -> error::Result<()> {
let metrics_enabled =
load_value_from_global_settings_with_conn(conn, EXPOSE_DEBUG_METRICS_SETTING, true).await;
let metrics_enabled = load_value_from_global_settings_with_conn(conn, EXPOSE_DEBUG_METRICS_SETTING, true).await;
match metrics_enabled {
Ok(Some(serde_json::Value::Bool(t))) => {
METRICS_DEBUG_ENABLED.store(t, Ordering::Relaxed);
@@ -599,9 +575,7 @@ async fn send_log_file_to_object_store(
};
let exists = LAST_LOG_FILE_SENT.lock().map(|last_log_file_sent| {
last_log_file_sent
.map(|last_log_file_sent| last_log_file_sent >= ts)
.unwrap_or(false)
last_log_file_sent.map(|last_log_file_sent| last_log_file_sent >= ts).unwrap_or(false)
});
if exists.unwrap_or(false) {
@@ -638,7 +612,7 @@ async fn send_log_file_to_object_store(
let (ok_lines, err_lines) = read_log_counters(ts_str);
if let Some(db) = conn.as_sql() {
if let Err(e) = sqlx::query!("INSERT INTO log_file (hostname, mode, worker_group, log_ts, file_path, ok_lines, err_lines, json_fmt) VALUES ($1, $2::text::LOG_MODE, $3, $4, $5, $6, $7, $8)",
if let Err(e) = sqlx::query!("INSERT INTO log_file (hostname, mode, worker_group, log_ts, file_path, ok_lines, err_lines, json_fmt) VALUES ($1, $2::text::LOG_MODE, $3, $4, $5, $6, $7, $8)",
hostname, mode.to_string(), worker_group.clone(), ts, highest_file, ok_lines as i64, err_lines as i64, *JSON_FMT)
.execute(db)
.await {
@@ -1028,21 +1002,11 @@ pub async fn reload_nuget_config_setting(conn: &Connection) {
.await;
}
pub async fn reload_maven_repos_setting(conn: &Connection) {
reload_option_setting_with_tracing(
conn,
windmill_common::global_settings::MAVEN_REPOS_SETTING,
"MAVEN_REPOS",
MAVEN_REPOS.clone(),
)
.await;
reload_option_setting_with_tracing(conn, windmill_common::global_settings::MAVEN_REPOS_SETTING, "MAVEN_REPOS", MAVEN_REPOS.clone())
.await;
}
pub async fn reload_no_default_maven_setting(conn: &Connection) {
let value = load_value_from_global_settings_with_conn(
conn,
windmill_common::global_settings::NO_DEFAULT_MAVEN_SETTING,
true,
)
.await;
let value = load_value_from_global_settings_with_conn(conn, windmill_common::global_settings::NO_DEFAULT_MAVEN_SETTING, true).await;
match value {
Ok(Some(serde_json::Value::Bool(t))) => NO_DEFAULT_MAVEN.store(t, Ordering::Relaxed),
Err(e) => {
@@ -1211,6 +1175,7 @@ pub async fn load_value_from_global_settings(
Ok(r)
}
pub async fn load_value_from_global_settings_with_conn(
conn: &Connection,
setting_name: &str,
@@ -1220,18 +1185,14 @@ pub async fn load_value_from_global_settings_with_conn(
Connection::Sql(db) => Ok(load_value_from_global_settings(db, setting_name).await?),
Connection::Http(client) => {
if load_from_http {
client
.get::<Option<serde_json::Value>>(&format!(
"/api/agent_workers/get_global_setting/{}",
setting_name
))
.await
.map_err(|e| anyhow::anyhow!("Error loading setting {}: {}", setting_name, e))
client.get::<Option<serde_json::Value>>(&format!("/api/agent_workers/get_global_setting/{}", setting_name)).await
.map_err(|e| anyhow::anyhow!("Error loading setting {}: {}", setting_name, e))
} else {
Ok(None)
}
}
}
}
pub async fn reload_option_setting<T: FromStr + DeserializeOwned>(
@@ -1353,12 +1314,12 @@ pub async fn monitor_db(
let zombie_jobs_f = async {
if server_mode && !initial_load && !*DISABLE_ZOMBIE_JOBS_MONITORING {
if let Some(db) = conn.as_sql() {
handle_zombie_jobs(db, base_internal_url, "server").await;
match handle_zombie_flows(db).await {
Err(err) => {
tracing::error!("Error handling zombie flows: {:?}", err);
}
_ => {}
handle_zombie_jobs(db, base_internal_url, "server").await;
match handle_zombie_flows(db).await {
Err(err) => {
tracing::error!("Error handling zombie flows: {:?}", err);
},
_ => {}
}
}
}
@@ -1366,7 +1327,7 @@ pub async fn monitor_db(
let expired_items_f = async {
if server_mode && !initial_load {
if let Some(db) = conn.as_sql() {
delete_expired_items(&db).await;
delete_expired_items(&db).await;
}
}
};
@@ -1535,7 +1496,11 @@ pub async fn reload_indexer_config(db: &Pool<Postgres>) {
}
}
pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: bool) {
pub async fn reload_worker_config(
db: &DB,
tx: KillpillSender,
kill_if_change: bool,
) {
let config = load_worker_config(db, tx.clone()).await;
if let Err(e) = config {
tracing::error!("Error reloading worker config: {:?}", e)
@@ -1578,8 +1543,7 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b
}
pub async fn load_base_url(conn: &Connection) -> error::Result<String> {
let q_base_url =
load_value_from_global_settings_with_conn(conn, BASE_URL_SETTING, false).await?;
let q_base_url = load_value_from_global_settings_with_conn(conn, BASE_URL_SETTING, false).await?;
let std_base_url = std::env::var("BASE_URL")
.ok()
@@ -1610,9 +1574,10 @@ pub async fn load_base_url(conn: &Connection) -> error::Result<String> {
}
pub async fn reload_base_url_setting(conn: &Connection) -> error::Result<()> {
#[cfg(feature = "oauth2")]
let oauths = if let Some(db) = conn.as_sql() {
let q_oauth = load_value_from_global_settings(db, OAUTH_SETTING).await?;
let q_oauth = load_value_from_global_settings (db, OAUTH_SETTING).await?;
if let Some(q) = q_oauth {
if let Ok(v) = serde_json::from_value::<
@@ -1687,7 +1652,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
increment_counter AS (
INSERT INTO zombie_job_counter (job_id, counter)
SELECT id, 1 FROM to_update WHERE counter < $2
ON CONFLICT (job_id) DO UPDATE
ON CONFLICT (job_id) DO UPDATE
SET counter = zombie_job_counter.counter + 1
),
update_concurrency AS (
@@ -1779,7 +1744,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
let same_worker_timeout_jobs = {
let long_same_worker_jobs = sqlx::query!(
"SELECT worker, array_agg(v2_job_queue.id) as ids FROM v2_job_queue LEFT JOIN v2_job ON v2_job_queue.id = v2_job.id LEFT JOIN v2_job_runtime ON v2_job_queue.id = v2_job_runtime.id WHERE v2_job_queue.created_at < now() - ('60 seconds')::interval
"SELECT worker, array_agg(v2_job_queue.id) as ids FROM v2_job_queue LEFT JOIN v2_job ON v2_job_queue.id = v2_job.id LEFT JOIN v2_job_runtime ON v2_job_queue.id = v2_job_runtime.id WHERE v2_job_queue.created_at < now() - ('60 seconds')::interval
AND running = true AND ping IS NULL AND same_worker = true AND worker IS NOT NULL GROUP BY worker",
)
.fetch_all(db)
@@ -1793,9 +1758,9 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
.collect::<Vec<_>>();
let long_dead_workers: std::collections::HashSet<String> = sqlx::query_scalar!(
"WITH worker_ids AS (SELECT unnest($1::text[]) as worker)
SELECT worker_ids.worker FROM worker_ids
LEFT JOIN worker_ping ON worker_ids.worker = worker_ping.worker
"WITH worker_ids AS (SELECT unnest($1::text[]) as worker)
SELECT worker_ids.worker FROM worker_ids
LEFT JOIN worker_ping ON worker_ids.worker = worker_ping.worker
WHERE worker_ping.worker IS NULL OR worker_ping.ping_at < now() - ('60 seconds')::interval",
&worker_ids[..]
)
@@ -1839,7 +1804,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
let non_restartable_jobs = if *RESTART_ZOMBIE_JOBS {
vec![]
} else {
sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE last_ping < now() - ($1 || ' seconds')::interval
sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE last_ping < now() - ($1 || ' seconds')::interval
AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false")
.bind(ZOMBIE_JOB_TIMEOUT.as_str())
.fetch_all(db)
@@ -1898,8 +1863,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
mpsc::channel::<SameWorkerPayload>(1);
let same_worker_tx_never_used =
SameWorkerSender(same_worker_tx_never_used, Arc::new(AtomicU16::new(0)));
let (send_result_never_used, _send_result_rx_never_used) =
JobCompletedSender::new_never_used();
let (send_result_never_used, _send_result_rx_never_used) = JobCompletedSender::new_never_used();
let label = if job.permissioned_as != format!("u/{}", job.created_by)
&& job.permissioned_as != job.created_by
@@ -1950,7 +1914,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
worker_name,
send_result_never_used,
#[cfg(feature = "benchmark")]
&mut windmill_common::bench::BenchmarkIter::new(),
&mut windmill_worker::bench::BenchmarkIter::new(),
)
.await;
}
@@ -2037,7 +2001,7 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
}
);
report_critical_error(reason.clone(), db.clone(), Some(&flow.workspace_id), None).await;
cancel_zombie_flow_job(db, flow.id, &flow.workspace_id,
cancel_zombie_flow_job(db, flow.id, &flow.workspace_id,
format!(r#"{reason}
This would happen if a worker was interrupted, killed or crashed while doing a state transition at the end of a job which is always an unexpected behavior that should never happen.
Please check your worker logs for more details and feel free to report it to the Windmill team on our Discord or support@windmill.dev (response for non EE customers will be best effort) with as much context as possible, ideally:
@@ -2054,7 +2018,7 @@ Please check your worker logs for more details and feel free to report it to the
r#"
DELETE
FROM parallel_monitor_lock
WHERE last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval
WHERE last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval
RETURNING parent_flow_id, job_id, last_ping, (SELECT workspace_id FROM v2_job_queue q
WHERE q.id = parent_flow_id AND q.running = true AND q.canceled_by IS NULL
) AS workspace_id
@@ -2109,12 +2073,8 @@ async fn cancel_zombie_flow_job(
Ok(())
}
pub async fn reload_hub_base_url_setting(
conn: &Connection,
server_mode: bool,
) -> error::Result<()> {
let hub_base_url =
load_value_from_global_settings_with_conn(conn, HUB_BASE_URL_SETTING, true).await?;
pub async fn reload_hub_base_url_setting(conn: &Connection, server_mode: bool) -> error::Result<()> {
let hub_base_url = load_value_from_global_settings_with_conn(conn, HUB_BASE_URL_SETTING, true).await?;
let base_url = if let Some(q) = hub_base_url {
if let Ok(v) = serde_json::from_value::<String>(q.clone()) {
@@ -2166,7 +2126,7 @@ pub async fn reload_critical_error_channels_setting(conn: &DB) -> error::Result<
v
} else {
tracing::error!(
"Could not parse critical_error_channels setting as an array of channels, found: {:#?}",
"Could not parse critical_error_emails setting as an array of channels, found: {:#?}",
&q
);
vec![]

View File

@@ -1,51 +0,0 @@
INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES (
'test-workspace',
'test-user',
'
# requirements:
# microdot==2.2.0
import pandas
import requests
import tiny # pin: tiny==0.1.2
def main():
pass
',
'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
'',
'',
'f/system/requirements', 12346, 'python3', '');
INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES (
'test-workspace',
'test-user',
'
# extra_requirements:
# bottle==0.13.2
import tiny
def main():
pass
',
'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
'',
'',
'f/system/extra_requirements', 12347, 'python3', '');
INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES (
'test-workspace',
'test-user',
'
import tiny # pin: bottle==0.13.2
import simplejson # pin: simplejson==3.19.3
def main():
return [test1(), test2(), test3(), test4()]
',
'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
'',
'',
'f/system/pins', 12348, 'python3', '');

View File

@@ -137,7 +137,6 @@ impl ApiServer {
rx,
port_tx,
false,
false,
format!("http://localhost:{}", addr.port()),
));
@@ -3832,204 +3831,6 @@ def main():
run_preview_relative_imports(&db, content, ScriptLang::Python3).await;
}
async fn assert_lockfile(
db: &Pool<Postgres>,
script_content: String,
language: ScriptLang,
expected_lines: Vec<&str>,
) {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await;
let port = server.addr.port();
let client = windmill_api_client::create_client(
&format!("http://localhost:{port}"),
"SECRET_TOKEN".to_string(),
);
client
.create_script(
"test-workspace",
&NewScript {
language: NewScriptLanguage::from_str(language.as_str()).unwrap(),
content: script_content,
path: "f/system/test_import".to_string(),
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
description: "".to_string(),
draft_only: None,
envs: vec![],
is_template: None,
kind: None,
parent_hash: None,
lock: None,
summary: "".to_string(),
tag: None,
schema: std::collections::HashMap::new(),
ws_error_handler_muted: Some(false),
priority: None,
delete_after_use: None,
timeout: None,
restart_unless_cancelled: None,
deployment_message: None,
concurrency_key: None,
visible_to_runner_only: None,
no_main_func: None,
codebase: None,
has_preprocessor: None,
on_behalf_of_email: None,
},
)
.await
.unwrap();
let mut completed = listen_for_completed_jobs(&db).await;
let db2 = db.clone();
in_test_worker(
&db,
async move {
completed.next().await; // deployed script
let script = sqlx::query!(
"SELECT hash FROM script WHERE path = $1",
"f/system/test_import".to_string()
)
.fetch_one(&db2)
.await
.unwrap();
let job = RunJob::from(JobPayload::Dependencies {
path: "f/system/test_import".to_string(),
hash: ScriptHash(script.hash),
dedicated_worker: None,
language,
})
.push(&db2)
.await;
completed.next().await; // completed job
let result = completed_job(job, &db2).await.json_result().unwrap();
assert_eq!(
result,
json!({
"lock": expected_lines.join("\n"),
"status": "Successful lock file generation"
})
);
},
port,
)
.await;
}
#[sqlx::test(fixtures("base", "lockfile_python"))]
async fn test_requirements_python(db: Pool<Postgres>) {
let content = r#"
# py311
# requirements:
# tiny==0.1.3
import bar
import baz # pin: foo
import baz # repin: fee
import bug # repin: free
def main():
pass
"#
.to_string();
assert_lockfile(
&db,
content,
ScriptLang::Python3,
vec!["# py311", "tiny==0.1.3"],
)
.await;
}
#[sqlx::test(fixtures("base", "lockfile_python"))]
async fn test_extra_requirements_python(db: Pool<Postgres>) {
{
let content = r#"
# py311
# extra_requirements:
# tiny
import f.system.extra_requirements
import tiny # pin: tiny==0.1.0
import tiny # pin: tiny==0.1.1
import tiny # repin: tiny==0.1.2
def main():
pass
"#
.to_string();
assert_lockfile(
&db,
content,
ScriptLang::Python3,
vec!["# py311", "bottle==0.13.2", "tiny==0.1.2"],
)
.await;
}
}
#[sqlx::test(fixtures("base", "lockfile_python"))]
async fn test_extra_requirements_python2(db: Pool<Postgres>) {
let content = r#"
# py311
# extra_requirements:
# tiny==0.1.3
import simplejson # pin: simplejson==3.20.1
def main():
pass
"#
.to_string();
assert_lockfile(
&db,
content,
ScriptLang::Python3,
vec!["# py311", "simplejson==3.20.1", "tiny==0.1.3"],
)
.await;
}
#[sqlx::test(fixtures("base", "lockfile_python"))]
async fn test_pins_python(db: Pool<Postgres>) {
let content = r#"
# py311
# extra_requirements:
# tiny==0.1.3
# bottle==0.13.2
import f.system.requirements
import f.system.pins
import tiny # repin: tiny==0.1.3
import simplejson
def main():
pass
"#
.to_string();
assert_lockfile(
&db,
content,
ScriptLang::Python3,
vec![
"# py311",
"bottle==0.13.2",
"microdot==2.2.0",
"simplejson==3.19.3",
"tiny==0.1.3",
],
)
.await;
}
#[sqlx::test(fixtures("base", "result_format"))]
async fn test_result_format(db: Pool<Postgres>) {
let ordered_result_job_id = "1eecb96a-c8b0-4a3d-b1b6-087878c55e41";

View File

@@ -13,7 +13,7 @@ default = []
enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker/enterprise"]
stripe = []
agent_worker_server = []
enterprise_saml = ["dep:samael", "dep:libxml"]
enterprise_saml = ["dep:samael"]
benchmark = []
embedding = ["dep:tinyvector", "dep:hf-hub", "dep:tokenizers", "dep:candle-core", "dep:candle-transformers", "dep:candle-nn", "dep:half"]
parquet = ["dep:datafusion", "dep:object_store", "dep:url", "windmill-common/parquet", "windmill-worker/parquet"]
@@ -32,13 +32,10 @@ static_frontend = ["dep:rust-embed"]
postgres_trigger = ["dep:rust-postgres", "dep:pg_escape", "dep:byteorder", "dep:thiserror", "dep:rust_decimal", "dep:rust-postgres-native-tls"]
mqtt_trigger = ["dep:thiserror", "dep:rumqttc"]
sqs_trigger = ["dep:aws-sdk-sqs", "dep:thiserror", "dep:aws-config"]
deno_core = ["dep:deno_core", "dep:deno_error"]
gcp_trigger = ["dep:thiserror", "dep:google-cloud-pubsub", "dep:google-cloud-googleapis", "dep:tonic"]
cloud = ["windmill-common/cloud"]
mcp = ["dep:rmcp"]
[dependencies]
rmcp = { git = "https://github.com/windmill-labs/rust-sdk", features = ["transport-sse-server"], optional = true }
windmill-queue.workspace = true
windmill-common = { workspace = true, default-features = false }
windmill-audit.workspace = true
@@ -50,7 +47,6 @@ windmill-git-sync.workspace = true
windmill-indexer = { workspace = true, optional = true }
windmill-worker.workspace = true
tokio.workspace = true
tokio-stream.workspace = true
anyhow.workspace = true
argon2.workspace = true
axum.workspace = true
@@ -102,7 +98,6 @@ async_zip = { workspace = true, optional = true }
regex.workspace = true
bytes.workspace = true
samael = { workspace = true, optional = true }
libxml = { workspace = true, optional = true }
async-recursion.workspace = true
rsa = { workspace = true, optional = true}
uuid.workspace = true
@@ -136,12 +131,6 @@ rust-postgres-native-tls = { workspace = true, optional = true}
rumqttc = { workspace = true, optional = true }
aws-sdk-sqs = { workspace = true, optional = true }
aws-config = { workspace = true, optional = true }
aws-sdk-sts = { workspace = true, optional = true }
google-cloud-pubsub = { workspace = true, optional = true }
google-cloud-googleapis = { workspace = true , optional = true }
tonic = { workspace = true, optional = true }
deno_error = { workspace = true, optional = true }
deno_core = { workspace = true, optional = true }
[build-dependencies]
deno_core = { workspace = true, optional = true }
tonic = { workspace = true, optional = true }

File diff suppressed because it is too large Load Diff

View File

@@ -21,8 +21,8 @@ pub fn workspaced_service(
_base_internal_url: String,
) -> (
Router,
Vec<tokio::task::JoinHandle<()>>,
Option<windmill_worker::JobCompletedSender>,
Option<tokio::task::JoinHandle<()>>,
windmill_worker::JobCompletedSender,
) {
use windmill_common::worker::Connection;
use windmill_worker::JobCompletedSender;
@@ -32,7 +32,7 @@ pub fn workspaced_service(
let router = Router::new();
(router, vec![], Some(job_completed_tx))
(router, None, job_completed_tx)
}
#[derive(Clone, Debug, Deserialize, Serialize)]

View File

@@ -20,14 +20,15 @@ use crate::{
use crate::{
job_helpers_ee::{
download_s3_file_internal, get_random_file_name, get_s3_resource,
get_workspace_s3_resource, upload_file_from_req, DownloadFileQuery,
get_workspace_s3_resource, load_image_preview_internal, upload_file_from_req,
DownloadFileQuery, LoadImagePreviewQuery,
},
users::fetch_api_authed_from_permissioned_as,
};
#[cfg(feature = "parquet")]
use axum::response::Response;
use axum::{
body::Body,
extract::{Extension, Json, Multipart, Path, Query},
extract::{Extension, Json, Path, Query},
response::IntoResponse,
routing::{delete, get, post},
Router,
@@ -50,6 +51,7 @@ use sqlx::{types::Uuid, FromRow};
use std::str;
use windmill_audit::audit_ee::audit_log;
use windmill_audit::ActionKind;
use windmill_common::variables::encrypt;
use windmill_common::{
apps::{AppScriptId, ListAppQuery},
cache::{self, future::FutureCachedExt},
@@ -61,24 +63,16 @@ use windmill_common::{
http_get_from_hub, not_found_if_none, paginate, query_elems_from_hub, require_admin,
Pagination, StripPath,
},
variables::{build_crypt, build_crypt_with_key_suffix, encrypt},
variables::{build_crypt, build_crypt_with_key_suffix},
worker::{to_raw_value, CLOUD_HOSTED},
HUB_BASE_URL,
};
#[cfg(feature = "parquet")]
use windmill_common::{jwt, s3_helpers::build_object_store_client};
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
use windmill_queue::{push, PushArgs, PushArgsOwned, PushIsolationLevel};
#[cfg(feature = "parquet")]
use hmac::Mac;
#[cfg(feature = "parquet")]
use windmill_common::{
jwt,
oauth2::HmacSha256,
s3_helpers::{build_object_store_client, S3Object},
variables::get_workspace_key,
};
pub fn workspaced_service() -> Router {
Router::new()
.route("/list", get(list_apps))
@@ -88,13 +82,10 @@ pub fn workspaced_service() -> Router {
.route("/get/draft/*path", get(get_app_w_draft))
.route("/secret_of/*path", get(get_secret_id))
.route("/get/v/*id", get(get_app_by_id))
.route("/get_data/v/*id", get(get_raw_app_data))
.route("/exists/*path", get(exists_app))
.route("/update/*path", post(update_app))
.route("/update_raw/*path", post(update_app_raw))
.route("/delete/*path", delete(delete_app))
.route("/create", post(create_app))
.route("/create_raw", post(create_app_raw))
.route("/history/p/*path", get(get_app_history))
.route("/get_latest_version/*path", get(get_latest_version))
.route("/history_update/a/:id/v/:version", post(update_app_history))
@@ -103,7 +94,6 @@ pub fn workspaced_service() -> Router {
get(list_paths_from_workspace_runnable),
)
.route("/custom_path_exists/*custom_path", get(custom_path_exists))
.route("/sign_s3_objects", post(sign_s3_objects))
}
pub fn unauthed_service() -> Router {
@@ -112,6 +102,10 @@ pub fn unauthed_service() -> Router {
.route("/upload_s3_file/*path", post(upload_s3_file_from_app))
.route("/delete_s3_file", delete(delete_s3_file_from_app))
.route("/download_s3_file/*path", get(download_s3_file_from_app))
.route(
"/load_image_preview/*path",
get(load_s3_file_image_preview_from_app),
)
.route("/public_app/:secret", get(get_public_app_by_secret))
.route("/public_resource/*path", get(get_public_resource))
}
@@ -138,12 +132,6 @@ pub struct ListableApp {
#[sqlx(default)]
#[serde(skip_serializing_if = "Option::is_none")]
pub deployment_msg: Option<String>,
#[serde(skip_serializing_if = "is_false")]
pub raw_app: bool,
}
fn is_false(b: &bool) -> bool {
!b
}
#[derive(FromRow, Serialize, Deserialize)]
@@ -244,8 +232,7 @@ pub struct S3Input {
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct S3Key {
s3_path: String,
#[serde(skip_serializing_if = "Option::is_none")]
storage: Option<String>,
resource: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
@@ -337,8 +324,7 @@ async fn list_apps(
"app.extra_perms",
"favorite.path IS NOT NULL as starred",
"draft.path IS NOT NULL as has_draft",
"draft_only",
"app_version.raw_app",
"draft_only"
])
.left()
.join("favorite")
@@ -397,44 +383,6 @@ async fn list_apps(
Ok(Json(rows))
}
async fn get_raw_app_data(Path((w_id, version_id)): Path<(String, String)>) -> Result<Response> {
let file_path = format!("/tmp/wmill/{}/{}", w_id, version_id);
let file = tokio::fs::File::open(file_path).await?;
let stream = tokio_util::io::ReaderStream::new(file);
let res = Response::builder().header(
http::header::CONTENT_TYPE,
if version_id.ends_with(".css") {
"text/css"
} else {
"text/javascript"
},
);
Ok(res.body(Body::from_stream(stream)).unwrap())
}
// async fn get_app_version(
// authed: ApiAuthed,
// Extension(user_db): Extension<UserDB>,
// Path((w_id, path)): Path<(String, StripPath)>,
// ) -> JsonResult<i64> {
// let path = path.to_path();
// let mut tx = user_db.begin(&authed).await?;
// let version_o = sqlx::query_scalar!(
// "SELECT app.versions[array_upper(app.versions, 1)] as version FROM app
// WHERE app.path = $1 AND app.workspace_id = $2",
// path,
// &w_id,
// )
// .fetch_optional(&mut *tx)
// .await?
// .flatten();
// tx.commit().await?;
// let version = not_found_if_none(version_o, "App", path)?;
// Ok(Json(version))
// }
async fn get_app(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
@@ -775,97 +723,6 @@ async fn get_secret_id(
Ok(hx)
}
macro_rules! process_app_multipart {
($authed:expr, $user_db:expr, $db:expr, $w_id:expr, $path:expr, $multipart:expr, $internal_fn:expr) => {
async {
let mut saved_app = None;
let mut uploaded_js = false;
//todo: use s3 instead
let file_path = format!("/tmp/wmill/{}", $w_id);
std::fs::create_dir_all(&file_path).unwrap();
let mut multipart = $multipart;
while let Some(field) = multipart.next_field().await.unwrap() {
let name = field.name().unwrap().to_string();
let data = field.bytes().await.unwrap();
if name == "app" {
let app = serde_json::from_slice(&data).map_err(to_anyhow)?;
let (ntx, npath, nid) = $internal_fn(
$authed.clone(),
$db.clone(),
$user_db.clone(),
$w_id,
$path,
true,
app,
)
.await?;
saved_app = Some((npath, nid, ntx));
} else if name == "js" {
if let Some((_npath, id, _tx)) = saved_app.as_ref() {
let file_path = format!("{}/{}.js", file_path, id);
std::fs::write(file_path, data).unwrap();
uploaded_js = true;
} else {
return Err(Error::BadRequest(
"App payload need to be created first".to_string(),
));
}
} else if name == "css" {
if let Some((_npath, id, _tx)) = saved_app.as_ref() {
let file_path = format!("{}/{}.css", file_path, id);
std::fs::write(file_path, data).unwrap();
} else {
return Err(Error::BadRequest(
"App payload need to be created first".to_string(),
));
}
} else {
return Err(Error::BadRequest(format!("Unsupported field: {}", name)));
}
}
if !uploaded_js {
return Err(Error::BadRequest("js or css file not uploaded".to_string()));
}
if let Some((npath, id, tx)) = saved_app {
tx.commit().await?;
Ok((npath, id))
} else {
Err(Error::BadRequest("App not created".to_string()))
}
}
};
}
async fn create_app_raw<'a>(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Extension(webhook): Extension<WebhookShared>,
Path(w_id): Path<String>,
multipart: Multipart,
) -> Result<(StatusCode, String)> {
let (path, _id) = process_app_multipart!(
authed,
user_db,
db,
&w_id,
"",
multipart,
|authed, db, user_db, w_id, _path, raw_app, app| create_app_internal(
authed, db, user_db, w_id, raw_app, app
)
)
.await?;
webhook.send_message(
w_id.clone(),
WebhookMessage::CreateApp { workspace: w_id, path: path.clone() },
);
Ok((StatusCode::CREATED, path))
}
async fn list_paths_from_workspace_runnable(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
@@ -894,36 +751,17 @@ async fn create_app(
Extension(db): Extension<DB>,
Extension(webhook): Extension<WebhookShared>,
Path(w_id): Path<String>,
Json(app): Json<CreateApp>,
Json(mut app): Json<CreateApp>,
) -> Result<(StatusCode, String)> {
let path = app.path.clone();
let (new_tx, _path, _id) = create_app_internal(authed, db, user_db, &w_id, false, app).await?;
new_tx.commit().await?;
webhook.send_message(
w_id.clone(),
WebhookMessage::CreateApp { workspace: w_id, path: path.clone() },
);
Ok((StatusCode::CREATED, path))
}
async fn create_app_internal<'a>(
authed: ApiAuthed,
db: sqlx::Pool<sqlx::Postgres>,
user_db: UserDB,
w_id: &String,
raw_app: bool,
mut app: CreateApp,
) -> Result<(sqlx::Transaction<'a, sqlx::Postgres>, String, i64)> {
let mut tx = user_db.clone().begin(&authed).await?;
app.policy.on_behalf_of = Some(username_to_permissioned_as(&authed.username));
app.policy.on_behalf_of_email = Some(authed.email.clone());
let path = app.path.clone();
if &app.path == "" {
return Err(Error::BadRequest("App path cannot be empty".to_string()));
}
let exists = sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM app WHERE path = $1 AND workspace_id = $2)",
&app.path,
@@ -932,19 +770,21 @@ async fn create_app_internal<'a>(
.fetch_one(&mut *tx)
.await?
.unwrap_or(false);
if exists {
return Err(Error::BadRequest(format!(
"App with path {} already exists",
&app.path
)));
}
if let Some(custom_path) = &app.custom_path {
require_admin(authed.is_admin, &authed.username)?;
let exists = sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM app WHERE custom_path = $1 AND ($2::TEXT IS NULL OR workspace_id = $2))",
custom_path,
if *CLOUD_HOSTED { Some(w_id) } else { None }
if *CLOUD_HOSTED { Some(&w_id) } else { None }
)
.fetch_one(&mut *tx)
.await?.unwrap_or(false);
@@ -956,6 +796,7 @@ async fn create_app_internal<'a>(
)));
}
}
sqlx::query!(
"DELETE FROM draft WHERE path = $1 AND workspace_id = $2 AND typ = 'app'",
&app.path,
@@ -963,6 +804,7 @@ async fn create_app_internal<'a>(
)
.execute(&mut *tx)
.await?;
let id = sqlx::query_scalar!(
"INSERT INTO app
(workspace_id, path, summary, policy, versions, draft_only, custom_path)
@@ -973,24 +815,24 @@ async fn create_app_internal<'a>(
json!(app.policy),
app.draft_only,
app.custom_path
.as_ref()
.map(|s| if s.is_empty() { None } else { Some(s) })
.flatten()
)
.fetch_one(&mut *tx)
.await?;
let v_id = sqlx::query_scalar!(
"INSERT INTO app_version
(app_id, value, created_by, raw_app)
VALUES ($1, $2::text::json, $3, $4) RETURNING id",
(app_id, value, created_by)
VALUES ($1, $2::text::json, $3) RETURNING id",
id,
//to preserve key orders
serde_json::to_string(&app.value).unwrap(),
authed.username,
raw_app
)
.fetch_one(&mut *tx)
.await?;
sqlx::query!(
"UPDATE app SET versions = array_append(versions, $1::bigint) WHERE id = $2",
v_id,
@@ -1004,20 +846,22 @@ async fn create_app_internal<'a>(
&authed,
"apps.create",
ActionKind::Create,
w_id,
&w_id,
Some(&app.path),
None,
)
.await?;
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = &app.deployment_message {
if let Some(dm) = app.deployment_message {
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
let tx = PushIsolationLevel::Transaction(tx);
let (dependency_job_uuid, new_tx) = push(
&db,
tx,
w_id,
&w_id,
JobPayload::AppDependencies { path: app.path.clone(), version: v_id },
PushArgs { args: &args, extra: None },
&authed.username,
@@ -1041,7 +885,14 @@ async fn create_app_internal<'a>(
.await?;
tracing::info!("Pushed app dependency job {}", dependency_job_uuid);
Ok((new_tx, path, v_id))
new_tx.commit().await?;
webhook.send_message(
w_id.clone(),
WebhookMessage::CreateApp { workspace: w_id, path: app.path.clone() },
);
Ok((StatusCode::CREATED, app.path))
}
async fn list_hub_apps(Extension(db): Extension<DB>) -> impl IntoResponse {
@@ -1162,76 +1013,12 @@ async fn update_app(
Path((w_id, path)): Path<(String, StripPath)>,
Json(ns): Json<EditApp>,
) -> Result<String> {
// create_app_internal(authed, user_db, db, &w_id, &mut app).await?;
let path = path.to_path();
let opath = path.to_string();
let (new_tx, npath, _v_id) =
update_app_internal(authed, db, user_db, &w_id, path, false, ns).await?;
new_tx.commit().await?;
webhook.send_message(
w_id.clone(),
WebhookMessage::UpdateApp {
workspace: w_id.clone(),
old_path: opath.clone(),
new_path: npath.clone(),
},
);
Ok(format!("app {} updated (npath: {:?})", opath, npath))
}
async fn update_app_raw<'a>(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Extension(webhook): Extension<WebhookShared>,
Path((w_id, path)): Path<(String, StripPath)>,
multipart: Multipart,
) -> Result<String> {
let path = path.to_path();
let opath = path.to_string();
let (npath, _id) = process_app_multipart!(
authed,
user_db,
db,
&w_id,
path,
multipart,
update_app_internal
)
.await?;
webhook.send_message(
w_id.clone(),
WebhookMessage::UpdateApp {
workspace: w_id.clone(),
old_path: opath.to_owned(),
new_path: npath.clone(),
},
);
Ok(format!("app {} updated (npath: {:?})", opath, npath))
}
// async fn create_app_internal<'a>(
// authed: ApiAuthed,
// db: sqlx::Pool<sqlx::Postgres>,
// user_db: UserDB,
// w_id: &String,
// app: &mut CreateApp,
// )
async fn update_app_internal<'a>(
authed: ApiAuthed,
db: sqlx::Pool<sqlx::Postgres>,
user_db: UserDB,
w_id: &str,
path: &str,
raw_app: bool,
ns: EditApp,
) -> Result<(sqlx::Transaction<'a, sqlx::Postgres>, String, i64)> {
use sql_builder::prelude::*;
let path = path.to_path();
let mut tx = user_db.clone().begin(&authed).await?;
let npath = if ns.policy.is_some()
|| ns.path.is_some()
|| ns.summary.is_some()
@@ -1278,7 +1065,7 @@ async fn update_app_internal<'a>(
let exists = sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM app WHERE custom_path = $1 AND ($2::TEXT IS NULL OR workspace_id = $2) AND NOT (path = $3 AND workspace_id = $4))",
ncustom_path,
if *CLOUD_HOSTED { Some(w_id) } else { None },
if *CLOUD_HOSTED { Some(&w_id) } else { None },
path,
w_id
)
@@ -1325,13 +1112,12 @@ async fn update_app_internal<'a>(
let v_id = sqlx::query_scalar!(
"INSERT INTO app_version
(app_id, value, created_by, raw_app)
VALUES ($1, $2::text::json, $3, $4) RETURNING id",
(app_id, value, created_by)
VALUES ($1, $2::text::json, $3) RETURNING id",
app_id,
//to preserve key orders
serde_json::to_string(&nvalue).unwrap(),
authed.username,
raw_app
)
.fetch_one(&mut *tx)
.await?;
@@ -1362,6 +1148,7 @@ async fn update_app_internal<'a>(
)));
}
};
sqlx::query!(
"DELETE FROM draft WHERE path = $1 AND workspace_id = $2 AND typ = 'app'",
path,
@@ -1369,26 +1156,29 @@ async fn update_app_internal<'a>(
)
.execute(&mut *tx)
.await?;
audit_log(
&mut *tx,
&authed,
"apps.update",
ActionKind::Update,
w_id,
&w_id,
Some(&npath),
None,
)
.await?;
let tx = PushIsolationLevel::Transaction(tx);
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = ns.deployment_message {
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
args.insert("parent_path".to_string(), to_raw_value(&path));
let (dependency_job_uuid, new_tx) = push(
&db,
tx,
w_id,
&w_id,
JobPayload::AppDependencies { path: npath.clone(), version: v_id },
PushArgs { args: &args, extra: None },
&authed.username,
@@ -1411,7 +1201,18 @@ async fn update_app_internal<'a>(
)
.await?;
tracing::info!("Pushed app dependency job {}", dependency_job_uuid);
Ok((new_tx, npath, v_id))
new_tx.commit().await?;
webhook.send_message(
w_id.clone(),
WebhookMessage::UpdateApp {
workspace: w_id,
old_path: path.to_owned(),
new_path: npath.clone(),
},
);
Ok(format!("app {} updated (npath: {:?})", path, npath))
}
#[derive(Debug, Deserialize, Clone)]
@@ -1753,97 +1554,15 @@ struct UploadFileToS3Query {
#[cfg(feature = "parquet")]
#[derive(Serialize, Deserialize)]
struct S3DeleteTokenClaims {
struct DeleteTokenClaims {
file_key: String,
on_behalf_of_email: String,
permissioned_as: String,
username: String,
s3_resource_path: Option<String>,
workspace: String,
pub exp: usize,
}
#[cfg(feature = "parquet")]
#[derive(Deserialize)]
struct S3TokenRequestBody {
s3_objects: Vec<S3Object>,
}
#[cfg(feature = "parquet")]
async fn sign_s3_objects(
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Json(body): Json<S3TokenRequestBody>,
) -> Result<Json<Vec<S3Object>>> {
let workspace_key = get_workspace_key(&w_id, &db).await?;
let futures = body.s3_objects.into_iter().map(|s3_object| async {
let exp = (chrono::Utc::now() + chrono::Duration::hours(12)).timestamp();
let mut message = format!("file_key={}&exp={}", s3_object.s3.clone(), exp);
if let Some(ref storage) = s3_object.storage {
message = format!("{}&storage={}", message, storage);
}
let mut max = HmacSha256::new_from_slice(workspace_key.as_bytes())
.map_err(|err| Error::internal_err(format!("Failed to create hmac: {}", err)))?;
max.update(message.as_bytes());
let result = max.finalize();
let signature = hex::encode(result.into_bytes());
let presigned = format!("exp={}&sig={}", exp, signature);
Ok::<_, Error>(S3Object { presigned: Some(presigned), ..s3_object })
});
let signed_s3_objects = futures::future::try_join_all(futures).await?;
Ok(Json(signed_s3_objects))
}
#[cfg(feature = "parquet")]
async fn validate_s3_signature(file_query: &AppS3FileQuery, w_id: &str, db: &DB) -> Result<()> {
let workspace_key = get_workspace_key(w_id, &db).await?;
let Some(exp) = file_query
.exp
.as_ref()
.map(|e| e.parse::<i64>().unwrap_or_default())
else {
return Err(Error::BadRequest("Missing exp".to_string()));
};
let Some(ref sig) = file_query.sig else {
return Err(Error::BadRequest("Missing signature".to_string()));
};
let mut message = format!("file_key={}&exp={}", file_query.s3, exp);
if let Some(ref storage) = file_query.storage {
message = format!("{}&storage={}", message, storage);
}
let mut mac = HmacSha256::new_from_slice(workspace_key.as_bytes())
.map_err(|err| Error::internal_err(format!("Failed to create hmac: {}", err)))?;
mac.update(message.as_bytes());
let sig_bytes = hex::decode(sig)?;
mac.verify_slice(&sig_bytes)
.map_err(|err| Error::BadRequest(format!("Invalid signature: {}", err)))?;
if exp < chrono::Utc::now().timestamp() {
return Err(Error::BadRequest("Signature expired".to_string()));
}
Ok(())
}
#[cfg(not(feature = "parquet"))]
async fn sign_s3_objects() -> Result<()> {
return Err(Error::BadRequest(
"This endpoint requires the parquet feature to be enabled".to_string(),
));
}
#[cfg(feature = "parquet")]
#[derive(Serialize)]
struct AppUploadFileResponse {
@@ -2098,14 +1817,13 @@ async fn upload_s3_file_from_app(
upload_file_from_req(s3_client, &file_key, request, options).await?;
let delete_token = jwt::encode_with_internal_secret(S3DeleteTokenClaims {
let delete_token = jwt::encode_with_internal_secret(DeleteTokenClaims {
file_key: file_key.clone(),
on_behalf_of_email,
permissioned_as,
username,
s3_resource_path: query.s3_resource_path,
workspace: w_id.clone(),
exp: (chrono::Utc::now() + chrono::Duration::hours(12)).timestamp() as usize,
exp: (chrono::Utc::now() + chrono::Duration::seconds(3600 * 24)).timestamp() as usize,
})
.await?;
@@ -2125,19 +1843,14 @@ async fn delete_s3_file_from_app(
Path(w_id): Path<String>,
Query(query): Query<DeleteS3FileQuery>,
) -> Result<()> {
let S3DeleteTokenClaims {
let DeleteTokenClaims {
file_key,
on_behalf_of_email,
permissioned_as,
username,
s3_resource_path,
workspace,
..
} = jwt::decode_with_internal_secret::<S3DeleteTokenClaims>(&query.delete_token).await?;
if workspace != w_id {
return Err(Error::BadRequest("Invalid workspace".to_string()));
}
} = jwt::decode_with_internal_secret::<DeleteTokenClaims>(&query.delete_token).await?;
let on_behalf_authed = fetch_api_authed_from_permissioned_as(
permissioned_as,
@@ -2245,7 +1958,7 @@ async fn get_on_behalf_authed_from_app(
async fn check_if_allowed_to_access_s3_file_from_app(
db: &DB,
opt_authed: &Option<ApiAuthed>,
file_query: &AppS3FileQuery,
file_key: &str,
w_id: &str,
path: &str,
policy: &Policy,
@@ -2253,59 +1966,40 @@ async fn check_if_allowed_to_access_s3_file_from_app(
// if anonymous, check that the file was the result of an app script ran by an anonymous user in the last 3 hours
// otherwise, if logged in, allow any file (TODO: change that when we implement better s3 policy)
if file_query.sig.is_some() {
validate_s3_signature(file_query, w_id, &db).await
} else if opt_authed.is_some() {
Ok(())
} else {
let allowed = policy
.allowed_s3_keys
.as_ref()
.unwrap()
.iter()
.any(|key| key.s3_path == file_query.s3 && key.storage == file_query.storage)
|| {
sqlx::query_scalar!(
r#"SELECT EXISTS (
SELECT 1 FROM v2_as_completed_job
WHERE workspace_id = $2
let allowed = opt_authed.is_some()
|| sqlx::query_scalar!(
r#"SELECT EXISTS (
SELECT 1 FROM v2_as_completed_job
WHERE workspace_id = $2
AND (job_kind = 'appscript' OR job_kind = 'preview')
AND created_by = 'anonymous'
AND created_by = 'anonymous'
AND started_at > now() - interval '3 hours'
AND script_path LIKE $3 || '/%'
AND result @> ('{"s3":"' || $1 || '"}')::jsonb
AND script_path LIKE $3 || '/%'
AND result @> ('{"s3":"' || $1 || '"}')::jsonb
)"#,
file_query.s3,
w_id,
path,
)
.fetch_one(db)
.await?
.unwrap_or(false)
};
file_key,
w_id,
path,
)
.fetch_one(db)
.await?
.unwrap_or(false)
if !allowed {
Err(Error::BadRequest("File restricted".to_string()))
} else {
Ok(())
}
// check if the file is allowed by the allowed_s3_keys policy
|| policy.allowed_s3_keys.as_ref().unwrap().iter().any(|key| key.s3_path == file_key);
if !allowed {
Err(Error::BadRequest("File restricted".to_string()))
} else {
Ok(())
}
}
#[cfg(feature = "parquet")]
#[derive(Deserialize, Debug)]
struct AppS3FileQuery {
s3: String,
storage: Option<String>,
sig: Option<String>,
exp: Option<String>,
}
#[cfg(feature = "parquet")]
#[derive(Deserialize, Debug)]
struct AppS3FileQueryWithForceViewerAllowedS3Keys {
#[derive(Deserialize)]
pub struct DownloadFileQueryWithForceViewerAllowedS3Keys {
#[serde(flatten)]
pub file_query: AppS3FileQuery,
pub file_query: DownloadFileQuery,
pub force_viewer_allowed_s3_keys: Option<String>,
}
@@ -2314,7 +2008,7 @@ async fn download_s3_file_from_app(
OptAuthed(opt_authed): OptAuthed,
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
Query(query): Query<AppS3FileQueryWithForceViewerAllowedS3Keys>,
Query(query): Query<DownloadFileQueryWithForceViewerAllowedS3Keys>,
) -> Result<Response> {
let path = path.to_path();
@@ -2333,26 +2027,46 @@ async fn download_s3_file_from_app(
check_if_allowed_to_access_s3_file_from_app(
&db,
&opt_authed,
&query.file_query,
&query.file_query.file_key,
&w_id,
&path,
&policy,
)
.await?;
download_s3_file_internal(
on_behalf_authed,
download_s3_file_internal(on_behalf_authed, &db, None, "", &w_id, query.file_query).await
}
#[cfg(not(feature = "parquet"))]
async fn load_s3_file_image_preview_from_app() -> Result<()> {
return Err(Error::BadRequest(
"This endpoint requires the parquet feature to be enabled".to_string(),
));
}
#[cfg(feature = "parquet")]
async fn load_s3_file_image_preview_from_app(
OptAuthed(opt_authed): OptAuthed,
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
Query(query): Query<LoadImagePreviewQuery>,
) -> Result<Response> {
let path = path.to_path();
let (on_behalf_authed, policy) =
get_on_behalf_authed_from_app(&db, &path, &w_id, &opt_authed, None).await?;
check_if_allowed_to_access_s3_file_from_app(
&db,
None,
"",
&opt_authed,
&query.file_key,
&w_id,
DownloadFileQuery {
file_key: query.file_query.s3,
s3_resource_path: None,
storage: query.file_query.storage,
},
&path,
&policy,
)
.await
.await?;
load_image_preview_internal(on_behalf_authed, &db, "", &w_id, query).await
}
fn get_on_behalf_of(policy: &Policy) -> Result<(String, String)> {

View File

@@ -511,13 +511,6 @@ where
let path_vec: Vec<&str> = original_uri.path().split("/").collect();
let workspace_id = if path_vec.len() >= 4 && path_vec[0] == "" && path_vec[2] == "w" {
Some(path_vec[3].to_owned())
} else if path_vec.len() >= 5
&& path_vec[0] == ""
&& path_vec[1] == "api"
&& path_vec[2] == "mcp"
&& path_vec[3] == "w"
{
Some(path_vec[4].to_string())
} else {
if path_vec.len() >= 5
&& path_vec[0] == ""

View File

@@ -18,22 +18,16 @@ use {
#[cfg(all(feature = "enterprise", feature = "gcp_trigger"))]
use crate::gcp_triggers_ee::{
manage_google_subscription, process_google_push_request, validate_jwt_token,
CreateUpdateConfig, SubscriptionMode,
manage_google_subscription, process_google_push_request, validate_jwt_token, SubscriptionMode,
};
#[cfg(all(feature = "enterprise", feature = "sqs_trigger"))]
use windmill_common::auth::aws::AwsAuthResourceType;
#[cfg(any(
feature = "http_trigger",
all(feature = "enterprise", feature = "gcp_trigger")
))]
use {
axum::extract::Request,
http::HeaderMap,
serde::de::DeserializeOwned,
windmill_common::{error::Error, utils::empty_string_as_none},
axum::extract::Request, http::HeaderMap, serde::de::DeserializeOwned,
windmill_common::error::Error,
};
#[cfg(all(feature = "enterprise", feature = "kafka"))]
@@ -156,20 +150,14 @@ pub struct SqsTriggerConfig {
pub queue_url: String,
pub aws_resource_path: String,
pub message_attributes: Option<Vec<String>>,
pub aws_auth_resource_type: AwsAuthResourceType,
}
#[cfg(all(feature = "enterprise", feature = "gcp_trigger"))]
#[derive(Debug, Serialize, Deserialize)]
pub struct GcpTriggerConfig {
pub gcp_resource_path: String,
pub subscription_mode: SubscriptionMode,
#[serde(default, deserialize_with = "empty_string_as_none")]
pub subscription_id: Option<String>,
#[serde(default, deserialize_with = "empty_string_as_none")]
pub base_endpoint: Option<String>,
#[serde(flatten)]
pub create_update: Option<CreateUpdateConfig>,
pub subscription_mode: SubscriptionMode,
pub topic_id: String,
}
@@ -378,15 +366,11 @@ async fn set_gcp_trigger_config(
&gcp_config.gcp_resource_path,
&capture_config.path,
&gcp_config.topic_id,
&mut gcp_config.subscription_id,
&mut gcp_config.base_endpoint,
gcp_config.subscription_mode,
gcp_config.create_update,
false,
)
.await?;
gcp_config.create_update = Some(config);
gcp_config.subscription_mode = SubscriptionMode::CreateUpdate;
gcp_config.subscription_mode = SubscriptionMode::CreateUpdate(config);
capture_config.trigger_config = Some(TriggerConfig::Gcp(gcp_config));
Ok(capture_config)
@@ -903,7 +887,7 @@ async fn gcp_payload(
let authed = fetch_api_authed(owner.clone(), email, &w_id, &db, None).await?;
let Some(config) = &gcp_trigger_config.create_update else {
let SubscriptionMode::CreateUpdate(config) = &gcp_trigger_config.subscription_mode else {
return Err(Error::BadConfig("Bad config".to_string()));
};

View File

@@ -52,12 +52,11 @@ pub struct ExistingGcpSubscription {
pub base_endpoint: String,
}
#[derive(Debug, Deserialize, Serialize, sqlx::Type)]
#[serde(rename_all = "snake_case")]
#[sqlx(type_name = "GCP_SUBSCRIPTION_MODE", rename_all = "snake_case")]
#[derive(Debug, Deserialize, Serialize)]
#[serde(tag = "subscription_mode", rename_all = "snake_case")]
pub enum SubscriptionMode {
Existing,
CreateUpdate,
Existing(ExistingGcpSubscription),
CreateUpdate(CreateUpdateConfig),
}
pub fn workspaced_service() -> Router {
@@ -78,11 +77,7 @@ pub async fn manage_google_subscription(
_gcp_resource_path: &str,
_path: &str,
_topic_id: &str,
_subscription_id: &mut Option<String>,
_base_endpoint: &mut Option<String>,
_subscription_mode: SubscriptionMode,
_create_update_config: Option<CreateUpdateConfig>,
_trigger_mode: bool,
) -> WindmillResult<CreateUpdateConfig> {
Ok(CreateUpdateConfig::default())
}
@@ -122,7 +117,6 @@ pub struct GcpTrigger {
pub subscription_id: String,
pub delivery_type: DeliveryType,
pub delivery_config: Option<SqlxJson<PushConfig>>,
pub subscription_mode: SubscriptionMode,
pub topic_id: String,
pub path: String,
pub script_path: String,

View File

@@ -253,8 +253,8 @@ mod zoom {
use super::*;
#[derive(Debug, Deserialize)]
#[serde(rename_all = "snake_case")]
struct ZoomPayload {
#[serde(rename = "plainToken")]
plain_token: String,
}

View File

@@ -30,20 +30,12 @@ pub struct UploadFileResponse {
#[derive(Deserialize)]
pub struct LoadImagePreviewQuery {
#[allow(dead_code)]
pub file_key: String,
#[allow(dead_code)]
pub storage: Option<String>,
}
#[derive(Deserialize)]
pub struct DownloadFileQuery {
#[allow(dead_code)]
pub file_key: String,
#[allow(dead_code)]
pub storage: Option<String>,
#[allow(dead_code)]
pub s3_resource_path: Option<String>,
}
pub fn workspaced_service() -> Router {
@@ -119,3 +111,16 @@ pub async fn download_s3_file_internal(
"Not implemented in Windmill's Open Source repository".to_string(),
))
}
#[cfg(feature = "parquet")]
pub async fn load_image_preview_internal(
_authed: ApiAuthed,
_db: &DB,
_token: &str,
_w_id: &str,
_query: LoadImagePreviewQuery,
) -> error::Result<Response> {
Err(error::Error::internal_err(
"Not implemented in Windmill's Open Source repository".to_string(),
))
}

View File

@@ -8,9 +8,7 @@
use axum::body::Body;
use axum::http::HeaderValue;
#[cfg(feature = "deno_core")]
use deno_core::{op2, serde_v8, v8, JsRuntime, OpState};
use futures::{StreamExt, TryFutureExt};
use futures::TryFutureExt;
use http::{HeaderMap, HeaderName};
use itertools::Itertools;
use quick_cache::sync::Cache;
@@ -142,13 +140,6 @@ pub fn workspaced_service() -> Router {
.layer(cors.clone())
.layer(ce_headers.clone()),
)
.route(
"/run/batch_rerun_jobs",
post(batch_rerun_jobs)
.head(|| async { "" })
.layer(cors.clone())
.layer(ce_headers.clone()),
)
.route(
"/run/workflow_as_code/:job_id/:entrypoint",
post(run_workflow_as_code)
@@ -212,13 +203,6 @@ pub fn workspaced_service() -> Router {
"/list",
get(list_jobs).layer(Extension(api_list_jobs_query_duration)),
)
.route(
"/list_selected_job_groups",
// We use post because sending a huge array as a query param can produce
// URLs that may be too long
post(list_selected_job_groups),
)
.route("/list_filtered_uuids", get(list_filtered_job_uuids))
.route("/queue/list", get(list_queue_jobs))
.route("/queue/count", get(count_queue_jobs))
.route("/queue/list_filtered_uuids", get(list_filtered_uuids))
@@ -661,48 +645,6 @@ async fn get_flow_job_debug_info(
}
}
async fn list_selected_job_groups(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path(w_id): Path<String>,
Json(uuids): Json<Vec<Uuid>>,
) -> error::Result<Response> {
let mut tx = user_db.begin(&authed).await?;
let results = sqlx::query_scalar!(
r#"SELECT jsonb_build_object(
'kind', jb.kind,
'script_path', jb.runnable_path,
'latest_schema', COALESCE(
(SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.workspace_id = $1 AND s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),
(SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.workspace_id = $1 AND flow.path = jb.runnable_path AND jb.kind = 'flow')
),
'schemas', ARRAY(
SELECT jsonb_build_object(
'script_hash', LPAD(TO_HEX(COALESCE(s.hash, f.id)), 16, '0'),
'job_ids', ARRAY_AGG(DISTINCT j.id),
'schema', (ARRAY_AGG(COALESCE(s.schema, f.schema)))[1]
) FROM v2_job j
LEFT JOIN script s ON s.hash = j.runnable_id AND j.kind = 'script'
LEFT JOIN flow_version f ON f.id = j.runnable_id AND j.kind = 'flow'
WHERE j.id = ANY(ARRAY_AGG(jb.id))
GROUP BY COALESCE(s.hash, f.id)
)
) FROM v2_job jb
WHERE (jb.kind = 'flow' OR jb.kind = 'script')
AND jb.workspace_id = $1 AND jb.id = ANY($2)
GROUP BY jb.kind, jb.runnable_path"#,
&w_id,
&uuids
)
.fetch_all(&mut *tx)
.await?;
tx.commit().await?;
Ok(Json(results).into_response())
}
#[derive(Deserialize)]
struct GetJobQuery {
pub no_logs: Option<bool>,
@@ -1803,37 +1745,6 @@ async fn cancel_selection(
cancel_jobs(jobs_to_cancel, &db, authed.username.as_str(), w_id.as_str()).await
}
async fn list_filtered_job_uuids(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Query(lq): Query<ListCompletedQuery>,
) -> error::JsonResult<Vec<Uuid>> {
require_admin(authed.is_admin, &authed.username)?;
check_scopes(&authed, || format!("jobs:listjobs"))?;
let mut sqlb = list_completed_jobs_query(
w_id.as_str(),
None,
0,
&lq,
&["v2_job.id"],
false,
get_scope_tags(&authed),
);
let sqlb2 = list_queue_jobs_query(
w_id.as_str(),
&lq.into(),
&["v2_job.id"],
Pagination { page: None, per_page: None },
false,
get_scope_tags(&authed),
);
let query = sqlb.union_all(sqlb2.subquery()?).subquery()?;
let ids = sqlx::query_scalar(query.as_str()).fetch_all(&db).await?;
Ok(Json(ids))
}
async fn list_filtered_uuids(
authed: ApiAuthed,
Extension(db): Extension<DB>,
@@ -1990,7 +1901,7 @@ async fn list_jobs(
let sqlc = if lq.running.is_none() {
Some(list_completed_jobs_query(
&w_id,
Some(per_page + offset),
per_page + offset,
0,
&ListCompletedQuery { order_desc: Some(true), ..lqc },
UnifiedJob::completed_job_fields(),
@@ -2029,9 +1940,7 @@ async fn list_jobs(
} else {
if sqlc.is_none() {
return Err(error::Error::BadRequest(
"cannot specify success, label, created_or_started_before, or starte
d_before with running"
.to_string(),
"cannot specify success, label, created_or_started_before, or started_before with running".to_string(),
));
}
sqlc.unwrap().limit(per_page).offset(offset).query()?
@@ -3095,17 +3004,16 @@ struct CancelJob {
enum PreviewKind {
Code,
Identity,
Http,
Noop,
Bundle,
Tarbundle,
ScriptHash,
}
#[derive(Deserialize)]
struct Preview {
content: Option<String>,
kind: Option<PreviewKind>,
script_hash: Option<String>,
path: Option<String>,
args: Option<HashMap<String, Box<JsonRawValue>>>,
language: Option<ScriptLang>,
@@ -3244,272 +3152,6 @@ pub async fn check_license_key_valid() -> error::Result<()> {
Ok(())
}
use windmill_common::flows::InputTransform;
#[derive(Deserialize)]
struct BatchReRunJobsBodyArgs {
job_ids: Vec<Uuid>,
script_options_by_path: HashMap<String, BatchReRunOptions>,
flow_options_by_path: HashMap<String, BatchReRunOptions>,
}
#[derive(Deserialize)]
struct BatchReRunOptions {
input_transforms: Option<HashMap<String, InputTransform>>,
use_latest_version: Option<bool>,
}
#[derive(sqlx::FromRow, Serialize, Clone)]
struct BatchReRunQueryReturnType {
id: Uuid,
kind: JobKind,
script_path: String,
script_hash: ScriptHash,
input: serde_json::Value,
scheduled_for: chrono::DateTime<Utc>,
#[serde(skip_serializing_if = "Option::is_none")]
schema: Option<serde_json::Value>,
}
#[cfg(feature = "deno_core")]
#[op2]
#[string]
fn get_deno_core_job_value(state: &mut OpState) -> Option<String> {
let obj = state.borrow::<BatchReRunQueryReturnType>();
let str = serde_json::to_string(&obj).ok()?;
Some(str)
}
#[cfg(feature = "deno_core")]
async fn batch_rerun_compute_js_expression(
expr: String,
job: BatchReRunQueryReturnType,
) -> error::Result<Box<RawValue>> {
let ext = deno_core::Extension {
name: "batch_rerun_arg_transform_ext",
ops: vec![get_deno_core_job_value()].into(),
..Default::default()
};
let mut isolate =
JsRuntime::new(deno_core::RuntimeOptions { extensions: vec![ext], ..Default::default() });
{
let op_state = isolate.op_state();
let mut op_state = op_state.borrow_mut();
op_state.put(BatchReRunQueryReturnType { schema: None, ..job });
}
isolate
.execute_script(
"<batch_rerun_arg_transform>",
"let job = JSON.parse(Deno.core.ops.get_deno_core_job_value());",
)
.map_err(|e| Error::ExecutionErr(e.to_string()))?;
// Run user expr
let result = isolate
.execute_script("<batch_rerun_arg_transform>", expr)
.map_err(|e| Error::ExecutionErr(e.to_string()))?;
let mut scope = isolate.handle_scope();
let result = v8::Local::new(&mut scope, result);
let result: serde_json::Value =
serde_v8::from_v8(&mut scope, result).map_err(|e| Error::ExecutionErr(e.to_string()))?;
let result = JsonRawValue::from_string(result.to_string())?;
Ok(result)
}
async fn batch_rerun_jobs(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path(w_id): Path<String>,
Json(body): Json<BatchReRunJobsBodyArgs>,
) -> Response {
let stream = batch_rerun_jobs_inner(authed, db, user_db, w_id, body);
let body = axum::body::Body::from_stream(stream.map(Result::<_, std::convert::Infallible>::Ok));
Response::builder()
.status(201)
.header("Content-Type", "text/event-stream")
.header("Cache-Control", "no-cache")
.body(body)
.unwrap()
}
fn batch_rerun_jobs_inner(
authed: ApiAuthed,
db: DB,
user_db: UserDB,
w_id: String,
body: BatchReRunJobsBodyArgs,
) -> impl futures::Stream<Item = String> {
let (tx, rx) = tokio::sync::mpsc::channel(10);
tokio::spawn(async move {
let mut job_stream = sqlx::query_as!(
BatchReRunQueryReturnType,
r#"SELECT
j.id,
j.kind AS "kind: _",
COALESCE(s.path, f.path) AS "script_path!",
COALESCE(s.hash, f.id) AS "script_hash!: _",
COALESCE(jc.started_at, jq.scheduled_for, make_date(1970, 1, 1)) AS "scheduled_for!: _",
args AS input,
COALESCE(s.schema, f.schema) AS "schema: _"
FROM v2_job j
LEFT JOIN script s ON j.runnable_id = s.hash AND j.kind = 'script'
LEFT JOIN flow_version f ON j.runnable_id = f.id AND j.runnable_path = f.path AND j.kind = 'flow'
LEFT JOIN v2_job_completed jc ON jc.id = j.id
LEFT JOIN v2_job_queue jq ON jq.id = j.id
WHERE j.id = ANY($1)
AND j.workspace_id = $2
AND COALESCE(s.hash, f.id) IS NOT NULL
AND COALESCE(s.path, f.path) IS NOT NULL"#,
&body.job_ids,
w_id
).fetch(&db);
while let Some(Ok(job)) = job_stream.next().await {
let job_result =
batch_rerun_handle_job(&job, &authed, &db, &user_db, &w_id, &body).await;
let send_to_stream_result = tx
.send(match job_result {
Ok(uuid) => format!("{}\n", uuid),
Err(err) => format!("Error: {}\n", err.to_string()),
})
.await;
match send_to_stream_result {
Ok(_) => {}
Err(e) => tracing::error!("Couldn't re-run job {}: {}", job.id, e.to_string()),
}
}
});
tokio_stream::wrappers::ReceiverStream::new(rx)
}
async fn batch_rerun_handle_job(
job: &BatchReRunQueryReturnType,
authed: &ApiAuthed,
db: &DB,
user_db: &UserDB,
w_id: &String,
body: &BatchReRunJobsBodyArgs,
) -> error::Result<String> {
let options = if matches!(job.kind, JobKind::Script) {
&body.script_options_by_path
} else {
&body.flow_options_by_path
}
.get(&job.script_path);
let mut args: HashMap<String, Box<RawValue>> = serde_json::from_value(job.input.clone())?;
let use_latest_version = options.and_then(|o| o.use_latest_version).unwrap_or(false);
let input_transforms = options
.and_then(|o| o.input_transforms.as_ref())
.map(|t| t.iter())
.into_iter()
.flatten();
let latest_schema;
let schema = if use_latest_version {
latest_schema = sqlx::query_scalar!(
r#"SELECT COALESCE(
(SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),
(SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = jb.runnable_path AND jb.kind = 'flow')
) FROM v2_job jb
WHERE jb.id = $1 AND jb.workspace_id = $2
GROUP BY jb.kind, jb.runnable_path"#,
&job.id,
&w_id
).fetch_optional(db).await?.flatten();
latest_schema.as_ref()
} else {
job.schema.as_ref()
};
let schema = schema
.and_then(serde_json::Value::as_object)
.and_then(|s| s.get("properties"))
.and_then(serde_json::Value::as_object);
for (property_name, transform) in input_transforms {
let schema_has_key = schema
.map(|s| s.contains_key(property_name))
.unwrap_or(false);
if !schema_has_key {
continue;
}
match transform {
InputTransform::Static { value } => {
args.insert(property_name.clone(), value.clone());
}
InputTransform::Javascript { expr } => {
#[cfg(not(feature = "deno_core"))]
Err(error::Error::ExecutionErr(
format!("deno_core feature is not activated, cannot evaluate: {expr}")
.to_string(),
))?;
#[cfg(feature = "deno_core")]
args.insert(
property_name.clone(),
batch_rerun_compute_js_expression(expr.clone(), job.clone()).await?,
);
}
}
}
// Call appropriate function to push job to queue
match job.kind {
JobKind::Flow => {
let result = run_flow_by_path_inner(
authed.clone(),
db.clone(),
user_db.clone(),
w_id.clone(),
StripPath(job.script_path.clone()),
RunJobQuery { ..Default::default() },
PushArgsOwned { extra: None, args },
None,
)
.await;
if let Ok((_, uuid)) = result {
return Ok(uuid);
}
}
JobKind::Script => {
let result = if use_latest_version {
run_script_by_path_inner(
authed.clone(),
db.clone(),
user_db.clone(),
w_id.clone(),
StripPath(job.script_path.clone()),
RunJobQuery { ..Default::default() },
PushArgsOwned { extra: None, args },
None,
)
.await
} else {
run_job_by_hash_inner(
authed.clone(),
db.clone(),
user_db.clone(),
w_id.clone(),
job.script_hash,
RunJobQuery { ..Default::default() },
PushArgsOwned { extra: None, args },
None,
)
.await
};
if let Ok((_, uuid)) = result {
return Ok(uuid);
}
}
_ => {}
}
Err(error::Error::ExecutionErr(
format!("Couldn't re-run job {}", job.id).to_string(),
))
}
pub async fn run_flow_by_path(
authed: ApiAuthed,
Extension(db): Extension<DB>,
@@ -4622,6 +4264,7 @@ pub async fn run_wait_result_script_by_hash(
check_license_key_valid().await?;
let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
check_queue_too_long(&db, run_query.queue_limit).await?;
let hash = script_hash.0;
@@ -4855,10 +4498,7 @@ async fn run_preview_script(
Some(PreviewKind::Identity) => JobPayload::Identity,
Some(PreviewKind::Noop) => JobPayload::Noop,
_ => JobPayload::Code(RawCode {
hash: preview
.script_hash
.as_ref()
.and_then(|s| windmill_common::scripts::to_i64(s).ok()),
hash: None,
content: preview.content.unwrap_or_default(),
path: preview.path,
language: preview.language.unwrap_or(ScriptLang::Deno),
@@ -5455,21 +5095,6 @@ async fn add_batch_jobs(
.execute(&mut *tx)
.await?;
sqlx::query!(
"INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id)
SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8",
&uuids,
authed.email,
authed.username,
authed.is_admin,
authed.is_operator,
&[],
&[],
w_id,
)
.execute(&mut *tx)
.await?;
if let Some(flow_status) = flow_status {
sqlx::query!(
"INSERT INTO v2_job_status (id, flow_status)
@@ -5838,7 +5463,7 @@ pub fn filter_list_completed_query(
if let Some(label) = &lq.label {
if lq.allow_wildcards.unwrap_or(false) {
let wh = format!(
"EXISTS (SELECT 1 FROM jsonb_array_elements_text(result->'wm_labels') label WHERE jsonb_typeof(result->'wm_labels') = 'array' AND label LIKE '{}')",
"EXISTS (SELECT 1 FROM jsonb_array_elements_text(result->'wm_labels') label WHERE label LIKE '{}')",
&label.replace("*", "%").replace("'", "''")
);
sqlb.and_where("result ? 'wm_labels'");
@@ -5969,7 +5594,7 @@ pub fn filter_list_completed_query(
pub fn list_completed_jobs_query(
w_id: &str,
per_page: Option<usize>,
per_page: usize,
offset: usize,
lq: &ListCompletedQuery,
fields: &[&str],
@@ -5980,10 +5605,8 @@ pub fn list_completed_jobs_query(
.fields(fields)
.order_by("v2_job.created_at", lq.order_desc.unwrap_or(true))
.offset(offset)
.limit(per_page)
.clone();
if let Some(per_page) = per_page {
sqlb.limit(per_page);
}
if let Some(tags) = tags {
sqlb.and_where_in(
@@ -6044,7 +5667,7 @@ async fn list_completed_jobs(
let sql = list_completed_jobs_query(
&w_id,
Some(per_page),
per_page,
offset,
&lq,
&[

View File

@@ -18,8 +18,6 @@ use crate::oauth2_ee::SlackVerifier;
#[cfg(feature = "smtp")]
use crate::smtp_server_ee::SmtpServer;
#[cfg(feature = "mcp")]
use crate::mcp::{setup_mcp_server, Runner as McpRunner};
use crate::tracing_init::MyOnFailure;
use crate::{
tracing_init::{MyMakeSpan, MyOnResponse},
@@ -29,7 +27,6 @@ use crate::{
#[cfg(feature = "agent_worker_server")]
use agent_workers_ee::AgentCache;
use anyhow::Context;
use argon2::Argon2;
use axum::extract::DefaultBodyLimit;
@@ -142,9 +139,6 @@ mod workspaces_ee;
mod workspaces_export;
mod workspaces_extra;
#[cfg(feature = "mcp")]
mod mcp;
pub const DEFAULT_BODY_LIMIT: usize = 2097152 * 100; // 200MB
lazy_static::lazy_static! {
@@ -225,7 +219,6 @@ pub async fn run_server(
mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
port_tx: tokio::sync::oneshot::Sender<String>,
server_mode: bool,
mcp_mode: bool,
_base_internal_url: String,
) -> anyhow::Result<()> {
let user_db = UserDB::new(db.clone());
@@ -411,7 +404,7 @@ pub async fn run_server(
Router::new()
};
if !*CLOUD_HOSTED && server_mode && !mcp_mode {
if !*CLOUD_HOSTED && server_mode {
#[cfg(feature = "websocket")]
{
let ws_killpill_rx = killpill_rx.resubscribe();
@@ -455,41 +448,9 @@ pub async fn run_server(
}
}
let listener = tokio::net::TcpListener::bind(addr)
.await
.context("binding main windmill server")?;
let port = listener.local_addr().map(|x| x.port()).unwrap_or(8000);
let ip = listener
.local_addr()
.map(|x| x.ip().to_string())
.unwrap_or("localhost".to_string());
// Setup MCP server
#[allow(unused_variables)]
let (mcp_router, mcp_main_ct, mcp_service_ct) = {
#[cfg(feature = "mcp")]
if server_mode || mcp_mode {
let (mcp_sse_server, mcp_router) = setup_mcp_server(addr, "/api/mcp/w/:workspace_id")?;
#[cfg(feature = "mcp")]
let mcp_main_ct = mcp_sse_server.config.ct.clone(); // Token to signal shutdown *to* MCP
#[cfg(feature = "mcp")]
let mcp_service_ct = mcp_sse_server.with_service(McpRunner::new); // Token to wait for MCP *service* shutdown
(mcp_router, Some(mcp_main_ct), Some(mcp_service_ct))
} else {
(Router::new(), None, None)
}
#[cfg(not(feature = "mcp"))]
(Router::new(), None::<()>, None::<()>)
};
#[cfg(feature = "agent_worker_server")]
let (agent_workers_router, agent_workers_bg_processor, agent_workers_killpill_tx) =
if server_mode {
agent_workers_ee::workspaced_service(db.clone(), _base_internal_url.clone())
} else {
(Router::new(), vec![], None)
};
agent_workers_ee::workspaced_service(db.clone(), _base_internal_url.clone());
#[cfg(feature = "agent_worker_server")]
let agent_cache = Arc::new(AgentCache::new());
@@ -570,6 +531,26 @@ pub async fn run_server(
.nest("/ai", ai::global_service())
.route_layer(from_extractor::<ApiAuthed>())
.route_layer(from_extractor::<users::Tokened>())
.nest("/agent_workers", {
#[cfg(feature = "agent_worker_server")]
{
agent_workers_ee::global_service().layer(Extension(agent_cache.clone()))
}
#[cfg(not(feature = "agent_worker_server"))]
{
Router::new()
}
})
.nest("/w/:workspace_id/agent_workers", {
#[cfg(feature = "agent_worker_server")]
{
agent_workers_router.layer(Extension(agent_cache.clone()))
}
#[cfg(not(feature = "agent_worker_server"))]
{
Router::new()
}
})
.nest("/jobs", jobs::global_root_service())
.nest(
"/srch/w/:workspace_id/index",
@@ -605,28 +586,6 @@ pub async fn run_server(
.layer(from_extractor::<OptAuthed>())
.layer(cors.clone()),
)
.nest("/mcp/w/:workspace_id", mcp_router)
.layer(from_extractor::<OptAuthed>())
.nest("/agent_workers", {
#[cfg(feature = "agent_worker_server")]
{
agent_workers_ee::global_service().layer(Extension(agent_cache.clone()))
}
#[cfg(not(feature = "agent_worker_server"))]
{
Router::new()
}
})
.nest("/w/:workspace_id/agent_workers", {
#[cfg(feature = "agent_worker_server")]
{
agent_workers_router.layer(Extension(agent_cache.clone()))
}
#[cfg(not(feature = "agent_worker_server"))]
{
Router::new()
}
})
.nest(
"/w/:workspace_id/jobs_u",
jobs::workspace_unauthed_service().layer(cors.clone()),
@@ -735,6 +694,14 @@ pub async fn run_server(
.on_failure(MyOnFailure {}),
)
};
let listener = tokio::net::TcpListener::bind(addr)
.await
.context("binding main windmill server")?;
let port = listener.local_addr().map(|x| x.port()).unwrap_or(8000);
let ip = listener
.local_addr()
.map(|x| x.ip().to_string())
.unwrap_or("localhost".to_string());
let server = axum::serve(listener, app.into_make_service());
@@ -752,33 +719,19 @@ pub async fn run_server(
let server = server.with_graceful_shutdown(async move {
killpill_rx.recv().await.ok();
#[cfg(feature = "agent_worker_server")]
if let Some(agent_workers_killpill_tx) = agent_workers_killpill_tx {
if let Err(e) = agent_workers_killpill_tx.kill().await {
tracing::error!("Error killing agent workers: {e:#}");
}
if let Err(e) = agent_workers_killpill_tx.kill().await {
tracing::error!("Error killing agent workers: {e:#}");
}
tracing::info!("Graceful shutdown of server");
#[cfg(feature = "mcp")]
{
if let Some(mcp_main_ct) = mcp_main_ct {
tracing::info!("Received shutdown signal, cancelling MCP server...");
mcp_main_ct.cancel();
}
if let Some(mcp_service_ct) = mcp_service_ct {
tracing::info!("Waiting for MCP service cancellation...");
mcp_service_ct.cancelled().await;
tracing::info!("MCP service cancelled.");
}
}
});
server.await?;
#[cfg(feature = "agent_worker_server")]
for (i, bg_processor) in agent_workers_bg_processor.into_iter().enumerate() {
tracing::info!("server off. shutting down agent worker bg processor {i}");
if let Some(bg_processor) = agent_workers_bg_processor {
tracing::info!("server off. shutting down agent workers bg processor");
bg_processor.await?;
tracing::info!("agent worker bg processor {i} shut down");
tracing::info!("agent workers bg processor shut down");
}
Ok(())
}

View File

@@ -1,773 +0,0 @@
use std::borrow::Cow;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use axum::body::to_bytes;
use axum::Router;
use rmcp::transport::sse_server::{SseServer, SseServerConfig};
use rmcp::{
handler::server::ServerHandler,
model::*,
service::{RequestContext, RoleServer},
Error,
};
use serde::Serialize;
use serde_json::Value;
use sql_builder::prelude::*;
use sqlx::FromRow;
use tokio::try_join;
use tokio_util::sync::CancellationToken;
use windmill_common::db::UserDB;
use windmill_common::scripts::Schema;
use windmill_common::worker::to_raw_value;
use windmill_common::DB;
use crate::db::ApiAuthed;
use crate::jobs::{
run_wait_result_flow_by_path_internal, run_wait_result_script_by_path_internal, RunJobQuery,
};
use windmill_common::utils::StripPath;
#[derive(Clone)]
pub struct Runner {}
#[derive(serde::Deserialize, serde::Serialize)]
struct SchemaType {
r#type: String,
properties: std::collections::HashMap<String, serde_json::Value>,
required: Vec<String>,
}
impl Default for SchemaType {
fn default() -> Self {
Self {
r#type: "object".to_string(),
properties: std::collections::HashMap::new(),
required: vec![],
}
}
}
#[derive(Serialize, FromRow, Debug)]
struct ScriptInfo {
path: String,
summary: Option<String>,
description: Option<String>,
schema: Option<Schema>,
}
#[derive(Serialize, FromRow)]
struct ItemSchema {
schema: Option<Schema>,
}
#[derive(Serialize, FromRow, Debug)]
struct FlowInfo {
path: String,
summary: Option<String>,
description: Option<String>,
schema: Option<Schema>,
}
#[derive(Serialize, FromRow, Debug)]
struct ResourceInfo {
path: String,
description: Option<String>,
resource_type: String,
}
#[derive(Serialize, FromRow, Debug, Clone)]
struct ResourceType {
name: String,
description: Option<String>,
}
impl Runner {
pub fn new() -> Self {
Self {}
}
async fn get_item_schema(
path: &str,
user_db: &UserDB,
authed: &ApiAuthed,
workspace_id: &str,
item_type: &str,
) -> Result<ItemSchema, Error> {
let mut sqlb = SqlBuilder::select_from(&format!("{} as o", item_type));
sqlb.fields(&["o.schema"]);
sqlb.and_where("o.path = ?".bind(&path));
sqlb.and_where("o.workspace_id = ?".bind(&workspace_id));
sqlb.and_where("o.archived = false");
sqlb.and_where("o.draft_only IS NOT TRUE");
let sql = sqlb.sql().map_err(|_e| {
tracing::error!("failed to build sql: {}", _e);
Error::internal_error("failed to build sql", None)
})?;
let mut tx = user_db
.clone()
.begin(authed)
.await
.map_err(|_e| Error::internal_error("failed to begin transaction", None))?;
let rows = sqlx::query_as::<_, ItemSchema>(&sql)
.fetch_one(&mut *tx)
.await
.map_err(|_e| {
tracing::error!("failed to fetch item schema: {}", _e);
Error::internal_error("failed to fetch item schema", None)
})?;
tx.commit()
.await
.map_err(|_e| Error::internal_error("failed to commit transaction", None))?;
Ok(rows)
}
fn transform_path(path: &str, type_str: &str) -> Result<String, String> {
if type_str != "script" && type_str != "flow" {
return Err(format!("Invalid type: {}", type_str));
}
// Only apply special underscore escaping for paths starting with "f/"
let transformed = if path.starts_with("f/") {
let escaped_path = path.replace('_', "__");
escaped_path.replace('/', "_")
} else {
path.replace('/', "_")
};
// first letter of type_str is used as prefix, only one letter to avoid reaching 60 char name limit
Ok(format!("{}-{}", &type_str[..1], transformed))
}
fn reverse_transform(transformed_path: &str) -> Result<(&str, String), String> {
let type_str = if transformed_path.starts_with("s-") {
"script"
} else if transformed_path.starts_with("f-") {
"flow"
} else {
return Err(format!(
"Invalid prefix in transformed path: {}",
transformed_path
));
};
let mangled_path = &transformed_path[2..];
// Check if this path was previously transformed with special underscore handling
let is_special_path = mangled_path.starts_with("f_");
let original_path = if is_special_path {
const TEMP_PLACEHOLDER: &str = "@@UNDERSCORE@@";
let path_with_placeholder = mangled_path.replace("__", TEMP_PLACEHOLDER);
let path_with_slashes = path_with_placeholder.replace('_', "/");
path_with_slashes.replace(TEMP_PLACEHOLDER, "_")
} else {
mangled_path.replacen('_', "/", 2)
};
Ok((type_str, original_path))
}
async fn inner_get_resources_types(
user_db: &UserDB,
authed: &ApiAuthed,
workspace_id: &str,
) -> Result<Vec<ResourceType>, Error> {
let mut sqlb = SqlBuilder::select_from("resource_type as o");
sqlb.fields(&["o.name", "o.description"]);
sqlb.and_where("o.workspace_id = ?".bind(&workspace_id));
let sql = sqlb.sql().map_err(|_e| {
tracing::error!("failed to build sql: {}", _e);
Error::internal_error("failed to build sql", None)
})?;
let mut tx = user_db
.clone()
.begin(authed)
.await
.map_err(|_e| Error::internal_error("failed to begin transaction", None))?;
let rows = sqlx::query_as::<_, ResourceType>(&sql)
.fetch_all(&mut *tx)
.await
.map_err(|_e| {
tracing::error!("Failed to fetch resource types: {}", _e);
Error::internal_error("failed to fetch resource types", None)
})?;
tx.commit()
.await
.map_err(|_e| Error::internal_error("failed to commit transaction", None))?;
Ok(rows)
}
async fn inner_get_resources(
user_db: &UserDB,
authed: &ApiAuthed,
workspace_id: &str,
resource_type: &str,
) -> Result<Vec<ResourceInfo>, Error> {
let mut sqlb = SqlBuilder::select_from("resource as o");
sqlb.fields(&["o.path", "o.description", "o.resource_type"]);
sqlb.and_where("o.workspace_id = ?".bind(&workspace_id));
sqlb.and_where("o.resource_type = ?".bind(&resource_type));
let sql = sqlb.sql().map_err(|_e| {
tracing::error!("failed to build sql: {}", _e);
Error::internal_error("failed to build sql", None)
})?;
let mut tx = user_db
.clone()
.begin(authed)
.await
.map_err(|_e| Error::internal_error("failed to begin transaction", None))?;
let rows = sqlx::query_as::<_, ResourceInfo>(&sql)
.fetch_all(&mut *tx)
.await
.map_err(|_e| {
tracing::error!("Failed to fetch resources: {}", _e);
Error::internal_error("failed to fetch resources", None)
})?;
tx.commit()
.await
.map_err(|_e| Error::internal_error("failed to commit transaction", None))?;
Ok(rows)
}
async fn inner_get_items<T: for<'a> sqlx::FromRow<'a, sqlx::postgres::PgRow> + Send + Unpin>(
user_db: &UserDB,
authed: &ApiAuthed,
workspace_id: &str,
scope_type: &str,
item_type: &str,
) -> Result<Vec<T>, Error> {
let mut sqlb = SqlBuilder::select_from(&format!("{} as o", item_type));
sqlb.fields(&["o.path", "o.summary", "o.description", "o.schema"]);
if scope_type == "favorites" {
sqlb.join("favorite")
.on("favorite.favorite_kind = ? AND favorite.workspace_id = o.workspace_id AND favorite.path = o.path AND favorite.usr = ?".bind(&item_type)
.bind(&authed.username));
}
sqlb.and_where("o.workspace_id = ?".bind(&workspace_id))
.and_where("o.archived = false")
.and_where("o.draft_only IS NOT TRUE")
.order_by(
if item_type == "flow" {
"o.edited_at"
} else {
"o.created_at"
},
false,
)
.limit(100);
let sql = sqlb.sql().map_err(|_e| {
tracing::error!("failed to build sql: {}", _e);
Error::internal_error("failed to build sql", None)
})?;
let mut tx = user_db
.clone()
.begin(authed)
.await
.map_err(|_e| Error::internal_error("failed to begin transaction", None))?;
let rows = sqlx::query_as::<_, T>(&sql)
.fetch_all(&mut *tx)
.await
.map_err(|_e| {
tracing::error!("Failed to fetch {}: {}", item_type, _e);
Error::internal_error(format!("failed to fetch {}", item_type), None)
})?;
tx.commit()
.await
.map_err(|_e| Error::internal_error("failed to commit transaction", None))?;
Ok(rows)
}
fn transform_value_if_object(
key: &str,
value: &Value,
schema_obj: &Option<SchemaType>,
) -> Value {
if value.is_string() && value.as_str().unwrap().starts_with("$res:") {
return value.clone();
}
let schema_obj = match schema_obj {
Some(s) => s,
None => return value.clone(),
};
// Check if property is defined in schema and is an object type
let is_obj_type = match schema_obj.properties.get(key) {
Some(property) => {
let prop_type = property.get("type").and_then(|t| t.as_str());
prop_type == Some("object")
}
None => false,
};
// If it's an object type and we received a string, try to parse it
if is_obj_type && value.is_string() {
if let Some(str_val) = value.as_str() {
if let Ok(obj_val) = serde_json::from_str::<serde_json::Value>(str_val) {
return obj_val;
}
}
}
value.clone()
}
fn reverse_transform_key(transformed_key: &str, schema_obj: &Option<SchemaType>) -> String {
let schema_obj = match schema_obj {
Some(s) => s,
None => {
// No schema available, return the key as is (best guess)
return transformed_key.to_string();
}
};
for original_key_in_schema in schema_obj.properties.keys() {
// Apply the SAME forward transformation to the schema key
let potential_transformed_key =
Runner::apply_key_transformation(original_key_in_schema);
// If it matches the key we received, we found the likely original
if potential_transformed_key == transformed_key {
return original_key_in_schema.clone();
}
}
transformed_key.to_string()
}
fn apply_key_transformation(key: &str) -> String {
key.replace(' ', "_")
.chars()
.filter(|c| c.is_alphanumeric() || *c == '_')
.collect::<String>()
}
async fn transform_schema_for_resources(
schema: &Schema,
user_db: &UserDB,
authed: &ApiAuthed,
w_id: &str,
resources_cache: &mut HashMap<String, Vec<ResourceInfo>>,
resources_types: &Vec<ResourceType>,
) -> Result<SchemaType, Error> {
let mut schema_obj: SchemaType = match serde_json::from_str(schema.0.get()) {
Ok(val) => val,
Err(_) => SchemaType::default(),
};
// replace invalid char in property key with underscore
let replacements: Vec<(String, String, serde_json::Value)> = schema_obj
.properties
.iter()
.filter_map(|(key, value)| {
if key.chars().any(|c| !c.is_alphanumeric() && c != '_') {
let new_key = Runner::apply_key_transformation(key);
Some((key.clone(), new_key, value.clone()))
} else {
None
}
})
.collect();
for (old_key, new_key, value) in replacements {
schema_obj.properties.remove(&old_key);
schema_obj.properties.insert(new_key, value);
}
for (_key, prop_value) in schema_obj.properties.iter_mut() {
if let serde_json::Value::Object(prop_map) = prop_value {
// transform object properties to string because some client does not support object, might change in the future
if let Some(type_value) = prop_map.get("type") {
if let serde_json::Value::String(type_str) = type_value {
if type_str == "object" {
prop_map.insert(
"type".to_string(),
serde_json::Value::String("string".to_string()),
);
}
}
}
// if property is a resource, fetch the resource type infos, and add each available resource to the description
if let Some(format_value) = prop_map.get("format") {
if let serde_json::Value::String(format_str) = format_value {
if format_str.starts_with("resource-") {
let resource_type_key =
format_str.split("-").last().unwrap_or_default().to_string();
let resource_type = resources_types
.iter()
.find(|rt| rt.name == resource_type_key);
let resource_type_obj = resource_type.cloned().unwrap_or_else(|| {
tracing::info!("Resource type not found: {}", resource_type_key);
ResourceType { name: resource_type_key.clone(), description: None }
});
if !resources_cache.contains_key(&resource_type_key) {
let available_resources = Runner::inner_get_resources(
user_db,
authed,
&w_id,
&resource_type_key,
)
.await;
match available_resources {
Ok(cache_data) => {
resources_cache
.insert(resource_type_key.clone(), cache_data);
}
Err(e) => {
tracing::error!(
"Failed to fetch resource cache data: {}",
e
);
continue; // Skip this property if fetching failed
}
}
}
if let Some(resource_cache) = resources_cache.get(&resource_type_key) {
let resources_count = resource_cache.len();
let description = format!(
"This is a resource named `{}` with the following description: `{}`.\nThe path of the resource should be used to specify the resource.\n{}",
resource_type_obj.name,
resource_type_obj.description.as_deref().unwrap_or("No description"),
if resources_count == 0 {
"This resource does not have any available instances, you should create one from your windmill workspace."
} else if resources_count > 1 {
"This resource has multiple available instances, you should precisely select the one you want to use."
} else {
"There is 1 resource available."
}
);
prop_map.insert(
"type".to_string(),
serde_json::Value::String("string".to_string()),
);
prop_map.insert(
"description".to_string(),
serde_json::Value::String(description),
);
if resources_count > 0 {
let resources_description = resource_cache
.iter()
.map(|resource| {
format!(
"{}: $res:{}",
resource
.description
.as_deref()
.unwrap_or("No title"),
resource.path
)
})
.collect::<Vec<String>>()
.join("\n");
prop_map.insert(
"description".to_string(),
serde_json::Value::String(format!(
"{}\nHere are the available resources, in the format title:path. Title can be empty. Path should be used to specify the resource:\n{}",
prop_map.get("description").unwrap_or(&serde_json::Value::String("No description".to_string())),
resources_description
)),
);
}
}
}
}
}
} else {
tracing::warn!(
"Schema property value is not a JSON object: {:?}",
prop_value
);
}
}
Ok(schema_obj)
}
}
impl ServerHandler for Runner {
async fn call_tool(
&self,
request: CallToolRequestParam,
context: RequestContext<RoleServer>,
) -> Result<CallToolResult, Error> {
let parse_args = |args_opt: Option<JsonObject>| -> Result<Value, Error> {
args_opt.map(Value::Object).ok_or_else(|| {
Error::invalid_params(
"Missing arguments for tool",
Some(request.name.clone().into()),
)
})
};
let authed = context
.req_extensions
.get::<ApiAuthed>()
.ok_or_else(|| Error::internal_error("ApiAuthed not found", None))?;
let db = context
.req_extensions
.get::<DB>()
.ok_or_else(|| Error::internal_error("DB not found", None))?;
let user_db = context
.req_extensions
.get::<UserDB>()
.ok_or_else(|| Error::internal_error("UserDB not found", None))?;
let args = parse_args(request.arguments)?;
let (tool_type, path) = Runner::reverse_transform(&request.name).unwrap_or_default();
let item_info =
Runner::get_item_schema(&path, user_db, authed, &context.workspace_id, &tool_type)
.await?;
let schema = item_info.schema;
let schema_obj = if let Some(ref s) = schema {
match serde_json::from_str::<SchemaType>(s.0.get()) {
Ok(val) => Some(val),
Err(e) => {
tracing::warn!("Failed to parse schema: {}", e);
None
}
}
} else {
None
};
let push_args = if let Value::Object(map) = args.clone() {
let mut args_hash = HashMap::new();
for (k, v) in map {
// need to transform back the key to the original key
let original_key = Runner::reverse_transform_key(&k, &schema_obj);
// object properties are transformed to string because some client does not support object, might change in the future
let transformed_v = Runner::transform_value_if_object(&k, &v, &schema_obj);
args_hash.insert(original_key, to_raw_value(&transformed_v));
}
windmill_queue::PushArgsOwned { extra: None, args: args_hash }
} else {
windmill_queue::PushArgsOwned::default()
};
let w_id = context.workspace_id.clone();
let script_or_flow_path = StripPath(path);
let run_query = RunJobQuery::default();
let result = if tool_type == "script" {
run_wait_result_script_by_path_internal(
db.clone(),
run_query,
script_or_flow_path,
authed.clone(),
user_db.clone(),
w_id.clone(),
push_args,
None,
)
.await
} else {
run_wait_result_flow_by_path_internal(
db.clone(),
run_query,
script_or_flow_path,
authed.clone(),
user_db.clone(),
push_args,
w_id.clone(),
None,
)
.await
};
match result {
Ok(response) => {
let body_bytes = to_bytes(response.into_body(), usize::MAX)
.await
.map_err(|e| {
Error::internal_error(format!("Failed to read response body: {}", e), None)
})?;
let body_str = String::from_utf8(body_bytes.to_vec()).map_err(|e| {
Error::internal_error(format!("Failed to decode response body: {}", e), None)
})?;
Ok(CallToolResult::success(vec![Content::text(body_str)]))
}
Err(e) => Err(Error::internal_error(
format!("Failed to run script: {}", e),
None,
)),
}
}
async fn list_tools(
&self,
_request: Option<PaginatedRequestParam>,
mut _context: RequestContext<RoleServer>,
) -> Result<ListToolsResult, Error> {
let workspace_id = _context.workspace_id.clone();
let user_db = _context
.req_extensions
.get::<UserDB>()
.ok_or_else(|| Error::internal_error("UserDB not found", None))?;
let authed = _context
.req_extensions
.get::<ApiAuthed>()
.ok_or_else(|| Error::internal_error("ApiAuthed not found", None))?;
let scope = authed
.scopes
.as_ref()
.and_then(|scopes| scopes.iter().find(|scope| scope.starts_with("mcp:")));
let scope_type = scope.map_or("all", |scope| scope.split(":").last().unwrap_or("all"));
let scripts_fn = Runner::inner_get_items::<ScriptInfo>(
user_db,
authed,
&workspace_id,
scope_type,
"script",
);
let flows_fn =
Runner::inner_get_items::<FlowInfo>(user_db, authed, &workspace_id, scope_type, "flow");
let resources_types_fn = Runner::inner_get_resources_types(user_db, authed, &workspace_id);
let (scripts, flows, resources_types) =
try_join!(scripts_fn, flows_fn, resources_types_fn)?;
let mut resources_cache: HashMap<String, Vec<ResourceInfo>> = HashMap::new();
let mut script_tools: Vec<Tool> = Vec::with_capacity(scripts.len());
for script in scripts {
let name = Runner::transform_path(&script.path, "script").unwrap_or_default();
let description = format!(
"This is a script named `{}` with the following description: `{}`.",
script.summary.as_deref().unwrap_or("No summary"),
script.description.as_deref().unwrap_or("No description")
);
let schema_obj = if let Some(schema) = script.schema {
Runner::transform_schema_for_resources(
&schema,
user_db,
authed,
&workspace_id,
&mut resources_cache,
&resources_types,
)
.await?
} else {
SchemaType::default()
};
script_tools.push(Tool {
name: Cow::Owned(name),
description: Some(Cow::Owned(description)),
input_schema: {
let value = serde_json::to_value(schema_obj).unwrap_or_default();
if let serde_json::Value::Object(map) = value {
Arc::new(map)
} else {
Arc::new(serde_json::Map::new())
}
},
annotations: None,
});
}
let mut flow_tools: Vec<Tool> = Vec::with_capacity(flows.len());
for flow in flows {
let name = Runner::transform_path(&flow.path, "flow").unwrap_or_default();
let description = format!(
"This is a flow named `{}` with the following description: `{}`.",
flow.summary.as_deref().unwrap_or("No summary"),
flow.description.as_deref().unwrap_or("No description")
);
let schema_obj = if let Some(schema) = flow.schema {
Runner::transform_schema_for_resources(
&schema,
user_db,
authed,
&workspace_id,
&mut resources_cache,
&resources_types,
)
.await?
} else {
SchemaType::default()
};
flow_tools.push(Tool {
name: Cow::Owned(name),
description: Some(Cow::Owned(description)),
input_schema: {
let value = serde_json::to_value(schema_obj).unwrap_or_default();
if let serde_json::Value::Object(map) = value {
Arc::new(map)
} else {
Arc::new(serde_json::Map::new())
}
},
annotations: None,
});
}
let tools = [script_tools, flow_tools].concat();
Ok(ListToolsResult { tools, next_cursor: None })
}
fn get_info(&self) -> ServerInfo {
ServerInfo {
protocol_version: Default::default(),
capabilities: ServerCapabilities::builder()
.enable_tools()
.enable_tool_list_changed()
.build(),
server_info: Implementation::from_build_env(),
instructions: Some("This server provides a list of scripts and flows the user can run on Windmill. Each flow and script is a tool callable with their respective arguments.".to_string()),
}
}
async fn initialize(
&self,
_request: InitializeRequestParam,
_context: RequestContext<RoleServer>,
) -> Result<InitializeResult, Error> {
Ok(self.get_info())
}
async fn list_resources(
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListResourcesResult, Error> {
Ok(ListResourcesResult { resources: vec![], next_cursor: None })
}
async fn list_prompts(
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListPromptsResult, Error> {
Ok(ListPromptsResult::default())
}
async fn list_resource_templates(
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListResourceTemplatesResult, Error> {
Ok(ListResourceTemplatesResult::default())
}
}
pub fn setup_mcp_server(addr: SocketAddr, path: &str) -> anyhow::Result<(SseServer, Router)> {
let config = SseServerConfig {
bind: addr,
sse_path: "/sse".to_string(),
post_path: "/message".to_string(),
full_message_path: path.to_string(),
ct: CancellationToken::new(),
sse_keep_alive: None,
};
Ok(SseServer::new(config))
}

View File

@@ -6,7 +6,6 @@ use std::collections::{
use crate::{
db::{ApiAuthed, DB},
postgres_triggers::mapper::{Mapper, MappingInfo},
resources::try_get_resource_from_db_as,
};
use axum::{
extract::{Path, Query},
@@ -30,9 +29,9 @@ use windmill_common::{
};
use super::{
create_logical_replication_slot_query, create_publication_query, drop_publication_query,
generate_random_string, get_database_connection, get_raw_postgres_connection,
ERROR_PUBLICATION_NAME_NOT_EXISTS, ERROR_REPLICATION_SLOT_NOT_EXISTS,
create_logical_replication_slot_query, create_publication_query,
drop_logical_replication_slot_query, drop_publication_query, generate_random_string,
get_database_connection, ERROR_PUBLICATION_NAME_NOT_EXISTS, ERROR_REPLICATION_SLOT_NOT_EXISTS,
};
use lazy_static::lazy_static;
@@ -395,7 +394,7 @@ pub async fn create_postgres_trigger(
"Missing replication slot name".to_string(),
));
}
(publication_name.unwrap(), replication_slot_name.unwrap())
(replication_slot_name.unwrap(), publication_name.unwrap())
};
let mut tx = user_db.begin(&authed).await?;
@@ -629,7 +628,7 @@ pub async fn create_slot(
sqlx::query(&query).execute(&mut connection).await?;
Ok(format!("Replication slot {} created!", name))
Ok(format!("Slot {} created!", name))
}
pub async fn drop_slot_name(
@@ -639,43 +638,19 @@ pub async fn drop_slot_name(
Path((w_id, postgres_resource_path)): Path<(String, String)>,
Json(Slot { name }): Json<Slot>,
) -> Result<String> {
let database = try_get_resource_from_db_as::<Postgres>(
authed,
Some(user_db),
let mut connection = get_database_connection(
authed.clone(),
Some(user_db.clone()),
&db,
&postgres_resource_path,
&w_id,
)
.await?;
let mut connection = get_raw_postgres_connection(&database).await?;
let query = drop_logical_replication_slot_query(&name);
sqlx::query(&query).execute(&mut connection).await?;
let active_pid = sqlx::query_scalar!(
r#"SELECT
active_pid
FROM
pg_replication_slots
WHERE
slot_name = $1
"#,
&name
)
.fetch_optional(&mut connection)
.await?
.flatten();
if let Some(pid) = active_pid {
sqlx::query("SELECT pg_terminate_backend($1)")
.bind(pid)
.execute(&mut connection)
.await?;
}
sqlx::query("SELECT pg_drop_replication_slot($1)")
.bind(&name)
.execute(&mut connection)
.await?;
Ok(format!("Replication slot {} deleted!", name))
Ok(format!("Slot name {} deleted!", name))
}
#[derive(Debug, Serialize)]
struct PublicationName {

View File

@@ -26,8 +26,8 @@ fn postgres_to_typescript_type(postgres_type: Option<Type>) -> String {
Type::DATE_ARRAY => "Array<string>",
Type::TIME => "string",
Type::TIME_ARRAY => "Array<string>",
Type::TIMESTAMPTZ | Type::TIMESTAMP => "string",
Type::TIMESTAMPTZ_ARRAY | Type::TIMESTAMP_ARRAY => "Array<string>",
Type::TIMESTAMPTZ | Type::TIMESTAMP => "Date",
Type::TIMESTAMPTZ_ARRAY | Type::TIMESTAMP_ARRAY => "Array<Date>",
Type::UUID => "string",
Type::UUID_ARRAY => "Array<string>",
Type::JSON | Type::JSONB | Type::JSON_ARRAY | Type::JSONB_ARRAY => "unknown",
@@ -124,13 +124,11 @@ export async function main(
transaction_type: "insert" | "update" | "delete",
schema_name: string,
table_name: string,
row: {},
old_row?: {}
row: {}
) {{
}}
"#,
&struct_definition,
&struct_definition
struct_definition
)
}
}

View File

@@ -47,7 +47,7 @@ impl RelationConverter {
.ok_or(RelationConversionError::FailToFindMatchingTable)
}
pub fn row_to_json(
pub fn body_to_json(
&self,
to_decode: (Oid, Vec<TupleData>),
) -> Result<Map<String, Value>, RelationConversionError> {

View File

@@ -467,7 +467,6 @@ impl PostgresConfig {
let client = PostgresSimpleClient::new(&database).await?;
let publication = client
.execute_query(&format!(
"SELECT pubname FROM pg_publication WHERE pubname = {}",
@@ -606,7 +605,6 @@ async fn listen_to_transactions(
}
};
let message = match message {
Ok(message) => message,
Err(err) => {
@@ -650,79 +648,37 @@ async fn listen_to_transactions(
None
}
Insert(insert) => {
Some((insert.o_id, Ok(None), relations.row_to_json((insert.o_id, insert.tuple)), "insert"))
Some((insert.o_id, relations.body_to_json((insert.o_id, insert.tuple)), "insert"))
}
Update(update) => {
let old_row = update.old_tuple.map(|old_tuple| relations.row_to_json((update.o_id, old_tuple))).transpose();
let row = relations.row_to_json((update.o_id, update.new_tuple));
Some((update.o_id, old_row, row, "update"))
Some((update.o_id, relations.body_to_json((update.o_id, update.new_tuple)), "update"))
}
Delete(delete) => {
let row = delete.old_tuple.unwrap_or_else(|| delete.key_tuple.unwrap());
Some((delete.o_id, Ok(None), relations.row_to_json((delete.o_id, row)), "delete"))
let body = delete.old_tuple.unwrap_or_else(|| delete.key_tuple.unwrap());
Some((delete.o_id, relations.body_to_json((delete.o_id, body)), "delete"))
}
};
match json {
Some((o_id, Ok(old_row), Ok(row), transaction_type)) => {
let relation = match relations.get_relation(o_id) {
Ok(relation) => relation,
Err(err) => {
tracing::error!("Postgres trigger named: {}, error: {}", pg.get_path(), err.to_string());
continue;
}
};
let database_info = HashMap::from([
("schema_name".to_string(), to_raw_value(&relation.namespace)),
("table_name".to_string(), to_raw_value(&relation.name)),
("transaction_type".to_string(), to_raw_value(&transaction_type)),
("old_row".to_string(), to_raw_value(&old_row)),
("row".to_string(), to_raw_value(&row)),
]);
let extra = Some(HashMap::from([(
"wm_trigger".to_string(),
to_raw_value(&serde_json::json!({"kind": "postgres", })),
)]));
let _ = pg.handle(&db, Some(database_info), extra).await;
}
Some((o_id, old_row, row, transaction_type)) => {
let relation = match relations.get_relation(o_id) {
Ok(relation) => relation,
Err(err) => {
tracing::error!("Postgres trigger named: {}, error: {}", pg.get_path(), err.to_string());
continue;
}
};
if let Err(err) = old_row {
tracing::error!(
transaction_type = ?transaction_type,
schema = %relation.namespace,
table = %relation.name,
error = %err,
"Failed to decode OLD row for {} transaction on {}.{}",
transaction_type,
relation.namespace,
relation.name,
);
if let Some((o_id, Ok(body), transaction_type)) = json {
let relation = match relations.get_relation(o_id) {
Ok(relation) => relation,
Err(err) => {
tracing::error!("Postgres trigger named: {}, error: {}", pg.get_path(), err.to_string());
continue;
}
if let Err(err) = row {
tracing::error!(
transaction_type = ?transaction_type,
schema = %relation.namespace,
table = %relation.name,
error = %err,
"Failed to decode NEW row for {} transaction on {}.{}",
transaction_type,
relation.namespace,
relation.name,
);
}
}
_ => {}
};
let database_info = HashMap::from([
("schema_name".to_string(), to_raw_value(&relation.namespace)),
("table_name".to_string(), to_raw_value(&relation.name)),
("transaction_type".to_string(), to_raw_value(&transaction_type)),
("row".to_string(), to_raw_value(&body)),
]);
let extra = Some(HashMap::from([(
"wm_trigger".to_string(),
to_raw_value(&serde_json::json!({"kind": "postgres", })),
)]));
let _ = pg.handle(&db, Some(database_info), extra).await;
}
}

View File

@@ -40,7 +40,6 @@ use std::{
};
use windmill_audit::audit_ee::audit_log;
use windmill_audit::ActionKind;
use windmill_worker::process_relative_imports;
use windmill_common::error::to_anyhow;
@@ -895,38 +894,6 @@ async fn create_script_internal<'c>(
.await?;
Ok((hash, new_tx))
} else {
let db2 = db.clone();
let w_id2 = w_id.clone();
let authed2 = authed.clone();
let permissioned_as2 = permissioned_as.clone();
let script_path2 = script_path.clone();
let parent_path = p_path_opt.clone();
let deployment_message = ns.deployment_message.clone();
let content = ns.content.clone();
let language = ns.language.clone();
tokio::spawn(async move {
// wait for 10 seconds to make sure the script is deployed and that the CLI sync that pushed it (f one) is complete
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
if let Err(e) = process_relative_imports(
&db2,
None,
None,
&w_id2,
&script_path2,
parent_path,
deployment_message,
&content,
&Some(language),
&authed2.email,
&authed2.username,
&permissioned_as2,
)
.await
{
tracing::error!(%e, "error processing relative imports");
}
});
handle_deployment_metadata(
&authed.email,
&authed.username,

View File

@@ -1,7 +1,6 @@
use crate::db::DB;
use axum::Router;
use serde::{Deserialize, Serialize};
use windmill_common::auth::aws::AwsAuthResourceType;
pub fn workspaced_service() -> Router {
@@ -15,7 +14,6 @@ pub fn start_sqs(_db: DB, mut _killpill_rx: tokio::sync::broadcast::Receiver<()>
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct SqsTrigger {
pub queue_url: String,
pub aws_auth_resource_type: AwsAuthResourceType,
pub aws_resource_path: String,
pub message_attributes: Option<Vec<String>>,
pub path: String,

View File

@@ -501,7 +501,7 @@ pub(crate) async fn tarball_workspace(
"SELECT app.id, app.path, app.summary, app.versions, app.policy, app.custom_path,
app.extra_perms, app_version.value,
app_version.created_at, app_version.created_by from app, app_version
WHERE app.workspace_id = $1 AND app_version.id = app.versions[array_upper(app.versions, 1)] AND app_version.raw_app IS false",
WHERE app.workspace_id = $1 AND app_version.id = app.versions[array_upper(app.versions, 1)]",
)
.bind(&w_id)
.fetch_all(&mut *tx)
@@ -577,35 +577,13 @@ pub(crate) async fn tarball_workspace(
#[cfg(feature = "websocket")]
{
let websocket_triggers = sqlx::query_as!(
crate::websocket_triggers::WebsocketTrigger,
r#"
SELECT
workspace_id,
path,
url,
script_path,
is_flow,
edited_by,
email,
edited_at,
server_id,
last_server_ping,
extra_perms,
error,
enabled,
filters AS "filters: _",
initial_messages AS "initial_messages: _",
url_runnable_args AS "url_runnable_args: _",
can_return_message
FROM
websocket_trigger
WHERE
workspace_id = $1
"#,
&w_id
)
.fetch_all(&mut *tx)
.await?;
crate::websocket_triggers::WebsocketTrigger,
"SELECT workspace_id, path, url, script_path, is_flow, edited_by, email, edited_at, server_id, last_server_ping, extra_perms, error, enabled, filters as \"filters: _\", initial_messages as \"initial_messages: _\", url_runnable_args as \"url_runnable_args: _\", can_return_message FROM websocket_trigger
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
.await?;
for trigger in websocket_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
@@ -644,29 +622,8 @@ pub(crate) async fn tarball_workspace(
{
let sqs_triggers = sqlx::query_as!(
crate::sqs_triggers_ee::SqsTrigger,
r#"
SELECT
aws_auth_resource_type AS "aws_auth_resource_type: _",
aws_resource_path,
message_attributes,
queue_url,
workspace_id,
path,
script_path,
is_flow,
edited_by,
email,
edited_at,
server_id,
last_server_ping,
extra_perms,
error,
enabled
FROM
sqs_trigger
WHERE
workspace_id = $1
"#,
"SELECT * FROM sqs_trigger
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
@@ -692,7 +649,6 @@ pub(crate) async fn tarball_workspace(
workspace_id,
delivery_type AS "delivery_type: _",
delivery_config AS "delivery_config: _",
subscription_mode AS "subscription_mode: _",
path,
script_path,
is_flow,

View File

@@ -13,7 +13,6 @@ prometheus = ["dep:prometheus"]
loki = ["dep:tracing-loki"]
benchmark = []
parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts"]
aws_auth = ["dep:aws-sdk-sts", "dep:aws-config"]
otel = ["dep:opentelemetry-semantic-conventions", "dep:opentelemetry-otlp", "dep:opentelemetry_sdk",
"dep:opentelemetry", "dep:tracing-opentelemetry", "dep:opentelemetry-appender-tracing", "dep:tonic"]
smtp = ["dep:mail-send"]

View File

@@ -6,8 +6,6 @@
* LICENSE-AGPL for a copy of the license.
*/
use std::collections::HashMap;
use serde::{Deserialize, Serialize};
/// Id in the `app_script` table.
@@ -23,8 +21,3 @@ pub struct ListAppQuery {
pub include_draft_only: Option<bool>,
pub with_deployment_msg: Option<bool>,
}
#[derive(Deserialize)]
pub struct RawAppValue {
pub files: HashMap<String, String>,
}

Some files were not shown because too many files have changed in this diff Show More