chore(main): release 1.446.0 (#5054)

* chore(main): release 1.446.0

* Apply automatic changes

---------

Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com>

.github/workflows/build-publish-rh-image.yml

@@ -64,7 +64,7 @@ jobs:
           platforms: linux/amd64
           push: true
           build-args: |
-            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core,license,http_trigger,zip,oauth2,kafka,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
+            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core,license,http_trigger,zip,oauth2,kafka,nats,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
           secrets: |
             rh_username=${{ secrets.RH_USERNAME }}
             rh_password=${{ secrets.RH_PASSWORD }}
@@ -81,7 +81,7 @@ jobs:
           platforms: linux/arm64
           push: true
           build-args: |
-            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core,license,http_trigger,zip,oauth2,kafka,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
+            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core,license,http_trigger,zip,oauth2,kafka,nats,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
           secrets: |
             rh_username=${{ secrets.RH_USERNAME }}
             rh_password=${{ secrets.RH_PASSWORD }}

.github/workflows/build_windows_worker_.yml

@@ -45,7 +45,7 @@ jobs:
           $env:OPENSSL_DIR="${Env:VCPKG_INSTALLATION_ROOT}\installed\x64-windows-static"
           mkdir frontend/build && cd backend
           New-Item -Path . -Name "windmill-api/openapi-deref.yaml" -ItemType "File" -Force
-          cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,license,http_trigger,zip,oauth2,kafka,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
+          cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,license,http_trigger,zip,oauth2,kafka,nats,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
 
       - name: Rename binary with corresponding architecture
         run: |

.github/workflows/docker-image.yml

@@ -149,7 +149,7 @@ jobs:
           platforms: linux/amd64,linux/arm64
           push: true
           build-args: |
-            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,license,http_trigger,zip,oauth2,kafka,otel,dind,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
+            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,license,http_trigger,zip,oauth2,kafka,nats,otel,dind,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
           tags: |
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee:${{ env.DEV_SHA }}
             ${{ steps.meta-ee-public.outputs.tags }}
@@ -211,7 +211,7 @@ jobs:
           platforms: linux/amd64
           push: true
           build-args: |
-            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,license,http_trigger,zip,oauth2,kafka,otel,dind,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
+            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,license,http_trigger,zip,oauth2,kafka,nats,otel,dind,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
             PYTHON_IMAGE=python:3.12.2-slim-bookworm
           tags: |
             ${{ steps.meta-ee-public-py312.outputs.tags }}

.github/workflows/publish_windows_worker.yml

@@ -47,7 +47,7 @@ jobs:
           $env:OPENSSL_DIR="${Env:VCPKG_INSTALLATION_ROOT}\installed\x64-windows-static"
           mkdir frontend/build && cd backend
           New-Item -Path . -Name "windmill-api/openapi-deref.yaml" -ItemType "File" -Force
-          cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,license,http_trigger,zip,oauth2,kafka,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
+          cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,license,http_trigger,zip,oauth2,kafka,nats,php,mysql,mssql,bigquery,websocket,python,smtp,csharp,static_frontend,rust
 
       - name: Rename binary with corresponding architecture
         run: |

CHANGELOG.md

@@ -1,5 +1,73 @@
 # Changelog
 
+## [1.446.0](https://github.com/windmill-labs/windmill/compare/v1.445.1...v1.446.0) (2025-01-14)
+
+
+### Features
+
+* support gpg signing commits with git sync ([#5053](https://github.com/windmill-labs/windmill/issues/5053)) ([010928b](https://github.com/windmill-labs/windmill/commit/010928b37f776839dc088d3b570bcfbb30a3b347))
+
+
+### Bug Fixes
+
+* **ui:** fix section height taking h-full ([c717a91](https://github.com/windmill-labs/windmill/commit/c717a915d6d48fa1ec697c24a4c85c04f4b70d84))
+
+## [1.445.1](https://github.com/windmill-labs/windmill/compare/v1.445.0...v1.445.1) (2025-01-13)
+
+
+### Bug Fixes
+
+* **ui:** capture v2 UX ([#4954](https://github.com/windmill-labs/windmill/issues/4954)) ([ebfde19](https://github.com/windmill-labs/windmill/commit/ebfde197fc11d243bbc9ba9b36d4f8a5e72d1f71))
+
+## [1.445.0](https://github.com/windmill-labs/windmill/compare/v1.444.0...v1.445.0) (2025-01-10)
+
+
+### Features
+
+* **frontend:** allow workspace admin to set workspace color ([#5032](https://github.com/windmill-labs/windmill/issues/5032)) ([0c39137](https://github.com/windmill-labs/windmill/commit/0c391372cca96da820e8438a3f685f9895dbff73))
+* nats triggers ([#5039](https://github.com/windmill-labs/windmill/issues/5039)) ([e66fcf9](https://github.com/windmill-labs/windmill/commit/e66fcf927097cb56d90d9f95c6d1f8ef10f45ff7))
+
+
+### Bug Fixes
+
+* **backend:** multiple routes with same path but different methods ([#5040](https://github.com/windmill-labs/windmill/issues/5040)) ([7b808c3](https://github.com/windmill-labs/windmill/commit/7b808c39771efb7c6b440b2a15427b4b1a7700ab))
+* support html in github markdown plugin ([#5031](https://github.com/windmill-labs/windmill/issues/5031)) ([53c62f2](https://github.com/windmill-labs/windmill/commit/53c62f2dbc4d43f02860fc9606a57996228be37c))
+
+## [1.444.0](https://github.com/windmill-labs/windmill/compare/v1.443.0...v1.444.0) (2025-01-06)
+
+
+### Features
+
+* captures ([#4807](https://github.com/windmill-labs/windmill/issues/4807)) ([01c6f84](https://github.com/windmill-labs/windmill/commit/01c6f845206b70a37163b8e67ff769cd5ad181ce))
+
+
+### Bug Fixes
+
+* **backend:** fix `has_failure_module` only looking for `raw_flow` ([#5018](https://github.com/windmill-labs/windmill/issues/5018)) ([abf63a6](https://github.com/windmill-labs/windmill/commit/abf63a6560817cfd392d3df7a74ba937255cb507))
+* recognize `forloop-<n>` in `use_flow_root_path` ([#5016](https://github.com/windmill-labs/windmill/issues/5016)) ([3eaef46](https://github.com/windmill-labs/windmill/commit/3eaef46a4fac92951e15a8e2177dbd25227c7250))
+
+## [1.443.0](https://github.com/windmill-labs/windmill/compare/v1.442.0...v1.443.0) (2025-01-04)
+
+
+### Features
+
+* allow s3 file download/preview from inside apps ([#5004](https://github.com/windmill-labs/windmill/issues/5004)) ([0c19171](https://github.com/windmill-labs/windmill/commit/0c19171f579cdd5d2753bd58dcc87b326cb6c09e))
+
+## [1.442.0](https://github.com/windmill-labs/windmill/compare/v1.441.2...v1.442.0) (2025-01-03)
+
+
+### Features
+
+* update git sync scripts with url redacted ([#5008](https://github.com/windmill-labs/windmill/issues/5008)) ([bbf7fd6](https://github.com/windmill-labs/windmill/commit/bbf7fd695b75403714eaa68764137a5fa560c92a))
+
+
+### Bug Fixes
+
+* **apps:** check for auth when executing as publisher ([#4979](https://github.com/windmill-labs/windmill/issues/4979)) ([f5c85d7](https://github.com/windmill-labs/windmill/commit/f5c85d7db994a269b63940a1445d1c2d659c3bc5))
+* **backend:** allow multiple files per field when using multipart ([#5002](https://github.com/windmill-labs/windmill/issues/5002)) ([383ecf8](https://github.com/windmill-labs/windmill/commit/383ecf846754d40ceaee256360a042d881984690))
+* handle `'flownode'` job kind where missing ([#4990](https://github.com/windmill-labs/windmill/issues/4990)) ([eeece84](https://github.com/windmill-labs/windmill/commit/eeece84a4b2f09cf50f7a04fac71912953f6e7f1))
+* **python:** Cancel and Start again within 1s caused module not found [v2] ([#5007](https://github.com/windmill-labs/windmill/issues/5007)) ([c998d2c](https://github.com/windmill-labs/windmill/commit/c998d2c8c50cc8ec38fb771bd55743fa3e651bcd))
+
 ## [1.441.2](https://github.com/windmill-labs/windmill/compare/v1.441.1...v1.441.2) (2024-12-27)
 
 

Dockerfile

@@ -159,7 +159,7 @@ ENV PATH="${PATH}:/usr/local/go/bin"
 ENV GO_PATH=/usr/local/go/bin/go
 
 # Install UV
-RUN curl --proto '=https' --tlsv1.2 -LsSf https://github.com/astral-sh/uv/releases/download/0.4.18/uv-installer.sh | sh && mv /root/.cargo/bin/uv /usr/local/bin/uv
+RUN curl --proto '=https' --tlsv1.2 -LsSf https://github.com/astral-sh/uv/releases/download/0.5.15/uv-installer.sh | sh && mv /root/.local/bin/uv /usr/local/bin/uv
 
 RUN curl -sL https://deb.nodesource.com/setup_20.x | bash - 
 RUN apt-get -y update && apt-get install -y curl procps nodejs awscli && apt-get clean \
@@ -196,6 +196,21 @@ COPY ./frontend/src/lib/hubPaths.json ${APP}/hubPaths.json
 
 RUN windmill cache ${APP}/hubPaths.json && rm ${APP}/hubPaths.json && chmod -R 777 /tmp/windmill
 
+# Create a non-root user 'windmill' with UID and GID 1000
+RUN addgroup --gid 1000 windmill && \
+    adduser --disabled-password --gecos "" --uid 1000 --gid 1000 windmill
+
+RUN cp -r /root/.cache /home/windmill/.cache
+
+RUN mkdir -p /tmp/windmill/logs && \
+    mkdir -p /tmp/windmill/search
+
+RUN chown -R windmill:windmill ${APP} && \
+     chown -R windmill:windmill /tmp/windmill && \
+     chown -R windmill:windmill /home/windmill/.cache
+
+USER root
+
 EXPOSE 8000
 
 CMD ["windmill"]

backend/.sqlx/query-031d0d70b0aff52feaad487bddb74e5ef0aaa2505facbea8c764003dfc8fffb1.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'websocket' AND server_id = $5 AND last_client_ping > NOW() - INTERVAL '10 seconds' RETURNING 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Bool",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "031d0d70b0aff52feaad487bddb74e5ef0aaa2505facbea8c764003dfc8fffb1"
+}

backend/.sqlx/query-07da723ce5c9ee2d7c236e8eabe254c783fc34b617c8a9a95a0eb0cda535dab5.json

@@ -0,0 +1,34 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, payload, trigger_extra, created_by)\n        VALUES ($1, $2, $3, $4, $5, $6, $7)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Bool",
+        {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats"
+              ]
+            }
+          }
+        },
+        "Jsonb",
+        "Jsonb",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "07da723ce5c9ee2d7c236e8eabe254c783fc34b617c8a9a95a0eb0cda535dab5"
+}

backend/.sqlx/query-0a9a191273c735c41d56ea46a39ffca075a0550eada87df7162c5037164ad6bf.json

@@ -1,16 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n       INSERT INTO capture\n                   (workspace_id, path, created_by)\n            VALUES ($1, $2, $3)\n       ON CONFLICT (workspace_id, path)\n     DO UPDATE SET created_at = now()\n        ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        "Varchar"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "0a9a191273c735c41d56ea46a39ffca075a0550eada87df7162c5037164ad6bf"
-}

backend/.sqlx/query-0dbd664c906ee3c65856520c22f0828357b36bbcdcc151bd97605d7d7a0489e8.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM capture_config WHERE path = $1 AND workspace_id = $2 AND is_flow IS TRUE",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "0dbd664c906ee3c65856520c22f0828357b36bbcdcc151bd97605d7d7a0489e8"
+}

backend/.sqlx/query-1730f39fd1793d45fbb41b21389c61296a3ff7489ae12f52a19f9543173ac597.json

@@ -122,6 +122,11 @@
         "ordinal": 23,
         "name": "mute_critical_alerts",
         "type_info": "Bool"
+      },
+      {
+        "ordinal": 24,
+        "name": "color",
+        "type_info": "Varchar"
       }
     ],
     "parameters": {
@@ -153,6 +158,7 @@
       false,
       true,
       true,
+      true,
       true
     ]
   },

backend/.sqlx/query-1c0f95a069891f7214505aaa9428e11e25c1d3a30294cbe78f01ed75edf68da4.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "UPDATE kafka_trigger SET kafka_resource_path = $1, group_id = $2, topics = $3, script_path = $4, path = $5, is_flow = $6, edited_by = $7, email = $8, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL\n            WHERE workspace_id = $9 AND path = $10",
+  "query": "UPDATE kafka_trigger SET kafka_resource_path = $1, group_id = $2, topics = $3, script_path = $4, path = $5, is_flow = $6, edited_by = $7, email = $8, edited_at = now(), server_id = NULL, error = NULL\n            WHERE workspace_id = $9 AND path = $10",
   "describe": {
     "columns": [],
     "parameters": {
@@ -19,5 +19,5 @@
     },
     "nullable": []
   },
-  "hash": "5962733746c81480abe9ab3a6ccf5664130ae7500279055c08db42d67b2822f6"
+  "hash": "1c0f95a069891f7214505aaa9428e11e25c1d3a30294cbe78f01ed75edf68da4"
 }

backend/.sqlx/query-1ef48cc430870ab6c062b046bdbf0a2db057141619e9818ddb904d7c242efec3.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT \n            EXISTS(SELECT 1 FROM websocket_trigger WHERE workspace_id = $1) as \"websocket_used!\", \n            EXISTS(SELECT 1 FROM http_trigger WHERE workspace_id = $1) as \"http_routes_used!\",\n            EXISTS(SELECT 1 FROM kafka_trigger WHERE workspace_id = $1) as \"kafka_used!\"",
+  "query": "SELECT \n            EXISTS(SELECT 1 FROM websocket_trigger WHERE workspace_id = $1) as \"websocket_used!\", \n            EXISTS(SELECT 1 FROM http_trigger WHERE workspace_id = $1) as \"http_routes_used!\",\n            EXISTS(SELECT 1 FROM kafka_trigger WHERE workspace_id = $1) as \"kafka_used!\",\n            EXISTS(SELECT 1 FROM nats_trigger WHERE workspace_id = $1) as \"nats_used!\"",
   "describe": {
     "columns": [
       {
@@ -17,6 +17,11 @@
         "ordinal": 2,
         "name": "kafka_used!",
         "type_info": "Bool"
+      },
+      {
+        "ordinal": 3,
+        "name": "nats_used!",
+        "type_info": "Bool"
       }
     ],
     "parameters": {
@@ -25,10 +30,11 @@
       ]
     },
     "nullable": [
+      null,
       null,
       null,
       null
     ]
   },
-  "hash": "cc8a71abdeccf0787695af32aa21fe73aba65cfdc8329c02b4d257de4d2d168a"
+  "hash": "1ef48cc430870ab6c062b046bdbf0a2db057141619e9818ddb904d7c242efec3"
 }

backend/.sqlx/query-203fa78d423ec5a8c5ff6166aed591b28cbf9ea8f61d379b84ee6e14c033035d.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET error = $1, server_id = NULL, last_server_ping = NULL WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'kafka'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Bool"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "203fa78d423ec5a8c5ff6166aed591b28cbf9ea8f61d379b84ee6e14c033035d"
+}

backend/.sqlx/query-25975935d59f88df117bae0fa5016cf95ad56f790f0eddb6e2fbe3d83d3accaa.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "UPDATE websocket_trigger SET enabled = $1, email = $2, edited_by = $3, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL\n        WHERE path = $4 AND workspace_id = $5 RETURNING 1",
+  "query": "UPDATE websocket_trigger SET enabled = $1, email = $2, edited_by = $3, edited_at = now(), server_id = NULL, error = NULL\n        WHERE path = $4 AND workspace_id = $5 RETURNING 1",
   "describe": {
     "columns": [
       {
@@ -22,5 +22,5 @@
       null
     ]
   },
-  "hash": "12a86755706ce030a0a9142da78d035d0c7d361240b60005cc864e4345eb0bc7"
+  "hash": "25975935d59f88df117bae0fa5016cf95ad56f790f0eddb6e2fbe3d83d3accaa"
 }

backend/.sqlx/query-28a389a93a3d2472b13d956ec55eb357c6147e186e00e371f7374d166903ef64.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\", static_asset_config as \"static_asset_config: _\" FROM http_trigger",
+  "query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, static_asset_config as \"static_asset_config: _\" FROM http_trigger WHERE http_method = $1",
   "describe": {
     "columns": [
       {
@@ -50,8 +50,13 @@
       },
       {
         "ordinal": 9,
-        "name": "http_method: _",
-        "type_info": {
+        "name": "static_asset_config: _",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        {
           "Custom": {
             "name": "http_method",
             "kind": {
@@ -65,15 +70,7 @@
             }
           }
         }
-      },
-      {
-        "ordinal": 10,
-        "name": "static_asset_config: _",
-        "type_info": "Jsonb"
-      }
-    ],
-    "parameters": {
-      "Left": []
+      ]
     },
     "nullable": [
       false,
@@ -85,9 +82,8 @@
       false,
       false,
       false,
-      false,
       true
     ]
   },
-  "hash": "11b698f82a54aac68b3617047dfe2b18dd6da7d962118fee276af354218baac2"
+  "hash": "28a389a93a3d2472b13d956ec55eb357c6147e186e00e371f7374d166903ef64"
 }

backend/.sqlx/query-2f440ab6083764b49e309c1f8dde0c5508da110f54bc5f465f1892afdf851af0.json

@@ -0,0 +1,77 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, created_at, trigger_kind as \"trigger_kind: _\", payload as \"payload: _\", trigger_extra as \"trigger_extra: _\"\n        FROM capture\n        WHERE workspace_id = $1\n            AND path = $2 AND is_flow = $3\n            AND ($4::trigger_kind IS NULL OR trigger_kind = $4)\n        ORDER BY created_at DESC",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 2,
+        "name": "trigger_kind: _",
+        "type_info": {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 3,
+        "name": "payload: _",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 4,
+        "name": "trigger_extra: _",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Bool",
+        {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats"
+              ]
+            }
+          }
+        }
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "2f440ab6083764b49e309c1f8dde0c5508da110f54bc5f465f1892afdf851af0"
+}

backend/.sqlx/query-3651ed42be75d41ab0387f1551012d72c90235bb40d2f66e6fb235c990d78352.json

@@ -0,0 +1,52 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace.id, workspace.name, workspace.owner, workspace.deleted, workspace.premium, workspace_settings.color\n         FROM workspace\n         LEFT JOIN workspace_settings ON workspace.id = workspace_settings.workspace_id\n         JOIN usr ON usr.workspace_id = workspace.id\n         WHERE usr.email = $1 AND workspace.deleted = false",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "name",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "owner",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "deleted",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 4,
+        "name": "premium",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 5,
+        "name": "color",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "3651ed42be75d41ab0387f1551012d72c90235bb40d2f66e6fb235c990d78352"
+}

backend/.sqlx/query-3b6ddfe4df620d5e34a01dbbc95067d7dee6a7108cb8a46a20f26236009481da.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM capture_config WHERE path = $1 AND workspace_id = $2 AND is_flow IS FALSE",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "3b6ddfe4df620d5e34a01dbbc95067d7dee6a7108cb8a46a20f26236009481da"
+}

backend/.sqlx/query-3c9fc4d8579767f3ce7c3633fca770e6341624e98117d21b9f01e68b4e0ce033.json

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

backend/.sqlx/query-3d8fdf7adf42b27808a5b223df3a660bfd5a8c0a451a0deac24632f0ccf66165.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE nats_trigger SET enabled = $1, email = $2, edited_by = $3, edited_at = now(), server_id = NULL, error = NULL\n        WHERE path = $4 AND workspace_id = $5 RETURNING 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "3d8fdf7adf42b27808a5b223df3a660bfd5a8c0a451a0deac24632f0ccf66165"
+}

backend/.sqlx/query-41e557e1b63b13c9fcc195901c0bd0de7e03c539ee046955543d9693551246f7.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM capture WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "41e557e1b63b13c9fcc195901c0bd0de7e03c539ee046955543d9693551246f7"
+}

backend/.sqlx/query-42030372693d8d6b8d03947bb6702024cfa226735eeeb720bbab46bd0e3cedcc.json

@@ -0,0 +1,50 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT path, is_flow, workspace_id, trigger_config as \"trigger_config!: _\", owner, email FROM capture_config WHERE trigger_kind = 'nats' AND last_client_ping > NOW() - INTERVAL '10 seconds' AND trigger_config IS NOT NULL AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "is_flow",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 2,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "trigger_config!: _",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 4,
+        "name": "owner",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "email",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true,
+      false,
+      false
+    ]
+  },
+  "hash": "42030372693d8d6b8d03947bb6702024cfa226735eeeb720bbab46bd0e3cedcc"
+}

backend/.sqlx/query-47560aa3d3af93167663e763ad873088eb4f57785df3c383bbf6e5ceab268980.json

@@ -0,0 +1,50 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT path, is_flow, workspace_id, trigger_config as \"trigger_config!: _\", owner, email FROM capture_config WHERE trigger_kind = 'websocket' AND last_client_ping > NOW() - INTERVAL '10 seconds' AND trigger_config IS NOT NULL AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "is_flow",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 2,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "trigger_config!: _",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 4,
+        "name": "owner",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "email",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true,
+      false,
+      false
+    ]
+  },
+  "hash": "47560aa3d3af93167663e763ad873088eb4f57785df3c383bbf6e5ceab268980"
+}

backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json

@@ -122,6 +122,11 @@
         "ordinal": 23,
         "name": "mute_critical_alerts",
         "type_info": "Bool"
+      },
+      {
+        "ordinal": 24,
+        "name": "color",
+        "type_info": "Varchar"
       }
     ],
     "parameters": {
@@ -153,6 +158,7 @@
       false,
       true,
       true,
+      true,
       true
     ]
   },

backend/.sqlx/query-561b7935d687f6b9f3d6488f8489f55e6737ffa2d3716f803d32f8b68cc1e915.json

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

backend/.sqlx/query-6c845f168b6265e6cf92e4d33c5409edfdc1847d0348df2ea55504ddaaa67736.json

@@ -0,0 +1,40 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace.id, workspace.name, usr.username, workspace_settings.color\n         FROM workspace\n         JOIN usr ON usr.workspace_id = workspace.id\n         JOIN workspace_settings ON workspace_settings.workspace_id = workspace.id\n         WHERE usr.email = $1 AND workspace.deleted = false",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "name",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "username",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "color",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "6c845f168b6265e6cf92e4d33c5409edfdc1847d0348df2ea55504ddaaa67736"
+}

