Compare commits

...
32 Commits
Author SHA1 Message Date
centdix 4013aa9098 fix 2025-05-17 15:04:12 +02:00
centdix 1c620db0c0 internal: React to merged pr (#5768)
* react to merged pr

* fix

* fix
2025-05-17 14:55:39 +02:00
centdix 5b58ce3ea9 do not redirect error to probe output (#5767) 2025-05-17 14:00:29 +02:00
Ruben Fiszelandrubenfiszel b2088a0805 chore(main): release 1.491.5 (#5765)
* chore(main): release 1.491.5

* Apply automatic changes

---------

Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com>
2025-05-17 11:00:36 +02:00
Ruben Fiszel beb658757f nit error 2025-05-17 10:56:16 +02:00
Ruben Fiszel 59afa493fa fix: improve handling of custom concurrency key/tag with preprocessors (#5762)
* fix

* rm spurrious diff

* handle failing preprocessors

* nits

* update

* update
2025-05-17 10:52:28 +02:00
Diego Imbert b29c6e7636 fix: S3 sql mode returns S3Object (#5764)
* fix upload_s3_file not checking response

* Return S3Object instead of path string
2025-05-16 19:28:57 +02:00
HugoCasa 74ff8e313f preprocessor template nits (#5760) 2025-05-16 10:47:11 +02:00
dieriba aba6f450aa add null typing to openapi spec for easier gen (#5759) 2025-05-16 09:59:45 +02:00
Ruben Fiszelandrubenfiszel 4496006a56 chore(main): release 1.491.4 (#5758)
* chore(main): release 1.491.4

* Apply automatic changes

---------

Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com>
2025-05-16 01:19:58 +02:00
HugoCasa 9b1c30eeff fix: add v1 preprocessor support to workspace preprocessor script (#5757)
* fix: add v1 preprocessor support to workspace preprocessor script

* nit

* build
2025-05-16 01:10:51 +02:00
Ruben Fiszelandrubenfiszel 6390eee792 chore(main): release 1.491.3 (#5754)
* chore(main): release 1.491.3

* Apply automatic changes

---------

Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com>
2025-05-15 22:07:45 +02:00
HugoCasa d53bceb800 fix: http trigger routers cache version sequence (#5755)
* fix: http trigger routers cache version sequence

* nit
2025-05-15 21:59:30 +02:00
Ruben Fiszel f488903635 fix(frontend): fix accordeon tabs initialization 2025-05-15 20:37:30 +02:00
Ruben Fiszelandrubenfiszel 747c53dfed chore(main): release 1.491.2 (#5747)
* chore(main): release 1.491.2

* Apply automatic changes

---------

Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com>
2025-05-15 20:27:36 +02:00
Ruben Fiszel 0dcb78307a fix 2025-05-15 20:11:00 +02:00
Ruben Fiszel b15964b8a3 fix 2025-05-15 20:10:03 +02:00
dieriba 9e9514b9af fix: http trigger signature validation (#5753)
* fix

* fix import

* nits: force raw string on signature authentication

* fix
2025-05-15 20:07:05 +02:00
Ruben Fiszel f8f201564f fix(cli): --version improvement 2025-05-15 19:49:16 +02:00
Ruben Fiszel c6d512d054 add alerts by workspace index 2025-05-15 19:19:53 +02:00
centdix 0429d5a6d6 bump helm chart on new release (#5751) 2025-05-15 19:08:52 +02:00
Ruben Fiszel f23878feb8 nit sqlx 2025-05-15 18:24:11 +02:00
Ruben Fiszel 72501cbf2e add dependency_map grants 2025-05-15 17:54:49 +02:00
Ruben Fiszel 62b6540a78 nit delete performance 2025-05-15 17:52:25 +02:00
wendrul b12feaf50a fix: Improve indexer performance by factoring required queries to the DB # (#5749)
* search modal UI improvements

* Add ee repo ref

* Prepare sqlx

* Remove unused variable
2025-05-15 17:38:57 +02:00
HugoCasaandRuben Fiszel ddd18d22a6 perf: cache http trigger routers and auth (#5748)
* perf: cache http trigger routers and auth

* fix build

* fix

* fix build

* fix build

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
2025-05-15 17:38:45 +02:00
centdix 1d2d589125 internal: Add create docs command (#5750)
* add create docs command

* use good repo

* remove unused
2025-05-15 17:38:34 +02:00
Ruben Fiszel 38907f74c6 nit delete jobs perf 2025-05-15 17:08:05 +02:00
Ruben Fiszel f135f8e420 update sqlx 2025-05-15 16:07:21 +02:00
Ruben Fiszel 0efba945ba fix: improve perf of job deletion 2025-05-15 15:44:55 +02:00
centdixandgraphite-app[bot] eccefb0dc6 nit: Use shareable discord flow to use it from other repo (#5746)
* cleaning

* cleaning

* fix

* Update .github/workflows/discord-notification.yml

Co-authored-by: graphite-app[bot] <96075541+graphite-app[bot]@users.noreply.github.com>

---------

Co-authored-by: graphite-app[bot] <96075541+graphite-app[bot]@users.noreply.github.com>
2025-05-15 12:16:03 +02:00
dieriba e05e1834cc nit queue metrics drawer (#5745) 2025-05-15 11:03:16 +02:00
86 changed files with 2144 additions and 929 deletions
+1 -1
View File
@@ -237,7 +237,7 @@ jobs:
'{ "message": "I'\''m giving you a request that needs to be implemented. Your role is ONLY to give me the files that are relevant to the request and nothing else. The request is prepended with the word REQUEST.\\nREQUEST: \($prompt_escaped). Give me all the files relevant to this request. Your output MUST be a single json array that can be parsed with programatic json parsing, with the relevant files. Files can be rust or typescript or javascript files. DO NOT INCLUDE ANY OTHER TEXT IN YOUR OUTPUT. ONLY THE JSON ARRAY. Example of output: [\"file1.py\", \"file2.py\"]" }' | jq -r .message)
set -o pipefail
PROBE_OUTPUT=$(npx --yes @buger/probe-chat@latest --max-iterations 50 --model-name gemini-2.5-pro-preview-05-06 --message "$MESSAGE_FOR_PROBE" 2>&1) || {
PROBE_OUTPUT=$(npx --yes @buger/probe-chat@latest --max-iterations 50 --model-name gemini-2.5-pro-preview-05-06 --message "$MESSAGE_FOR_PROBE") || {
echo "::error::probe-chat command failed. Output:"
echo "$PROBE_OUTPUT"
exit 1
+15
View File
@@ -0,0 +1,15 @@
on:
issue_comment:
types: [created]
jobs:
trigger-docs:
if: ${{ github.event.issue.pull_request && startsWith(github.event.comment.body, '/docs') }}
uses: windmill-labs/windmilldocs/.github/workflows/create-docs.yml@main
with:
pr_number: ${{ github.event.issue.number }}
repo: ${{ github.event.repository.name }}
comment_text: ${{ github.event.comment.body }}
secrets:
DOCS_TOKEN: ${{ secrets.DOCS_TOKEN }}
GOOGLE_API_KEY: ${{ secrets.GOOGLE_API_KEY }}
@@ -0,0 +1,32 @@
name: Create discord thread when a PR is opened, react with green checkmark when PR is merged
on:
pull_request:
types:
- opened
- ready_for_review
- closed
jobs:
notify_discord_when_pr_opened:
if: (github.event.pull_request.draft == false) && (github.event.action == 'opened' || github.event.action == 'ready_for_review')
uses: ./.github/workflows/shareable-discord-notification.yml
with:
PR_TITLE: ${{ github.event.pull_request.title }}
PR_URL: ${{ github.event.pull_request.html_url }}
PR_AUTHOR: ${{ github.event.pull_request.user.login }}
PR_STATUS: "opened"
PR_NUMBER: ${{ github.event.pull_request.number }}
secrets:
DISCORD_WEBHOOK_URL: ${{ secrets.DISCORD_PR_REVIEWS_WEBHOOK }}
merge_success_emoji:
if: github.event.pull_request.merged == true
uses: ./.github/workflows/shareable-discord-notification.yml
with:
PR_STATUS: "merged"
DISCORD_CHANNEL_ID: "1372204995868491786"
DISCORD_GUILD_ID: "930051556043276338"
PR_NUMBER: ${{ github.event.pull_request.number }}
secrets:
DISCORD_BOT_TOKEN: ${{ secrets.DISCORD_PR_BOT_TOKEN }}
@@ -0,0 +1,66 @@
name: Publish Helm Chart on Release
on:
release:
types: [published]
jobs:
bump-helm-version:
runs-on: ubicloud-standard-2
steps:
- name: Checkout on helm repository
uses: actions/checkout@v3
with:
repository: windmill-labs/windmill-helm-charts
token: ${{ secrets.DOCS_TOKEN }}
- name: Get version
id: get_version
run: |
echo "VERSION=${GITHUB_REF#refs/tags/v}" >> $GITHUB_ENV
- name: Create new branch
run: |
# Check if branch already exists remotely
if git ls-remote --heads origin bump-helm-version-${{ env.VERSION }} | grep -q bump-helm-version-${{ env.VERSION }}; then
# Branch exists, check it out
git fetch origin bump-helm-version-${{ env.VERSION }}
git checkout bump-helm-version-${{ env.VERSION }}
else
# Create new branch
git checkout -b bump-helm-version-${{ env.VERSION }}
fi
git config --local user.email "action@github.com"
git config --local user.name "GitHub Action"
- name: Bump helm version
run: |
# Get current version and increment it by 1
CURRENT_VERSION=$(grep "version:" ./charts/windmill/Chart.yaml | awk '{print $2}' | head -n 1)
NEW_VERSION=$(echo "$CURRENT_VERSION" | awk -F. '{$NF = $NF + 1;} 1' | sed 's/ /./g')
sed -i "s/^version: .*/version: $NEW_VERSION/" ./charts/windmill/Chart.yaml
# Get the app version from the version
VERSION=${{ env.VERSION }}
APP_VERSION=${VERSION#refs/tag/}
APP_VERSION=${APP_VERSION#v}
APP_VERSION=${APP_VERSION%/}
sed -i "s/appVersion: .*/appVersion: $APP_VERSION/" ./charts/windmill/Chart.yaml
- name: Commit and push
run: |
git add .
git commit -m "Bump helm version to ${{ env.VERSION }}"
git push origin bump-helm-version-${{ env.VERSION }}
- name: Create PR
env:
GH_TOKEN: ${{ secrets.DOCS_TOKEN }}
run: |
gh pr create \
--title "helm: bump version to ${{ env.VERSION }}" \
--body "This PR was auto-generated to bring the helm chart up to date for [release ${{ env.VERSION }}](https://github.com/windmill-labs/windmill/releases/tag/v${{ env.VERSION }}) in the main repo." \
--head bump-helm-version-${{ env.VERSION }} \
--base main
-34
View File
@@ -1,34 +0,0 @@
name: "Notify Discord on New PR (with Thread)"
on:
pull_request:
types:
- opened
- ready_for_review
jobs:
discord_notification:
# still guard out any drafts (just in case)
if: github.event.pull_request.draft == false
runs-on: ubicloud-standard-2
steps:
- name: Send Discord notification and start a thread
env:
WEBHOOK_URL: ${{ secrets.DISCORD_PR_REVIEWS_WEBHOOK }}
PR_TITLE: ${{ github.event.pull_request.title }}
PR_URL: ${{ github.event.pull_request.html_url }}
PR_AUTHOR: ${{ github.event.pull_request.user.login }}
run: |
payload=$(jq -n \
--arg content "${PR_URL}" \
--arg thread "$PR_TITLE by \`${PR_AUTHOR}\`" \
'{
content: $content,
thread_name: $thread,
auto_archive_duration: 10080
}'
)
curl -H "Content-Type: application/json" \
-X POST \
--data "$payload" \
"$WEBHOOK_URL"
@@ -0,0 +1,98 @@
name: "Notify Discord when a PR is opened or merged"
on:
workflow_call:
inputs:
PR_TITLE:
description: "The title of the PR"
type: string
PR_URL:
description: "The URL of the PR"
type: string
PR_AUTHOR:
description: "The author of the PR"
type: string
PR_STATUS:
description: "The status of the PR"
type: string
DISCORD_CHANNEL_ID:
description: "The Discord channel ID"
type: string
PR_NUMBER:
description: "The number of the PR"
type: string
DISCORD_GUILD_ID:
description: "The Discord guild ID"
type: string
secrets:
DISCORD_WEBHOOK_URL:
description: "Discord Webhook URL"
DISCORD_BOT_TOKEN:
description: "Discord Bot Token"
jobs:
open_thread:
runs-on: ubicloud-standard-2
if: ${{ inputs.PR_STATUS == 'opened' }}
steps:
- name: Send Discord notification and start a thread
env:
WEBHOOK_URL: ${{ secrets.DISCORD_WEBHOOK_URL }}
PR_TITLE: ${{ inputs.PR_TITLE }}
PR_NUMBER: ${{ inputs.PR_NUMBER }}
PR_URL: ${{ inputs.PR_URL }}
PR_AUTHOR: ${{ inputs.PR_AUTHOR }}
run: |
payload=$(jq -n \
--arg content "${PR_URL}" \
--arg thread "#${PR_NUMBER}: $PR_TITLE by \`${PR_AUTHOR}\`" \
'{
content: $content,
thread_name: $thread,
auto_archive_duration: 10080
}'
)
curl -H "Content-Type: application/json" \
-X POST \
-d "$payload" \
"$WEBHOOK_URL"
merge_success_emoji:
runs-on: ubuntu-latest
if: ${{ inputs.PR_STATUS == 'merged' }}
steps:
- name: React
env:
BOT_TOKEN: ${{ secrets.DISCORD_BOT_TOKEN }}
CHANNEL_ID: ${{ inputs.DISCORD_CHANNEL_ID }}
GUILD_ID: ${{ inputs.DISCORD_GUILD_ID }}
PR_NUMBER: ${{ inputs.PR_NUMBER }}
run: |
# 1) get PR thread
threads=$(curl -H "Authorization: Bot $BOT_TOKEN" "https://discord.com/api/v10/guilds/${GUILD_ID}/threads/active")
thread_id=$(
echo "$threads" \
| jq -r --arg cid "$CHANNEL_ID" \
--arg pref "#${PR_NUMBER}:" \
'.threads[]
| select(.parent_id == $cid and (.name | startswith($pref)))
| .id'
)
if [ -z "$thread_id" ]; then
echo "Thread not found"
exit 1
fi
# 2) get the first message in that thread
messages=$(curl -H "Authorization: Bot $BOT_TOKEN" \
"https://discord.com/api/v10/channels/$thread_id/messages?limit=1")
message_id=$(echo "$messages" | jq -r '.[-1].id')
if [ -z "$message_id" ]; then
echo "Message not found"
exit 1
fi
# 3) add the ✅ reaction
curl -X PUT \
-H "Authorization: Bot $BOT_TOKEN" \
"https://discord.com/api/v10/channels/$thread_id/messages/$message_id/reactions/%E2%9C%85/@me"
+38
View File
@@ -1,5 +1,43 @@
# Changelog
## [1.491.5](https://github.com/windmill-labs/windmill/compare/v1.491.4...v1.491.5) (2025-05-17)
### Bug Fixes
* improve handling of custom concurrency key/tag with preprocessors ([#5762](https://github.com/windmill-labs/windmill/issues/5762)) ([59afa49](https://github.com/windmill-labs/windmill/commit/59afa493fa20cc70b6825e6356713cef84d75312))
* S3 sql mode returns S3Object ([#5764](https://github.com/windmill-labs/windmill/issues/5764)) ([b29c6e7](https://github.com/windmill-labs/windmill/commit/b29c6e7636bb21c4d977bdaf89ac90e2a1a1086c))
## [1.491.4](https://github.com/windmill-labs/windmill/compare/v1.491.3...v1.491.4) (2025-05-15)
### Bug Fixes
* add v1 preprocessor support to workspace preprocessor script ([#5757](https://github.com/windmill-labs/windmill/issues/5757)) ([9b1c30e](https://github.com/windmill-labs/windmill/commit/9b1c30eeff35291ad50f3ddeb64831eac88e2f66))
## [1.491.3](https://github.com/windmill-labs/windmill/compare/v1.491.2...v1.491.3) (2025-05-15)
### Bug Fixes
* **frontend:** fix accordeon tabs initialization ([f488903](https://github.com/windmill-labs/windmill/commit/f488903635a1457f839ca641ed4f8d0891ef8212))
* http trigger routers cache version sequence ([#5755](https://github.com/windmill-labs/windmill/issues/5755)) ([d53bceb](https://github.com/windmill-labs/windmill/commit/d53bceb8004541b79d33220ae8de06d25521da91))
## [1.491.2](https://github.com/windmill-labs/windmill/compare/v1.491.1...v1.491.2) (2025-05-15)
### Bug Fixes
* **cli:** --version improvement ([f8f2015](https://github.com/windmill-labs/windmill/commit/f8f201564f7a323eb96f6dc684a525a0784d41f2))
* http trigger signature validation ([#5753](https://github.com/windmill-labs/windmill/issues/5753)) ([9e9514b](https://github.com/windmill-labs/windmill/commit/9e9514b9af2337e143a9e4cf1e915e1477032e80))
* Improve indexer performance by factoring required queries to the DB # ([#5749](https://github.com/windmill-labs/windmill/issues/5749)) ([b12feaf](https://github.com/windmill-labs/windmill/commit/b12feaf50ae0ef03816719ff39157fcf55159dbf))
* improve perf of job deletion ([0efba94](https://github.com/windmill-labs/windmill/commit/0efba945bac9b84a489c6ef552e834593f209fe1))
### Performance Improvements
* cache http trigger routers and auth ([#5748](https://github.com/windmill-labs/windmill/issues/5748)) ([ddd18d2](https://github.com/windmill-labs/windmill/commit/ddd18d22a615408a9f57f910d0a58f17e6d6e29d))
## [1.491.1](https://github.com/windmill-labs/windmill/compare/v1.491.0...v1.491.1) (2025-05-15)
@@ -1,35 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n value->'preprocessor_module' IS NOT NULL as has_preprocessor,\n value->'preprocessor_module'->'value'->'input_transforms'->'wm_trigger' IS NOT NULL as is_v1_preprocessor,\n schema as \"schema: _\"\n FROM flow \n WHERE workspace_id = $1 \n AND path = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "has_preprocessor",
"type_info": "Bool"
},
{
"ordinal": 1,
"name": "is_v1_preprocessor",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "schema: _",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null,
null,
true
]
},
"hash": "0e296134f05593edc989c628c00cbb60a5446993217baffa83f843bc12a5ac73"
}
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "SELECT nextval('http_trigger_version_seq')",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "nextval",
"type_info": "Int8"
}
],
"parameters": {
"Left": []
},
"nullable": [
null
]
},
"hash": "16be720bf1c88ecfa2a4bf6adbb1924df4817b6236b3a949209465f3a2c42bb9"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM v2_job_completed c\n USING v2_job j\n WHERE\n created_at <= now() - ($1::bigint::text || ' s')::interval\n AND completed_at + ($1::bigint::text || ' s')::interval <= now()\n AND c.id = j.id\n RETURNING c.id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
false
]
},
"hash": "1d819b829cd92995c39d29540df8cffbcc3334bada244a331a0bd8db06029d42"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n path, \n script_path, \n is_flow, \n route_path, \n authentication_resource_path,\n workspace_id, \n is_async, \n authentication_method AS \"authentication_method: _\", \n edited_by, \n email, \n static_asset_config AS \"static_asset_config: _\",\n wrap_body,\n raw_string,\n workspaced_route,\n is_static_website\n FROM \n http_trigger \n WHERE \n http_method = $1\n ",
"query": "\n SELECT \n path, \n script_path, \n is_flow, \n route_path, \n authentication_resource_path,\n workspace_id, \n is_async, \n authentication_method AS \"authentication_method: _\", \n edited_by, \n email, \n static_asset_config AS \"static_asset_config: _\",\n wrap_body,\n raw_string,\n workspaced_route,\n is_static_website\n FROM \n http_trigger \n WHERE \n http_method = $1\n ",
"describe": {
"columns": [
{
@@ -129,5 +129,5 @@
false
]
},
"hash": "4053f0bb30f651ddf2214115748daca0ea457da8252394eeeead0897d184f6da"
"hash": "1eeb218c30c0a6b0f7633813c764f57f8968894b4786b94109a057796ff500e4"
}
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "SELECT last_value FROM http_trigger_version_seq",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "last_value",
"type_info": "Int8"
}
],
"parameters": {
"Left": []
},
"nullable": [
false
]
},
"hash": "20fd50949796913dd48f67ee75b11272ce5b3046f87b9efd504e013fba9724f5"
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT EXISTS(\n SELECT 1\n FROM \n http_trigger \n WHERE \n workspace_id = $1 AND \n path = $2\n )\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "234acda79d470e99e9cbde5c7401d6f7894c25f90e39ec8606f79b8be56d1c17"
}
@@ -1,41 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT created_by, coalesce(job_logs.logs, '') as logs, job_logs.log_offset, job_logs.log_file_index\n FROM v2_as_completed_job\n LEFT JOIN job_logs ON job_logs.job_id = v2_as_completed_job.id\n WHERE v2_as_completed_job.id = $1 AND v2_as_completed_job.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "created_by",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "logs",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "log_offset",
"type_info": "Int4"
},
{
"ordinal": 3,
"name": "log_file_index",
"type_info": "TextArray"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
null,
false,
true
]
},
"hash": "282afbff89d3186d47ef5dbd0b65026ad37fb31b485fc44b6ec257dd77825428"
}
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "DROP INDEX CONCURRENTLY IF EXISTS log_file_hostname_log_ts_idx",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "2a33a35afc1ba4c31a5713cfd1a2c662f25cda387197aaf9f35000df31b8b07d"
}
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_completed_completed_at ON v2_job_completed (completed_at DESC)",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "49943f69ed74bc889120dcd2571e8e868a4f4795933044ff95b20d8df45cd145"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH job_result AS (\n SELECT result \n FROM v2_job_completed \n WHERE id = $1\n )\n UPDATE v2_job \n SET args = COALESCE(\n CASE \n WHEN job_result.result IS NULL THEN NULL\n WHEN jsonb_typeof(job_result.result) = 'object' \n THEN job_result.result\n WHEN jsonb_typeof(job_result.result) = 'null'\n THEN NULL\n ELSE jsonb_build_object('value', job_result.result)\n END, \n '{}'::jsonb\n ),\n preprocessed = TRUE\n FROM job_result\n WHERE v2_job.id = $2;\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Uuid"
]
},
"nullable": []
},
"hash": "52ad0c838d19cbd9e90b8368abe71dd12655179f41f43896e7d30fdfb3ae5939"
}
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM v2_job_completed c\n WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval \n RETURNING c.id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
false
]
},
"hash": "5820d34be1a7f7b72e656c692f53146f45ad4a6e584e917a0a86280d8f473c10"
}
@@ -0,0 +1,248 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n cj.id AS \"id!\",\n cj.workspace_id AS \"workspace_id!\",\n cj.parent_job,\n cj.created_by AS \"created_by!\",\n cj.duration_ms AS \"duration_ms!\",\n cj.success AS \"success!\",\n cj.script_hash AS \"script_hash!: Option<ScriptHash>\",\n cj.script_path,\n cj.args AS \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n cj.result AS \"result: sqlx::types::Json<Box<RawValue>>\",\n cj.deleted AS \"deleted!\",\n cj.canceled AS \"canceled!\",\n cj.canceled_by,\n cj.canceled_reason,\n cj.job_kind AS \"job_kind!: JobKind\",\n cj.schedule_path,\n cj.permissioned_as AS \"permissioned_as!\",\n cj.is_flow_step AS \"is_flow_step!\",\n cj.language AS \"language: ScriptLang\",\n cj.is_skipped AS \"is_skipped!\",\n cj.email AS \"email!\",\n cj.visible_to_owner AS \"visible_to_owner!\",\n cj.mem_peak,\n cj.tag AS \"tag!\",\n cj.created_at AS \"created_at!\",\n cj.started_at,\n job_logs.logs,\n job_logs.log_offset AS \"log_offset?\",\n job_logs.log_file_index\n\n FROM v2_as_completed_job AS cj\n LEFT JOIN job_logs ON cj.id = job_logs.job_id\n WHERE (cj.created_at > $1 AND cj.created_at < $3)\n OR cj.id = ANY($2)\n ORDER BY cj.created_at ASC LIMIT $4",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "workspace_id!",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "parent_job",
"type_info": "Uuid"
},
{
"ordinal": 3,
"name": "created_by!",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "duration_ms!",
"type_info": "Int8"
},
{
"ordinal": 5,
"name": "success!",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_hash!: Option<ScriptHash>",
"type_info": "Int8"
},
{
"ordinal": 7,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 8,
"name": "args: sqlx::types::Json<HashMap<String, Box<RawValue>>>",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "result: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 10,
"name": "deleted!",
"type_info": "Bool"
},
{
"ordinal": 11,
"name": "canceled!",
"type_info": "Bool"
},
{
"ordinal": 12,
"name": "canceled_by",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "canceled_reason",
"type_info": "Text"
},
{
"ordinal": 14,
"name": "job_kind!: JobKind",
"type_info": {
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode",
"appscript"
]
}
}
}
},
{
"ordinal": 15,
"name": "schedule_path",
"type_info": "Varchar"
},
{
"ordinal": 16,
"name": "permissioned_as!",
"type_info": "Varchar"
},
{
"ordinal": 17,
"name": "is_flow_step!",
"type_info": "Bool"
},
{
"ordinal": 18,
"name": "language: ScriptLang",
"type_info": {
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb",
"nu",
"java"
]
}
}
}
},
{
"ordinal": 19,
"name": "is_skipped!",
"type_info": "Bool"
},
{
"ordinal": 20,
"name": "email!",
"type_info": "Varchar"
},
{
"ordinal": 21,
"name": "visible_to_owner!",
"type_info": "Bool"
},
{
"ordinal": 22,
"name": "mem_peak",
"type_info": "Int4"
},
{
"ordinal": 23,
"name": "tag!",
"type_info": "Varchar"
},
{
"ordinal": 24,
"name": "created_at!",
"type_info": "Timestamptz"
},
{
"ordinal": 25,
"name": "started_at",
"type_info": "Timestamptz"
},
{
"ordinal": 26,
"name": "logs",
"type_info": "Text"
},
{
"ordinal": 27,
"name": "log_offset?",
"type_info": "Int4"
},
{
"ordinal": 28,
"name": "log_file_index",
"type_info": "TextArray"
}
],
"parameters": {
"Left": [
"Timestamptz",
"UuidArray",
"Timestamptz",
"Int8"
]
},
"nullable": [
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
false,
true
]
},
"hash": "6e2564dc37ee967c634deade67ceabc4c516418e595dee9cd7d651acc1f4af6e"
}
@@ -0,0 +1,29 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n value->'preprocessor_module'->'value' as \"preprocessor_module: _\",\n schema as \"schema: _\"\n FROM flow \n WHERE workspace_id = $1\n AND path = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "preprocessor_module: _",
"type_info": "Jsonb"
},
{
"ordinal": 1,
"name": "schema: _",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null,
true
]
},
"hash": "72916f8e490f8252e0a51b7f562ccc3be832b12102eb86a07d8405a4fa9287d5"
}
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT result as \"result: Json<HashMap<String, Box<RawValue>>>\"\n FROM v2_job_completed \n WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "result: Json<HashMap<String, Box<RawValue>>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "91f23fcc27777c279c79e2682fc15c026e55f9ec3799be65a2e8920fe6174a17"
}
@@ -1,134 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n path, \n script_path, \n is_flow, \n route_path, \n workspace_id, \n is_async, \n authentication_method AS \"authentication_method: _\", \n edited_by, \n email,\n static_asset_config AS \"static_asset_config: _\",\n wrap_body,\n raw_string,\n workspaced_route,\n is_static_website,\n authentication_resource_path\n FROM \n http_trigger \n WHERE \n workspace_id = $1 AND \n http_method = $2\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 3,
"name": "route_path",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "is_async",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "authentication_method: _",
"type_info": {
"Custom": {
"name": "authentication_method",
"kind": {
"Enum": [
"none",
"windmill",
"api_key",
"basic_http",
"custom_script",
"signature"
]
}
}
}
},
{
"ordinal": 7,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 8,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 9,
"name": "static_asset_config: _",
"type_info": "Jsonb"
},
{
"ordinal": 10,
"name": "wrap_body",
"type_info": "Bool"
},
{
"ordinal": 11,
"name": "raw_string",
"type_info": "Bool"
},
{
"ordinal": 12,
"name": "workspaced_route",
"type_info": "Bool"
},
{
"ordinal": 13,
"name": "is_static_website",
"type_info": "Bool"
},
{
"ordinal": 14,
"name": "authentication_resource_path",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "http_method",
"kind": {
"Enum": [
"get",
"post",
"put",
"delete",
"patch"
]
}
}
}
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
false,
true
]
},
"hash": "927149213e0f8ae983652ef80464f646d6be80e702193f1acdd40dd6033652e4"
}
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "CREATE INDEX CONCURRENTLY IF NOT EXISTS alerts_by_workspace ON alerts (workspace_id);",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "a8ca4e588e0bf3c4bba2fe4b68a5364e4cba99964513599f8a012a5680d3dca8"
}
@@ -18,8 +18,8 @@
"Left": []
},
"nullable": [
true,
false
false,
true
]
},
"hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76"
@@ -0,0 +1,46 @@
{
"db_name": "PostgreSQL",
"query": "\n WITH j AS (\n SELECT \n raw_flow->>'concurrency_key' as concurrency_key, \n raw_flow->>'concurrency_time_window_s' as concurrency_time_window_s,\n raw_flow->>'concurrency_limit' as concurrent_limit,\n runnable_path, \n runnable_id as version FROM v2_job\n WHERE id = $1\n )\n SELECT tag, j.concurrency_key, j.concurrency_time_window_s::int, j.concurrent_limit::int, j.version\n FROM flow, j\n WHERE path = j.runnable_path\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "concurrency_key",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 3,
"name": "concurrent_limit",
"type_info": "Int4"
},
{
"ordinal": 4,
"name": "version",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true,
null,
null,
null,
true
]
},
"hash": "b7335ac24702c86fbb4ab95916a6aa1648082287b09122755df2462dc71ce831"
}
@@ -0,0 +1,18 @@
{
"db_name": "PostgreSQL",
"query": "WITH job_result AS (\n SELECT result \n FROM v2_job_completed \n WHERE id = $1\n ),\n updated_queue AS (\n UPDATE v2_job_queue\n SET running = false,\n tag = COALESCE($3, tag)\n WHERE id = $2\n )\n UPDATE v2_job \n SET \n tag = COALESCE($3, tag),\n concurrent_limit = COALESCE($4, concurrent_limit),\n concurrency_time_window_s = COALESCE($5, concurrency_time_window_s),\n args = COALESCE(\n CASE \n WHEN job_result.result IS NULL THEN NULL\n WHEN jsonb_typeof(job_result.result) = 'object' \n THEN job_result.result\n WHEN jsonb_typeof(job_result.result) = 'null'\n THEN NULL\n ELSE jsonb_build_object('value', job_result.result)\n END, \n '{}'::jsonb\n ),\n preprocessed = TRUE\n FROM job_result\n WHERE v2_job.id = $2;\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Uuid",
"Varchar",
"Int4",
"Int4"
]
},
"nullable": []
},
"hash": "e07660e8d2a265cb6a83f3a2bb8e7e6330f09ab116e9837f6f16f8fdef938004"
}
@@ -0,0 +1,246 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n cj.id AS \"id!\",\n cj.workspace_id AS \"workspace_id!\",\n cj.parent_job,\n cj.created_by AS \"created_by!\",\n cj.duration_ms AS \"duration_ms!\",\n cj.success AS \"success!\",\n cj.script_hash AS \"script_hash!: Option<ScriptHash>\",\n cj.script_path,\n cj.args AS \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n cj.result AS \"result: sqlx::types::Json<Box<RawValue>>\",\n cj.deleted AS \"deleted!\",\n cj.canceled AS \"canceled!\",\n cj.canceled_by,\n cj.canceled_reason,\n cj.job_kind AS \"job_kind!: JobKind\",\n cj.schedule_path,\n cj.permissioned_as AS \"permissioned_as!\",\n cj.is_flow_step AS \"is_flow_step!\",\n cj.language AS \"language: ScriptLang\",\n cj.is_skipped AS \"is_skipped!\",\n cj.email AS \"email!\",\n cj.visible_to_owner AS \"visible_to_owner!\",\n cj.mem_peak,\n cj.tag AS \"tag!\",\n cj.created_at AS \"created_at!\",\n cj.started_at,\n job_logs.logs,\n job_logs.log_offset AS \"log_offset?\",\n job_logs.log_file_index\n\n FROM v2_as_completed_job AS cj\n LEFT JOIN job_logs ON cj.id = job_logs.job_id\n WHERE cj.created_at < $1\n ORDER BY cj.created_at ASC LIMIT $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "workspace_id!",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "parent_job",
"type_info": "Uuid"
},
{
"ordinal": 3,
"name": "created_by!",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "duration_ms!",
"type_info": "Int8"
},
{
"ordinal": 5,
"name": "success!",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_hash!: Option<ScriptHash>",
"type_info": "Int8"
},
{
"ordinal": 7,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 8,
"name": "args: sqlx::types::Json<HashMap<String, Box<RawValue>>>",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "result: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 10,
"name": "deleted!",
"type_info": "Bool"
},
{
"ordinal": 11,
"name": "canceled!",
"type_info": "Bool"
},
{
"ordinal": 12,
"name": "canceled_by",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "canceled_reason",
"type_info": "Text"
},
{
"ordinal": 14,
"name": "job_kind!: JobKind",
"type_info": {
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode",
"appscript"
]
}
}
}
},
{
"ordinal": 15,
"name": "schedule_path",
"type_info": "Varchar"
},
{
"ordinal": 16,
"name": "permissioned_as!",
"type_info": "Varchar"
},
{
"ordinal": 17,
"name": "is_flow_step!",
"type_info": "Bool"
},
{
"ordinal": 18,
"name": "language: ScriptLang",
"type_info": {
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb",
"nu",
"java"
]
}
}
}
},
{
"ordinal": 19,
"name": "is_skipped!",
"type_info": "Bool"
},
{
"ordinal": 20,
"name": "email!",
"type_info": "Varchar"
},
{
"ordinal": 21,
"name": "visible_to_owner!",
"type_info": "Bool"
},
{
"ordinal": 22,
"name": "mem_peak",
"type_info": "Int4"
},
{
"ordinal": 23,
"name": "tag!",
"type_info": "Varchar"
},
{
"ordinal": 24,
"name": "created_at!",
"type_info": "Timestamptz"
},
{
"ordinal": 25,
"name": "started_at",
"type_info": "Timestamptz"
},
{
"ordinal": 26,
"name": "logs",
"type_info": "Text"
},
{
"ordinal": 27,
"name": "log_offset?",
"type_info": "Int4"
},
{
"ordinal": 28,
"name": "log_file_index",
"type_info": "TextArray"
}
],
"parameters": {
"Left": [
"Timestamptz",
"Int8"
]
},
"nullable": [
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
false,
true
]
},
"hash": "f22964772dc2d67aee437bbbd08b64792c00da1d713d7ca8f9904ccce7bfdae7"
}
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT EXISTS(\n SELECT 1 \n FROM \n http_trigger \n WHERE \n workspace_id = $1 AND \n path = $2\n )\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "f632d08a8d3df691fff9f57fdf926f787287c3cd181a6853056077edba10473d"
}
@@ -41,11 +41,11 @@
]
},
"nullable": [
true,
true,
true,
true,
true,
false,
false,
false,
false,
false,
true
]
},
+78 -72
View File
@@ -744,9 +744,9 @@ checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26"
[[package]]
name = "aws-config"
version = "1.6.2"
version = "1.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6fcc63c9860579e4cb396239570e979376e70aab79e496621748a09913f8b36"
checksum = "02a18fd934af6ae7ca52410d4548b98eb895aab0f1ea417d168d85db1434a141"
dependencies = [
"aws-credential-types",
"aws-runtime",
@@ -833,9 +833,9 @@ dependencies = [
[[package]]
name = "aws-sdk-sqs"
version = "1.67.0"
version = "1.68.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c6f15bedfb1c4385fccc474f0fe46dffb0335d0b3d6b4413df06fb30d90caba8"
checksum = "5b484821a335b02b109c17623b8347e692583c2229f8db2f029edd0fdbbd3bea"
dependencies = [
"aws-credential-types",
"aws-runtime",
@@ -849,16 +849,15 @@ dependencies = [
"bytes",
"fastrand",
"http 0.2.12",
"once_cell",
"regex-lite",
"tracing",
]
[[package]]
name = "aws-sdk-sso"
version = "1.67.0"
version = "1.68.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0d4863da26489d1e6da91d7e12b10c17e86c14f94c53f416bd10e0a9c34057ba"
checksum = "bd5f01ea61fed99b5fe4877abff6c56943342a56ff145e9e0c7e2494419008be"
dependencies = [
"aws-credential-types",
"aws-runtime",
@@ -872,16 +871,15 @@ dependencies = [
"bytes",
"fastrand",
"http 0.2.12",
"once_cell",
"regex-lite",
"tracing",
]
[[package]]
name = "aws-sdk-ssooidc"
version = "1.68.0"
version = "1.69.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "95caa3998d7237789b57b95a8e031f60537adab21fa84c91e35bef9455c652e4"
checksum = "27454e4c55aaa4ef65647e3a1cf095cb834ca6d54e959e2909f1fef96ad87860"
dependencies = [
"aws-credential-types",
"aws-runtime",
@@ -895,16 +893,15 @@ dependencies = [
"bytes",
"fastrand",
"http 0.2.12",
"once_cell",
"regex-lite",
"tracing",
]
[[package]]
name = "aws-sdk-sts"
version = "1.68.0"
version = "1.69.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4939f6f449a37308a78c5a910fd91265479bd2bb11d186f0b8fc114d89ec828d"
checksum = "ffd6ef5d00c94215960fabcdf2d9fe7c090eed8be482d66d47b92d4aba1dd4aa"
dependencies = [
"aws-credential-types",
"aws-runtime",
@@ -919,16 +916,15 @@ dependencies = [
"aws-types",
"fastrand",
"http 0.2.12",
"once_cell",
"regex-lite",
"tracing",
]
[[package]]
name = "aws-sigv4"
version = "1.3.1"
version = "1.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3503af839bd8751d0bdc5a46b9cac93a003a353e635b0c12cf2376b5b53e41ea"
checksum = "3734aecf9ff79aa401a6ca099d076535ab465ff76b46440cf567c8e70b65dc13"
dependencies = [
"aws-credential-types",
"aws-smithy-http",
@@ -1843,9 +1839,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.2.22"
version = "1.2.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32db95edf998450acc7881c932f94cd9b05c87b4b2599e8bab064753da4acfd1"
checksum = "5f4ac86a9e5bc1e2b3449ab9d7d3a6a405e3d1bb28d7b9be8614f55846ae3766"
dependencies = [
"jobserver",
"libc",
@@ -3215,7 +3211,7 @@ dependencies = [
"once_cell",
"percent-encoding",
"serde",
"sourcemap 9.2.0",
"sourcemap 9.2.1",
"swc_atoms",
"swc_common",
"swc_config",
@@ -4152,9 +4148,9 @@ dependencies = [
[[package]]
name = "deno_unsync"
version = "0.4.2"
version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d774fd83f26b24f0805a6ab8b26834a0d06ceac0db517b769b1e4633c96a2057"
checksum = "47c618b51088b3ac67f15c69b3ed7620ba3a7d495e5a090186df9424b5ab623e"
dependencies = [
"futures",
"parking_lot 0.12.3",
@@ -4871,9 +4867,9 @@ dependencies = [
[[package]]
name = "errno"
version = "0.3.11"
version = "0.3.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "976dd42dc7e85965fe702eb8164f21f450704bdde31faefd6471dba214cb594e"
checksum = "cea14ef9355e3beab063703aa9dab15afd25f0667c341310c1e5274bb1d0da18"
dependencies = [
"libc",
"windows-sys 0.59.0",
@@ -6494,7 +6490,7 @@ dependencies = [
"js-sys",
"log",
"wasm-bindgen",
"windows-core 0.61.0",
"windows-core 0.61.1",
]
[[package]]
@@ -8823,9 +8819,9 @@ dependencies = [
[[package]]
name = "owo-colors"
version = "4.2.0"
version = "4.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1036865bb9422d3300cf723f657c2851d0e9ab12567854b1f4eba3d77decf564"
checksum = "26995317201fa17f3656c36716aed4a7c81743a9634ac4c99c0eeda495db0cec"
[[package]]
name = "p224"
@@ -11497,9 +11493,9 @@ dependencies = [
[[package]]
name = "sourcemap"
version = "9.2.0"
version = "9.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dd430118acc9fdd838557649b9b43fd0a78e3834d84a283b466f8e84720d6101"
checksum = "bdee719193ae5c919a3ee43f64c2c0dd87f9b9a451d67918a2a5ec2e3c70561c"
dependencies = [
"base64-simd 0.8.0",
"bitvec",
@@ -11980,7 +11976,7 @@ dependencies = [
"rustc-hash 1.1.0",
"serde",
"siphasher 0.3.11",
"sourcemap 9.2.0",
"sourcemap 9.2.1",
"swc_allocator",
"swc_atoms",
"swc_eq_ignore_macros",
@@ -12044,7 +12040,7 @@ dependencies = [
"num-bigint",
"once_cell",
"serde",
"sourcemap 9.2.0",
"sourcemap 9.2.1",
"swc_allocator",
"swc_atoms",
"swc_common",
@@ -14401,7 +14397,7 @@ checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
[[package]]
name = "windmill"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"axum",
@@ -14450,7 +14446,7 @@ dependencies = [
[[package]]
name = "windmill-api"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"argon2",
@@ -14559,7 +14555,7 @@ dependencies = [
[[package]]
name = "windmill-api-client"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"base64 0.22.1",
"chrono",
@@ -14574,7 +14570,7 @@ dependencies = [
[[package]]
name = "windmill-audit"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"chrono",
"serde",
@@ -14587,7 +14583,7 @@ dependencies = [
[[package]]
name = "windmill-autoscaling"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"serde",
@@ -14601,7 +14597,7 @@ dependencies = [
[[package]]
name = "windmill-common"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"async-stream",
@@ -14671,7 +14667,7 @@ dependencies = [
[[package]]
name = "windmill-git-sync"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"regex",
"serde",
@@ -14685,7 +14681,7 @@ dependencies = [
[[package]]
name = "windmill-indexer"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"bytes",
@@ -14708,7 +14704,7 @@ dependencies = [
[[package]]
name = "windmill-macros"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"itertools 0.14.0",
"lazy_static",
@@ -14720,7 +14716,7 @@ dependencies = [
[[package]]
name = "windmill-parser"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"convert_case 0.6.0",
"serde",
@@ -14729,7 +14725,7 @@ dependencies = [
[[package]]
name = "windmill-parser-bash"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"lazy_static",
@@ -14741,7 +14737,7 @@ dependencies = [
[[package]]
name = "windmill-parser-csharp"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"serde_json",
@@ -14753,7 +14749,7 @@ dependencies = [
[[package]]
name = "windmill-parser-go"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"gosyn",
@@ -14765,7 +14761,7 @@ dependencies = [
[[package]]
name = "windmill-parser-graphql"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"lazy_static",
@@ -14777,7 +14773,7 @@ dependencies = [
[[package]]
name = "windmill-parser-java"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"serde_json",
@@ -14789,7 +14785,7 @@ dependencies = [
[[package]]
name = "windmill-parser-nu"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"nu-parser",
@@ -14800,7 +14796,7 @@ dependencies = [
[[package]]
name = "windmill-parser-php"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"itertools 0.14.0",
@@ -14811,7 +14807,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"itertools 0.14.0",
@@ -14822,7 +14818,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py-imports"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"async-recursion",
@@ -14842,7 +14838,7 @@ dependencies = [
[[package]]
name = "windmill-parser-rust"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"convert_case 0.6.0",
@@ -14859,7 +14855,7 @@ dependencies = [
[[package]]
name = "windmill-parser-sql"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"lazy_static",
@@ -14871,7 +14867,7 @@ dependencies = [
[[package]]
name = "windmill-parser-ts"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"lazy_static",
@@ -14889,7 +14885,7 @@ dependencies = [
[[package]]
name = "windmill-parser-wasm"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"getrandom 0.2.16",
@@ -14913,7 +14909,7 @@ dependencies = [
[[package]]
name = "windmill-parser-yaml"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"serde_json",
@@ -14923,7 +14919,7 @@ dependencies = [
[[package]]
name = "windmill-queue"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"async-recursion",
@@ -14956,7 +14952,7 @@ dependencies = [
[[package]]
name = "windmill-sql-datatype-parser-wasm"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"wasm-bindgen",
"wasm-bindgen-test",
@@ -14966,7 +14962,7 @@ dependencies = [
[[package]]
name = "windmill-worker"
version = "1.491.1"
version = "1.491.5"
dependencies = [
"anyhow",
"async-recursion",
@@ -15083,7 +15079,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c5ee8f3d025738cb02bad7868bbb5f8a6327501e870bf51f1b455b0a2454a419"
dependencies = [
"windows-collections",
"windows-core 0.61.0",
"windows-core 0.61.1",
"windows-future",
"windows-link",
"windows-numerics",
@@ -15095,7 +15091,7 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8"
dependencies = [
"windows-core 0.61.0",
"windows-core 0.61.1",
]
[[package]]
@@ -15124,25 +15120,26 @@ dependencies = [
[[package]]
name = "windows-core"
version = "0.61.0"
version = "0.61.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4763c1de310c86d75a878046489e2e5ba02c649d185f21c67d4cf8a56d098980"
checksum = "46ec44dc15085cea82cf9c78f85a9114c463a369786585ad2882d1ff0b0acf40"
dependencies = [
"windows-implement 0.60.0",
"windows-interface 0.59.1",
"windows-link",
"windows-result 0.3.2",
"windows-strings 0.4.0",
"windows-result 0.3.3",
"windows-strings 0.4.1",
]
[[package]]
name = "windows-future"
version = "0.2.0"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7a1d6bbefcb7b60acd19828e1bc965da6fcf18a7e39490c5f8be71e54a19ba32"
checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e"
dependencies = [
"windows-core 0.61.0",
"windows-core 0.61.1",
"windows-link",
"windows-threading",
]
[[package]]
@@ -15223,7 +15220,7 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1"
dependencies = [
"windows-core 0.61.0",
"windows-core 0.61.1",
"windows-link",
]
@@ -15233,7 +15230,7 @@ version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4286ad90ddb45071efd1a66dfa43eb02dd0dfbae1545ad6cc3c51cf34d7e8ba3"
dependencies = [
"windows-result 0.3.2",
"windows-result 0.3.3",
"windows-strings 0.3.1",
"windows-targets 0.53.0",
]
@@ -15249,9 +15246,9 @@ dependencies = [
[[package]]
name = "windows-result"
version = "0.3.2"
version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c64fd11a4fd95df68efcfee5f44a294fe71b8bc6a91993e2791938abcc712252"
checksum = "4b895b5356fc36103d0f64dd1e94dfa7ac5633f1c9dd6e80fe9ec4adef69e09d"
dependencies = [
"windows-link",
]
@@ -15267,9 +15264,9 @@ dependencies = [
[[package]]
name = "windows-strings"
version = "0.4.0"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7a2ba9642430ee452d5a7aa78d72907ebe8cfda358e8cb7918a2050581322f97"
checksum = "2a7ab927b2637c19b3dbe0965e75d8f2d30bdd697a1516191cad2ec4df8fb28a"
dependencies = [
"windows-link",
]
@@ -15348,6 +15345,15 @@ dependencies = [
"windows_x86_64_msvc 0.53.0",
]
[[package]]
name = "windows-threading"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6"
dependencies = [
"windows-link",
]
[[package]]
name = "windows_aarch64_gnullvm"
version = "0.48.5"
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "windmill"
version = "1.491.1"
version = "1.491.5"
authors.workspace = true
edition.workspace = true
@@ -32,7 +32,7 @@ members = [
]
[workspace.package]
version = "1.491.1"
version = "1.491.5"
authors = ["Ruben Fiszel <ruben@windmill.dev>"]
edition = "2021"
+1 -1
View File
@@ -1 +1 @@
b77d145e278de3bd4079e3228733df24cc4c0070
3efa7fa51e9f93f60e141fef5b8b9338528cf955
@@ -0,0 +1,4 @@
-- Add down migration script here
DROP TRIGGER http_trigger_change_trigger ON http_trigger;
DROP FUNCTION notify_http_trigger_change();
DROP SEQUENCE http_trigger_version_seq;
@@ -0,0 +1,15 @@
-- Add up migration script here
CREATE OR REPLACE FUNCTION notify_http_trigger_change()
RETURNS TRIGGER AS $$
BEGIN
PERFORM pg_notify('notify_http_trigger_change', NEW.workspace_id || ':' || NEW.path);
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER http_trigger_change_trigger
AFTER INSERT OR UPDATE OR DELETE ON http_trigger
FOR EACH ROW
EXECUTE FUNCTION notify_http_trigger_change();
CREATE SEQUENCE http_trigger_version_seq;
@@ -0,0 +1 @@
-- Add down migration script here
@@ -0,0 +1,3 @@
-- Add up migration script here
GRANT ALL ON dependency_map TO windmill_user;
GRANT ALL ON dependency_map TO windmill_admin;
@@ -0,0 +1 @@
-- Add down migration script here
@@ -0,0 +1,9 @@
-- Add up migration script here
-- this makes sure that the first time nextval is called, 2 is returned
-- otherwise, `SELECT last_value from http_trigger_version_seq;` would return 1 before and after the first nextval call
-- which would not refresh the routers cache after the first create/update/delete
SELECT setval(
'http_trigger_version_seq',
(SELECT last_value FROM http_trigger_version_seq),
true
);
+19
View File
@@ -810,6 +810,21 @@ Windmill Community Edition {GIT_VERSION}
}
}
},
#[cfg(feature = "http_trigger")]
"notify_http_trigger_change" => {
tracing::info!("HTTP trigger change detected: {}", n.payload());
match windmill_api::http_triggers::refresh_routers(&db).await {
Ok((true, _)) => {
tracing::info!("Refreshed HTTP routers (trigger change)");
},
Ok((false, _)) => {
tracing::warn!("Should have refreshed HTTP routers (trigger change) but did not");
},
Err(err) => {
tracing::error!("Error refreshing HTTP routers (trigger change): {err:#}");
}
};
},
"notify_global_setting_change" => {
tracing::info!("Global setting change detected: {}", n.payload());
match n.payload() {
@@ -1133,6 +1148,10 @@ async fn listen_pg(url: &str) -> Option<PgListener> {
"notify_workspace_envs_change",
"notify_runnable_version_change",
];
#[cfg(feature = "http_trigger")]
channels.push("notify_http_trigger_change");
#[cfg(feature = "cloud")]
channels.push("notify_workspace_premium_change");
+6 -6
View File
@@ -62,7 +62,11 @@ use windmill_common::{
users::truncate_token,
utils::{empty_as_none, now_from_db, rd_string, report_critical_error, Mode},
worker::{
load_env_vars, load_init_bash_from_env, load_whitelist_env_vars_from_env, load_worker_config, reload_custom_tags_setting, store_pull_query, store_suspended_pull_query, update_min_version, Connection, WorkerConfig, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, SCRIPT_TOKEN_EXPIRY, SMTP_CONFIG, TMP_DIR, WORKER_CONFIG, WORKER_GROUP
load_env_vars, load_init_bash_from_env, load_whitelist_env_vars_from_env,
load_worker_config, reload_custom_tags_setting, store_pull_query,
store_suspended_pull_query, update_min_version, Connection, WorkerConfig,
DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, SCRIPT_TOKEN_EXPIRY,
SMTP_CONFIG, TMP_DIR, WORKER_CONFIG, WORKER_GROUP,
},
KillpillSender, BASE_URL, CRITICAL_ALERTS_ON_DB_OVERSIZE, CRITICAL_ALERT_MUTE_UI_ENABLED,
CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS,
@@ -829,11 +833,7 @@ pub async fn delete_expired_items(db: &DB) -> () {
Ok(mut tx) => {
let deleted_jobs = sqlx::query_scalar!(
"DELETE FROM v2_job_completed c
USING v2_job j
WHERE
created_at <= now() - ($1::bigint::text || ' s')::interval
AND completed_at + ($1::bigint::text || ' s')::interval <= now()
AND c.id = j.id
WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval
RETURNING c.id",
job_retention_secs
)
+34 -2
View File
@@ -1,7 +1,7 @@
openapi: "3.0.3"
info:
version: 1.491.1
version: 1.491.5
title: Windmill API
contact:
@@ -11028,7 +11028,8 @@ paths:
description: a config
content:
application/json:
schema: {}
schema:
$ref: "#/components/schemas/Configs"
/configs/update/{name}:
post:
@@ -13183,6 +13184,37 @@ components:
code_completion_model:
$ref: "#/components/schemas/AIProviderModel"
Alert:
type: object
properties:
name:
type: string
tags_to_monitor:
type: array
items:
type: string
jobs_num_threshold:
type: integer
alert_cooldown_seconds:
type: integer
alert_time_threshold_seconds:
type: integer
required:
- name
- tags_to_monitor
- jobs_num_threshold
- alert_cooldown_seconds
- alert_time_threshold_seconds
Configs:
type: object
nullable: true
properties:
alerts:
type: array
items:
$ref: '#/components/schemas/Alert'
Script:
type: object
properties:
+3 -3
View File
@@ -29,14 +29,14 @@ pub enum RawBody {
Empty,
}
#[derive(Clone, Serialize)]
#[derive(Debug, Clone, Serialize)]
#[serde(untagged)]
pub enum Body {
HashMap(HashMap<String, Box<RawValue>>),
NoHashMap(Box<RawValue>),
}
#[derive(Clone, Default)]
#[derive(Debug, Clone, Default)]
pub struct WebhookArgsMetadata {
pub raw_string: Option<String>,
pub headers: HashMap<String, Box<RawValue>>,
@@ -51,7 +51,7 @@ pub struct RawWebhookArgs {
pub metadata: WebhookArgsMetadata,
}
#[derive(Clone)]
#[derive(Debug, Clone)]
pub struct WebhookArgs {
pub body: Body,
pub metadata: WebhookArgsMetadata,
+11 -9
View File
@@ -14,11 +14,18 @@ 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,
use {
crate::gcp_triggers_ee::{
manage_google_subscription, process_google_push_request, validate_jwt_token,
CreateUpdateConfig, SubscriptionMode,
},
axum::extract::Request,
http::HeaderMap,
};
#[cfg(all(feature = "enterprise", feature = "gcp_trigger"))]
use windmill_common::utils::empty_as_none;
#[cfg(all(feature = "enterprise", feature = "sqs_trigger"))]
use windmill_common::auth::aws::AwsAuthResourceType;
@@ -26,12 +33,7 @@ use windmill_common::auth::aws::AwsAuthResourceType;
feature = "http_trigger",
all(feature = "enterprise", feature = "gcp_trigger")
))]
use {
axum::extract::Request,
http::HeaderMap,
serde::de::DeserializeOwned,
windmill_common::{error::Error, utils::empty_as_none},
};
use {serde::de::DeserializeOwned, windmill_common::error::Error};
#[cfg(all(feature = "enterprise", feature = "kafka"))]
use crate::kafka_triggers_ee::KafkaTriggerConfigConnection;
+23 -1
View File
@@ -782,10 +782,32 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
.execute(db)
.await?;
});
run_windmill_migration!("job_completed_completed_at", db, |tx| {
sqlx::query!(
"CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_completed_completed_at ON v2_job_completed (completed_at DESC)"
)
.execute(db)
.await?;
});
run_windmill_migration!("alerts_by_workspace", db, |tx| {
sqlx::query!(
"CREATE INDEX CONCURRENTLY IF NOT EXISTS alerts_by_workspace ON alerts (workspace_id);"
)
.execute(db)
.await?;
});
run_windmill_migration!("remove_redundant_log_file_index", db, |tx| {
sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS log_file_hostname_log_ts_idx")
.execute(db)
.await?;
});
Ok(())
}
#[derive(Clone, Debug)]
#[derive(Clone, Debug, Hash, Eq, PartialEq)]
pub struct ApiAuthed {
pub email: String,
pub username: String,
@@ -17,7 +17,7 @@ use crate::{
pub struct RawHttpTriggerArgs(pub RawWebhookArgs);
#[derive(Serialize, Deserialize, sqlx::Type, Debug)]
#[derive(Serialize, Deserialize, sqlx::Type, Debug, Clone, Hash, Eq, PartialEq)]
#[sqlx(type_name = "HTTP_METHOD", rename_all = "lowercase")]
#[serde(rename_all = "lowercase")]
pub enum HttpMethod {
@@ -56,7 +56,7 @@ where
}
}
#[derive(Clone)]
#[derive(Debug, Clone)]
pub struct HttpTriggerArgs(pub WebhookArgs);
impl RawHttpTriggerArgs {
@@ -414,7 +414,7 @@ pub enum Encoding {
Base64Uri,
Hex,
}
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct SignatureAuthenticationMethod {
algorithm: HmacAlgorithm,
encoding: Encoding,
@@ -426,20 +426,20 @@ pub struct SignatureConfigData<'config> {
secret_key: &'config str,
}
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct SignatureAuthentication {
signature_provider: WebhookType,
secret_key: String,
authentication_config: Option<SignatureAuthenticationMethod>,
}
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct BasicAuthAuthentication {
pub username: String,
pub password: String,
}
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct ApiKeyAuthentication {
api_key_header: String,
api_key_secret: String,
@@ -558,7 +558,7 @@ pub fn verify_hmac_signature(
Ok(())
}
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(untagged)]
pub enum AuthenticationMethod {
Signature(SignatureAuthentication),
+266 -149
View File
@@ -4,7 +4,7 @@ use crate::http_trigger_args::{HttpMethod, RawHttpTriggerArgs};
use crate::job_helpers_ee::get_workspace_s3_resource;
use crate::resources::try_get_resource_from_db_as;
use crate::trigger_helpers::{get_runnable_format, RunnableId};
use crate::utils::non_empty_str;
use crate::utils::{non_empty_str, ExpiringCacheEntry};
use crate::{
auth::{AuthCache, OptTokened},
db::{ApiAuthed, DB},
@@ -24,11 +24,14 @@ use axum::{
#[cfg(feature = "parquet")]
use http::header::IF_NONE_MATCH;
use http::{HeaderMap, StatusCode};
use quick_cache::sync::Cache;
use serde::{Deserialize, Serialize};
use sql_builder::{bind::Bind, SqlBuilder};
use sqlx::prelude::FromRow;
use sqlx::PgTransaction;
use std::borrow::Cow;
use std::{collections::HashMap, sync::Arc};
use tokio::sync::{RwLock, RwLockReadGuard};
use tower_http::cors::CorsLayer;
use windmill_audit::{audit_ee::audit_log, ActionKind};
use windmill_common::error::Error;
@@ -283,6 +286,16 @@ fn validate_authentication_method(
Ok(())
}
async fn increase_trigger_version_and_commit(mut tx: PgTransaction<'_>) -> error::Result<()> {
sqlx::query!("SELECT nextval('http_trigger_version_seq')",)
.fetch_one(&mut *tx)
.await?;
tx.commit().await?;
Ok(())
}
async fn create_trigger(
authed: ApiAuthed,
Extension(db): Extension<DB>,
@@ -382,7 +395,7 @@ async fn create_trigger(
)
.await?;
tx.commit().await?;
increase_trigger_version_and_commit(tx).await?;
Ok((StatusCode::CREATED, format!("{}", ct.path)))
}
@@ -538,7 +551,7 @@ async fn update_trigger(
)
.await?;
tx.commit().await?;
increase_trigger_version_and_commit(tx).await?;
Ok(path.to_string())
}
@@ -572,7 +585,7 @@ async fn delete_trigger(
)
.await?;
tx.commit().await?;
increase_trigger_version_and_commit(tx).await?;
Ok(format!("HTTP trigger {path} deleted"))
}
@@ -685,8 +698,8 @@ async fn exists_route(
Ok(Json(exists))
}
#[derive(Debug, Deserialize)]
struct TriggerRoute {
#[derive(Debug, Deserialize, Clone)]
pub struct TriggerRoute {
path: String,
script_path: String,
is_flow: bool,
@@ -704,6 +717,145 @@ struct TriggerRoute {
raw_string: bool,
}
pub struct RoutersCache {
routers: HashMap<HttpMethod, matchit::Router<TriggerRoute>>,
version: i64,
}
lazy_static::lazy_static! {
static ref HTTP_ACCESS_CACHE: Cache<(String, String, ApiAuthed), ExpiringCacheEntry<()>> = Cache::new(100);
static ref HTTP_AUTH_CACHE: Cache<(String, String, ApiAuthed), ExpiringCacheEntry<crate::http_trigger_auth::AuthenticationMethod>> = Cache::new(100);
static ref HTTP_ROUTERS_CACHE: RwLock<RoutersCache> = RwLock::new(RoutersCache {
routers: HashMap::new(),
version: 0,
});
}
pub async fn refresh_routers_loop(
db: &DB,
mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> () {
match refresh_routers(db).await {
Ok(_) => {
tracing::info!("Loaded HTTP routers");
}
Err(err) => {
tracing::error!("Error loading HTTP routers: {err:#}");
}
};
let db = db.clone();
tokio::spawn(async move {
loop {
tokio::select! {
_ = killpill_rx.recv() => {
break;
}
_ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {
match refresh_routers(&db).await {
Ok((true, _)) => {
tracing::info!("Refreshed HTTP routers");
}
Err(err) => {
tracing::error!("Error refreshing HTTP routers: {err:#}");
}
_ => {}
}
}
}
}
});
}
pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, RoutersCache>), Error> {
let version = sqlx::query_scalar!("SELECT last_value FROM http_trigger_version_seq",)
.fetch_one(db)
.await?;
let routers_cache = HTTP_ROUTERS_CACHE.read().await;
if routers_cache.version == 0 || version > routers_cache.version {
drop(routers_cache);
let mut routers = HashMap::new();
for http_method in [
HttpMethod::Get,
HttpMethod::Post,
HttpMethod::Put,
HttpMethod::Patch,
HttpMethod::Delete,
] {
let triggers = sqlx::query_as!(
TriggerRoute,
r#"
SELECT
path,
script_path,
is_flow,
route_path,
authentication_resource_path,
workspace_id,
is_async,
authentication_method AS "authentication_method: _",
edited_by,
email,
static_asset_config AS "static_asset_config: _",
wrap_body,
raw_string,
workspaced_route,
is_static_website
FROM
http_trigger
WHERE
http_method = $1
"#,
&http_method as &HttpMethod
)
.fetch_all(db)
.await?;
let mut router = matchit::Router::new();
for trigger in triggers {
let full_path = if trigger.workspaced_route || *CLOUD_HOSTED {
format!("/{}/{}", trigger.workspace_id, trigger.route_path)
} else {
format!("/{}", trigger.route_path)
};
if trigger.is_static_website {
router
.insert(format!("{}/*wm_subpath", full_path), trigger.clone())
.unwrap_or_else(|e| {
tracing::warn!(
"Failed to consider http trigger route {}/*wm_subpath: {:?}",
full_path,
e,
);
});
}
router
.insert(full_path.clone(), trigger.clone())
.unwrap_or_else(|e| {
tracing::warn!(
"Failed to consider http trigger route {}: {:?}",
full_path,
e,
);
});
}
routers.insert(http_method, router);
}
let mut routers_cache = HTTP_ROUTERS_CACHE.write().await;
*routers_cache = RoutersCache { routers, version };
Ok((true, routers_cache.downgrade()))
} else {
tracing::debug!("No HTTP routers refresh needed");
Ok((false, routers_cache))
}
}
async fn get_http_route_trigger(
route_path: &str,
auth_cache: &Arc<AuthCache>,
@@ -713,111 +865,30 @@ async fn get_http_route_trigger(
method: &http::Method,
) -> error::Result<(TriggerRoute, String, HashMap<String, String>, ApiAuthed)> {
let http_method: HttpMethod = method.try_into()?;
let (mut triggers, route_path) = if *CLOUD_HOSTED {
let mut splitted = route_path.split("/");
let w_id = splitted.next().ok_or_else(|| {
error::Error::BadRequest("Missing workspace id in route path".to_string())
})?;
let route_path = StripPath(splitted.collect::<Vec<_>>().join("/"));
let triggers = sqlx::query_as!(
TriggerRoute,
r#"
SELECT
path,
script_path,
is_flow,
route_path,
workspace_id,
is_async,
authentication_method AS "authentication_method: _",
edited_by,
email,
static_asset_config AS "static_asset_config: _",
wrap_body,
raw_string,
workspaced_route,
is_static_website,
authentication_resource_path
FROM
http_trigger
WHERE
workspace_id = $1 AND
http_method = $2
"#,
w_id,
http_method as HttpMethod
)
.fetch_all(db)
.await?;
(triggers, route_path)
let requested_path = format!("/{}", route_path);
let routers_cache = HTTP_ROUTERS_CACHE.read().await;
let routers_cache = if routers_cache.routers.is_empty() {
tracing::warn!("HTTP routers are not loaded, loading from db");
let (_, routers_cache) = refresh_routers(db).await?;
routers_cache
} else {
let triggers = sqlx::query_as!(
TriggerRoute,
r#"
SELECT
path,
script_path,
is_flow,
route_path,
authentication_resource_path,
workspace_id,
is_async,
authentication_method AS "authentication_method: _",
edited_by,
email,
static_asset_config AS "static_asset_config: _",
wrap_body,
raw_string,
workspaced_route,
is_static_website
FROM
http_trigger
WHERE
http_method = $1
"#,
http_method as HttpMethod
)
.fetch_all(db)
.await?;
(triggers, StripPath(route_path.to_string()))
routers_cache
};
let mut router = matchit::Router::new();
let router = routers_cache
.routers
.get(&http_method)
.ok_or(error::Error::internal_err(
"HTTP routers could not be loaded".to_string(),
))?;
for (idx, trigger) in triggers.iter().enumerate() {
let route_path = match trigger.workspaced_route {
true => format!("{}/{}", &trigger.workspace_id, &trigger.route_path),
_ => trigger.route_path.clone(),
};
if trigger.is_static_website {
router
.insert(format!("/{}/*wm_subpath", route_path), idx)
.unwrap_or_else(|e| {
tracing::warn!(
"Failed to consider http trigger route {}: {:?}",
route_path,
e,
);
});
}
router
.insert(format!("/{}", route_path), idx)
.unwrap_or_else(|e| {
tracing::warn!(
"Failed to consider http trigger route {}: {:?}",
route_path,
e,
);
});
}
let trigger_match = router.at(requested_path.as_str()).ok();
let requested_path = format!("/{}", route_path.0);
let trigger_idx = router.at(requested_path.as_str()).ok();
let matchit::Match { value: trigger_idx, params } =
not_found_if_none(trigger_idx, "Trigger", requested_path.as_str())?;
let trigger = triggers.remove(trigger_idx.to_owned());
let matchit::Match { value: trigger, params } =
not_found_if_none(trigger_match, "Trigger", requested_path.as_str())?;
let params: HashMap<String, String> = params
.iter()
@@ -834,25 +905,49 @@ async fn get_http_route_trigger(
};
if let Some(authed) = opt_authed {
// check that the user has access to the trigger
let mut tx = user_db.begin(&authed).await?;
let exists = sqlx::query_scalar!(
r#"
SELECT EXISTS(
SELECT 1
FROM
http_trigger
WHERE
workspace_id = $1 AND
path = $2
)
"#,
trigger.workspace_id,
trigger.path
)
.fetch_one(&mut *tx)
.await?
.unwrap_or(false);
tx.commit().await?;
let cache_key = (
trigger.workspace_id.clone(),
trigger.path.clone(),
authed.clone(),
);
let exists = match HTTP_ACCESS_CACHE.get(&cache_key) {
Some(cache_entry) if cache_entry.expiry > std::time::Instant::now() => {
tracing::debug!("HTTP access cache hit for trigger {}", trigger.path);
true
}
_ => {
tracing::debug!("HTTP access cache miss for trigger {}", trigger.path);
let mut tx = user_db.begin(&authed).await?;
let exists = sqlx::query_scalar!(
r#"
SELECT EXISTS(
SELECT 1
FROM
http_trigger
WHERE
workspace_id = $1 AND
path = $2
)
"#,
trigger.workspace_id,
trigger.path
)
.fetch_one(&mut *tx)
.await?
.unwrap_or(false);
if exists {
HTTP_ACCESS_CACHE.insert(
cache_key,
ExpiringCacheEntry {
value: (),
expiry: std::time::Instant::now()
+ std::time::Duration::from_secs(10),
},
);
}
exists
}
};
if exists {
Some(authed.display_username().to_owned())
} else {
@@ -876,7 +971,7 @@ async fn get_http_route_trigger(
)
.await?;
Ok((trigger, route_path.0, params, authed))
Ok((trigger.clone(), route_path.to_string(), params, authed))
}
async fn route_job(
@@ -901,7 +996,15 @@ async fn route_job(
.map_err(|e| e.into_response())?;
let args = args
.process_args(&authed, &db, &trigger.workspace_id, trigger.raw_string)
.process_args(
&authed,
&db,
&trigger.workspace_id,
match trigger.authentication_method {
AuthenticationMethod::CustomScript | AuthenticationMethod::Signature => true,
_ => trigger.raw_string,
},
)
.await
.map_err(|e| e.into_response())?;
@@ -920,31 +1023,45 @@ async fn route_job(
}
};
let authentication_method =
try_get_resource_from_db_as::<crate::http_trigger_auth::AuthenticationMethod>(
authed.clone(),
Some(user_db.clone()),
&db,
&resource_path,
&trigger.workspace_id,
)
.await
.map_err(|e| e.into_response())?;
let cache_key = (
trigger.workspace_id.clone(),
resource_path.clone(),
authed.clone(),
);
let raw_payload = args
.0
.metadata
.raw_string
.as_ref()
.map(|raw_payload| serde_json::from_str::<String>(raw_payload))
.transpose()
.map_err(|e| {
windmill_common::error::Error::SerdeJson { location: e.to_string(), error: e }
.into_response()
})?;
let authentication_method = match HTTP_AUTH_CACHE.get(&cache_key) {
Some(cache_entry) if cache_entry.expiry > std::time::Instant::now() => {
tracing::debug!("HTTP auth method cache hit for trigger {}", trigger.path);
cache_entry.value
}
_ => {
tracing::debug!("HTTP auth method cache miss for trigger {}", trigger.path);
let auth_method = try_get_resource_from_db_as::<
crate::http_trigger_auth::AuthenticationMethod,
>(
authed.clone(),
Some(user_db.clone()),
&db,
&resource_path,
&trigger.workspace_id,
)
.await
.map_err(|e| e.into_response())?;
HTTP_AUTH_CACHE.insert(
cache_key,
ExpiringCacheEntry {
value: auth_method.clone(),
expiry: std::time::Instant::now() + std::time::Duration::from_secs(60),
},
);
auth_method
}
};
let raw_payload = args.0.metadata.raw_string.as_ref();
let response = authentication_method
.authenticate_http_request(&headers, raw_payload.as_ref())
.authenticate_http_request(&headers, raw_payload)
.map_err(|e| e.into_response())?;
if let Some(response) = response {
+9 -3
View File
@@ -85,13 +85,14 @@ mod http_trigger_args;
#[cfg(feature = "http_trigger")]
mod http_trigger_auth;
#[cfg(feature = "http_trigger")]
mod http_triggers;
pub mod http_triggers;
mod indexer_ee;
mod inputs;
mod integration;
#[cfg(feature = "postgres_trigger")]
mod postgres_triggers;
mod approvals;
#[cfg(feature = "enterprise")]
mod apps_ee;
#[cfg(all(feature = "enterprise", feature = "gcp_trigger"))]
@@ -120,12 +121,11 @@ mod scripts;
mod service_logs;
mod settings;
mod slack_approvals;
mod approvals;
mod teams_approvals_ee;
#[cfg(feature = "smtp")]
mod smtp_server_ee;
#[cfg(all(feature = "enterprise", feature = "sqs_trigger"))]
mod sqs_triggers_ee;
mod teams_approvals_ee;
mod trigger_helpers;
mod static_assets;
@@ -406,6 +406,12 @@ pub async fn run_server(
Router::new()
};
#[cfg(feature = "http_trigger")]
{
let http_killpill_rx = killpill_rx.resubscribe();
http_triggers::refresh_routers_loop(&db, http_killpill_rx).await;
}
let postgres_triggers_service = {
#[cfg(feature = "postgres_trigger")]
{
+118 -63
View File
@@ -4,6 +4,7 @@ use serde_json::value::RawValue;
use std::collections::HashMap;
use windmill_common::{
error::Result,
flows::FlowModuleValue,
get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path,
scripts::{ScriptHash, ScriptLang},
worker::to_raw_value,
@@ -38,12 +39,6 @@ struct ScriptInfo {
schema: Option<sqlx::types::Json<PartialSchema>>,
}
struct FlowInfo {
has_preprocessor: Option<bool>,
is_v1_preprocessor: Option<bool>,
schema: Option<sqlx::types::Json<PartialSchema>>,
}
#[derive(Debug, Deserialize)]
struct PropertyDefinition {
r#type: Option<String>,
@@ -104,9 +99,8 @@ async fn get_script_info(
.await
}
fn runnable_format_from_schema(
fn runnable_format_from_schema_without_preprocessor(
trigger_kind: &TriggerKind,
has_preprocessor: bool,
schema: Option<sqlx::types::Json<PartialSchema>>,
) -> RunnableFormat {
match trigger_kind {
@@ -119,7 +113,7 @@ fn runnable_format_from_schema(
})
}) =>
{
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor }
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: false }
}
TriggerKind::Kafka | TriggerKind::Nats
if schema.as_ref().is_some_and(|schema| {
@@ -129,18 +123,46 @@ fn runnable_format_from_schema(
.is_some_and(|properties| properties.keys().any(|key| key == "msg"))
}) =>
{
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor }
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: false }
}
_ => RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor },
_ => RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor: false },
}
}
fn runnable_format_from_preprocessor_args(
args: Option<Vec<windmill_parser::Arg>>,
) -> RunnableFormat {
if let Some(args) = args {
if args.iter().any(|arg| arg.name == "wm_trigger")
|| (args.len() > 0 && args.iter().all(|arg| arg.name != "event"))
{
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: true }
} else {
RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor: true }
}
} else {
RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor: true }
}
}
enum PreprocessorInfo {
Preprocessor { content: String, language: ScriptLang },
NoPreprocessor { schema: Option<sqlx::types::Json<PartialSchema>> },
}
#[derive(Debug, Deserialize)]
struct FlowInfo {
preprocessor_module: Option<sqlx::types::Json<FlowModuleValue>>,
schema: Option<sqlx::types::Json<PartialSchema>>,
}
pub async fn get_runnable_format(
runnable_id: RunnableId,
workspace_id: &str,
db: &DB,
trigger_kind: &TriggerKind,
) -> Result<RunnableFormat> {
match runnable_id {
let (key, preprocessor_info) = match runnable_id {
RunnableId::FlowPath(path) => {
let FlowVersionInfo { version, .. } =
get_latest_flow_version_info_for_path(db, workspace_id, &path, true).await?;
@@ -157,11 +179,10 @@ pub async fn get_runnable_format(
let flow_info = sqlx::query_as!(
FlowInfo,
"SELECT
value->'preprocessor_module' IS NOT NULL as has_preprocessor,
value->'preprocessor_module'->'value'->'input_transforms'->'wm_trigger' IS NOT NULL as is_v1_preprocessor,
value->'preprocessor_module'->'value' as \"preprocessor_module: _\",
schema as \"schema: _\"
FROM flow
WHERE workspace_id = $1
WHERE workspace_id = $1
AND path = $2",
workspace_id,
path
@@ -169,18 +190,40 @@ pub async fn get_runnable_format(
.fetch_one(db)
.await?;
let has_preprocessor = flow_info.has_preprocessor.unwrap_or(false);
let is_v1_preprocessor = flow_info.is_v1_preprocessor.unwrap_or(false);
let runnable_format = if has_preprocessor && is_v1_preprocessor {
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: true }
if let Some(preprocessor_module) = flow_info.preprocessor_module {
match preprocessor_module.0 {
FlowModuleValue::RawScript { content, language, .. } => {
(key, PreprocessorInfo::Preprocessor { content, language })
}
FlowModuleValue::Script { path, hash, .. } => {
let hash = if let Some(hash) = hash {
hash.0
} else {
let script_hash =
get_latest_deployed_hash_for_path(db, workspace_id, &path).await?;
script_hash.hash
};
let script_info = get_script_info(db, workspace_id, hash).await?;
(
key,
PreprocessorInfo::Preprocessor {
content: script_info.content,
language: script_info.language,
},
)
}
_ => {
return Err(windmill_common::error::Error::internal_err(
"Unsupported preprocessor module".to_string(),
));
}
}
} else {
runnable_format_from_schema(trigger_kind, has_preprocessor, flow_info.schema)
};
RUNNABLE_FORMAT_VERSION_CACHE.insert(key, runnable_format);
Ok(runnable_format)
(
key,
PreprocessorInfo::NoPreprocessor { schema: flow_info.schema },
)
}
}
RunnableId::ScriptId(script_id) => {
let hash = script_id.get_script_hash(workspace_id, db).await?;
@@ -194,47 +237,59 @@ pub async fn get_runnable_format(
let script_info = get_script_info(db, workspace_id, hash).await?;
let has_preprocessor = script_info.has_preprocessor.unwrap_or(false);
let runnable_format = if has_preprocessor {
let args = match script_info.language {
ScriptLang::Bun
| ScriptLang::Bunnative
| ScriptLang::Deno
| ScriptLang::Nativets => {
let args = windmill_parser_ts::parse_deno_signature(
&script_info.content,
true,
false,
Some("preprocessor".to_string()),
)?;
Some(args.args)
}
ScriptLang::Python3 => {
let args = windmill_parser_py::parse_python_signature(
&script_info.content,
Some("preprocessor".to_string()),
false,
)?;
Some(args.args)
}
_ => None,
};
if args.is_some_and(|args| args.iter().any(|arg| arg.name == "wm_trigger")) {
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: true }
} else {
runnable_format_from_schema(trigger_kind, has_preprocessor, script_info.schema)
}
if script_info.has_preprocessor.unwrap_or(false) {
(
key,
PreprocessorInfo::Preprocessor {
content: script_info.content,
language: script_info.language,
},
)
} else {
runnable_format_from_schema(trigger_kind, has_preprocessor, script_info.schema)
(
key,
PreprocessorInfo::NoPreprocessor { schema: script_info.schema },
)
}
}
};
let runnable_format = match preprocessor_info {
PreprocessorInfo::Preprocessor { content, language } => {
let args = match language {
ScriptLang::Bun
| ScriptLang::Bunnative
| ScriptLang::Deno
| ScriptLang::Nativets => {
let args = windmill_parser_ts::parse_deno_signature(
&content,
true,
false,
Some("preprocessor".to_string()),
)?;
Some(args.args)
}
ScriptLang::Python3 => {
let args = windmill_parser_py::parse_python_signature(
&content,
Some("preprocessor".to_string()),
false,
)?;
Some(args.args)
}
_ => None,
};
RUNNABLE_FORMAT_VERSION_CACHE.insert(key, runnable_format);
Ok(runnable_format)
runnable_format_from_preprocessor_args(args)
}
}
PreprocessorInfo::NoPreprocessor { schema } => {
runnable_format_from_schema_without_preprocessor(trigger_kind, schema)
}
};
RUNNABLE_FORMAT_VERSION_CACHE.insert(key, runnable_format);
Ok(runnable_format)
}
#[allow(dead_code)]
+45 -13
View File
@@ -8,6 +8,8 @@
#![allow(non_snake_case)]
use quick_cache::sync::Cache;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use std::time::Duration;
@@ -20,7 +22,8 @@ use crate::utils::{
generate_instance_wide_unique_username, get_instance_username_or_create_pending,
};
use crate::{
db::DB, utils::require_super_admin, webhook_util::WebhookShared, COOKIE_DOMAIN, IS_SECURE,
auth::ExpiringAuthCache, db::DB, utils::require_super_admin, webhook_util::WebhookShared,
COOKIE_DOMAIN, IS_SECURE,
};
use argon2::{Argon2, PasswordHash, PasswordVerifier};
use axum::{
@@ -214,6 +217,10 @@ pub async fn fetch_api_authed(
fetch_api_authed_from_permissioned_as(permissioned_as, email, w_id, db, username_override).await
}
lazy_static::lazy_static! {
static ref API_AUTHED_CACHE: Cache<(String,String,String), ExpiringAuthCache> = Cache::new(300);
}
#[allow(unused)]
pub async fn fetch_api_authed_from_permissioned_as(
permissioned_as: String,
@@ -222,18 +229,43 @@ pub async fn fetch_api_authed_from_permissioned_as(
db: &DB,
username_override: Option<String>,
) -> error::Result<ApiAuthed> {
let authed =
fetch_authed_from_permissioned_as(permissioned_as, email.clone(), w_id, db).await?;
Ok(ApiAuthed {
username: authed.username,
email: email,
is_admin: authed.is_admin,
is_operator: authed.is_operator,
groups: authed.groups,
folders: authed.folders,
scopes: authed.scopes,
username_override: username_override,
})
let key = (w_id.to_string(), permissioned_as.clone(), email.clone());
let mut api_authed = match API_AUTHED_CACHE.get(&key) {
Some(expiring_authed) if expiring_authed.expiry > chrono::Utc::now() => {
tracing::debug!("API authed cache hit for user {}", email);
expiring_authed.authed
}
_ => {
tracing::debug!("API authed cache miss for user {}", email);
let authed =
fetch_authed_from_permissioned_as(permissioned_as, email.clone(), w_id, db).await?;
let api_authed = ApiAuthed {
username: authed.username,
email: email,
is_admin: authed.is_admin,
is_operator: authed.is_operator,
groups: authed.groups,
folders: authed.folders,
scopes: authed.scopes,
username_override: None,
};
API_AUTHED_CACHE.insert(
key,
ExpiringAuthCache {
authed: api_authed.clone(),
expiry: chrono::Utc::now() + chrono::Duration::try_seconds(120).unwrap(),
},
);
api_authed
}
};
api_authed.username_override = username_override;
Ok(api_authed)
}
#[derive(FromRow, Serialize)]
+7 -1
View File
@@ -5,7 +5,6 @@
* Please see the included NOTICE for copyright information and
* LICENSE-AGPL for a copy of the license.
*/
use axum::{body::Body, response::Response};
use regex::Regex;
use serde::{Deserialize, Deserializer};
@@ -415,3 +414,10 @@ pub async fn acknowledge_all_critical_alerts(
);
Ok("All unacknowledged critical alerts acknowledged".to_string())
}
#[cfg(feature = "http_trigger")]
#[derive(Clone)]
pub struct ExpiringCacheEntry<T> {
pub value: T,
pub expiry: std::time::Instant,
}
+32 -24
View File
@@ -140,6 +140,7 @@ pub struct CanceledBy {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobCompleted {
pub job: Arc<MiniPulledJob>,
pub preprocessed_args: Option<HashMap<String, Box<RawValue>>>,
pub result: Arc<Box<RawValue>>,
pub result_columns: Option<Vec<String>>,
pub mem_peak: i32,
@@ -2664,7 +2665,7 @@ async fn concurrency_key(db: &Pool<Postgres>, id: &Uuid) -> windmill_common::err
)
}
fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String {
pub fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String {
// Save this value to avoid parsing twice
let workspaced = x.as_str().replace("$workspace", workspace_id).to_string();
if RE_ARG_TAG.is_match(&workspaced) {
@@ -2702,7 +2703,6 @@ fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String {
.trim_matches('"')
.to_string()
};
tracing::error!("arg_value: {}", arg_value);
interpolated =
interpolated.replace(format!("$args[{}]", arg_name).as_str(), &arg_value);
}
@@ -3887,11 +3887,13 @@ pub async fn push<'c, 'd>(
let cache_ttl = value.cache_ttl.map(|x| x as i32);
let custom_concurrency_key = value.concurrency_key.clone();
let concurrency_time_window_s = value.concurrency_time_window_s;
let concurrent_limit = value.concurrent_limit;
let mut concurrent_limit = value.concurrent_limit;
if !apply_preprocessor {
value.preprocessor_module = None;
} else {
tag = None;
concurrent_limit = None;
preprocessed = Some(false);
}
@@ -4173,27 +4175,7 @@ pub async fn push<'c, 'd>(
};
if concurrent_limit.is_some() {
let concurrency_key = custom_concurrency_key
.map(|x| interpolate_args(x, &args, workspace_id))
.unwrap_or(fullpath_with_workspace(
workspace_id,
script_path.as_ref(),
&job_kind,
));
sqlx::query!(
"WITH inserted_concurrency_counter AS (
INSERT INTO concurrency_counter (concurrency_id, job_uuids)
VALUES ($1, '{}'::jsonb)
ON CONFLICT DO NOTHING
)
INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)",
concurrency_key,
job_id,
)
.execute(&mut *tx)
.warn_after_seconds(3)
.await
.map_err(|e| Error::internal_err(format!("Could not insert concurrency_key={concurrency_key} for job_id={job_id} script_path={script_path:?} workspace_id={workspace_id}: {e:#}")))?;
insert_concurrency_key(workspace_id, &args, &script_path, job_kind, custom_concurrency_key, &mut tx, job_id).await?;
}
let stringified_args = if *JOB_ARGS_AUDIT_LOGS {
@@ -4211,6 +4193,7 @@ pub async fn push<'c, 'd>(
Some("preprocessor") => Some(false),
_ => None,
});
let job_authed = match authed {
Some(authed)
@@ -4433,6 +4416,31 @@ pub async fn push<'c, 'd>(
Ok((job_id, tx))
}
pub async fn insert_concurrency_key<'d, 'c>(workspace_id: &str, args: &PushArgs<'d>, script_path: &Option<String>, job_kind: JobKind, custom_concurrency_key: Option<String>, tx: &mut Transaction<'c, Postgres>, job_id: Uuid) -> Result<(), Error> {
let concurrency_key = custom_concurrency_key
.map(|x| interpolate_args(x, args, workspace_id))
.unwrap_or(fullpath_with_workspace(
workspace_id,
script_path.as_ref(),
&job_kind,
));
sqlx::query!(
"WITH inserted_concurrency_counter AS (
INSERT INTO concurrency_counter (concurrency_id, job_uuids)
VALUES ($1, '{}'::jsonb)
ON CONFLICT DO NOTHING
)
INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)",
concurrency_key,
job_id,
)
.execute(&mut **tx)
.warn_after_seconds(3)
.await
.map_err(|e| Error::internal_err(format!("Could not insert concurrency_key={concurrency_key} for job_id={job_id} script_path={script_path:?} workspace_id={workspace_id}: {e:#}")))?;
Ok(())
}
pub fn canceled_job_to_result(job: &MiniPulledJob) -> serde_json::Value {
let reason = job
.canceled_reason
@@ -206,7 +206,7 @@ fn do_bigquery_inner<'a>(
convert_json_line_stream(rows_stream.boxed(), s3.format).await?;
s3.upload(stream.boxed()).await?;
return Ok(to_raw_value(&s3.object_key));
return Ok(to_raw_value(&s3.to_return_s3_obj()));
}
Ok(to_raw_value(&rows))
+9 -1
View File
@@ -1591,7 +1591,7 @@ pub struct S3ModeWorkerData {
}
impl S3ModeWorkerData {
pub async fn upload<S>(&self, stream: S) -> error::Result<reqwest::Response>
pub async fn upload<S>(&self, stream: S) -> error::Result<()>
where
S: futures::stream::TryStream + Send + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
@@ -1606,6 +1606,14 @@ impl S3ModeWorkerData {
)
.await
}
pub fn to_return_s3_obj(&self) -> windmill_common::s3_helpers::S3Object {
windmill_common::s3_helpers::S3Object {
s3: self.object_key.clone(),
storage: self.storage.clone(),
..Default::default()
}
}
}
pub fn s3_mode_args_to_worker_data(
@@ -187,14 +187,14 @@ pub async fn handle_dedicated_process(
let result = Arc::new(result);
append_logs(&job.id, &job.workspace_id, logs.clone(), &db.into()).await;
if line.starts_with("wm_res[success]:") {
job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string(), duration: None }, true).await.unwrap()
job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None }, true).await.unwrap()
} else {
job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None }, true).await.unwrap()
job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None }, true).await.unwrap()
}
},
Err(e) => {
tracing::error!("Could not deserialize job result `{line}`: {e:?}");
job_completed_tx.send_job(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))), result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None }, true).await.unwrap();
job_completed_tx.send_job(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))), result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None }, true).await.unwrap();
},
};
logs = init_log.clone();
@@ -214,7 +214,7 @@ pub async fn do_mssql(
let stream = convert_json_line_stream(rows_stream.boxed(), s3.format).await?;
s3.upload(stream.boxed()).await?;
Ok(serde_json::value::to_raw_value(&s3.object_key)?)
Ok(to_raw_value(&s3.to_return_s3_obj()))
} else {
let stream = prepared_query.query(&mut client).await.map_err(to_anyhow)?;
let results = stream.into_results().await.map_err(to_anyhow)?;
@@ -105,7 +105,7 @@ fn do_mysql_inner<'a>(
let stream = convert_json_line_stream(rows_stream.boxed(), s3.format).await?;
s3.upload(stream.boxed()).await?;
Ok(serde_json::value::to_raw_value(&s3.object_key)?)
Ok(to_raw_value(&s3.to_return_s3_obj()))
} else {
let rows: Vec<Row> = conn
.lock()
@@ -159,7 +159,7 @@ pub fn do_oracledb_inner<'a>(
if let Some(s3) = s3 {
let stream = convert_json_line_stream(rows_stream.boxed(), s3.format).await?;
s3.upload(stream.boxed()).await?;
return Ok(serde_json::value::to_raw_value(&s3.object_key)?);
return Ok(to_raw_value(&s3.to_return_s3_obj()));
} else {
let rows: Vec<_> = rows_stream.collect().await;
Ok(to_raw_value(
+1 -1
View File
@@ -124,7 +124,7 @@ fn do_postgresql_inner<'a>(
let stream = convert_json_line_stream(rows_stream.boxed(), s3.format).await?;
s3.upload(stream.boxed()).await?;
return Ok(serde_json::value::to_raw_value(&s3.object_key)?);
return Ok(to_raw_value(&s3.to_return_s3_obj()));
} else {
let rows = client
.query_raw(&query, query_params)
+42 -54
View File
@@ -29,7 +29,8 @@ use windmill_common::{
use windmill_common::bench::{BenchmarkInfo, BenchmarkIter};
use windmill_queue::{
append_logs, get_queued_job, CanceledBy, JobCompleted, MiniPulledJob, WrappedError,
append_logs, get_queued_job, CanceledBy, JobCompleted, MiniPulledJob,
WrappedError,
};
use serde_json::{json, value::RawValue};
@@ -274,27 +275,9 @@ pub fn start_background_processor(
async fn send_job_completed(
job_completed_tx: JobCompletedSender,
job: Arc<MiniPulledJob>,
result: Arc<Box<RawValue>>,
result_columns: Option<Vec<String>>,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
success: bool,
cached_res_path: Option<String>,
token: &str,
duration: Option<i64>,
jc: JobCompleted,
) {
let jc = JobCompleted {
job,
result,
result_columns,
mem_peak,
canceled_by,
success,
cached_res_path,
token: token.to_string(),
duration,
};
job_completed_tx
.send_job(jc, true)
.with_context(windmill_common::otel_ee::otel_ctx())
@@ -311,37 +294,28 @@ pub async fn process_result(
canceled_by: Option<CanceledBy>,
cached_res_path: Option<String>,
token: &str,
column_order: Option<Vec<String>>,
new_args: Option<HashMap<String, Box<RawValue>>>,
result_columns: Option<Vec<String>>,
preprocessed_args: Option<HashMap<String, Box<RawValue>>>,
conn: &Connection,
duration: Option<i64>,
) -> error::Result<bool> {
match result {
Ok(r) => {
// Update script args to preprocessed args
if let Connection::Sql(db) = conn {
if let Some(preprocessed_args) = new_args {
sqlx::query!(
"UPDATE v2_job SET args = $1, preprocessed = TRUE WHERE id = $2",
Json(preprocessed_args) as Json<HashMap<String, Box<RawValue>>>,
job.id
)
.execute(db)
.await?;
}
}
Ok(result) => {
send_job_completed(
job_completed_tx,
job,
r,
column_order,
mem_peak,
canceled_by,
true,
cached_res_path,
token,
duration,
JobCompleted {
job,
preprocessed_args,
result,
result_columns,
mem_peak,
canceled_by,
success: true,
cached_res_path,
token: token.to_string(),
duration,
},
)
.with_context(windmill_common::otel_ee::otel_ctx())
.await;
@@ -392,15 +366,18 @@ pub async fn process_result(
send_job_completed(
job_completed_tx,
job,
Arc::new(to_raw_value(&error_value)),
None,
mem_peak,
canceled_by,
false,
cached_res_path,
token,
duration,
JobCompleted {
job,
result: Arc::new(to_raw_value(&error_value)),
result_columns: None,
preprocessed_args: None,
mem_peak,
canceled_by,
success: false,
cached_res_path,
token: token.to_string(),
duration,
},
)
.with_context(windmill_common::otel_ee::otel_ctx())
.await;
@@ -476,6 +453,7 @@ pub async fn process_completed_job(
canceled_by,
duration,
result_columns,
preprocessed_args,
..
}: JobCompleted,
client: &AuthedClient,
@@ -500,6 +478,7 @@ pub async fn process_completed_job(
if job.flow_step_id.as_deref() == Some("preprocessor") {
// Do this before inserting to `v2_job_completed` for backwards compatibility
// when we set `flow_status->_metadata->preprocessed_args` to true.
sqlx::query!(
r#"UPDATE v2_job SET
args = '{"reason":"PREPROCESSOR_ARGS_ARE_DISCARDED"}'::jsonb,
@@ -514,6 +493,15 @@ pub async fn process_completed_job(
"error while deleting args of preprocessing step: {e:#}"
))
})?;
} else if let Some(preprocessed_args) = preprocessed_args {
// Update script args to preprocessed args
sqlx::query!(
"UPDATE v2_job SET args = $1, preprocessed = TRUE WHERE id = $2",
Json(preprocessed_args) as Json<HashMap<String, Box<RawValue>>>,
job.id
)
.execute(db)
.await?;
}
add_time!(bench, "pre add_completed_job");
@@ -256,7 +256,7 @@ fn do_snowflake_inner<'a>(
rows_stream.map(|r| serde_json::value::to_value(&r?).map_err(to_anyhow));
let stream = convert_json_line_stream(rows_stream.boxed(), s3.format).await?;
s3.upload(stream.boxed()).await?;
Ok(to_raw_value(&s3.object_key))
Ok(to_raw_value(&s3.to_return_s3_obj()))
} else {
let rows = rows_stream
.collect::<Vec<_>>()
+12 -3
View File
@@ -527,7 +527,7 @@ impl AuthedClient {
object_key: String,
storage: Option<String>,
body: S,
) -> error::Result<Response>
) -> error::Result<()>
where
S: futures::stream::TryStream + Send + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
@@ -537,7 +537,8 @@ impl AuthedClient {
if let Some(storage) = storage {
query.push(("storage", storage));
}
self.force_client
let response = self
.force_client
.as_ref()
.unwrap_or(&HTTP_CLIENT)
.post(format!(
@@ -558,7 +559,12 @@ impl AuthedClient {
.send()
.await
.context(format!("Sent upload_s3_file request",))
.map_err(error::Error::from)
.map_err(error::Error::from)?;
match response.status().as_u16() {
200u16 => Ok(()),
_ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default()))?,
}
}
}
@@ -1515,6 +1521,7 @@ pub async fn run_worker(
job_completed_tx
.send_job(
JobCompleted {
preprocessed_args: None,
job: Arc::new(job.job()),
success: true,
result: Arc::new(empty_result()),
@@ -1742,6 +1749,7 @@ pub async fn run_worker(
job_completed_tx
.send_job(
JobCompleted {
preprocessed_args: None,
job: arc_job.clone(),
result: Arc::new(
windmill_common::worker::to_raw_value(
@@ -2106,6 +2114,7 @@ async fn handle_queued_job(
job_completed_tx
.send_job(
JobCompleted {
preprocessed_args: None,
job,
result,
result_columns: None,
+132 -39
View File
@@ -13,6 +13,7 @@ use std::time::Duration;
use crate::common::{cached_result_path, save_in_cache};
use crate::js_eval::{eval_timeout, IdContext};
use crate::worker_utils::get_tag_and_concurrency;
use crate::{
AuthedClient, JobCompletedSender, PreviousResult, SameWorkerSender, SendResult, UpdateFlow,
KEEP_JOB_DIR,
@@ -59,8 +60,8 @@ use windmill_queue::flow_status::Step;
use windmill_queue::schedule::get_schedule_opt;
use windmill_queue::{
add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job,
handle_maybe_scheduled_job, CanceledBy, MiniPulledJob, PushArgs, PushIsolationLevel,
SameWorkerPayload, WrappedError,
handle_maybe_scheduled_job, insert_concurrency_key, interpolate_args, CanceledBy,
MiniPulledJob, PushArgs, PushIsolationLevel, SameWorkerPayload, WrappedError,
};
type DB = sqlx::Pool<sqlx::Postgres>;
@@ -162,6 +163,10 @@ pub async fn update_flow_status_after_job_completion(
add_time!(bench, "update flow status internal END");
return Ok(None);
}
UpdateFlowStatusAfterJobCompletion::PreprocessingStep => {
add_time!(bench, "update flow status preprocessing step END");
return Ok(None);
}
}
}
}
@@ -171,6 +176,7 @@ pub enum UpdateFlowStatusAfterJobCompletion {
Done(Arc<MiniPulledJob>),
NotDone,
NonLastParallelBranch,
PreprocessingStep,
}
pub struct RecUpdateFlowStatusAfterJobCompletion {
flow: uuid::Uuid,
@@ -426,41 +432,6 @@ pub async fn update_flow_status_after_job_completion_internal(
_ => false,
};
if matches!(module_step, Step::PreprocessorStep) {
sqlx::query!(
"WITH job_result AS (
SELECT result
FROM v2_job_completed
WHERE id = $1
)
UPDATE v2_job
SET args = COALESCE(
CASE
WHEN job_result.result IS NULL THEN NULL
WHEN jsonb_typeof(job_result.result) = 'object'
THEN job_result.result
WHEN jsonb_typeof(job_result.result) = 'null'
THEN NULL
ELSE jsonb_build_object('value', job_result.result)
END,
'{}'::jsonb
),
preprocessed = TRUE
FROM job_result
WHERE v2_job.id = $2;
",
job_id_for_status,
flow
)
.execute(db)
.await
.map_err(|e| {
Error::internal_err(format!(
"error while updating args in preprocessing step: {e:#}"
))
})?;
}
let mut tx = db.begin().await?;
add_time!(bench, "process module status START");
@@ -1010,6 +981,128 @@ pub async fn update_flow_status_after_job_completion_internal(
.ok_or_else(|| Error::internal_err(format!("requiring flow to be in the queue")))?;
tx.commit().await?;
if matches!(module_step, Step::PreprocessorStep) {
let tag_and_concurrency_key = get_tag_and_concurrency(&flow, db).await;
let require_args = tag_and_concurrency_key.as_ref().is_some_and(|x| {
x.tag.as_ref().is_some_and(|t| t.contains("$args"))
|| x.concurrency_key
.as_ref()
.is_some_and(|ck| ck.contains("$args"))
});
let mut tag = tag_and_concurrency_key
.as_ref()
.map(|x| x.tag.clone())
.flatten();
let concurrency_key = tag_and_concurrency_key
.as_ref()
.map(|x| x.concurrency_key.clone())
.flatten();
let concurrent_limit = tag_and_concurrency_key
.as_ref()
.map(|x| x.concurrent_limit)
.flatten();
let concurrency_time_window_s = tag_and_concurrency_key
.as_ref()
.map(|x| x.concurrency_time_window_s)
.flatten();
if require_args {
let args = sqlx::query_scalar!(
"SELECT result as \"result: Json<HashMap<String, Box<RawValue>>>\"
FROM v2_job_completed
WHERE id = $1",
job_id_for_status
)
.fetch_one(db)
.await
.map_err(|e| {
Error::internal_err(format!("error while fetching preprocessing args: {e:#}"))
})?;
let args_hm = args.unwrap_or_default().0;
let args = PushArgs::from(&args_hm);
if let Some(ck) = concurrency_key {
let mut tx = db.begin().await?;
insert_concurrency_key(
&flow_job.workspace_id,
&args,
&flow_job.runnable_path,
JobKind::Flow,
Some(ck),
&mut tx,
flow,
)
.await?;
tx.commit().await?;
}
if let Some(t) = tag {
tag = Some(interpolate_args(t, &args, &flow_job.workspace_id));
}
} else if let Some(ck) = concurrency_key {
let mut tx = db.begin().await?;
insert_concurrency_key(
&flow_job.workspace_id,
&PushArgs::from(&HashMap::new()),
&flow_job.runnable_path,
JobKind::Flow,
Some(ck),
&mut tx,
flow,
)
.await?;
tx.commit().await?;
}
// let tag = tag_and_concurrency_key.and_then(|tc| tc.tag.map(|t| interpolate_args(t.clone(), &args, &workspace_id)));
// let concurrency_key = tag_and_concurrency_key.and_then(|tc| tc.concurrency_key.map(|ck| interpolate_args(&ck, &args, &workspace_id)));
sqlx::query!(
"WITH job_result AS (
SELECT result
FROM v2_job_completed
WHERE id = $1
),
updated_queue AS (
UPDATE v2_job_queue
SET running = false,
tag = COALESCE($3, tag)
WHERE id = $2
)
UPDATE v2_job
SET
tag = COALESCE($3, tag),
concurrent_limit = COALESCE($4, concurrent_limit),
concurrency_time_window_s = COALESCE($5, concurrency_time_window_s),
args = COALESCE(
CASE
WHEN job_result.result IS NULL THEN NULL
WHEN jsonb_typeof(job_result.result) = 'object'
THEN job_result.result
WHEN jsonb_typeof(job_result.result) = 'null'
THEN NULL
ELSE jsonb_build_object('value', job_result.result)
END,
'{}'::jsonb
),
preprocessed = TRUE
FROM job_result
WHERE v2_job.id = $2;
",
job_id_for_status,
flow,
tag,
concurrent_limit,
concurrency_time_window_s,
)
.execute(db)
.await
.map_err(|e| {
Error::internal_err(format!(
"error while updating args in preprocessing step: {e:#}"
))
})?;
if success {
return Ok(UpdateFlowStatusAfterJobCompletion::PreprocessingStep);
}
}
let job_root = flow_job
.flow_innermost_root_job
.map(|x| x.to_string())
@@ -2649,8 +2742,8 @@ async fn push_next_flow_job(
};
tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}");
let tag = if flow_job.tag == "flow"
|| flow_job.tag == format!("flow-{}", flow_job.workspace_id)
let tag = if !matches!(step, Step::PreprocessorStep)
&& (flow_job.tag == "flow" || flow_job.tag == format!("flow-{}", flow_job.workspace_id))
{
payload_tag.tag.clone()
} else {
+70 -1
View File
@@ -3,13 +3,14 @@ use tracing::Instrument;
use uuid::Uuid;
use windmill_common::{
agent_workers::{PingJobStatus, PingJobStatusResponse},
cache,
worker::{
get_memory, get_vcpus, get_windmill_memory_usage, get_worker_memory_usage,
insert_ping_query, update_job_ping_query, update_worker_ping_from_job_query,
update_worker_ping_main_loop_query, Connection, Ping, PingType, WORKER_CONFIG,
WORKER_GROUP,
},
KillpillSender,
KillpillSender, DB,
};
use crate::{
@@ -320,3 +321,71 @@ pub(crate) async fn queue_vacuum(conn: &Connection, worker_name: &str, hostname:
}
}
}
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct TagAndConcurrencyKey {
pub tag: Option<String>,
pub concurrency_key: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub version: Option<i64>,
}
pub async fn get_tag_and_concurrency(job_id: &Uuid, db: &DB) -> Option<TagAndConcurrencyKey> {
let r = sqlx::query_as!(
TagAndConcurrencyKey,
"
WITH j AS (
SELECT
raw_flow->>'concurrency_key' as concurrency_key,
raw_flow->>'concurrency_time_window_s' as concurrency_time_window_s,
raw_flow->>'concurrency_limit' as concurrent_limit,
runnable_path,
runnable_id as version FROM v2_job
WHERE id = $1
)
SELECT tag, j.concurrency_key, j.concurrency_time_window_s::int, j.concurrent_limit::int, j.version
FROM flow, j
WHERE path = j.runnable_path
",
job_id
)
.fetch_optional(db)
.await
.ok()
.flatten();
if let Some(tag_and_concurrency_key) = r {
if tag_and_concurrency_key.concurrency_key.as_ref().is_some()
|| tag_and_concurrency_key.version.as_ref().is_none()
{
return Some(tag_and_concurrency_key);
} else {
let version = tag_and_concurrency_key.version.unwrap();
let r = cache::flow::fetch_version_lite(db, version).await;
let flow = match r {
Ok(data) => Ok(data),
Err(_) => cache::flow::fetch_version(db, version).await,
};
let flow_value = flow.map(|f| f.value().clone()).ok();
let concurrency_key = flow_value
.as_ref()
.map(|fv| fv.concurrency_key.clone())
.flatten();
let concurrent_limit = flow_value.as_ref().map(|fv| fv.concurrent_limit).flatten();
let concurrent_time_window_s = flow_value
.as_ref()
.map(|fv| fv.concurrency_time_window_s)
.flatten();
Some(TagAndConcurrencyKey {
tag: tag_and_concurrency_key.tag,
concurrency_key,
concurrent_limit,
concurrency_time_window_s: concurrent_time_window_s,
version: None,
})
}
} else {
None
}
}
+1 -1
View File
@@ -2,7 +2,7 @@ import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts";
import * as windmill from "https://deno.land/x/windmill@v1.174.0/mod.ts";
import * as api from "https://deno.land/x/windmill@v1.174.0/windmill-api/index.ts";
export const VERSION = "v1.491.1";
export const VERSION = "v1.491.5";
export async function login(email: string, password: string): Promise<string> {
return await windmill.UserService.login({
+26 -6
View File
@@ -63,7 +63,7 @@ export {
// }
// });
export const VERSION = "1.491.1";
export const VERSION = "1.491.5";
const command = new Command()
.name("wmill")
@@ -94,6 +94,7 @@ const command = new Command()
"Specify headers to use for all requests. e.g: \"HEADERS='h1: v1, h2: v2'\""
)
.version(VERSION)
.versionOption(false)
.command("init", "Bootstrap a windmill project with a wmill.yaml file")
.action(async () => {
if (await Deno.stat("wmill.yaml").catch(() => null)) {
@@ -134,15 +135,34 @@ const command = new Command()
.command("worker-groups", workerGroups)
.command("workers", workers)
.command("queues", queues)
.command("version", "Show version information")
.command("version --version", "Show version information")
.action(async (opts) => {
console.log("CLI build against " + VERSION);
console.log("CLI version: " + VERSION);
try {
const provider = new NpmProvider({ package: "windmill-cli" });
const versions = await provider.getVersions("windmill-cli");
if (versions.latest !== VERSION) {
console.log(
`CLI is outdated. Latest version ${versions.latest} is available. Run \`wmill upgrade\` to update.`
);
} else {
console.log("CLI is up to date");
}
} catch (e) {
console.warn(
`Cannot fetch latest CLI version on npmjs to check if up-to-date: ${e}`
);
}
const workspace = await getActiveWorkspace(opts as GlobalOptions);
if (workspace) {
const backendVersion = await fetchVersion(workspace.remote);
console.log("Backend Version: " + backendVersion);
try {
const backendVersion = await fetchVersion(workspace.remote);
console.log("Backend Version: " + backendVersion);
} catch (e) {
console.warn("Cannot fetch backend version: " + e);
}
} else {
console.log(
console.warn(
"Cannot fetch backend version: no active workspace selected, choose one to pick a remote to fetch version of"
);
}
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "windmill-components",
"version": "1.491.1",
"version": "1.491.5",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "windmill-components",
"version": "1.491.1",
"version": "1.491.5",
"hasInstallScript": true,
"license": "AGPL-3.0",
"dependencies": {
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "windmill-components",
"version": "1.491.1",
"version": "1.491.5",
"scripts": {
"dev": "vite dev",
"build": "vite build",
+2 -2
View File
@@ -69,7 +69,7 @@
? truncateContent.substring(
s3LogPrefixes[prefixIndex]?.length,
end == -1 ? undefined : end + 1
)
)
: undefined
}
@@ -109,7 +109,7 @@
? truncatedContent.substring(
truncatedContent.substring(1).indexOf('\n') + 2,
truncatedContent.length
)
)
: truncatedContent
)
export function scrollToBottom() {
@@ -2,7 +2,7 @@
import { onMount } from 'svelte'
import { Drawer, DrawerContent, Button } from './common'
import QueueMetricsDrawerInner from './QueueMetricsDrawerInner.svelte'
import { ConfigService } from '$lib/gen'
import { ConfigService, type Alert } from '$lib/gen'
import Section from './Section.svelte'
import { sendUserToast } from '$lib/toast'
import { Pencil, Trash, Check, PlusCircle, SaveIcon } from 'lucide-svelte'
@@ -21,14 +21,6 @@
}
}
type Alert = {
name: string
tags_to_monitor: string[]
jobs_num_threshold: number
alert_cooldown_seconds: number
alert_time_threshold_seconds: number
}
let drawer: Drawer
export function openDrawer() {
drawer?.openDrawer()
@@ -55,8 +47,8 @@
async function fetchConfig() {
try {
const response = (await ConfigService.getConfig({ name: configName })) as { alerts: Alert[] }
alerts = response.alerts || []
const response = await ConfigService.getConfig({ name: configName })
alerts = response?.alerts || []
originalAlerts = JSON.parse(JSON.stringify(alerts))
} catch (error) {
console.error('Failed to fetch config:', error)
@@ -298,7 +290,9 @@
<input
type="text"
bind:value={newTag}
placeholder="{workerTags.length === alert.tags_to_monitor.length ? 'All tags already added' : 'Add tag from dropdown' }"
placeholder={workerTags.length === alert.tags_to_monitor.length
? 'All tags already added'
: 'Add tag from dropdown'}
on:input={(e) => filterTags(e)}
disabled={workerTags.length === alert.tags_to_monitor.length}
class="p-1 flex-grow mr-1"
@@ -321,15 +315,15 @@
>
{#each filteredTags as tag}
{#if !alert.tags_to_monitor.includes(tag)}
<li>
<button
type="button"
class="w-full text-left p-2 cursor-pointer hover:bg-slate-200 dark:hover:bg-slate-700"
on:click={() => addTag(index, tag)}
>
{tag}
</button>
</li>
<li>
<button
type="button"
class="w-full text-left p-2 cursor-pointer hover:bg-slate-200 dark:hover:bg-slate-700"
on:click={() => addTag(index, tag)}
>
{tag}
</button>
</li>
{/if}
{/each}
</ul>
@@ -41,9 +41,9 @@
export let fromWorkspaceSettings: boolean = false
export let readOnlyMode: boolean
export let initialFileKey: { s3: string } | undefined = undefined
let initialFileKeyInternalCopy: { s3: string }
export let selectedFileKey: { s3: string } | undefined = undefined
export let initialFileKey: { s3: string; storage?: string } | undefined = undefined
let initialFileKeyInternalCopy: { s3: string; storage?: string }
export let selectedFileKey: { s3: string; storage?: string } | undefined = undefined
export let folderOnly = false
export let regexFilter: RegExp | undefined = undefined
@@ -298,7 +298,7 @@
deletionModalOpen = false
}
sendUserToast(`${fileKey} deleted from S3 bucket`)
selectedFileKey = { s3: '' }
selectedFileKey = { s3: '', storage }
const currentPage = page
await clearAndLoadFiles()
for (let i = 0; i < currentPage; i++) {
@@ -359,7 +359,7 @@
moveModalOpen = false
}
sendUserToast(`${srcFileKey} moved to ${destFileKey}`)
selectedFileKey = { s3: destFileKey! }
selectedFileKey = { s3: destFileKey!, storage }
await clearAndLoadFiles()
await loadFileMetadataPlusPreviewAsync(selectedFileKey.s3)
}
@@ -397,7 +397,7 @@
await clearAndLoadFiles()
if (selectedFileKey !== undefined) {
if (allFilesByKey[selectedFileKey.s3] === undefined) {
selectedFileKey = { s3: '' }
selectedFileKey = { s3: '', storage }
} else {
loadFileMetadataPlusPreviewAsync(selectedFileKey.s3)
}
@@ -421,7 +421,8 @@
if (item.type === 'folder') {
if (folderOnly) {
selectedFileKey = {
s3: item_key
s3: item_key,
storage
}
}
if (toggleCollapsed) {
@@ -456,7 +457,8 @@
displayedFileKeys = displayedFileKeys.sort()
} else {
selectedFileKey = {
s3: item_key
s3: item_key,
storage
}
loadFileMetadataPlusPreviewAsync(selectedFileKey.s3)
}
@@ -841,7 +843,7 @@
on:close={async (evt) => {
uploadModalOpen = false
if (evt.detail !== undefined && evt.detail !== null) {
selectedFileKey = { s3: evt.detail }
selectedFileKey = { s3: evt.detail, storage }
await clearAndLoadFiles()
loadFileMetadataPlusPreviewAsync(evt.detail)
}
@@ -199,24 +199,22 @@
<span class="mr-2 w-8 font-mono">{selected == res ? '-' : '+'}</span>
{res}
</button>
{#if selected == res}
<div class="border-t">
<SubGridEditor
{id}
visible={render && index === selectedIndex}
subGridId={`${id}-${index}`}
class={twMerge(css?.container?.class, 'wm-tabs-container')}
style={css?.container?.style}
containerHeight={componentContainerHeight - (titleBarHeight * tabs.length + 40)}
on:focus={() => {
if (!$connectingInput.opened) {
$selectedComponent = [id]
handleTabSelection()
}
}}
/>
</div>
{/if}
<div class={selected == res ? 'border-t' : ''}>
<SubGridEditor
{id}
visible={render && index === selectedIndex}
subGridId={`${id}-${index}`}
class={twMerge(css?.container?.class, 'wm-tabs-container')}
style={css?.container?.style}
containerHeight={componentContainerHeight - (titleBarHeight * tabs.length + 40)}
on:focus={() => {
if (!$connectingInput.opened) {
$selectedComponent = [id]
handleTabSelection()
}
}}
/>
</div>
</div>
{/each}
</div>
@@ -262,6 +262,7 @@
}
let selectedItem: any
async function handleKeydown(event: KeyboardEvent) {
if ((!isMac() ? event.ctrlKey : event.metaKey) && event.key === 'k') {
event.preventDefault()
@@ -641,11 +642,11 @@
<div class="w-4/12 overflow-y-auto max-h-[70vh]">
{#each itemMap['runs'] ?? [] as r}
<QuickMenuItem
on:hover={() => {
on:select={() => {
selectedItem = r
selectedWorkspace = r?.document.workspace_id[0]
}}
on:select={() => {
on:keyboardOnlySelect={() => {
open = false
goto(`/run/${r?.document.id[0]}`)
}}
@@ -657,10 +658,7 @@
>
<svelte:fragment slot="itemReplacement">
<div
class={twMerge(
`w-full flex flex-row items-center gap-4 transition-all`,
r?.document.id === selectedItem?.document?.id ? 'bg-surface-hover' : ''
)}
class="w-full flex flex-row items-center gap-4 transition-all"
>
<div
class="rounded-full w-2 h-2 {r?.document.success[0]
@@ -24,6 +24,7 @@
async function handleKeydown(event: KeyboardEvent) {
if (hovered && event.key === 'Enter') {
event.preventDefault()
dispatch('keyboardOnlySelect')
runAction()
}
}
@@ -263,7 +263,7 @@
btnClasses="ml-4 mt-2"
color="dark"
size="xs"
href={itemKind === 'flow' ? '/flows/add?hub=68' : '/scripts/add?hub=hub%2F11446'}
href={itemKind === 'flow' ? '/flows/add?hub=68' : '/scripts/add?hub=hub%2F19662'}
target="_blank">Create from template</Button
>
{/if}
@@ -268,6 +268,7 @@
bind:selectedFileKey={static_asset_config}
on:close={() => {
s3Editor?.setCode(JSON.stringify(static_asset_config, null, 2))
s3FileUploadRawMode = true
}}
readOnlyMode={false}
/>
@@ -366,6 +367,7 @@
disabled={!can_write}
/>
{/if}
{s3FileUploadRawMode}
{#if s3FileUploadRawMode}
{#if can_write}
<JsonEditor
@@ -396,7 +398,6 @@
s3: evt.detail?.path ?? '',
filename: evt.detail?.filename ?? undefined
}
s3FileUploadRawMode = true
}}
on:deletion={(evt) => {
static_asset_config = {
@@ -447,7 +448,7 @@
size="xs"
href={itemKind === 'flow'
? '/flows/add?hub=62'
: '/scripts/add?hub=hub%2F11627'}
: '/scripts/add?hub=hub%2F19669'}
target="_blank">Create from template</Button
>
{/if}
@@ -622,7 +623,9 @@
{/if}
{/if}
<RouteBodyTransformerOption bind:raw_string bind:wrap_body />
{#if !static_asset_config}
<RouteBodyTransformerOption bind:raw_string bind:wrap_body />
{/if}
</div>
</Section>
{/if}
@@ -2,7 +2,7 @@ import { base } from '$lib/base'
import { isCloudHosted } from '$lib/cloud'
export const SECRET_KEY_PATH = 'secret_key_path'
export const HUB_SCRIPT_ID = 19661
export const HUB_SCRIPT_ID = 19670
export const SIGNATURE_TEMPLATE_SCRIPT_HUB_PATH: string = `hub/${HUB_SCRIPT_ID}`
export const SIGNATURE_TEMPLATE_FLOW_HUB_ID = '67'
+91 -72
View File
@@ -660,86 +660,94 @@ export const TS_PREPROCESSOR_FLOW_INTRO = `/**
export const TS_PREPROCESSOR_MODULE_CODE = `export async function preprocessor(
event:
| {
kind: "webhook";
body: any,
raw_string: string | null,
query: Record<string, string>;
headers: Record<string, string>;
}
kind: "webhook";
body: any;
raw_string: string | null;
query: Record<string, string>;
headers: Record<string, string>;
}
| {
kind: "http";
body: any,
raw_string: string | null,
route: string;
path: string;
method: string;
params: Record<string, string>;
query: Record<string, string>;
headers: Record<string, string>;
}
kind: "http";
body: any;
raw_string: string | null;
route: string;
path: string;
method: string;
params: Record<string, string>;
query: Record<string, string>;
headers: Record<string, string>;
}
| {
kind: "email";
parsed_email: any,
raw_email: string,
}
kind: "email";
parsed_email: any;
raw_email: string;
}
| { kind: "websocket"; msg: string; url: string }
| {
kind: "kafka";
payload: string;
brokers: string[];
topic: string;
group_id: string;
}
kind: "kafka";
payload: string;
brokers: string[];
topic: string;
group_id: string;
}
| {
kind: "nats";
payload: string;
servers: string[];
subject: string;
headers?: Record<string, string[]>;
status?: number;
description?: string;
length: number;
}
kind: "nats";
payload: string;
servers: string[];
subject: string;
headers?: Record<string, string[]>;
status?: number;
description?: string;
length: number;
}
| {
kind: "sqs";
msg: string,
queue_url: string;
message_id?: string;
receipt_handle?: string;
attributes: Record<string, string>;
message_attributes?: Record<
string,
{ string_value?: string; data_type: string }
>;
}
kind: "sqs";
msg: string;
queue_url: string;
message_id?: string;
receipt_handle?: string;
attributes: Record<string, string>;
message_attributes?: Record<
string,
{ string_value?: string; data_type: string }
>;
}
| {
kind: "mqtt";
payload: string,
topic: string;
retain: boolean;
pkid: number;
qos: number;
v5?: {
payload_format_indicator?: number;
topic_alias?: number;
response_topic?: string;
correlation_data?: Array<number>;
user_properties?: Array<[string, string]>;
subscription_identifiers?: Array<number>;
content_type?: string;
};
}
kind: "mqtt";
payload: string;
topic: string;
retain: boolean;
pkid: number;
qos: number;
v5?: {
payload_format_indicator?: number;
topic_alias?: number;
response_topic?: string;
correlation_data?: Array<number>;
user_properties?: Array<[string, string]>;
subscription_identifiers?: Array<number>;
content_type?: string;
};
}
| {
kind: "gcp";
payload: string,
message_id: string;
subscription: string;
ordering_key?: string;
attributes?: Record<string, string>;
delivery_type: "push" | "pull";
headers?: Record<string, string>;
publish_time?: string;
}
kind: "gcp";
payload: string;
message_id: string;
subscription: string;
ordering_key?: string;
attributes?: Record<string, string>;
delivery_type: "push" | "pull";
headers?: Record<string, string>;
publish_time?: string;
}
| {
kind: "postgres";
transaction_type: "insert" | "update" | "delete",
schema_name: string,
table_name: string,
old_row?: Record<string, any>,
row: Record<string, any>
}
) {
return {
// return the args to be passed to the runnable
@@ -898,6 +906,16 @@ class GcpEvent(TypedDict):
headers: Optional[dict[str, str]]
publish_time: Optional[str]
class PostgresEvent(TypedDict):
kind: Literal["postgres"]
transaction_type: Literal["insert", "update", "delete"]
schema_name: str
table_name: str
old_row: Optional[dict[str, any]]
row: dict[str, any]
Event = Union[
WebhookEvent,
HttpEvent,
@@ -908,6 +926,7 @@ Event = Union[
SqsEvent,
MqttEvent,
GcpEvent,
PostgresEvent,
]
+2 -2
View File
@@ -4,8 +4,8 @@ verify_ssl = true
name = "pypi"
[packages]
wmill = ">=1.491.1"
wmill_pg = ">=1.491.1"
wmill = ">=1.491.5"
wmill_pg = ">=1.491.5"
sendgrid = "*"
mysql-connector-python = "*"
pymongo = "*"
+1 -1
View File
@@ -1,7 +1,7 @@
openapi: "3.0.3"
info:
version: 1.491.1
version: 1.491.5
title: OpenFlow Spec
contact:
name: Ruben Fiszel
@@ -12,7 +12,7 @@
RootModule = 'WindmillClient.psm1'
# Version number of this module.
ModuleVersion = '1.491.1'
ModuleVersion = '1.491.5'
# Supported PSEditions
# CompatiblePSEditions = @()
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "wmill"
version = "1.491.1"
version = "1.491.5"
description = "A client library for accessing Windmill server wrapping the Windmill client API"
license = "Apache-2.0"
homepage = "https://windmill.dev"
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "wmill-pg"
version = "1.491.1"
version = "1.491.5"
description = "An extension client for the wmill client library focused on pg"
license = "Apache-2.0"
homepage = "https://windmill.dev"
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@windmill/windmill",
"version": "1.491.1",
"version": "1.491.5",
"exports": "./src/index.ts",
"publish": {
"exclude": ["!src", "./s3Types.ts", "./client.ts"]
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "windmill-client",
"description": "Windmill SDK client for browsers and Node.js",
"version": "1.491.1",
"version": "1.491.5",
"author": "Ruben Fiszel",
"license": "Apache 2.0",
"devDependencies": {
+1 -1
View File
@@ -1 +1 @@
1.491.1
1.491.5