backend/.sqlx/query-71d51bbc35da7b9930e3ea3a634451217ccb9f1bc35b1ad6e10d16bc19c41447.json

@@ -0,0 +1,45 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT owner, email\n        FROM capture_config\n        WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4 AND last_client_ping > NOW() - INTERVAL '10 seconds'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "owner",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "email",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Bool",
+        {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats"
+              ]
+            }
+          }
+        }
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "71d51bbc35da7b9930e3ea3a634451217ccb9f1bc35b1ad6e10d16bc19c41447"
+}

backend/.sqlx/query-746ee16a04267cd251d4cefbc44a37fcc985bf61978cc07167ed35d7dbf92d11.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT 1 FROM nats_trigger WHERE path = $1 AND workspace_id = $2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "746ee16a04267cd251d4cefbc44a37fcc985bf61978cc07167ed35d7dbf92d11"
+}

backend/.sqlx/query-78b182167ba19d1dc103577bd3492736e7088ed69b543adbcfd3b3b7a4ece430.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS FALSE",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "78b182167ba19d1dc103577bd3492736e7088ed69b543adbcfd3b3b7a4ece430"
+}

backend/.sqlx/query-7b36b58761cd459b808cc4f463dfc93f01fdbc19c66d7404f95e1f8444235a8e.json

@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE nats_trigger SET nats_resource_path = $1, subjects = $2, stream_name = $3, consumer_name = $4, use_jetstream = $5, script_path = $6, path = $7, is_flow = $8, edited_by = $9, email = $10, edited_at = now(), server_id = NULL, error = NULL\n            WHERE workspace_id = $11 AND path = $12",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "VarcharArray",
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "7b36b58761cd459b808cc4f463dfc93f01fdbc19c66d7404f95e1f8444235a8e"
+}

backend/.sqlx/query-7f717130a398c8d52f814a968c2b0bc4dbb9cd654307f5167d8dbe794f17a1cf.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT success FROM completed_job WHERE id = $1 AND workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "success",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "7f717130a398c8d52f814a968c2b0bc4dbb9cd654307f5167d8dbe794f17a1cf"
+}

backend/.sqlx/query-870b12a46f26c9e29889dd28a5c7832c5a674a5553674fd9762c981e0d03bb57.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM nats_trigger WHERE workspace_id = $1 AND path = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "870b12a46f26c9e29889dd28a5c7832c5a674a5553674fd9762c981e0d03bb57"
+}

backend/.sqlx/query-87e0bee8b78facc62d4c225dc51651d65b1135092067061e7f4050776b733b03.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM capture WHERE path = $1 AND workspace_id = $2 AND is_flow IS TRUE",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "87e0bee8b78facc62d4c225dc51651d65b1135092067061e7f4050776b733b03"
+}

backend/.sqlx/query-90ad280f6744937878b134142825ccb7d7071a80a3f5da59da8948c0106e959c.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS TRUE",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "90ad280f6744937878b134142825ccb7d7071a80a3f5da59da8948c0106e959c"
+}

backend/.sqlx/query-90e74c28b417e73924ed337f6634f574024e641e73072586aca88847259e8cfc.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS FALSE",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "90e74c28b417e73924ed337f6634f574024e641e73072586aca88847259e8cfc"
+}

backend/.sqlx/query-92e60af0d3ae8c73d74ae68d70e20ae18f79cc84626097b766307cd42722baa3.json

@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE websocket_trigger SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "92e60af0d3ae8c73d74ae68d70e20ae18f79cc84626097b766307cd42722baa3"
+}

backend/.sqlx/query-97942578df746c8c8103b403cfc4e44ef5a0f082bdde854900064325adc4dd77.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM capture\n            WHERE workspace_id = $1\n                AND created_at <=\n                    (\n                        SELECT created_at\n                            FROM capture\n                            WHERE workspace_id = $1\n                        ORDER BY created_at DESC\n                            OFFSET $2\n                            LIMIT 1\n                    )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "97942578df746c8c8103b403cfc4e44ef5a0f082bdde854900064325adc4dd77"
+}

backend/.sqlx/query-988b594f70ff7886985a0f90a095aeffac05016abc4dfed37d5ae2872e8da564.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS TRUE",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "988b594f70ff7886985a0f90a095aeffac05016abc4dfed37d5ae2872e8da564"
+}

backend/.sqlx/query-99a2c935acf5d6bbeb70ea1255679115b6e9042800d40899a36b3867049c5c46.json

@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE last_client_ping > NOW() - INTERVAL '10 seconds' AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'websocket' AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text",
+        "Bool"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "99a2c935acf5d6bbeb70ea1255679115b6e9042800d40899a36b3867049c5c46"
+}

backend/.sqlx/query-a17b1b8d4f58c58c253e63ec4c2fbb1df4bef54003b01fa901bfa782b5f83342.json

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

backend/.sqlx/query-a76be7f4e9e8b8c81afe50fcbf1f3d393ed0bb73314e5c430f32541c43bcac52.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\", static_asset_config as \"static_asset_config: _\" FROM http_trigger WHERE workspace_id = $1",
+  "query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, static_asset_config as \"static_asset_config: _\" FROM http_trigger WHERE workspace_id = $1 AND http_method = $2",
   "describe": {
     "columns": [
       {
@@ -50,8 +50,14 @@
       },
       {
         "ordinal": 9,
-        "name": "http_method: _",
-        "type_info": {
+        "name": "static_asset_config: _",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        {
           "Custom": {
             "name": "http_method",
             "kind": {
@@ -65,16 +71,6 @@
             }
           }
         }
-      },
-      {
-        "ordinal": 10,
-        "name": "static_asset_config: _",
-        "type_info": "Jsonb"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text"
       ]
     },
     "nullable": [
@@ -87,9 +83,8 @@
       false,
       false,
       false,
-      false,
       true
     ]
   },
-  "hash": "c9930fcfe79541af570eace58ba7e15a0816a6b4fd036cf7b991a210654b2633"
+  "hash": "a76be7f4e9e8b8c81afe50fcbf1f3d393ed0bb73314e5c430f32541c43bcac52"
 }

backend/.sqlx/query-ad07b57ad928a4e5260833f4fe2a4bf51211f2f73d78c546c894c8e147677ac7.json

@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COUNT(*) FROM nats_trigger WHERE script_path = $1 AND is_flow = $2 AND workspace_id = $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Bool",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "ad07b57ad928a4e5260833f4fe2a4bf51211f2f73d78c546c894c8e147677ac7"
+}

backend/.sqlx/query-add48c8e7c6fa2c549ad6293cbee22889d35e919d3267c1d2a265d868fa8a7d1.json

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

backend/.sqlx/query-b9468b9e16f55db11b33d8e9793e6e3ae6c5add6ca02414140adb724120a6800.json

@@ -1,16 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n       UPDATE capture\n          SET payload = $3\n        WHERE workspace_id = $1\n          AND path = $2\n        ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text",
-        "Jsonb"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "b9468b9e16f55db11b33d8e9793e6e3ae6c5add6ca02414140adb724120a6800"
-}

backend/.sqlx/query-c223f8b7fa4ef1aa06e1ba2a56d677774aa237508d5610714efd2e9b8b93c7b8.json

@@ -0,0 +1,56 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT trigger_config as \"trigger_config: _\", trigger_kind as \"trigger_kind: _\", error, last_server_ping\n        FROM capture_config\n        WHERE workspace_id = $1 AND path = $2 AND is_flow = $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "trigger_config: _",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 1,
+        "name": "trigger_kind: _",
+        "type_info": {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 2,
+        "name": "error",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "last_server_ping",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Bool"
+      ]
+    },
+    "nullable": [
+      true,
+      false,
+      true,
+      true
+    ]
+  },
+  "hash": "c223f8b7fa4ef1aa06e1ba2a56d677774aa237508d5610714efd2e9b8b93c7b8"
+}

backend/.sqlx/query-c5270ee815689e42b65df507b850da43239c9a5aaea41c9aed7ed33a6219a534.json

@@ -0,0 +1,31 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET last_client_ping = now() WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Bool",
+        {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats"
+              ]
+            }
+          }
+        }
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "c5270ee815689e42b65df507b850da43239c9a5aaea41c9aed7ed33a6219a534"
+}

backend/.sqlx/query-c73dd6226a154a54debe99f8ae64fd75f8d7ce551c5955aa6bdd8dab2dcbd7ff.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT 1 FROM app WHERE path = $1 AND workspace_id = $2 LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "c73dd6226a154a54debe99f8ae64fd75f8d7ce551c5955aa6bdd8dab2dcbd7ff"
+}

backend/.sqlx/query-cd5d62d456b74237b941bc72ea8de007185263167bf1dfa2e469d319cc4da674.json

@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE nats_trigger SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "cd5d62d456b74237b941bc72ea8de007185263167bf1dfa2e469d319cc4da674"
+}

backend/.sqlx/query-d06efdc24706e0d7479bffc0b19a0c5976ee60125289e9f7b0b04090bce4a3a3.json

@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE last_client_ping > NOW() - INTERVAL '10 seconds'  AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'kafka' AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text",
+        "Bool"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "d06efdc24706e0d7479bffc0b19a0c5976ee60125289e9f7b0b04090bce4a3a3"
+}

backend/.sqlx/query-d08f34000c3d96ccd0f44ca8520f966d751a4dda554d8215eedb8f65be98e100.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET error = $1, server_id = NULL, last_server_ping = NULL WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'websocket'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Bool"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d08f34000c3d96ccd0f44ca8520f966d751a4dda554d8215eedb8f65be98e100"
+}

backend/.sqlx/query-d2668c2d4ece82a3617f098fe993d3e218fa224bd0bb311e7db776c7fd69cf0b.json

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

backend/.sqlx/query-d6615719bf8db4b333ed55c9a3c160ad1962668fc3305d8e80ae91ef73614a80.json

@@ -1,24 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE kafka_trigger SET server_id = $1, last_server_ping = now() WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "?column?",
-        "type_info": "Bool"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "d6615719bf8db4b333ed55c9a3c160ad1962668fc3305d8e80ae91ef73614a80"
-}

backend/.sqlx/query-d78ecf85c1e1e95650c380c9488aa90d8d8c5c76f3971484e4c515ff60293d3b.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "UPDATE websocket_trigger SET server_id = $1, last_server_ping = now() WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
+  "query": "UPDATE kafka_trigger SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
   "describe": {
     "columns": [
       {
@@ -20,5 +20,5 @@
       null
     ]
   },
-  "hash": "02424907504848e983bfa89eec343061932dc5b4b17cf13d5cf8d833aedbe6d5"
+  "hash": "d78ecf85c1e1e95650c380c9488aa90d8d8c5c76f3971484e4c515ff60293d3b"
 }

backend/.sqlx/query-d8ea17fba0e417333e9c6cf82ad36a57830a791c8358e30e9be012b815f5e8e3.json

@@ -0,0 +1,50 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT path, is_flow, workspace_id, trigger_config as \"trigger_config!: _\", owner, email FROM capture_config WHERE trigger_kind = 'kafka' AND last_client_ping > NOW() - INTERVAL '10 seconds' AND trigger_config IS NOT NULL AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "is_flow",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 2,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "trigger_config!: _",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 4,
+        "name": "owner",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "email",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true,
+      false,
+      false
+    ]
+  },
+  "hash": "d8ea17fba0e417333e9c6cf82ad36a57830a791c8358e30e9be012b815f5e8e3"
+}

backend/.sqlx/query-d9a6f75e4c4a1f61e55b313cc09bceffac637548841897341672da427a9140fc.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'kafka' AND server_id = $5 AND last_client_ping > NOW() - INTERVAL '10 seconds' RETURNING 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Bool",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "d9a6f75e4c4a1f61e55b313cc09bceffac637548841897341672da427a9140fc"
+}

backend/.sqlx/query-da9114fc6689ebc78422b3572de1ed44050bd28212540db50fe235463c15a900.json

@@ -0,0 +1,91 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT\n            queue.job_kind AS \"job_kind: JobKind\",\n            queue.script_hash AS \"script_hash: ScriptHash\",\n            queue.raw_flow AS \"raw_flow: sqlx::types::Json<Box<RawValue>>\",\n            completed_job.parent_job AS \"parent_job: Uuid\",\n            completed_job.created_at AS \"created_at: chrono::NaiveDateTime\",\n            completed_job.created_by AS \"created_by!\",\n            queue.script_path,\n            queue.args AS \"args: sqlx::types::Json<Box<RawValue>>\"\n        FROM queue\n        JOIN completed_job ON completed_job.parent_job = queue.id\n        WHERE completed_job.id = $1 AND completed_job.workspace_id = $2\n        LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "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": 1,
+        "name": "script_hash: ScriptHash",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 2,
+        "name": "raw_flow: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 3,
+        "name": "parent_job: Uuid",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 4,
+        "name": "created_at: chrono::NaiveDateTime",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 5,
+        "name": "created_by!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 6,
+        "name": "script_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 7,
+        "name": "args: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      true,
+      true,
+      true,
+      false,
+      false,
+      true,
+      true
+    ]
+  },
+  "hash": "da9114fc6689ebc78422b3572de1ed44050bd28212540db50fe235463c15a900"
+}

backend/.sqlx/query-dcf03ff4b922b93be37d2be5da6884ae3e8c6cf7eaa2d9c62056366eb42f2276.json

@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE nats_trigger SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND server_id = $4 AND enabled IS TRUE RETURNING 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "dcf03ff4b922b93be37d2be5da6884ae3e8c6cf7eaa2d9c62056366eb42f2276"
+}

backend/.sqlx/query-de0735f0f5b59ebb72fd876f7f46156e26b352511f408fd6464303301383461f.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM capture WHERE path = $1 AND workspace_id = $2 AND is_flow IS FALSE",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "de0735f0f5b59ebb72fd876f7f46156e26b352511f408fd6464303301383461f"
+}

backend/.sqlx/query-df0454f75e819d7d3d03fef0a7d5606940a089f6179e5443090afa9aba7c5b24.json

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

backend/.sqlx/query-e02b99525cb1f8737acfec86809f59c8cff67bb1ec3926680fb691cc3573738a.json

@@ -1,16 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n   DELETE FROM capture\n         WHERE workspace_id = $1\n           AND created_by = $2\n           AND created_at <=\n              ( SELECT created_at\n                  FROM capture\n                 WHERE workspace_id = $1\n                   AND created_by = $2\n              ORDER BY created_at DESC\n                OFFSET $3\n                 LIMIT 1 )\n        ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text",
-        "Int8"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "e02b99525cb1f8737acfec86809f59c8cff67bb1ec3926680fb691cc3573738a"
-}

backend/.sqlx/query-e08dddf6af2656b561c453460ff928e7a158bd29d9035bee136f9952fb96bad4.json

@@ -0,0 +1,77 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, created_at, trigger_kind as \"trigger_kind: _\", payload as \"payload: _\", trigger_extra as \"trigger_extra: _\"\n        FROM capture\n        WHERE workspace_id = $1\n            AND path = $2 AND is_flow = $3\n            AND ($4::trigger_kind IS NULL OR trigger_kind = $4)\n        ORDER BY created_at DESC\n        OFFSET $5\n        LIMIT $6",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 2,
+        "name": "trigger_kind: _",
+        "type_info": {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 3,
+        "name": "payload: _",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 4,
+        "name": "trigger_extra: _",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Bool",
+        {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email"
+              ]
+            }
+          }
+        },
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "e08dddf6af2656b561c453460ff928e7a158bd29d9035bee136f9952fb96bad4"
+}

backend/.sqlx/query-e23e110e1f0438d21534fc4323e0e7bc1f0dbeca2e4f44ced05bae0ca5ca1039.json

@@ -0,0 +1,51 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT trigger_config as \"trigger_config: _\", owner, email\n        FROM capture_config\n        WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4 AND last_client_ping > NOW() - INTERVAL '10 seconds'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "trigger_config: _",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 1,
+        "name": "owner",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "email",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Bool",
+        {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats"
+              ]
+            }
+          }
+        }
+      ]
+    },
+    "nullable": [
+      true,
+      false,
+      false
+    ]
+  },
+  "hash": "e23e110e1f0438d21534fc4323e0e7bc1f0dbeca2e4f44ced05bae0ca5ca1039"
+}

backend/.sqlx/query-e37ba13aa3174931f0bfcff26dbc141fe8a346ab6a1a3bc924794bfe3c9af306.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'nats' AND server_id = $5 AND last_client_ping > NOW() - INTERVAL '10 seconds' RETURNING 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Bool",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "e37ba13aa3174931f0bfcff26dbc141fe8a346ab6a1a3bc924794bfe3c9af306"
+}

backend/.sqlx/query-e4e87539ae18f7e5c6bd9a28d16b7527ececad87d218182ff723e6a6c43ecd50.json

@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT \n                                        flow_status->>'step' = '0' \n                                        AND (\n                                            jsonb_array_length(flow_status->'modules') = 0 \n                                            OR flow_status->'modules'->0->>'type' = 'WaitingForPriorSteps' \n                                            OR (\n                                                flow_status->'modules'->0->>'type' = 'Failure' \n                                                AND flow_status->'modules'->0->>'job' = $1\n                                            )\n                                        )\n                                    FROM completed_job WHERE id = $2 AND workspace_id = $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "e4e87539ae18f7e5c6bd9a28d16b7527ececad87d218182ff723e6a6c43ecd50"
+}

backend/.sqlx/query-e54eb583f011cea6e4d533f4d2014cbf509b5afcc2647048c9bfc839d8f290ad.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO workspace_settings\n            (workspace_id, color)\n            VALUES ($1, $2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "e54eb583f011cea6e4d533f4d2014cbf509b5afcc2647048c9bfc839d8f290ad"
+}

backend/.sqlx/query-eda1e5d1109a13feb14be254eaca630be29176bd5d1246c974642d40201782fc.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "UPDATE kafka_trigger SET enabled = $1, email = $2, edited_by = $3, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL\n        WHERE path = $4 AND workspace_id = $5 RETURNING 1",
+  "query": "UPDATE kafka_trigger SET enabled = $1, email = $2, edited_by = $3, edited_at = now(), server_id = NULL, error = NULL\n        WHERE path = $4 AND workspace_id = $5 RETURNING 1",
   "describe": {
     "columns": [
       {
@@ -22,5 +22,5 @@
       null
     ]
   },
-  "hash": "6eb076afcf11dba0dfe35ee108c64814f95dbe5376f0539195d24a8ca96ca3e2"
+  "hash": "eda1e5d1109a13feb14be254eaca630be29176bd5d1246c974642d40201782fc"
 }

backend/.sqlx/query-eed16e356f3f36183c3db13fcc1950295e0d0fbdabb38434534fb3430eeddc25.json

@@ -0,0 +1,53 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace.id, workspace.name, workspace.owner, workspace.deleted, workspace.premium, workspace_settings.color\n         FROM workspace\n         LEFT JOIN workspace_settings ON workspace.id = workspace_settings.workspace_id\n         LIMIT $1 OFFSET $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "name",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "owner",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "deleted",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 4,
+        "name": "premium",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 5,
+        "name": "color",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "eed16e356f3f36183c3db13fcc1950295e0d0fbdabb38434534fb3430eeddc25"
+}

backend/.sqlx/query-ef299490c4674c4c76e18d84620a74407b78378d66d8a089407998074059e79b.json

@@ -0,0 +1,34 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO capture_config\n            (workspace_id, path, is_flow, trigger_kind, trigger_config, owner, email)\n        VALUES ($1, $2, $3, $4, $5, $6, $7)\n        ON CONFLICT (workspace_id, path, is_flow, trigger_kind)\n            DO UPDATE SET trigger_config = $5, owner = $6, email = $7, server_id = NULL, last_server_ping = NULL, error = NULL",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Bool",
+        {
+          "Custom": {
+            "name": "trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats"
+              ]
+            }
+          }
+        },
+        "Jsonb",
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "ef299490c4674c4c76e18d84620a74407b78378d66d8a089407998074059e79b"
+}

backend/.sqlx/query-f16b00bad2880f896f4452e5894fe101cd5c2fe3e4c143c0b979668143d85dd2.json

@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE last_client_ping > NOW() - INTERVAL '10 seconds'  AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'nats' AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text",
+        "Bool"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "f16b00bad2880f896f4452e5894fe101cd5c2fe3e4c143c0b979668143d85dd2"
+}

backend/.sqlx/query-ffb6b2a40b605f9c55826a0e519af80b100d2ffa58ec567223e27099d364fcc8.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET error = $1, server_id = NULL, last_server_ping = NULL WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'nats'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Bool"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "ffb6b2a40b605f9c55826a0e519af80b100d2ffa58ec567223e27099d364fcc8"
+}

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "windmill"
-version = "1.441.2"
+version = "1.446.0"
 authors.workspace = true
 edition.workspace = true
 
@@ -30,7 +30,7 @@ members = [
 ]
 
 [workspace.package]
-version = "1.441.2"
+version = "1.446.0"
 authors = ["Ruben Fiszel <ruben@windmill.dev>"]
 edition = "2021"
 
@@ -63,6 +63,7 @@ tantivy = ["dep:windmill-indexer", "windmill-api/tantivy", "windmill-indexer/ent
 sqlx = ["windmill-worker/sqlx"]
 deno_core = ["windmill-worker/deno_core", "dep:deno_core"]
 kafka = ["windmill-api/kafka"]
+nats = ["windmill-api/nats"]
 otel = ["windmill-common/otel", "windmill-worker/otel"]
 dind = ["windmill-worker/dind"]
 php = ["windmill-worker/php"]
@@ -260,7 +261,7 @@ getrandom = "0.2"
 tokio-postgres = {version = "^0.7", features = ["array-impls", "with-serde_json-1", "with-chrono-0_4", "with-uuid-1", "with-bit-vec-0_6"]}
 bit-vec = "=0.6.3"
 mappable-rc = "^0"
-mysql_async = { version = "*", default-features = false, features = ["minimal", "default", "native-tls-tls"]}
+mysql_async = { version = "*", default-features = false, features = ["minimal", "default", "native-tls-tls", "rust_decimal"]}
 postgres-native-tls = "^0"
 native-tls = "^0"
 # samael will break compilation on MacOS. Use this fork instead to make it work
@@ -285,6 +286,8 @@ openssl = "=0.10"
 mail-parser = "^0"
 matchit = "=0.7.3"
 rdkafka = { version = "0.36.2", features = ["cmake-build", "ssl-vendored"] }
+async-nats = "0.38.0"
+nkeys = "0.4.4"
 
 datafusion = "39.0.0"
 object_store = { version = "0.10.0", features = ["aws", "azure"] }

backend/ee-repo-ref.txt

@@ -1 +1 @@
-586b02014d57f862a5c4313dd1e529d50c315c30
+9cea7e44f1b6401662e22088b41e8a00d7e67b23

backend/migrations/20250102145420_more_captures.down.sql

@@ -0,0 +1,14 @@
+-- Add down migration script here
+DROP TABLE capture_config;
+DELETE FROM capture;
+DROP POLICY see_from_allowed_runnables ON capture;
+ALTER TABLE capture DROP CONSTRAINT capture_pkey;
+ALTER TABLE capture DROP COLUMN is_flow, DROP COLUMN trigger_kind, DROP COLUMN trigger_extra, DROP COLUMN id;
+ALTER TABLE capture ADD CONSTRAINT capture_pkey PRIMARY KEY (workspace_id, path);
+DROP TYPE TRIGGER_KIND;
+
+CREATE POLICY see_folder_extra_perms_user ON capture FOR ALL TO windmill_user
+USING (SPLIT_PART(capture.path, '/', 1) = 'f' AND SPLIT_PART(capture.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_read'), ',')::text[]))
+WITH CHECK (SPLIT_PART(capture.path, '/', 1) = 'f' AND SPLIT_PART(capture.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+CREATE POLICY see_member ON public.capture TO windmill_user USING (((split_part((path)::text, '/'::text, 1) = 'g'::text) AND (split_part((path)::text, '/'::text, 2) = ANY (regexp_split_to_array(current_setting('session.groups'::text), ','::text)))));
+CREATE POLICY see_own ON public.capture TO windmill_user USING (((split_part((path)::text, '/'::text, 1) = 'u'::text) AND (split_part((path)::text, '/'::text, 2) = current_setting('session.user'::text))));

backend/migrations/20250102145420_more_captures.up.sql

@@ -0,0 +1,65 @@
+-- Add up migration script here
+CREATE TYPE TRIGGER_KIND AS ENUM ('webhook', 'http', 'websocket', 'kafka', 'email');
+ALTER TABLE capture ADD COLUMN is_flow BOOLEAN NOT NULL DEFAULT TRUE, ADD COLUMN trigger_kind TRIGGER_KIND NOT NULL DEFAULT 'webhook', ADD COLUMN trigger_extra JSONB;
+ALTER TABLE capture ALTER COLUMN is_flow DROP DEFAULT, ALTER COLUMN trigger_kind DROP DEFAULT;
+ALTER TABLE capture DROP CONSTRAINT capture_pkey;
+ALTER TABLE capture ADD COLUMN id BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY;
+
+DROP POLICY see_own ON capture;
+DROP POLICY see_member ON capture;
+DROP POLICY see_folder_extra_perms_user ON capture;
+
+CREATE POLICY see_from_allowed_runnables ON capture FOR ALL TO windmill_user
+USING (
+    (capture.is_flow AND EXISTS (
+        SELECT 1
+        FROM flow
+        WHERE flow.workspace_id = capture.workspace_id
+        AND flow.path = capture.path
+    ))
+    OR (NOT capture.is_flow AND EXISTS (
+        SELECT 1
+        FROM script
+        WHERE script.workspace_id = capture.workspace_id
+        AND script.path = capture.path
+    ))
+);
+
+
+CREATE TABLE capture_config (
+    workspace_id VARCHAR(50) NOT NULL,
+    path VARCHAR(255) NOT NULL,
+    is_flow BOOLEAN NOT NULL,
+    trigger_kind TRIGGER_KIND NOT NULL,
+    trigger_config JSONB NULL,
+    owner VARCHAR(50) NOT NULL,
+    email VARCHAR(255) NOT NULL,
+    server_id VARCHAR(50) NULL,
+    last_client_ping TIMESTAMPTZ NULL,
+    last_server_ping TIMESTAMPTZ NULL,
+    error TEXT NULL,
+    PRIMARY KEY (workspace_id, path, is_flow, trigger_kind),
+    FOREIGN KEY (workspace_id) REFERENCES workspace(id) ON DELETE CASCADE
+);
+
+ALTER TABLE capture_config ENABLE ROW LEVEL SECURITY;
+
+CREATE POLICY see_from_allowed_runnables ON capture_config FOR ALL TO windmill_user
+USING (
+    (capture_config.is_flow AND EXISTS (
+        SELECT 1
+        FROM flow
+        WHERE flow.workspace_id = capture_config.workspace_id
+        AND flow.path = capture_config.path
+    ))
+    OR (NOT capture_config.is_flow AND EXISTS (
+        SELECT 1
+        FROM script
+        WHERE script.workspace_id = capture_config.workspace_id
+        AND script.path = capture_config.path
+    ))
+);
+
+
+GRANT ALL ON capture_config TO windmill_user;
+GRANT ALL ON capture_config TO windmill_admin;

backend/migrations/20250106164709_nats_triggers.down.sql

@@ -0,0 +1 @@
+DROP TABLE nats_trigger;

backend/migrations/20250106164709_nats_triggers.up.sql

@@ -0,0 +1,71 @@
+ALTER TYPE trigger_kind ADD VALUE IF NOT EXISTS 'nats';
+
+CREATE TABLE nats_trigger (
+    path VARCHAR(255) NOT NULL,
+    nats_resource_path VARCHAR(255) NOT NULL,
+    subjects VARCHAR(255)[] NOT NULL,
+    stream_name VARCHAR(255) NULL,
+    consumer_name VARCHAR(255) NULL,
+    use_jetstream BOOLEAN NOT NULL,
+    script_path VARCHAR(255) NOT NULL,
+    is_flow BOOLEAN NOT NULL,
+    workspace_id VARCHAR(50) NOT NULL,
+    edited_by VARCHAR(50) NOT NULL,
+    email VARCHAR(255) NOT NULL,
+    edited_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    extra_perms JSONB NOT NULL DEFAULT '{}',
+    server_id VARCHAR(50) NULL,
+    last_server_ping TIMESTAMPTZ NULL,
+    error TEXT NULL,
+    enabled BOOLEAN NOT NULL,
+    PRIMARY KEY (path, workspace_id),
+    FOREIGN KEY (workspace_id) REFERENCES workspace(id) ON DELETE CASCADE
+);
+
+GRANT ALL ON nats_trigger TO windmill_user;
+GRANT ALL ON nats_trigger TO windmill_admin;
+
+ALTER TABLE nats_trigger ENABLE ROW LEVEL SECURITY;
+
+CREATE POLICY admin_policy ON nats_trigger FOR ALL TO windmill_admin USING (true);
+
+CREATE POLICY see_folder_extra_perms_user_select ON nats_trigger FOR SELECT TO windmill_user
+USING (SPLIT_PART(nats_trigger.path, '/', 1) = 'f' AND SPLIT_PART(nats_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_read'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_insert ON nats_trigger FOR INSERT TO windmill_user
+WITH CHECK (SPLIT_PART(nats_trigger.path, '/', 1) = 'f' AND SPLIT_PART(nats_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_update ON nats_trigger FOR UPDATE TO windmill_user
+USING (SPLIT_PART(nats_trigger.path, '/', 1) = 'f' AND SPLIT_PART(nats_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_delete ON nats_trigger FOR DELETE TO windmill_user
+USING (SPLIT_PART(nats_trigger.path, '/', 1) = 'f' AND SPLIT_PART(nats_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+
+CREATE POLICY see_own ON nats_trigger FOR ALL TO windmill_user
+USING (SPLIT_PART(nats_trigger.path, '/', 1) = 'u' AND SPLIT_PART(nats_trigger.path, '/', 2) = current_setting('session.user'));
+CREATE POLICY see_member ON nats_trigger FOR ALL TO windmill_user
+USING (SPLIT_PART(nats_trigger.path, '/', 1) = 'g' AND SPLIT_PART(nats_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[]));
+
+CREATE POLICY see_extra_perms_user_select ON nats_trigger FOR SELECT TO windmill_user
+USING (extra_perms ? CONCAT('u/', current_setting('session.user')));
+CREATE POLICY see_extra_perms_user_insert ON nats_trigger FOR INSERT TO windmill_user
+WITH CHECK ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+CREATE POLICY see_extra_perms_user_update ON nats_trigger FOR UPDATE TO windmill_user
+USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+CREATE POLICY see_extra_perms_user_delete ON nats_trigger FOR DELETE TO windmill_user
+USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+
+CREATE POLICY see_extra_perms_groups_select ON nats_trigger FOR SELECT TO windmill_user
+USING (extra_perms ?| regexp_split_to_array(current_setting('session.pgroups'), ',')::text[]);
+CREATE POLICY see_extra_perms_groups_insert ON nats_trigger FOR INSERT TO windmill_user
+WITH CHECK (exists(
+    SELECT key, value FROM jsonb_each_text(extra_perms) 
+    WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
+    AND value::boolean));
+CREATE POLICY see_extra_perms_groups_update ON nats_trigger FOR UPDATE TO windmill_user
+USING (exists(
+    SELECT key, value FROM jsonb_each_text(extra_perms) 
+    WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
+    AND value::boolean));
+CREATE POLICY see_extra_perms_groups_delete ON nats_trigger FOR DELETE  TO windmill_user
+USING (exists(
+    SELECT key, value FROM jsonb_each_text(extra_perms) 
+    WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
+    AND value::boolean));

backend/migrations/20250107212922_workspace_color.down.sql

@@ -0,0 +1 @@
+ALTER TABLE workspace_settings DROP COLUMN color;

backend/migrations/20250107212922_workspace_color.up.sql

@@ -0,0 +1 @@
+ALTER TABLE workspace_settings ADD COLUMN color VARCHAR(7) DEFAULT NULL;

backend/windmill-api/Cargo.toml

@@ -20,6 +20,7 @@ prometheus = ["windmill-common/prometheus", "windmill-queue/prometheus", "dep:pr
 openidconnect = ["dep:openidconnect"]
 tantivy = ["dep:windmill-indexer"]
 kafka = ["dep:rdkafka"]
+nats = ["dep:async-nats", "dep:nkeys"]
 websocket = ["dep:tokio-tungstenite"]
 smtp = ["dep:mail-parser", "dep:openssl", "windmill-common/smtp"]
 license = ["dep:rsa"]
@@ -103,6 +104,8 @@ jsonwebtoken = { workspace = true }
 matchit = { workspace = true, optional = true }
 tokio-tungstenite = { workspace = true, optional = true}
 rdkafka = { workspace = true, optional = true }
+async-nats = { workspace = true, optional = true }
+nkeys = { workspace = true, optional = true }
 const_format.workspace = true
 
 pin-project.workspace = true

backend/windmill-api/src/apps.rs

@@ -20,11 +20,14 @@ use crate::{
 #[cfg(feature = "parquet")]
 use crate::{
     job_helpers_ee::{
-        get_random_file_name, get_s3_resource, get_workspace_s3_resource, upload_file_from_req,
-        UploadFileResponse,
+        download_s3_file_internal, get_random_file_name, get_s3_resource,
+        get_workspace_s3_resource, load_image_preview_internal, upload_file_from_req,
+        DownloadFileQuery, LoadImagePreviewQuery, UploadFileResponse,
     },
     users::fetch_api_authed_from_permissioned_as,
 };
+#[cfg(feature = "parquet")]
+use axum::response::Response;
 use axum::{
     extract::{Extension, Json, Path, Query},
     response::IntoResponse,
@@ -35,6 +38,7 @@ use futures::future::{FutureExt, TryFutureExt};
 use hyper::StatusCode;
 #[cfg(feature = "parquet")]
 use itertools::Itertools;
+use lazy_static::lazy_static;
 use magic_crypt::MagicCryptTrait;
 #[cfg(feature = "parquet")]
 use object_store::{Attribute, Attributes};
@@ -92,6 +96,11 @@ pub fn unauthed_service() -> Router {
     Router::new()
         .route("/execute_component/*path", post(execute_component))
         .route("/upload_s3_file/*path", post(upload_s3_file_from_app))
+        .route("/download_s3_file/*path", get(download_s3_file_from_app))
+        .route(
+            "/load_image_preview/*path",
+            get(load_s3_file_image_preview_from_app),
+        )
         .route("/public_app/:secret", get(get_public_app_by_secret))
         .route("/public_resource/*path", get(get_public_resource))
 }
@@ -190,7 +199,7 @@ pub type StaticFields = HashMap<String, Box<RawValue>>;
 pub type OneOfFields = HashMap<String, Vec<Box<RawValue>>>;
 pub type AllowUserResources = Vec<String>;
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Copy, Clone, Default)]
 #[serde(rename_all = "lowercase")]
 pub enum ExecutionMode {
     #[default]
@@ -1374,6 +1383,45 @@ async fn execute_component(
         }
     };
 
+    // Execution is publisher and an user is authenticated: check if the user is authorized to
+    // execute the app.
+    if let (ExecutionMode::Publisher, Some(authed)) = (policy.execution_mode, opt_authed.as_ref()) {
+        lazy_static! {
+            /// Cache for the permit to execute an app component.
+            static ref PERMIT_CACHE: cache::Cache<[u8; 32], bool> = cache::Cache::new(1000);
+        }
+
+        // Avoid allocation for the permit key using a sha256 hash of:
+        // - the user email,
+        // - the application path,
+        // - the workspace id.
+        let permit_key: [u8; 32] = [authed.email.as_bytes(), path.as_bytes(), &w_id.as_bytes()]
+            .iter()
+            .fold(Sha256::new(), |hasher, bytes| hasher.chain_update(bytes))
+            .finalize()
+            .into();
+        let permit_fut = PERMIT_CACHE.get_or_insert_async(&permit_key, async {
+            let mut tx = user_db.clone().begin(authed).await?;
+            // Permissions are checked by the database; just fetch a row from app using `user_db`:
+            let row = sqlx::query_scalar!(
+                "SELECT 1 FROM app WHERE path = $1 AND workspace_id = $2 LIMIT 1",
+                path,
+                &w_id,
+            )
+            .fetch_optional(&mut *tx)
+            .await?;
+            tx.commit().await?;
+            Result::Ok(row.is_some_and(|x| x.is_some()))
+        });
+
+        if !permit_fut.await? {
+            return Err(Error::NotAuthorized(format!(
+                "Missing read permissions on the `{}` app to execute `{}` runnable",
+                path, payload.component
+            )));
+        }
+    }
+
     let (username, permissioned_as, email) =
         get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?;
 
@@ -1400,7 +1448,7 @@ async fn execute_component(
         ),
         _ => unreachable!(),
     };
-    let tx = windmill_queue::PushIsolationLevel::IsolatedRoot(db.clone());
+    let tx = PushIsolationLevel::IsolatedRoot(db.clone());
 
     let (uuid, tx) = push(
         &db,
@@ -1678,6 +1726,128 @@ async fn upload_s3_file_from_app(
     return Ok(Json(UploadFileResponse { file_key }));
 }
 
+#[cfg(not(feature = "parquet"))]
+async fn download_s3_file_from_app() -> Result<()> {
+    return Err(Error::BadRequest(
+        "This endpoint requires the parquet feature to be enabled".to_string(),
+    ));
+}
+
+#[cfg(feature = "parquet")]
+async fn get_on_behalf_authed_from_app(
+    db: &DB,
+    path: &str,
+    w_id: &str,
+    opt_authed: &Option<ApiAuthed>,
+) -> Result<ApiAuthed> {
+    let policy_o = sqlx::query_scalar!(
+        "SELECT policy from app WHERE path = $1 AND workspace_id = $2",
+        path,
+        w_id
+    )
+    .fetch_optional(db)
+    .await?;
+
+    let policy = policy_o
+        .map(|p| serde_json::from_value::<Policy>(p).map_err(to_anyhow))
+        .transpose()?
+        .unwrap_or_else(|| Policy {
+            execution_mode: ExecutionMode::Viewer,
+            triggerables: None,
+            triggerables_v2: None,
+            on_behalf_of: None,
+            on_behalf_of_email: None,
+            s3_inputs: None,
+        });
+
+    let (username, permissioned_as, email) =
+        get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?;
+
+    let on_behalf_authed =
+        fetch_api_authed_from_permissioned_as(permissioned_as, email, &w_id, &db, Some(username))
+            .await?;
+
+    Ok(on_behalf_authed)
+}
+
+#[cfg(feature = "parquet")]
+async fn check_if_allowed_to_access_s3_file_from_app(
+    db: &DB,
+    opt_authed: &Option<ApiAuthed>,
+    file_key: &str,
+    w_id: &str,
+    path: &str,
+) -> Result<()> {
+    // if anonymous, check that the file was the result of an app script ran by an anonymous user in the last 3 hours
+    // otherwise, if logged in, allow any file (TODO: change that when we implement better s3 policy)
+
+    let allowed = opt_authed.is_some()
+        || sqlx::query_scalar!(
+            r#"SELECT EXISTS (
+                SELECT 1 FROM completed_job 
+                WHERE workspace_id = $2 
+                    AND (job_kind = 'appscript' OR job_kind = 'preview')
+                    AND created_by = 'anonymous' 
+                    AND started_at > now() - interval '3 hours'
+                    AND script_path LIKE $3 || '/%' 
+                    AND result @> ('{"s3":"' || $1 ||  '"}')::jsonb 
+            )"#,
+            file_key,
+            w_id,
+            path,
+        )
+        .fetch_one(db)
+        .await?
+        .unwrap_or(false);
+
+    if !allowed {
+        Err(Error::BadRequest("File restricted".to_string()))
+    } else {
+        Ok(())
+    }
+}
+
+#[cfg(feature = "parquet")]
+async fn download_s3_file_from_app(
+    OptAuthed(opt_authed): OptAuthed,
+    Extension(db): Extension<DB>,
+    Path((w_id, path)): Path<(String, StripPath)>,
+    Query(query): Query<DownloadFileQuery>,
+) -> Result<Response> {
+    let path = path.to_path();
+
+    let on_behalf_authed = get_on_behalf_authed_from_app(&db, &path, &w_id, &opt_authed).await?;
+
+    check_if_allowed_to_access_s3_file_from_app(&db, &opt_authed, &query.file_key, &w_id, &path)
+        .await?;
+
+    download_s3_file_internal(on_behalf_authed, &db, None, "", &w_id, query).await
+}
+
+#[cfg(not(feature = "parquet"))]
+async fn load_s3_file_image_preview_from_app() -> Result<()> {
+    return Err(Error::BadRequest(
+        "This endpoint requires the parquet feature to be enabled".to_string(),
+    ));
+}
+
+#[cfg(feature = "parquet")]
+async fn load_s3_file_image_preview_from_app(
+    OptAuthed(opt_authed): OptAuthed,
+    Extension(db): Extension<DB>,
+    Path((w_id, path)): Path<(String, StripPath)>,
+    Query(query): Query<LoadImagePreviewQuery>,
+) -> Result<Response> {
+    let path = path.to_path();
+
+    let on_behalf_authed = get_on_behalf_authed_from_app(&db, &path, &w_id, &opt_authed).await?;
+
+    check_if_allowed_to_access_s3_file_from_app(&db, &opt_authed, &query.file_key, &w_id, &path)
+        .await?;
+
+    load_image_preview_internal(on_behalf_authed, &db, "", &w_id, query).await
+}
+
 fn get_on_behalf_of(policy: &Policy) -> Result<(String, String)> {
     let permissioned_as = policy
         .on_behalf_of

backend/windmill-api/src/capture.rs

@@ -7,138 +7,572 @@
  */
 
 use axum::{
-    extract::{Extension, Path},
-    routing::{get, post, put},
-    Router,
+    extract::{Extension, Path, Query},
+    routing::{delete, get, head, post},
+    Json, Router,
 };
+#[cfg(feature = "http_trigger")]
+use http::HeaderMap;
 use hyper::StatusCode;
-use sqlx::types::Json;
+#[cfg(feature = "http_trigger")]
+use serde::de::DeserializeOwned;
+use serde::{Deserialize, Serialize};
+use serde_json::value::RawValue;
+use sqlx::types::Json as SqlxJson;
+#[cfg(feature = "http_trigger")]
+use std::collections::HashMap;
+use std::fmt;
+#[cfg(feature = "http_trigger")]
+use windmill_common::error::Error;
 use windmill_common::{
     db::UserDB,
     error::{JsonResult, Result},
-    utils::{not_found_if_none, StripPath},
+    utils::{not_found_if_none, paginate, Pagination, StripPath},
+    worker::{to_raw_value, CLOUD_HOSTED},
 };
-use windmill_queue::PushArgs;
+use windmill_queue::{PushArgs, PushArgsOwned};
 
+#[cfg(feature = "http_trigger")]
+use crate::http_triggers::{build_http_trigger_extra, HttpMethod};
+#[cfg(all(feature = "enterprise", feature = "kafka"))]
+use crate::kafka_triggers_ee::KafkaResourceSecurity;
+#[cfg(all(feature = "enterprise", feature = "nats"))]
+use crate::nats_triggers_ee::NatsResourceAuth;
 use crate::{
     args::WebhookArgs,
     db::{ApiAuthed, DB},
+    users::fetch_api_authed,
 };
 
-const KEEP_LAST: i64 = 8;
+const KEEP_LAST: i64 = 20;
 
 pub fn workspaced_service() -> Router {
     Router::new()
-        .route("/*path", put(new_payload))
-        .route("/*path", get(get_payload))
+        .route("/set_config", post(set_config))
+        .route(
+            "/ping_config/:trigger_kind/:runnable_kind/*path",
+            post(ping_config),
+        )
+        .route("/get_configs/:runnable_kind/*path", get(get_configs))
+        .route("/list/:runnable_kind/*path", get(list_captures))
+        .route("/:id", delete(delete_capture))
 }
 
-pub fn global_service() -> Router {
-    Router::new().route("/*path", post(update_payload))
+pub fn workspaced_unauthed_service() -> Router {
+    let router = Router::new().route(
+        "/webhook/:runnable_kind/*path",
+        head(|| async {}).post(webhook_payload),
+    );
+
+    #[cfg(feature = "http_trigger")]
+    {
+        router.route("/http/:runnable_kind/:path/*route_path", {
+            head(|| async {}).fallback(http_payload)
+        })
+    }
+
+    #[cfg(not(feature = "http_trigger"))]
+    {
+        router
+    }
 }
 
-pub async fn new_payload(
+#[derive(sqlx::Type, Serialize, Deserialize)]
+#[sqlx(type_name = "TRIGGER_KIND", rename_all = "lowercase")]
+#[serde(rename_all = "lowercase")]
+pub enum TriggerKind {
+    Webhook,
+    Http,
+    Websocket,
+    Kafka,
+    Email,
+    Nats,
+}
+
+impl fmt::Display for TriggerKind {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let s = match self {
+            TriggerKind::Webhook => "webhook",
+            TriggerKind::Http => "http",
+            TriggerKind::Websocket => "websocket",
+            TriggerKind::Kafka => "kafka",
+            TriggerKind::Email => "email",
+            TriggerKind::Nats => "nats",
+        };
+        write!(f, "{}", s)
+    }
+}
+
+#[cfg(feature = "http_trigger")]
+#[derive(Serialize, Deserialize)]
+struct HttpTriggerConfig {
+    route_path: String,
+    http_method: HttpMethod,
+}
+
+#[cfg(all(feature = "enterprise", feature = "kafka"))]
+#[derive(Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum KafkaTriggerConfigConnection {
+    Resource { kafka_resource_path: String },
+    Static { brokers: Vec<String>, security: KafkaResourceSecurity },
+}
+
+#[cfg(all(feature = "enterprise", feature = "kafka"))]
+#[derive(Serialize, Deserialize)]
+pub struct KafkaTriggerConfig {
+    #[serde(flatten)]
+    pub connection: KafkaTriggerConfigConnection,
+    pub topics: Vec<String>,
+    pub group_id: String,
+}
+
+#[cfg(all(feature = "enterprise", feature = "nats"))]
+#[derive(Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum NatsTriggerConfigConnection {
+    Resource { nats_resource_path: String },
+    Static { servers: Vec<String>, auth: NatsResourceAuth, require_tls: bool },
+}
+
+#[cfg(all(feature = "enterprise", feature = "nats"))]
+#[derive(Serialize, Deserialize)]
+pub struct NatsTriggerConfig {
+    #[serde(flatten)]
+    pub connection: NatsTriggerConfigConnection,
+    pub subjects: Vec<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub stream_name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub consumer_name: Option<String>,
+    pub use_jetstream: bool,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct WebsocketTriggerConfig {
+    pub url: String,
+    // have to use Value because RawValue is not supported inside untagged
+    pub url_runnable_args: Option<serde_json::Value>,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(untagged)]
+enum TriggerConfig {
+    #[cfg(feature = "http_trigger")]
+    Http(HttpTriggerConfig),
+    Websocket(WebsocketTriggerConfig),
+    #[cfg(all(feature = "enterprise", feature = "kafka"))]
+    Kafka(KafkaTriggerConfig),
+    #[cfg(all(feature = "enterprise", feature = "nats"))]
+    Nats(NatsTriggerConfig),
+}
+
+#[derive(Serialize, Deserialize)]
+struct NewCaptureConfig {
+    trigger_kind: TriggerKind,
+    path: String,
+    is_flow: bool,
+    trigger_config: Option<TriggerConfig>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CaptureConfig {
+    trigger_config: Option<SqlxJson<Box<RawValue>>>,
+    trigger_kind: TriggerKind,
+    error: Option<String>,
+    last_server_ping: Option<chrono::DateTime<chrono::Utc>>,
+}
+
+async fn get_configs(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
-    Path((w_id, path)): Path<(String, StripPath)>,
-) -> Result<StatusCode> {
+    Path((w_id, runnable_kind, path)): Path<(String, RunnableKind, StripPath)>,
+) -> JsonResult<Vec<CaptureConfig>> {
+    let mut tx = user_db.begin(&authed).await?;
+
+    let configs = sqlx::query_as!(
+        CaptureConfig,
+        r#"SELECT trigger_config as "trigger_config: _", trigger_kind as "trigger_kind: _", error, last_server_ping
+        FROM capture_config
+        WHERE workspace_id = $1 AND path = $2 AND is_flow = $3"#,
+        &w_id,
+        &path.to_path(),
+        matches!(runnable_kind, RunnableKind::Flow),
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
+    tx.commit().await?;
+
+    Ok(Json(configs))
+}
+
+async fn set_config(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path(w_id): Path<String>,
+    Json(nc): Json<NewCaptureConfig>,
+) -> Result<()> {
     let mut tx = user_db.begin(&authed).await?;
 
     sqlx::query!(
-        "
-       INSERT INTO capture
-                   (workspace_id, path, created_by)
-            VALUES ($1, $2, $3)
-       ON CONFLICT (workspace_id, path)
-     DO UPDATE SET created_at = now()
-        ",
+        "INSERT INTO capture_config
+            (workspace_id, path, is_flow, trigger_kind, trigger_config, owner, email)
+        VALUES ($1, $2, $3, $4, $5, $6, $7)
+        ON CONFLICT (workspace_id, path, is_flow, trigger_kind)
+            DO UPDATE SET trigger_config = $5, owner = $6, email = $7, server_id = NULL, last_server_ping = NULL, error = NULL",
         &w_id,
-        &path.to_path(),
+        &nc.path,
+        nc.is_flow,
+        nc.trigger_kind as TriggerKind,
+        nc.trigger_config.map(|x| SqlxJson(to_raw_value(&x))) as Option<SqlxJson<Box<RawValue>>>,
         &authed.username,
-    )
-    .execute(&mut *tx)
-    .await?;
-
-    /* Retain only KEEP_LAST most recent captures by this user in this workspace. */
-    sqlx::query!(
-        "
-   DELETE FROM capture
-         WHERE workspace_id = $1
-           AND created_by = $2
-           AND created_at <=
-              ( SELECT created_at
-                  FROM capture
-                 WHERE workspace_id = $1
-                   AND created_by = $2
-              ORDER BY created_at DESC
-                OFFSET $3
-                 LIMIT 1 )
-        ",
-        &w_id,
-        &authed.username,
-        KEEP_LAST,
+        &authed.email,
     )
     .execute(&mut *tx)
     .await?;
 
     tx.commit().await?;
 
-    Ok(StatusCode::CREATED)
+    Ok(())
 }
 
-pub async fn update_payload(
-    Extension(db): Extension<DB>,
-    Path((w_id, path)): Path<(String, StripPath)>,
-    args: WebhookArgs,
-) -> Result<StatusCode> {
-    let args = args.args;
-
-    let mut tx = db.begin().await?;
-
+async fn ping_config(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, trigger_kind, runnable_kind, path)): Path<(
+        String,
+        TriggerKind,
+        RunnableKind,
+        StripPath,
+    )>,
+) -> Result<()> {
+    let mut tx = user_db.begin(&authed).await?;
     sqlx::query!(
-        "
-       UPDATE capture
-          SET payload = $3
-        WHERE workspace_id = $1
-          AND path = $2
-        ",
+        "UPDATE capture_config SET last_client_ping = now() WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4",
         &w_id,
         &path.to_path(),
-        Json(PushArgs { args: &args.args, extra: args.extra }) as Json<PushArgs>,
+        matches!(runnable_kind, RunnableKind::Flow),
+        trigger_kind as TriggerKind,
     )
     .execute(&mut *tx)
     .await?;
+    tx.commit().await?;
+    Ok(())
+}
+
+#[derive(Serialize, Deserialize)]
+struct Capture {
+    id: i64,
+    created_at: chrono::DateTime<chrono::Utc>,
+    trigger_kind: TriggerKind,
+    payload: SqlxJson<Box<serde_json::value::RawValue>>,
+    trigger_extra: Option<SqlxJson<Box<serde_json::value::RawValue>>>,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "lowercase")]
+enum RunnableKind {
+    Script,
+    Flow,
+}
+
+#[derive(Deserialize)]
+struct ListCapturesQuery {
+    trigger_kind: Option<TriggerKind>,
+    page: Option<usize>,
+    per_page: Option<usize>,
+}
+
+async fn list_captures(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, runnable_kind, path)): Path<(String, RunnableKind, StripPath)>,
+    Query(query): Query<ListCapturesQuery>,
+) -> JsonResult<Vec<Capture>> {
+    let mut tx = user_db.begin(&authed).await?;
+
+    let (per_page, offset) = paginate(Pagination { page: query.page, per_page: query.per_page });
+
+    let captures = sqlx::query_as!(
+        Capture,
+        r#"SELECT id, created_at, trigger_kind as "trigger_kind: _", payload as "payload: _", trigger_extra as "trigger_extra: _"
+        FROM capture
+        WHERE workspace_id = $1
+            AND path = $2 AND is_flow = $3
+            AND ($4::trigger_kind IS NULL OR trigger_kind = $4)
+        ORDER BY created_at DESC
+        OFFSET $5
+        LIMIT $6"#,
+        &w_id,
+        &path.to_path(),
+        matches!(runnable_kind, RunnableKind::Flow),
+        query.trigger_kind as Option<TriggerKind>,
+        offset as i64,
+        per_page as i64,
+    )
+    .fetch_all(&mut *tx)
+    .await?;
 
     tx.commit().await?;
 
+    Ok(Json(captures))
+}
+
+async fn delete_capture(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path((_, id)): Path<(String, i64)>,
+) -> Result<()> {
+    let mut tx = user_db.begin(&authed).await?;
+    sqlx::query!("DELETE FROM capture WHERE id = $1", id)
+        .execute(&mut *tx)
+        .await?;
+    tx.commit().await?;
+    Ok(())
+}
+
+#[derive(Serialize, Deserialize)]
+struct ActiveCaptureOwner {
+    owner: String,
+    email: String,
+}
+
+pub async fn get_active_capture_owner_and_email(
+    db: &DB,
+    w_id: &str,
+    path: &str,
+    is_flow: bool,
+    kind: &TriggerKind,
+) -> Result<(String, String)> {
+    let capture_config = sqlx::query_as!(
+        ActiveCaptureOwner,
+        "SELECT owner, email
+        FROM capture_config
+        WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4 AND last_client_ping > NOW() - INTERVAL '10 seconds'",
+        &w_id,
+        &path,
+        is_flow,
+        kind as &TriggerKind,
+    )
+    .fetch_optional(db)
+    .await?;
+
+    let capture_config = not_found_if_none(
+        capture_config,
+        &format!("capture config for {} trigger", kind),
+        path,
+    )?;
+
+    Ok((capture_config.owner, capture_config.email))
+}
+
+#[cfg(feature = "http_trigger")]
+async fn get_capture_trigger_config_and_owner<T: DeserializeOwned>(
+    db: &DB,
+    w_id: &str,
+    path: &str,
+    is_flow: bool,
+    kind: &TriggerKind,
+) -> Result<(T, String, String)> {
+    #[derive(Deserialize)]
+    struct CaptureTriggerConfigAndOwner {
+        trigger_config: Option<SqlxJson<Box<RawValue>>>,
+        owner: String,
+        email: String,
+    }
+
+    let capture_config = sqlx::query_as!(
+        CaptureTriggerConfigAndOwner,
+        r#"SELECT trigger_config as "trigger_config: _", owner, email
+        FROM capture_config
+        WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4 AND last_client_ping > NOW() - INTERVAL '10 seconds'"#,
+        &w_id,
+        &path,
+        is_flow,
+        kind as &TriggerKind,
+    )
+    .fetch_optional(db)
+    .await?;
+
+    let capture_config = not_found_if_none(
+        capture_config,
+        &format!("capture config for {} trigger", kind),
+        path,
+    )?;
+
+    let trigger_config = not_found_if_none(
+        capture_config.trigger_config,
+        &format!("capture {} trigger config", kind),
+        path,
+    )?;
+
+    Ok((
+        serde_json::from_str(trigger_config.get()).map_err(|e| {
+            Error::InternalErr(format!(
+                "error parsing capture config for {} trigger: {}",
+                kind, e
+            ))
+        })?,
+        capture_config.owner,
+        capture_config.email,
+    ))
+}
+
+async fn clear_captures_history(db: &DB, w_id: &str) -> Result<()> {
+    if *CLOUD_HOSTED {
+        /* Retain only KEEP_LAST most recent captures in this workspace. */
+        sqlx::query!(
+            "DELETE FROM capture
+            WHERE workspace_id = $1
+                AND created_at <=
+                    (
+                        SELECT created_at
+                            FROM capture
+                            WHERE workspace_id = $1
+                        ORDER BY created_at DESC
+                            OFFSET $2
+                            LIMIT 1
+                    )",
+            &w_id,
+            KEEP_LAST,
+        )
+        .execute(db)
+        .await?;
+    }
+    Ok(())
+}
+
+pub async fn insert_capture_payload(
+    db: &DB,
+    w_id: &str,
+    path: &str,
+    is_flow: bool,
+    trigger_kind: &TriggerKind,
+    payload: PushArgsOwned,
+    trigger_extra: Option<Box<RawValue>>,
+    owner: &str,
+) -> Result<()> {
+    sqlx::query!(
+        "INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, payload, trigger_extra, created_by)
+        VALUES ($1, $2, $3, $4, $5, $6, $7)",
+        &w_id,
+        path,
+        is_flow,
+        trigger_kind as &TriggerKind,
+        SqlxJson(to_raw_value(&PushArgs {
+            args: &payload.args,
+            extra: payload.extra
+        })) as SqlxJson<Box<RawValue>>,
+        trigger_extra.map(SqlxJson) as Option<SqlxJson<Box<RawValue>>>,
+        owner,
+    )
+    .execute(db)
+    .await?;
+
+    clear_captures_history(db, &w_id).await?;
+
+    Ok(())
+}
+
+async fn webhook_payload(
+    Extension(db): Extension<DB>,
+    Path((w_id, runnable_kind, path)): Path<(String, RunnableKind, StripPath)>,
+    args: WebhookArgs,
+) -> Result<StatusCode> {
+    let (owner, email) = get_active_capture_owner_and_email(
+        &db,
+        &w_id,
+        &path.to_path(),
+        matches!(runnable_kind, RunnableKind::Flow),
+        &TriggerKind::Webhook,
+    )
+    .await?;
+
+    let authed = fetch_api_authed(owner.clone(), email, &w_id, &db, None).await?;
+    let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
+
+    insert_capture_payload(
+        &db,
+        &w_id,
+        &path.to_path(),
+        matches!(runnable_kind, RunnableKind::Flow),
+        &TriggerKind::Webhook,
+        args,
+        Some(to_raw_value(&serde_json::json!({
+            "wm_trigger": {
+                "kind": "webhook",
+            }
+        }))),
+        &owner,
+    )
+    .await?;
+
     Ok(StatusCode::NO_CONTENT)
 }
 
-#[derive(sqlx::FromRow)]
-struct Payload {
-    payload: sqlx::types::Json<Box<serde_json::value::RawValue>>,
-}
-pub async fn get_payload(
-    authed: ApiAuthed,
-    Extension(user_db): Extension<UserDB>,
-    Path((w_id, path)): Path<(String, StripPath)>,
-) -> JsonResult<Box<serde_json::value::RawValue>> {
-    let mut tx = user_db.begin(&authed).await?;
+#[cfg(feature = "http_trigger")]
+async fn http_payload(
+    Extension(db): Extension<DB>,
+    Path((w_id, kind, path, route_path)): Path<(String, RunnableKind, String, StripPath)>,
+    Query(query): Query<HashMap<String, String>>,
+    method: http::Method,
+    headers: HeaderMap,
+    args: WebhookArgs,
+) -> Result<StatusCode> {
+    let route_path = route_path.to_path();
+    let path = path.replace(".", "/");
 
-    let payload = sqlx::query_as::<_, Payload>(
-        "
-       SELECT payload
-         FROM capture
-        WHERE workspace_id = $1
-          AND path = $2
-        ",
+    let (http_trigger_config, owner, email): (HttpTriggerConfig, _, _) =
+        get_capture_trigger_config_and_owner(
+            &db,
+            &w_id,
+            &path,
+            matches!(kind, RunnableKind::Flow),
+            &TriggerKind::Http,
+        )
+        .await?;
+
+    let authed = fetch_api_authed(owner.clone(), email, &w_id, &db, None).await?;
+    let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
+
+    let mut router = matchit::Router::new();
+    router.insert(&http_trigger_config.route_path, ()).ok();
+    let match_ = router.at(route_path).ok();
+
+    let match_ = not_found_if_none(match_, "capture http trigger", &route_path)?;
+
+    let matchit::Match { params, .. } = match_;
+
+    let params: HashMap<String, String> = params
+        .iter()
+        .map(|(k, v)| (k.to_string(), v.to_string()))
+        .collect();
+
+    let extra: HashMap<String, Box<RawValue>> = HashMap::from_iter(vec![(
+        "wm_trigger".to_string(),
+        build_http_trigger_extra(
+            &http_trigger_config.route_path,
+            route_path,
+            &method,
+            &params,
+            &query,
+            &headers,
+        )
+        .await,
+    )]);
+
+    insert_capture_payload(
+        &db,
+        &w_id,
+        &path,
+        matches!(kind, RunnableKind::Flow),
+        &TriggerKind::Http,
+        args,
+        Some(to_raw_value(&extra)),
+        &owner,
     )
-    .bind(&w_id)
-    .bind(&path.to_path())
-    .fetch_optional(&mut *tx)
     .await?;
 
-    tx.commit().await?;
-
-    not_found_if_none(payload.map(|x| x.payload.0), "capture", path.to_path()).map(axum::Json)
+    Ok(StatusCode::NO_CONTENT)
 }

backend/windmill-api/src/flows.rs

@@ -740,6 +740,24 @@ async fn update_flow(
                 "Error updating flow due to deleting old flow: {e:#}"
             ))
         })?;
+
+        sqlx::query!(
+            "UPDATE capture_config SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS TRUE",
+            nf.path,
+            flow_path,
+            w_id
+        )
+        .execute(&mut *tx)
+        .await?;
+    
+        sqlx::query!(
+            "UPDATE capture SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS TRUE",
+            nf.path,
+            flow_path,
+            w_id
+        )
+        .execute(&mut *tx)
+        .await?;
     }
 
     let version = sqlx::query_scalar!(
@@ -1123,6 +1141,22 @@ async fn delete_flow_by_path(
     .execute(&mut *tx)
     .await?;
 
+    sqlx::query!(
+        "DELETE FROM capture_config WHERE path = $1 AND workspace_id = $2 AND is_flow IS TRUE",
+        path,
+        &w_id
+    )
+    .execute(&mut *tx)
+    .await?;
+
+    sqlx::query!(
+        "DELETE FROM capture WHERE path = $1 AND workspace_id = $2 AND is_flow IS TRUE",
+        path,
+        &w_id
+    )
+    .execute(&mut *tx)
+    .await?;
+
     audit_log(
         &mut *tx,
         &authed,

backend/windmill-api/src/granular_acls.rs

@@ -23,7 +23,7 @@ use windmill_common::{
     utils::{not_found_if_none, StripPath},
 };
 
-const KINDS: [&str; 12] = [
+const KINDS: [&str; 13] = [
     "script",
     "group_",
     "resource",
@@ -36,6 +36,7 @@ const KINDS: [&str; 12] = [
     "http_trigger",
     "websocket_trigger",
     "kafka_trigger",
+    "nats_trigger",
 ];
 
 pub fn workspaced_service() -> Router {

backend/windmill-api/src/http_triggers.rs

@@ -77,7 +77,7 @@ pub fn workspaced_service() -> Router {
 #[derive(Serialize, Deserialize, sqlx::Type)]
 #[sqlx(type_name = "HTTP_METHOD", rename_all = "lowercase")]
 #[serde(rename_all = "lowercase")]
-enum HttpMethod {
+pub enum HttpMethod {
     Get,
     Post,
     Put,
@@ -85,14 +85,16 @@ enum HttpMethod {
     Patch,
 }
 
-impl From<HttpMethod> for http::Method {
-    fn from(method: HttpMethod) -> Self {
+impl TryFrom<&http::Method> for HttpMethod {
+    type Error = error::Error;
+    fn try_from(method: &http::Method) -> Result<Self, Self::Error> {
         match method {
-            HttpMethod::Get => http::Method::GET,
-            HttpMethod::Post => http::Method::POST,
-            HttpMethod::Put => http::Method::PUT,
-            HttpMethod::Delete => http::Method::DELETE,
-            HttpMethod::Patch => http::Method::PATCH,
+            &http::Method::GET => Ok(HttpMethod::Get),
+            &http::Method::POST => Ok(HttpMethod::Post),
+            &http::Method::PUT => Ok(HttpMethod::Put),
+            &http::Method::DELETE => Ok(HttpMethod::Delete),
+            &http::Method::PATCH => Ok(HttpMethod::Patch),
+            _ => Err(error::Error::BadRequest("Invalid HTTP method".to_string())),
         }
     }
 }
@@ -417,7 +419,6 @@ struct TriggerRoute {
     requires_auth: bool,
     edited_by: String,
     email: String,
-    http_method: HttpMethod,
     static_asset_config: Option<sqlx::types::Json<S3Object>>,
 }
 
@@ -427,7 +428,9 @@ async fn get_http_route_trigger(
     token: Option<&String>,
     db: &DB,
     user_db: UserDB,
+    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(|| {
@@ -436,8 +439,9 @@ async fn get_http_route_trigger(
         let route_path = StripPath(splitted.collect::<Vec<_>>().join("/"));
         let triggers = sqlx::query_as!(
             TriggerRoute,
-            r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _", static_asset_config as "static_asset_config: _" FROM http_trigger WHERE workspace_id = $1"#,
-            w_id
+            r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, static_asset_config as "static_asset_config: _" FROM http_trigger WHERE workspace_id = $1 AND http_method = $2"#,
+            w_id,
+            http_method as HttpMethod
         )
         .fetch_all(db)
         .await?;
@@ -445,7 +449,8 @@ async fn get_http_route_trigger(
     } else {
         let triggers = sqlx::query_as!(
             TriggerRoute,
-            r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _", static_asset_config as "static_asset_config: _" FROM http_trigger"#,
+            r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, static_asset_config as "static_asset_config: _" FROM http_trigger WHERE http_method = $1"#,
+            http_method as HttpMethod
         )
         .fetch_all(db)
         .await?;
@@ -523,6 +528,32 @@ async fn get_http_route_trigger(
     Ok((trigger, route_path.0, params, authed))
 }
 
+pub async fn build_http_trigger_extra(
+    route_path: &str,
+    called_path: &str,
+    method: &http::Method,
+    params: &HashMap<String, String>,
+    query: &HashMap<String, String>,
+    headers: &HeaderMap,
+) -> Box<serde_json::value::RawValue> {
+    let headers = headers
+        .iter()
+        .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
+        .collect::<HashMap<String, String>>();
+
+    to_raw_value(&serde_json::json!({
+        "kind": "http",
+        "http": {
+            "route": route_path,
+            "path": called_path,
+            "method": method.to_string().to_lowercase(),
+            "params": params,
+            "query": query,
+            "headers": headers
+        },
+    }))
+}
+
 async fn route_job(
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
@@ -541,6 +572,7 @@ async fn route_job(
         token.as_ref(),
         &db,
         user_db.clone(),
+        &method,
     )
     .await
     {
@@ -639,34 +671,23 @@ async fn route_job(
         }
     }
 
-    let headers = headers
-        .iter()
-        .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
-        .collect::<HashMap<String, String>>();
     let extra = args.extra.get_or_insert_with(HashMap::new);
     extra.insert(
         "wm_trigger".to_string(),
-        to_raw_value(&serde_json::json!({
-            "kind": "http",
-            "http": {
-                "route": trigger.route_path,
-                "path": called_path,
-                "method": method.to_string().to_lowercase(),
-                "params": params,
-                "query": query,
-                "headers": headers
-            },
-        })),
+        build_http_trigger_extra(
+            &trigger.route_path,
+            &called_path,
+            &method,
+            &params,
+            &query,
+            &headers,
+        )
+        .await,
     );
-    let http_method = http::Method::from(trigger.http_method);
-
-    if http_method != method {
-        return error::Error::BadRequest("Invalid HTTP method".to_string()).into_response();
-    }
 
     let label_prefix = Some(format!(
         "http-{}-{}-",
-        http_method.as_str().to_lowercase(),
+        method.as_str().to_lowercase(),
         trigger.route_path
     ));
 

backend/windmill-api/src/inputs.rs

@@ -115,12 +115,18 @@ pub struct CompletedJobMini {
     success: bool,
 }
 
+#[derive(Deserialize)]
+struct GetInputHistory {
+    include_preview: Option<bool>,
+}
+
 async fn get_input_history(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
     Query(r): Query<RunnableParams>,
+    Query(g): Query<GetInputHistory>,
 ) -> JsonResult<Vec<Input>> {
     let (per_page, offset) = paginate(pagination);
 
@@ -128,7 +134,7 @@ async fn get_input_history(
 
     let sql = &format!(
         "select id, created_at, created_by, 'null'::jsonb as args, success from completed_job \
-        where {} = $1 and job_kind = $2 and workspace_id = $3 \
+        where {} = $1 and job_kind = any($2) and workspace_id = $3 \
         order by created_at desc limit $4 offset $5",
         r.runnable_type.column_name()
     );
@@ -140,8 +146,18 @@ async fn get_input_history(
         _ => query.bind(&r.runnable_id),
     };
 
+    let job_kinds = match r.runnable_type.job_kind() {
+        kind @ JobKind::Script if g.include_preview.unwrap_or(false) => {
+            vec![kind, JobKind::Preview]
+        }
+        kind @ JobKind::Flow if g.include_preview.unwrap_or(false) => {
+            vec![kind, JobKind::FlowPreview]
+        }
+        kind => vec![kind],
+    };
+
     let rows = query
-        .bind(r.runnable_type.job_kind())
+        .bind(job_kinds)
         .bind(&w_id)
         .bind(per_page as i32)
         .bind(offset as i32)

backend/windmill-api/src/job_helpers_ee.rs

@@ -18,11 +18,26 @@ use bytes::Bytes;
 #[cfg(feature = "parquet")]
 use futures::Stream;
 
+#[cfg(feature = "parquet")]
+use axum::response::Response;
+#[cfg(feature = "parquet")]
+use serde::Deserialize;
+
 #[derive(Serialize)]
 pub struct UploadFileResponse {
     pub file_key: String,
 }
 
+#[derive(Deserialize)]
+pub struct LoadImagePreviewQuery {
+    pub file_key: String,
+}
+
+#[derive(Deserialize)]
+pub struct DownloadFileQuery {
+    pub file_key: String,
+}
+
 pub fn workspaced_service() -> Router {
     Router::new()
 }
@@ -82,3 +97,30 @@ pub async fn upload_file_internal(
         "Not implemented in Windmill's Open Source repository".to_string(),
     ))
 }
+
+#[cfg(feature = "parquet")]
+pub async fn download_s3_file_internal(
+    _authed: ApiAuthed,
+    _db: &DB,
+    _user_db: Option<UserDB>,
+    _token: &str,
+    _w_id: &str,
+    _query: DownloadFileQuery,
+) -> error::Result<Response> {
+    Err(error::Error::InternalErr(
+        "Not implemented in Windmill's Open Source repository".to_string(),
+    ))
+}
+
+#[cfg(feature = "parquet")]
+pub async fn load_image_preview_internal(
+    _authed: ApiAuthed,
+    _db: &DB,
+    _token: &str,
+    _w_id: &str,
+    _query: LoadImagePreviewQuery,
+) -> error::Result<Response> {
+    Err(error::Error::InternalErr(
+        "Not implemented in Windmill's Open Source repository".to_string(),
+    ))
+}

backend/windmill-api/src/jobs.rs

@@ -85,7 +85,7 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED};
 
 use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL};
 use windmill_queue::{
-    cancel_job, get_queued_job, get_result_by_id_from_running_flow, job_is_complete, push,
+    cancel_job, get_queued_job, get_result_and_success_by_id_from_flow, job_is_complete, push,
     PushArgs, PushArgsOwned, PushIsolationLevel,
 };
 
@@ -2398,7 +2398,12 @@ pub async fn get_resume_urls(
     Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>,
     Query(approver): Query<QueryApprover>,
 ) -> error::JsonResult<ResumeUrls> {
-    get_resume_urls_internal(Extension(db), Path((w_id, job_id, resume_id)), Query(approver)).await
+    get_resume_urls_internal(
+        Extension(db),
+        Path((w_id, job_id, resume_id)),
+        Query(approver),
+    )
+    .await
 }
 
 pub async fn get_resume_urls_internal(
@@ -3483,6 +3488,7 @@ pub async fn run_wait_result(
     username: &str,
 ) -> error::Result<Response> {
     let mut result = None;
+    let mut success = false;
     let timeout = TIMEOUT_WAIT_RESULT.read().await.clone().unwrap_or(600);
     let timeout_ms = if timeout <= 0 {
         2000
@@ -3503,7 +3509,7 @@ pub async fn run_wait_result(
 
     loop {
         if let Some(node_id_for_empty_return) = node_id_for_empty_return.as_ref() {
-            result = get_result_by_id_from_running_flow(
+            let result_and_success = get_result_and_success_by_id_from_flow(
                 &db,
                 &w_id,
                 &uuid,
@@ -3512,11 +3518,15 @@ pub async fn run_wait_result(
             )
             .await
             .ok();
+            if let Some((r, s)) = result_and_success {
+                result = Some(r);
+                success = s;
+            }
         }
 
         if result.is_none() {
-            let row = sqlx::query_as::<_, RawResult>(
-                "SELECT null as created_by, result, language, flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2",
+            let row = sqlx::query_as::<_, RawResultWithSuccess>(
+                "SELECT '' as created_by, result, language, flow_status, success FROM completed_job WHERE id = $1 AND workspace_id = $2",
             )
             .bind(uuid)
             .bind(&w_id)
@@ -3529,6 +3539,7 @@ pub async fn run_wait_result(
                     raw_result.result.as_mut(),
                 );
                 result = raw_result.result.map(|x| x.0);
+                success = raw_result.success;
             }
         }
 
@@ -3558,7 +3569,15 @@ pub async fn run_wait_result(
                 result: result_value,
             }) => {
                 if windmill_content_type.is_none() && windmill_status_code.is_none() {
-                    return Ok(Json(result).into_response());
+                    return Ok((
+                        if success {
+                            StatusCode::OK
+                        } else {
+                            StatusCode::INTERNAL_SERVER_ERROR
+                        },
+                        Json(result),
+                    )
+                        .into_response());
                 }
 
                 let status_code_or_default = windmill_status_code
@@ -3566,10 +3585,14 @@ pub async fn run_wait_result(
                         Ok(sc) => Ok(sc),
                         Err(_) => Err(Error::ExecutionErr("Invalid status code".to_string())),
                     })
-                    .unwrap_or(if result_value.is_some() {
-                        Ok(StatusCode::OK)
-                    } else {
-                        Ok(StatusCode::NO_CONTENT)
+                    .unwrap_or_else(|| {
+                        if !success {
+                            Ok(StatusCode::INTERNAL_SERVER_ERROR)
+                        } else if result_value.is_some() {
+                            Ok(StatusCode::OK)
+                        } else {
+                            Ok(StatusCode::NO_CONTENT)
+                        }
                     })?;
 
                 if windmill_content_type.is_some() {
@@ -3598,7 +3621,15 @@ pub async fn run_wait_result(
                 )
                     .into_response());
             }
-            _ => Ok(Json(result).into_response()),
+            _ => Ok((
+                if success {
+                    StatusCode::OK
+                } else {
+                    StatusCode::INTERNAL_SERVER_ERROR
+                },
+                Json(result),
+            )
+                .into_response()),
         }
     } else {
         Err(Error::ExecutionErr(format!("timeout after {}s", timeout)))
@@ -5569,4 +5600,3 @@ async fn delete_completed_job<'a>(
     let response = Json(cj).into_response();
     Ok(response)
 }
-

backend/windmill-api/src/kafka_triggers_ee.rs

@@ -1,5 +1,9 @@
 use crate::db::DB;
 use axum::Router;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize)]
+pub struct KafkaResourceSecurity {}
 
 pub fn workspaced_service() -> Router {
     Router::new()

backend/windmill-api/src/lib.rs

@@ -82,6 +82,8 @@ pub mod job_metrics;
 pub mod jobs;
 #[cfg(all(feature = "enterprise", feature = "kafka"))]
 mod kafka_triggers_ee;
+#[cfg(all(feature = "enterprise", feature = "nats"))]
+mod nats_triggers_ee;
 #[cfg(feature = "oauth2")]
 pub mod oauth2_ee;
 mod oidc_ee;
@@ -93,6 +95,7 @@ mod scim_ee;
 mod scripts;
 mod service_logs;
 mod settings;
+mod slack_approvals;
 #[cfg(feature = "smtp")]
 mod smtp_server_ee;
 mod static_assets;
@@ -109,7 +112,6 @@ mod websocket_triggers;
 mod workers;
 mod workspaces;
 mod workspaces_ee;
-mod slack_approvals;
 mod workspaces_export;
 mod workspaces_extra;
 
@@ -250,9 +252,6 @@ pub async fn run_server(
         }
     }
 
-    // #[cfg(feature = "kafka")]
-    // start_listening().await;
-
     let job_helpers_service = {
         #[cfg(feature = "parquet")]
         {
@@ -277,6 +276,18 @@ pub async fn run_server(
         }
     };
 
+    let nats_triggers_service = {
+        #[cfg(all(feature = "enterprise", feature = "nats"))]
+        {
+            nats_triggers_ee::workspaced_service()
+        }
+
+        #[cfg(not(all(feature = "enterprise", feature = "nats")))]
+        {
+            Router::new()
+        }
+    };
+
     if !*CLOUD_HOSTED {
         #[cfg(feature = "websocket")]
         {
@@ -289,6 +300,12 @@ pub async fn run_server(
             let kafka_killpill_rx = rx.resubscribe();
             kafka_triggers_ee::start_kafka_consumers(db.clone(), kafka_killpill_rx).await;
         }
+
+        #[cfg(all(feature = "enterprise", feature = "nats"))]
+        {
+            let nats_killpill_rx = rx.resubscribe();
+            nats_triggers_ee::start_nats_consumers(db.clone(), nats_killpill_rx).await;
+        }
     }
 
     // build our application with a route
@@ -357,7 +374,8 @@ pub async fn run_server(
                             #[cfg(not(feature = "websocket"))]
                             Router::new()
                         })
-                        .nest("/kafka_triggers", kafka_triggers_service),
+                        .nest("/kafka_triggers", kafka_triggers_service)
+                        .nest("/nats_triggers", nats_triggers_service),
                 )
                 .nest("/workspaces", workspaces::global_service())
                 .nest(
@@ -417,14 +435,17 @@ pub async fn run_server(
                     jobs::workspace_unauthed_service().layer(cors.clone()),
                 )
                 .route("/slack", post(slack_approvals::slack_app_callback_handler))
-                .route("/w/:workspace_id/jobs/slack_approval/:job_id", get(slack_approvals::request_slack_approval))
+                .route(
+                    "/w/:workspace_id/jobs/slack_approval/:job_id",
+                    get(slack_approvals::request_slack_approval),
+                )
                 .nest(
                     "/w/:workspace_id/resources_u",
                     resources::public_service().layer(cors.clone()),
                 )
                 .nest(
                     "/w/:workspace_id/capture_u",
-                    capture::global_service().layer(cors.clone()),
+                    capture::workspaced_unauthed_service().layer(cors.clone()),
                 )
                 .nest(
                     "/auth",

backend/windmill-api/src/nats_triggers_ee.rs

@@ -0,0 +1,17 @@
+use crate::db::DB;
+use axum::Router;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize)]
+pub struct NatsResourceAuth {}
+
+pub fn workspaced_service() -> Router {
+    Router::new()
+}
+
+pub async fn start_nats_consumers(
+    _db: DB,
+    mut _killpill_rx: tokio::sync::broadcast::Receiver<()>,
+) -> () {
+    // implementation is not open source
+}

backend/windmill-api/src/schedule.rs

@@ -438,7 +438,7 @@ async fn get_schedule(
     let path = path.to_path();
     let mut tx = user_db.begin(&authed).await?;
 
-    let schedule_o = windmill_queue::schedule::get_schedule_opt(&mut tx, &w_id, path).await?;
+    let schedule_o = windmill_queue::schedule::get_schedule_opt(&mut *tx, &w_id, path).await?;
     let schedule = not_found_if_none(schedule_o, "Schedule", path)?;
     tx.commit().await?;
     Ok(Json(schedule))

backend/windmill-api/src/scripts.rs

@@ -662,6 +662,24 @@ async fn create_script_internal<'c>(
         .execute(&mut *tx)
         .await?;
 
+        sqlx::query!(
+            "UPDATE capture_config SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS FALSE",
+            ns.path,
+            p_path,
+            w_id
+        )
+        .execute(&mut *tx)
+        .await?;
+    
+        sqlx::query!(
+            "UPDATE capture SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS FALSE",
+            ns.path,
+            p_path,
+            w_id
+        )
+        .execute(&mut *tx)
+        .await?;
+
         let mut schedulables = sqlx::query_as::<_, Schedule>(
             "UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND path != $2 AND workspace_id = $3 AND is_flow IS false RETURNING *")
             .bind(&ns.path)
@@ -1471,6 +1489,22 @@ async fn delete_script_by_path(
     .execute(&db)
     .await?;
 
+    sqlx::query!(
+        "DELETE FROM capture_config WHERE path = $1 AND workspace_id = $2 AND is_flow IS FALSE",
+        path,
+        w_id
+    )
+    .execute(&db)
+    .await?;
+
+    sqlx::query!(
+        "DELETE FROM capture WHERE path = $1 AND workspace_id = $2 AND is_flow IS FALSE",
+        path,
+        w_id
+    )
+    .execute(&db)
+    .await?;
+
     audit_log(
         &mut *tx,
         &authed,

backend/windmill-api/src/slack_approvals.rs

@@ -8,7 +8,6 @@ use serde_json::value::{RawValue, Value};
 
 use sqlx::types::Uuid;
 use std::{collections::HashMap, str::FromStr};
-use windmill_common::error::{self, Error};
 
 use regex::Regex;
 use reqwest::Client;
@@ -20,6 +19,8 @@ use crate::jobs::{
 };
 
 use windmill_common::{
+    cache,
+    error::{self, Error},
     jobs::JobKind,
     scripts::ScriptHash,
     variables::{build_crypt, decrypt_value_with_mc},
@@ -30,22 +31,30 @@ pub struct SlackFormData {
     payload: String,
 }
 
-#[derive(Deserialize, Debug, Serialize)]
-struct Container {
-    message_ts: String,
-    channel_id: String,
-}
-
 #[derive(Deserialize, Debug)]
 struct Payload {
     actions: Option<Vec<Action>>,
     view: Option<View>,
     trigger_id: Option<String>,
-    #[serde(rename = "type")]
-    r#type: String,
+    r#type: PayloadType,
     container: Option<Container>,
 }
 
+#[derive(Deserialize, Debug)]
+#[serde(rename_all = "snake_case")]
+enum PayloadType {
+    ViewSubmission,
+    ViewClosed,
+    #[serde(other)]
+    Unknown,
+}
+
+#[derive(Deserialize, Debug, Serialize)]
+struct Container {
+    message_ts: String,
+    channel_id: String,
+}
+
 #[derive(Deserialize, Debug)]
 struct View {
     state: Option<State>,
@@ -53,9 +62,12 @@ struct View {
 }
 
 #[derive(Deserialize, Debug)]
-struct Action {
-    value: Option<String>,
-    action_id: String,
+#[serde(tag = "action_id")]
+enum Action {
+    #[serde(rename = "open_modal")]
+    OpenModal { value: String },
+    #[serde(other)]
+    Unknown,
 }
 
 #[derive(Deserialize, Debug)]
@@ -97,15 +109,24 @@ struct Schema {
     properties: HashMap<String, ResumeFormField>,
 }
 
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(rename_all = "lowercase")]
+enum FieldType {
+    Boolean,
+    String,
+    Number,
+    Integer,
+    #[serde(other)]
+    Unknown,
+}
+
 #[derive(Debug, Deserialize, Serialize)]
 struct ResumeFormField {
-    #[serde(rename = "type")]
-    r#type: String,
+    r#type: FieldType,
     format: Option<String>,
     default: Option<serde_json::Value>,
     description: Option<String>,
     title: Option<String>,
-    #[serde(rename = "enum")]
     r#enum: Option<Vec<String>>,
     #[serde(rename = "enumLabels")]
     enum_labels: Option<HashMap<String, String>>,
@@ -119,17 +140,27 @@ pub struct QueryMessage {
 
 #[derive(Deserialize)]
 pub struct QueryResourcePath {
-    slack_resource_path: Option<String>,
+    slack_resource_path: String,
 }
 
 #[derive(Deserialize)]
 pub struct QueryChannelId {
-    channel_id: Option<String>,
+    channel_id: String,
 }
 
 #[derive(Deserialize)]
 pub struct QueryFlowStepId {
-    flow_step_id: Option<String>,
+    flow_step_id: String,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct QueryDefaultArgsJson {
+    default_args_json: Option<serde_json::Value>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct QueryDynamicEnumJson {
+    dynamic_enums_json: Option<serde_json::Value>,
 }
 
 #[derive(Deserialize, Debug)]
@@ -140,6 +171,8 @@ struct ModalActionValue {
     approver: Option<String>,
     message: Option<String>,
     flow_step_id: Option<String>,
+    default_args_json: Option<String>,
+    dynamic_enums_json: Option<String>,
 }
 
 #[derive(Deserialize, Debug)]
@@ -147,6 +180,7 @@ struct PrivateMetadata {
     resume_url: String,
     resource_path: String,
     container: Container,
+    hide_cancel: Option<bool>,
 }
 
 pub async fn slack_app_callback_handler(
@@ -158,27 +192,20 @@ pub async fn slack_app_callback_handler(
     let payload: Payload = serde_json::from_str(&form_data.payload)?;
     tracing::debug!("Payload: {:#?}", payload);
 
-    match payload.r#type.as_str() {
-        "view_submission" => {
-            //print the container
-            handle_submission(authed, db, &payload, "resume").await?
-        }
-        "view_closed" => handle_submission(authed, db, &payload, "cancel").await?,
+    match payload.r#type {
+        PayloadType::ViewSubmission => handle_submission(authed, db, &payload, "resume").await?,
+        PayloadType::ViewClosed => handle_submission(authed, db, &payload, "cancel").await?,
         _ => {
-            if let Some(actions) = payload.actions.as_ref() {
+            if let Some(actions) = &payload.actions {
                 if let Some(action) = actions.first() {
-                    match action.action_id.as_str() {
-                        "open_modal" => {
+                    match action {
+                        Action::OpenModal { value } => {
                             let trigger_id = payload.trigger_id.as_deref().ok_or_else(|| {
-                                Error::BadRequest("No trigger_id found in payload.".to_string())
+                                Error::BadRequest("Missing trigger_id".to_string())
                             })?;
 
-                            let value_str = action.value.as_ref().ok_or_else(|| {
-                                Error::BadRequest("No action value found".to_string())
-                            })?;
-
-                            let parsed_value: ModalActionValue = serde_json::from_str(value_str)
-                                .map_err(|_| {
+                            let parsed_value: ModalActionValue =
+                                serde_json::from_str(value.as_str()).map_err(|_| {
                                     Error::BadRequest("Invalid JSON in action value".to_string())
                                 })?;
 
@@ -195,6 +222,31 @@ pub async fn slack_app_callback_handler(
                                 Error::BadRequest("No container found.".to_string())
                             })?;
 
+                            let default_args_json: Option<serde_json::Value> = parsed_value
+                                .default_args_json
+                                .as_deref()
+                                .map(|s| serde_json::from_str(s))
+                                .transpose()
+                                .map_err(|_| {
+                                    Error::BadRequest(
+                                        "Invalid JSON in default_args_json".to_string(),
+                                    )
+                                })?;
+
+                            let dynamic_enums_json: Option<serde_json::Value> = parsed_value
+                                .dynamic_enums_json
+                                .as_deref()
+                                .map(|s| serde_json::from_str(s))
+                                .transpose()
+                                .map_err(|_| {
+                                    Error::BadRequest(
+                                        "Invalid JSON in dynamic_enums_json".to_string(),
+                                    )
+                                })?;
+
+                            tracing::debug!("Default args json: {:#?}", default_args_json);
+                            tracing::debug!("Dynamic enum json: {:#?}", dynamic_enums_json);
+
                             open_modal_with_blocks(
                                 &client,
                                 slack_token.as_str(),
@@ -207,17 +259,19 @@ pub async fn slack_app_callback_handler(
                                 message,
                                 flow_step_id,
                                 container,
+                                default_args_json.as_ref(),
+                                dynamic_enums_json.as_ref(),
                             )
                             .await
-                            .map_err(|e| {
-                                windmill_common::error::Error::BadRequest(e.to_string())
-                            })?;
+                            .map_err(|e| Error::BadRequest(e.to_string()))?;
                         }
-                        _ => println!("Unknown action_id: {}", action.action_id),
+                        Action::Unknown => println!("Unknown action_id"),
                     }
+                } else {
+                    tracing::debug!("Unknown Slack Action!");
                 }
             } else {
-                tracing::debug!("Unkown Slack Action!");
+                tracing::debug!("Unknown Slack Action!");
             }
         }
     }
@@ -234,48 +288,22 @@ pub async fn request_slack_approval(
     Query(slack_resource_path): Query<QueryResourcePath>,
     Query(channel_id): Query<QueryChannelId>,
     Query(flow_step_id): Query<QueryFlowStepId>,
+    Query(default_args_json): Query<QueryDefaultArgsJson>,
+    Query(dynamic_enums_json): Query<QueryDynamicEnumJson>,
 ) -> Result<StatusCode, Error> {
-    let slack_resource_path = match slack_resource_path.slack_resource_path {
-        Some(path) => path,
-        None => {
-            return Err(windmill_common::error::Error::BadRequest(
-                "slack_resource_path is required".to_string(),
-            ))
-        }
-    };
-
-    let channel_id = match channel_id.channel_id {
-        Some(id) => id,
-        None => {
-            return Err(windmill_common::error::Error::BadRequest(
-                "Slack channel_id is required".to_string(),
-            ))
-        }
-    };
-
-    let flow_step_id = match flow_step_id.flow_step_id {
-        Some(id) => id,
-        None => {
-            return Err(windmill_common::error::Error::BadRequest(
-                "Slack flow_step_id is required".to_string(),
-            ))
-        }
-    };
+    let slack_resource_path = slack_resource_path.slack_resource_path;
+    let channel_id = channel_id.channel_id;
+    let flow_step_id = flow_step_id.flow_step_id;
 
     let slack_token = get_slack_token(&db, slack_resource_path.as_str(), &w_id).await?;
     let client = Client::new();
 
-    // Optional fields
-    let approver_str = approver.approver.as_deref();
-    let message_str = message.message.as_deref();
-
-    tracing::debug!("Approver: {:?}", approver_str);
-    tracing::debug!("Message: {:?}", message_str);
+    tracing::debug!("Approver: {:?}", approver.approver);
+    tracing::debug!("Message: {:?}", message.message);
     tracing::debug!("W ID: {:?}", w_id);
     tracing::debug!("Slack Resource Path: {:?}", slack_resource_path);
     tracing::debug!("Channel ID: {:?}", channel_id);
 
-    // Use approver_str and message_str in the function call
     send_slack_message(
         &client,
         slack_token.as_str(),
@@ -283,12 +311,14 @@ pub async fn request_slack_approval(
         &w_id,
         job_id,
         &slack_resource_path,
-        approver_str,
-        message_str,
+        approver.approver.as_deref(),
+        message.message.as_deref(),
         flow_step_id.as_str(),
+        default_args_json.default_args_json.as_ref(),
+        dynamic_enums_json.dynamic_enums_json.as_ref(),
     )
     .await
-    .map_err(|e| windmill_common::error::Error::BadRequest(e.to_string()))?;
+    .map_err(|e| Error::BadRequest(e.to_string()))?;
 
     Ok(StatusCode::OK)
 }
@@ -325,6 +355,12 @@ async fn handle_submission(
     let resume_url = private_metadata.resume_url;
     let resource_path = private_metadata.resource_path;
     let container: Container = private_metadata.container;
+    let hide_cancel = private_metadata.hide_cancel;
+
+    // If hide_cancel is true, we don't need to extract information from the private_metadata
+    if hide_cancel.unwrap_or(false) && action == "cancel" {
+        return Ok(());
+    }
 
     // Use regex to extract information from private_metadata
     let re = Regex::new(r"/api/w/(?P<w_id>[^/]+)/jobs_u/(?P<action>resume|cancel)/(?P<job_id>[^/]+)/(?P<resume_id>[^/]+)/(?P<secret>[a-fA-F0-9]+)(?:\?approver=(?P<approver>[^&]+))?").unwrap();
@@ -393,6 +429,8 @@ async fn transform_schemas(
     urls: &ResumeUrls,
     order: Option<&Vec<String>>,
     required: Option<&Vec<String>>,
+    default_args_json: Option<&serde_json::Value>,
+    dynamic_enums_json: Option<&serde_json::Value>,
 ) -> Result<serde_json::Value, Error> {
     tracing::debug!("Resume urls: {:#?}", urls);
 
@@ -408,7 +446,12 @@ async fn transform_schemas(
         for key in order.unwrap() {
             if let Some(schema) = properties.get(key) {
                 let is_required = required.unwrap().contains(key);
-                let input_block = create_input_block(key, schema, is_required);
+
+                let default_value = default_args_json.and_then(|json| json.get(key).cloned());
+                let dynamic_enums_value = dynamic_enums_json.and_then(|json| json.get(key).cloned());
+
+                let input_block =
+                    create_input_block(key, schema, is_required, default_value, dynamic_enums_value);
                 match input_block {
                     serde_json::Value::Array(arr) => blocks.extend(arr),
                     _ => blocks.push(input_block),
@@ -420,7 +463,13 @@ async fn transform_schemas(
     Ok(serde_json::Value::Array(blocks))
 }
 
-fn create_input_block(key: &str, schema: &ResumeFormField, required: bool) -> serde_json::Value {
+fn create_input_block(
+    key: &str,
+    schema: &ResumeFormField,
+    required: bool,
+    default_value: Option<serde_json::Value>,
+    dynamic_enums_value: Option<serde_json::Value>,
+) -> serde_json::Value {
     let placeholder = schema
         .description
         .as_deref()
@@ -435,16 +484,20 @@ fn create_input_block(key: &str, schema: &ResumeFormField, required: bool) -> se
     };
 
     // Handle boolean type
-    if schema.r#type == "boolean" {
-        let initial_value = schema
-            .default
+    if let FieldType::Boolean = schema.r#type {
+        let initial_value = default_value
             .as_ref()
-            .and_then(|default| default.as_bool())
+            .and_then(|v| v.as_bool())
+            .or_else(|| {
+                schema
+                    .default
+                    .as_ref()
+                    .and_then(|default| default.as_bool())
+            })
             .unwrap_or(false);
 
         let mut element = serde_json::json!({
             "type": "checkboxes",
-            "optional": !required,
             "options": [{
                 "text": {
                     "type": "plain_text",
@@ -471,6 +524,7 @@ fn create_input_block(key: &str, schema: &ResumeFormField, required: bool) -> se
 
         return serde_json::json!({
             "type": "input",
+            "optional": !required,
             "element": element,
             "label": {
                 "type": "plain_text",
@@ -481,80 +535,102 @@ fn create_input_block(key: &str, schema: &ResumeFormField, required: bool) -> se
     }
 
     // Handle date-time format
-    if schema.r#type == "string" && schema.format.as_deref() == Some("date-time") {
-        let now = chrono::Local::now();
-        let current_date = now.format("%Y-%m-%d").to_string();
-        let current_time = now.format("%H:%M").to_string();
+    if let FieldType::String = schema.r#type {
+        if schema.format.as_deref() == Some("date-time") {  
+            tracing::debug!("Date-time type");
+            let now = chrono::Local::now();
+            let current_date = now.format("%Y-%m-%d").to_string();
+            let current_time = now.format("%H:%M").to_string();
 
-        let (default_date, default_time) = if let Some(default) = &schema.default {
-            if let Ok(parsed_date) = chrono::DateTime::parse_from_rfc3339(default.as_str().unwrap())
-            {
-                (
-                    parsed_date.format("%Y-%m-%d").to_string(),
-                    parsed_date.format("%H:%M").to_string(),
-                )
-            } else {
-                (current_date.clone(), current_time.clone())
-            }
-        } else {
-            (current_date.clone(), current_time.clone())
-        };
+            let (default_date, default_time) = default_value
+                .as_ref()
+                .and_then(|v| v.as_str())
+                .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
+                .map(|parsed_date| {
+                    (
+                        parsed_date.format("%Y-%m-%d").to_string(),
+                        parsed_date.format("%H:%M").to_string(),
+                    )
+                })
+                .or_else(|| {
+                    schema
+                        .default
+                        .as_ref()
+                        .and_then(|default| {
+                            chrono::DateTime::parse_from_rfc3339(default.as_str().unwrap()).ok()
+                        })
+                        .map(|parsed_date| {
+                            (
+                                parsed_date.format("%Y-%m-%d").to_string(),
+                                parsed_date.format("%H:%M").to_string(),
+                            )
+                        })
+                })
+                .unwrap_or((current_date.clone(), current_time.clone()));
 
-        return serde_json::json!([
-            {
-                "type": "input",
-                "optional": !required,
-                "element": {
-                    "type": "datepicker",
-                    "initial_date": &default_date,
-                    "placeholder": {
-                        "type": "plain_text",
-                        "text": "Select a date",
-                        "emoji": true
+            return serde_json::json!([
+                {
+                    "type": "input",
+                    "optional": !required,
+                    "element": {
+                        "type": "datepicker",
+                        "initial_date": &default_date,
+                        "placeholder": {
+                            "type": "plain_text",
+                            "text": "Select a date",
+                            "emoji": true
+                        },
+                        "action_id": format!("{}_date", key)
                     },
-                    "action_id": format!("{}_date", key)
-                },
-                "label": {
-                    "type": "plain_text",
-                    "text": title_with_required,
-                    "emoji": true
-                }
-            },
-            {
-                "type": "input",
-                "optional": !required,
-                "element": {
-                    "type": "timepicker",
-                    "initial_time": &default_time,
-                    "placeholder": {
+                    "label": {
                         "type": "plain_text",
-                        "text": "Select time",
+                        "text": title_with_required,
                         "emoji": true
-                    },
-                    "action_id": format!("{}_time", key)
+                    }
                 },
-                "label": {
-                    "type": "plain_text",
-                    "text": " ",
-                    "emoji": true
+                {
+                    "type": "input",
+                    "optional": !required,
+                    "element": {
+                        "type": "timepicker",
+                        "initial_time": &default_time,
+                        "placeholder": {
+                            "type": "plain_text",
+                            "text": "Select time",
+                            "emoji": true
+                        },
+                        "action_id": format!("{}_time", key)
+                    },
+                    "label": {
+                        "type": "plain_text",
+                        "text": " ",
+                        "emoji": true
+                    }
                 }
-            }
-        ]);
+            ]);
+        }
     }
 
     // Handle enum type
     if let Some(enums) = &schema.r#enum {
+        tracing::debug!("Enum type");
+        let enums = dynamic_enums_value
+            .as_ref()
+            .and_then(|v| v.as_array())
+            .cloned()
+            .unwrap_or_else(|| enums.iter().map(|s| serde_json::json!(s)).collect());
+
         let initial_option = schema.default.as_ref().and_then(|default_value| {
             enums
                 .iter()
-                .find(|enum_value| enum_value == &default_value)
+                .find(|enum_value| enum_value == &&serde_json::json!(default_value))
                 .map(|enum_value| {
                     serde_json::json!({
                         "text": {
                             "type": "plain_text",
                             "text": schema.enum_labels.as_ref()
-                                .and_then(|labels| labels.get(enum_value))
-                                .unwrap_or(enum_value),
+                                .and_then(|labels| labels.get(enum_value.as_str().unwrap()))
+                                .unwrap_or(&enum_value.as_str().unwrap().to_string()),
                             "emoji": true
                         },
                         "value": enum_value
@@ -574,8 +650,8 @@ fn create_input_block(key: &str, schema: &ResumeFormField, required: bool) -> se
                     "text": {
                         "type": "plain_text",
                         "text": schema.enum_labels.as_ref()
-                            .and_then(|labels| labels.get(enum_value))
-                            .unwrap_or(enum_value),
+                            .and_then(|labels| labels.get(enum_value.as_str().unwrap()))
+                            .unwrap_or(&enum_value.as_str().unwrap().to_string()),
                         "emoji": true
                     },
                     "value": enum_value
@@ -598,15 +674,16 @@ fn create_input_block(key: &str, schema: &ResumeFormField, required: bool) -> se
                 "emoji": true
             }
         })
-    } else if schema.r#type == "number" || schema.r#type == "integer" {
+    } else if let FieldType::Number | FieldType::Integer = schema.r#type {
+        tracing::debug!("Number or integer type");
         // Handle number and integer types
-        let initial_value = schema
-            .default
+        let initial_value = default_value
             .as_ref()
-            .and_then(|default| default.as_f64())
+            .and_then(|v| v.as_f64())
+            .or_else(|| schema.default.as_ref().and_then(|default| default.as_f64()))
             .unwrap_or(0.0);
 
-        let action_id_suffix = if schema.r#type == "number" {
+        let action_id_suffix = if let FieldType::Number = schema.r#type {
             "_type_number"
         } else {
             "_type_integer"
@@ -627,11 +704,12 @@ fn create_input_block(key: &str, schema: &ResumeFormField, required: bool) -> se
             }
         })
     } else {
+        tracing::debug!("Other type");
         // Handle other types as string
-        let initial_value = schema
-            .default
+        let initial_value = default_value
             .as_ref()
-            .and_then(|default| default.as_str())
+            .and_then(|v| v.as_str())
+            .or_else(|| schema.default.as_ref().and_then(|default| default.as_str()))
             .unwrap_or("");
 
         serde_json::json!({
@@ -640,7 +718,7 @@ fn create_input_block(key: &str, schema: &ResumeFormField, required: bool) -> se
             "element": {
                 "type": "plain_text_input",
                 "action_id": key,
-                "initial_value": initial_value
+                "initial_value": initial_value.to_string()
             },
             "label": {
                 "type": "plain_text",
@@ -787,6 +865,8 @@ async fn send_slack_message(
     approver: Option<&str>,
     message: Option<&str>,
     flow_step_id: &str,
+    default_args_json: Option<&serde_json::Value>,
+    dynamic_enums_json: Option<&serde_json::Value>,
 ) -> Result<StatusCode, Box<dyn std::error::Error>> {
     let url = "https://slack.com/api/chat.postMessage";
 
@@ -806,6 +886,14 @@ async fn send_slack_message(
         value["message"] = serde_json::json!(message);
     }
 
+    if let Some(default_args_json) = default_args_json {
+        value["default_args_json"] = default_args_json.clone();
+    }
+
+    if let Some(dynamic_enums_json) = dynamic_enums_json {
+        value["dynamic_enums_json"] = dynamic_enums_json.clone();
+    }
+
     let payload = serde_json::json!({
         "channel": channel_id,
         "text": "A flow has been suspended. Please approve or reject the flow.",
@@ -867,7 +955,9 @@ async fn get_modal_blocks(
     flow_step_id: Option<&str>,
     resource_path: &str,
     container: Container,
-) -> Result<axum::Json<serde_json::Value>, windmill_common::error::Error> {
+    default_args_json: Option<&serde_json::Value>,
+    dynamic_enums_json: Option<&serde_json::Value>,
+) -> Result<axum::Json<serde_json::Value>, Error> {
     let res = get_resume_urls_internal(
         axum::Extension(db.clone()),
         Path((w_id.to_string(), job_id, resume_id)),
@@ -879,12 +969,16 @@ async fn get_modal_blocks(
 
     tracing::debug!("Job ID: {:?}", job_id);
 
-    let (job_kind, script_hash, raw_flow, parent_job_id) = sqlx::query!(
+    let (job_kind, script_hash, raw_flow, parent_job_id, created_at, created_by, script_path, args) = sqlx::query!(
         "SELECT
             queue.job_kind AS \"job_kind: JobKind\",
             queue.script_hash AS \"script_hash: ScriptHash\",
             queue.raw_flow AS \"raw_flow: sqlx::types::Json<Box<RawValue>>\",
-            completed_job.parent_job AS \"parent_job: Uuid\"
+            completed_job.parent_job AS \"parent_job: Uuid\",
+            completed_job.created_at AS \"created_at: chrono::NaiveDateTime\",
+            completed_job.created_by AS \"created_by!\",
+            queue.script_path,
+            queue.args AS \"args: sqlx::types::Json<Box<RawValue>>\"
         FROM queue
         JOIN completed_job ON completed_job.parent_job = queue.id
         WHERE completed_job.id = $1 AND completed_job.workspace_id = $2
@@ -896,15 +990,13 @@ async fn get_modal_blocks(
     .await
     .map_err(|e| error::Error::BadRequest(e.to_string()))?
     .ok_or_else(|| error::Error::BadRequest("This workflow is no longer running and has either already timed out or been cancelled or completed.".to_string()))
-    .map(|r| (r.job_kind, r.script_hash, r.raw_flow, r.parent_job))?;
+    .map(|r| (r.job_kind, r.script_hash, r.raw_flow, r.parent_job, r.created_at, r.created_by, r.script_path, r.args))?;
 
-    let flow_data = match windmill_common::cache::job::fetch_flow(&db, job_kind, script_hash).await
-    {
+    let flow_data = match cache::job::fetch_flow(&db, job_kind, script_hash).await {
         Ok(data) => data,
         Err(_) => {
             if let Some(parent_job_id) = parent_job_id.as_ref() {
-                windmill_common::cache::job::fetch_preview_flow(&db, parent_job_id, raw_flow)
-                    .await?
+                cache::job::fetch_preview_flow(&db, parent_job_id, raw_flow).await?
             } else {
                 return Err(error::Error::BadRequest(
                     "This workflow is no longer running and has either already timed out or been cancelled or completed.".to_string(),
@@ -918,6 +1010,7 @@ async fn get_modal_blocks(
     let module = flow_value.modules.iter().find(|m| m.id == flow_step_id);
 
     tracing::debug!("Module: {:#?}", module);
+
     let schema = module.and_then(|module| {
         module.suspend.as_ref().map(|suspend| ResumeFormRow {
             resume_form: suspend.resume_form.clone(),
@@ -925,14 +1018,39 @@ async fn get_modal_blocks(
         })
     });
 
-    let message_str =
-        message.unwrap_or("*A workflow has been suspended and is waiting for approval:*\n");
+    let args_str = args.map_or("None".to_string(), |a| a.get().to_string());
+    let parent_job_id_str = parent_job_id.map_or("None".to_string(), |id| id.to_string());
+    let script_path_str = script_path.as_deref().unwrap_or("None");
+
+    let created_at_formatted = created_at.format("%Y-%m-%d %H:%M:%S").to_string();
+
+    let mut message_str = format!(
+        "A workflow has been suspended and is waiting for approval:\n\n\
+        *Created by*: {created_by}\n\
+        *Created at*: {created_at_formatted}\n\
+        *Script path*: {script_path_str}\n\
+        *Args*: {args_str}\n\
+        *Flow ID*: {parent_job_id_str}\n\n"
+    );
+
+    // Append custom message if provided
+    if let Some(msg) = message {
+        message_str.push_str(msg);
+    }
 
     tracing::debug!("Schema: {:#?}", schema);
 
     if let Some(resume_schema) = schema {
         let hide_cancel = resume_schema.hide_cancel.unwrap_or(false);
 
+        // if hide cancel is false add note to message
+        if !hide_cancel {
+            message_str.push_str("\n\n*NOTE*: closing this modal will cancel the workflow.\n\n");
+        }
+
+        // Convert message_str back to &str when needed
+        let message_str_ref: &str = &message_str;
+
         if let Some(schema_obj) = resume_schema.resume_form {
             let inner_schema: ResumeSchema =
                 serde_json::from_value(schema_obj.clone()).map_err(|e| {
@@ -944,11 +1062,13 @@ async fn get_modal_blocks(
                 })?;
 
             let blocks = transform_schemas(
-                message_str,
+                message_str_ref,
                 Some(&inner_schema.schema.properties),
                 &urls,
                 Some(&inner_schema.schema.order),
                 Some(&inner_schema.schema.required),
+                default_args_json,
+                dynamic_enums_json,
             )
             .await?;
 
@@ -963,7 +1083,16 @@ async fn get_modal_blocks(
             )));
         } else {
             tracing::debug!("No suspend form found!");
-            let blocks = transform_schemas(message_str, None, &urls, None, None).await?;
+            let blocks = transform_schemas(
+                message_str_ref,
+                None,
+                &urls,
+                None,
+                None,
+                default_args_json,
+                dynamic_enums_json,
+            )
+            .await?;
             return Ok(axum::Json(construct_payload(
                 blocks,
                 hide_cancel,
@@ -1001,7 +1130,7 @@ fn construct_payload(
             "type": "plain_text",
             "text": "Resume Workflow"
         },
-        "private_metadata": serde_json::json!({ "resume_url": resume_url, "resource_path": resource_path, "container": container }).to_string(),
+        "private_metadata": serde_json::json!({ "resume_url": resume_url, "resource_path": resource_path, "container": container, "hide_cancel": hide_cancel }).to_string(),
     });
 
     if !hide_cancel {
@@ -1029,6 +1158,8 @@ async fn open_modal_with_blocks(
     message: Option<&str>,
     flow_step_id: Option<&str>,
     container: Container,
+    default_args_json: Option<&serde_json::Value>,
+    dynamic_enums_json: Option<&serde_json::Value>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     let resume_id = rand::random::<u32>();
     let blocks_json = match get_modal_blocks(
@@ -1042,6 +1173,8 @@ async fn open_modal_with_blocks(
         flow_step_id,
         resource_path,
         container,
+        default_args_json,
+        dynamic_enums_json,
     )
     .await
     {

backend/windmill-api/src/triggers.rs

@@ -19,6 +19,7 @@ pub struct TriggersCount {
     email_count: i64,
     websocket_count: i64,
     kafka_count: i64,
+    nats_count: i64,
 }
 pub(crate) async fn get_triggers_count_internal(
     db: &DB,
@@ -75,6 +76,16 @@ pub(crate) async fn get_triggers_count_internal(
     .await?
     .unwrap_or(0);
 
+    let nats_count = sqlx::query_scalar!(
+        "SELECT COUNT(*) FROM nats_trigger WHERE script_path = $1 AND is_flow = $2 AND workspace_id = $3",
+        path,
+        is_flow,
+        w_id
+    )
+    .fetch_one(db)
+    .await?
+    .unwrap_or(0);
+
     let webhook_count = (if is_flow {
         sqlx::query_scalar!(
             "SELECT COUNT(*) FROM token WHERE label LIKE 'webhook-%' AND workspace_id = $1 AND scopes @> ARRAY['run:flow/' || $2]::text[]",
@@ -117,6 +128,7 @@ pub(crate) async fn get_triggers_count_internal(
         email_count,
         websocket_count,
         kafka_count,
+        nats_count,
     }))
 }
 

backend/windmill-api/src/workspaces.rs

@@ -106,6 +106,7 @@ pub fn workspaced_service() -> Router {
         .route("/leave", post(leave_workspace))
         .route("/get_workspace_name", get(get_workspace_name))
         .route("/change_workspace_name", post(change_workspace_name))
+        .route("/change_workspace_color", post(change_workspace_color))
         .route(
             "/change_workspace_id",
             post(crate::workspaces_extra::change_workspace_id),
@@ -158,6 +159,7 @@ struct Workspace {
     owner: String,
     deleted: bool,
     premium: bool,
+    color: Option<String>,
 }
 
 #[derive(FromRow, Serialize, Debug)]
@@ -186,6 +188,7 @@ pub struct WorkspaceSettings {
     pub automatic_billing: bool,
     pub default_scripts: Option<serde_json::Value>,
     pub mute_critical_alerts: Option<bool>,
+    pub color: Option<String>,
 }
 
 #[derive(FromRow, Serialize, Debug)]
@@ -263,6 +266,7 @@ struct CreateWorkspace {
     id: String,
     name: String,
     username: Option<String>,
+    color: Option<String>,
 }
 
 #[derive(Deserialize)]
@@ -282,6 +286,7 @@ struct UserWorkspace {
     pub id: String,
     pub name: String,
     pub username: String,
+    pub color: Option<String>,
 }
 
 #[derive(Deserialize)]
@@ -376,8 +381,11 @@ async fn list_workspaces(
     let mut tx = user_db.begin(&authed).await?;
     let workspaces = sqlx::query_as!(
         Workspace,
-        "SELECT workspace.* FROM workspace, usr WHERE usr.workspace_id = workspace.id AND \
-         usr.email = $1 AND deleted = false",
+        "SELECT workspace.id, workspace.name, workspace.owner, workspace.deleted, workspace.premium, workspace_settings.color
+         FROM workspace
+         LEFT JOIN workspace_settings ON workspace.id = workspace_settings.workspace_id
+         JOIN usr ON usr.workspace_id = workspace.id
+         WHERE usr.email = $1 AND workspace.deleted = false",
         authed.email
     )
     .fetch_all(&mut *tx)
@@ -1295,6 +1303,7 @@ struct UsedTriggers {
     pub websocket_used: bool,
     pub http_routes_used: bool,
     pub kafka_used: bool,
+    pub nats_used: bool,
 }
 
 async fn get_used_triggers(
@@ -1308,7 +1317,8 @@ async fn get_used_triggers(
         r#"SELECT 
             EXISTS(SELECT 1 FROM websocket_trigger WHERE workspace_id = $1) as "websocket_used!", 
             EXISTS(SELECT 1 FROM http_trigger WHERE workspace_id = $1) as "http_routes_used!",
-            EXISTS(SELECT 1 FROM kafka_trigger WHERE workspace_id = $1) as "kafka_used!""#,
+            EXISTS(SELECT 1 FROM kafka_trigger WHERE workspace_id = $1) as "kafka_used!",
+            EXISTS(SELECT 1 FROM nats_trigger WHERE workspace_id = $1) as "nats_used!""#,
         w_id,
     )
     .fetch_one(&mut *tx)
@@ -1331,7 +1341,10 @@ async fn list_workspaces_as_super_admin(
     let mut tx = user_db.begin(&authed).await?;
     let workspaces = sqlx::query_as!(
         Workspace,
-        "SELECT * FROM workspace LIMIT $1 OFFSET $2",
+        "SELECT workspace.id, workspace.name, workspace.owner, workspace.deleted, workspace.premium, workspace_settings.color
+         FROM workspace
+         LEFT JOIN workspace_settings ON workspace.id = workspace_settings.workspace_id
+         LIMIT $1 OFFSET $2",
         per_page as i32,
         offset as i32
     )
@@ -1348,9 +1361,11 @@ async fn user_workspaces(
     let mut tx = db.begin().await?;
     let workspaces = sqlx::query_as!(
         UserWorkspace,
-        "SELECT workspace.id, workspace.name, usr.username
-     FROM workspace, usr WHERE usr.workspace_id = workspace.id AND usr.email = $1 AND deleted = \
-         false",
+        "SELECT workspace.id, workspace.name, usr.username, workspace_settings.color
+         FROM workspace
+         JOIN usr ON usr.workspace_id = workspace.id
+         JOIN workspace_settings ON workspace_settings.workspace_id = workspace.id
+         WHERE usr.email = $1 AND workspace.deleted = false",
         email
     )
     .fetch_all(&mut *tx)
@@ -1430,9 +1445,10 @@ async fn create_workspace(
     .await?;
     sqlx::query!(
         "INSERT INTO workspace_settings
-            (workspace_id)
-            VALUES ($1)",
-        nw.id
+            (workspace_id, color)
+            VALUES ($1, $2)",
+        nw.id,
+        nw.color,
     )
     .execute(&mut *tx)
     .await?;
@@ -1943,6 +1959,11 @@ struct ChangeWorkspaceName {
     new_name: String,
 }
 
+#[derive(Deserialize)]
+struct ChangeWorkspaceColor {
+    color: Option<String>,
+}
+
 async fn change_workspace_name(
     authed: ApiAuthed,
     Path(w_id): Path<String>,
@@ -1977,6 +1998,32 @@ async fn change_workspace_name(
     Ok(format!("updated workspace name to {}", &rw.new_name))
 }
 
+async fn change_workspace_color(
+    authed: ApiAuthed,
+    Path(w_id): Path<String>,
+    Extension(db): Extension<DB>,
+    Json(rw): Json<ChangeWorkspaceColor>,
+) -> Result<String> {
+    require_admin(authed.is_admin, &authed.username)?;
+
+    let mut tx = db.begin().await?;
+
+    sqlx::query!(
+        "UPDATE workspace_settings SET color = $1 WHERE workspace_id = $2",
+        rw.color,
+        &w_id
+    )
+    .execute(&mut *tx)
+    .await?;
+
+    tx.commit().await?;
+
+    Ok(format!(
+        "updated workspace color to {}",
+        rw.color.as_deref().unwrap_or("no color")
+    ))
+}
+
 async fn get_usage(Extension(db): Extension<DB>, Path(w_id): Path<String>) -> Result<String> {
     let usage = sqlx::query_scalar!(
         "

backend/windmill-api/src/workspaces_extra.rs

@@ -103,6 +103,14 @@ pub(crate) async fn change_workspace_id(
     .execute(&mut *tx)
     .await?;
 
+    sqlx::query!(
+        "UPDATE capture_config SET workspace_id = $1 WHERE workspace_id = $2",
+        &rw.new_id,
+        &old_id
+    )
+    .execute(&mut *tx)
+    .await?;
+
     sqlx::query!(
         "UPDATE completed_job SET workspace_id = $1 WHERE workspace_id = $2",
         &rw.new_id,
@@ -396,6 +404,9 @@ pub(crate) async fn delete_workspace(
     sqlx::query!("DELETE FROM capture WHERE workspace_id = $1", &w_id)
         .execute(&mut *tx)
         .await?;
+    sqlx::query!("DELETE FROM capture_config WHERE workspace_id = $1", &w_id)
+        .execute(&mut *tx)
+        .await?;
     sqlx::query!("DELETE FROM draft WHERE workspace_id = $1", &w_id)
         .execute(&mut *tx)
         .await?;