feat: nats triggers (#5039)

* feat: nats triggers * missing files * fix build * update feature flags * event stream fixes * update ee ref
2026-08-18 16:02:10 +00:00 · 2025-01-10 19:47:17 +01:00
parent 03450bd6f4
commit e66fcf9270
80 changed files with 2512 additions and 235 deletions
@@ -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 }}
@@ -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: |
@@ -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 }}
@@ -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: |
@@ -17,7 +17,8 @@
                "http",
                "websocket",
                "kafka",
-                "email"
+                "email",
+                "nats"
              ]
            }
          }
@@ -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"
 }
@@ -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"
 }
@@ -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"
 }
@@ -25,7 +25,8 @@
                "http",
                "websocket",
                "kafka",
-                "email"
+                "email",
+                "nats"
              ]
            }
          }
@@ -56,7 +57,8 @@
                "http",
                "websocket",
                "kafka",
-                "email"
+                "email",
+                "nats"
              ]
            }
          }
@@ -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"
+}
@@ -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"
+}
@@ -1,6 +1,6 @@
 {
  "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 (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')",
+  "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": [
      {
@@ -46,5 +46,5 @@
      false
    ]
  },
-  "hash": "4f1bb3713bc52fb8cd4088947de1e9905387a20bf4b137adf0a33274f820ddc2"
+  "hash": "47560aa3d3af93167663e763ad873088eb4f57785df3c383bbf6e5ceab268980"
 }
@@ -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"
 }
@@ -28,7 +28,8 @@
                "http",
                "websocket",
                "kafka",
-                "email"
+                "email",
+                "nats"
              ]
            }
          }
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -20,7 +20,8 @@
                "http",
                "websocket",
                "kafka",
-                "email"
+                "email",
+                "nats"
              ]
            }
          }
@@ -17,7 +17,8 @@
                "http",
                "websocket",
                "kafka",
-                "email"
+                "email",
+                "nats"
              ]
            }
          }
@@ -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"
+}
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "UPDATE capture_config SET server_id = $1, last_server_ping = now() WHERE last_client_ping > NOW() - INTERVAL '10 seconds'  AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'kafka' AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
+  "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": [
      {
@@ -21,5 +21,5 @@
      null
    ]
  },
-  "hash": "ee9adcbf82d3f62088a38ff65e8c90ac1c18b5df7aab6a143a328a6bddc6ad32"
+  "hash": "d06efdc24706e0d7479bffc0b19a0c5976ee60125289e9f7b0b04090bce4a3a3"
 }
@@ -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"
-}
@@ -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"
 }
@@ -1,6 +1,6 @@
 {
  "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 (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')",
+  "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": [
      {
@@ -46,5 +46,5 @@
      false
    ]
  },
-  "hash": "438b8bbf49781fc362eecbdc99cb9d2fafd03a59197477aafc915b0e34a00eb2"
+  "hash": "d8ea17fba0e417333e9c6cf82ad36a57830a791c8358e30e9be012b815f5e8e3"
 }
@@ -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"
+}
@@ -33,7 +33,8 @@
                "http",
                "websocket",
                "kafka",
-                "email"
+                "email",
+                "nats"
              ]
            }
          }
@@ -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"
+}
@@ -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"
 }
@@ -17,7 +17,8 @@
                "http",
                "websocket",
                "kafka",
-                "email"
+                "email",
+                "nats"
              ]
            }
          }
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "UPDATE capture_config SET server_id = $1, last_server_ping = now() WHERE last_client_ping > NOW() - INTERVAL '10 seconds' AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'websocket' AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
+  "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": [
      {
@@ -21,5 +21,5 @@
      null
    ]
  },
-  "hash": "e86295e181a82823ffce8234d413ab5a528b0468715238e17ffed7d75e9c0c5c"
+  "hash": "f16b00bad2880f896f4452e5894fe101cd5c2fe3e4c143c0b979668143d85dd2"
 }
@@ -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"
+}
@@ -474,6 +474,42 @@ dependencies = [
 "zstd-safe",
 ]

+[[package]]
+name = "async-nats"
+version = "0.38.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76433c4de73442daedb3a59e991d94e85c14ebfc33db53dfcd347a21cd6ef4f8"
+dependencies = [
+ "base64 0.22.1",
+ "bytes",
+ "futures",
+ "memchr",
+ "nkeys",
+ "nuid",
+ "once_cell",
+ "pin-project",
+ "portable-atomic",
+ "rand 0.8.5",
+ "regex",
+ "ring 0.17.8",
+ "rustls-native-certs 0.7.3",
+ "rustls-pemfile 2.2.0",
+ "rustls-webpki 0.102.8",
+ "serde",
+ "serde_json",
+ "serde_nanos",
+ "serde_repr",
+ "thiserror 1.0.69",
+ "time",
+ "tokio",
+ "tokio-rustls 0.26.1",
+ "tokio-util",
+ "tokio-websockets",
+ "tracing",
+ "tryhard",
+ "url",
+]
+
 [[package]]
 name = "async-oauth2"
 version = "0.5.0"
@@ -1429,6 +1465,9 @@ name = "bytes"
 version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "325918d6fe32f23b19878fe4b34794ae41fc19ddbe53b10571a4874d44ffd39b"
+dependencies = [
+ "serde",
+]

 [[package]]
 name = "bytes-utils"
@@ -3069,6 +3108,7 @@ dependencies = [
 "ed25519",
 "serde",
 "sha2 0.10.8",
+ "signature",
 "subtle",
 "zeroize",
 ]
@@ -5363,6 +5403,21 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "nkeys"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9f49e787f4c61cbd0f9320b31cc26e58719f6aa5068e34697dd3aea361412fe3"
+dependencies = [
+ "data-encoding",
+ "ed25519",
+ "ed25519-dalek",
+ "getrandom 0.2.15",
+ "log",
+ "rand 0.8.5",
+ "signatory",
+]
+
 [[package]]
 name = "nom"
 version = "7.1.3"
@@ -5383,6 +5438,15 @@ dependencies = [
 "winapi",
 ]

+[[package]]
+name = "nuid"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc895af95856f929163a0aa20c26a78d26bfdc839f51b9d5aa7a5b79e52b7e83"
+dependencies = [
+ "rand 0.8.5",
+]
+
 [[package]]
 name = "num"
 version = "0.4.3"
@@ -7657,6 +7721,15 @@ dependencies = [
 "serde",
 ]

+[[package]]
+name = "serde_nanos"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a93142f0367a4cc53ae0fead1bcda39e85beccfad3dcd717656cacab94b12985"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "serde_path_to_error"
 version = "0.1.16"
@@ -7866,6 +7939,18 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "signatory"
+version = "0.27.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c1e303f8205714074f6068773f0e29527e0453937fe837c9717d066635b65f31"
+dependencies = [
+ "pkcs8",
+ "rand_core 0.6.4",
+ "signature",
+ "zeroize",
+]
+
 [[package]]
 name = "signature"
 version = "2.2.0"
@@ -9503,6 +9588,27 @@ dependencies = [
 "tokio",
 ]

+[[package]]
+name = "tokio-websockets"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f591660438b3038dd04d16c938271c79e7e06260ad2ea2885a4861bfb238605d"
+dependencies = [
+ "base64 0.22.1",
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "http 1.2.0",
+ "httparse",
+ "rand 0.8.5",
+ "ring 0.17.8",
+ "rustls-native-certs 0.8.1",
+ "rustls-pki-types",
+ "tokio",
+ "tokio-rustls 0.26.1",
+ "tokio-util",
+]
+
 [[package]]
 name = "toml"
 version = "0.7.8"
@@ -9895,6 +10001,17 @@ version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"

+[[package]]
+name = "tryhard"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9c9f0a709784e86923586cff0d872dba54cd2d2e116b3bc57587d15737cfce9d"
+dependencies = [
+ "futures",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tungstenite"
 version = "0.24.0"
@@ -10638,6 +10755,7 @@ version = "1.444.0"
 dependencies = [
 "anyhow",
 "argon2",
+ "async-nats",
 "async-oauth2",
 "async-recursion",
 "async-stream",
@@ -10671,6 +10789,7 @@ dependencies = [
 "matchit",
 "mime_guess",
 "native-tls",
+ "nkeys",
 "object_store",
 "openidconnect",
 "openssl",
@@ -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"]
@@ -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"] }
@@ -1 +1 @@
-e9bb21fc9d651e4302f9c3b7b3e8f35e00679e4b
+9cea7e44f1b6401662e22088b41e8a00d7e67b23
@@ -0,0 +1 @@
+DROP TABLE nats_trigger;
@@ -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));
@@ -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
@@ -2379,10 +2379,13 @@ paths:
                    type: boolean
                  kafka_used:
                    type: boolean
+                  nats_used:
+                    type: boolean
                required:
                  - http_routes_used
                  - websocket_used
                  - kafka_used
+                  - nats_used

  /w/{workspace}/users/list:
    get:
@@ -8018,6 +8021,169 @@ paths:
              schema:
                type: string

+  /w/{workspace}/nats_triggers/create:
+    post:
+      summary: create nats trigger
+      operationId: createNatsTrigger
+      tags:
+        - nats_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      requestBody:
+        description: new nats trigger
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/NewNatsTrigger"
+      responses:
+        "201":
+          description: nats trigger created
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /w/{workspace}/nats_triggers/update/{path}:
+    post:
+      summary: update nats trigger
+      operationId: updateNatsTrigger
+      tags:
+        - nats_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      requestBody:
+        description: updated trigger
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/EditNatsTrigger"
+      responses:
+        "200":
+          description: nats trigger updated
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /w/{workspace}/nats_triggers/delete/{path}:
+    delete:
+      summary: delete nats trigger
+      operationId: deleteNatsTrigger
+      tags:
+        -  nats_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      responses:
+        "200":
+          description: nats trigger deleted
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /w/{workspace}/nats_triggers/get/{path}:
+    get:
+      summary: get nats trigger
+      operationId: getNatsTrigger
+      tags:
+        - nats_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      responses:
+        "200":
+          description: nats trigger deleted
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/NatsTrigger"
+
+  
+  /w/{workspace}/nats_triggers/list:
+    get:
+      summary: list nats triggers
+      operationId: listNatsTriggers
+      tags:
+        - nats_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+          required: true
+        - $ref: "#/components/parameters/Page"
+        - $ref: "#/components/parameters/PerPage"
+        - name: path
+          description: filter by path
+          in: query
+          schema:
+            type: string
+        - name: is_flow
+          in: query
+          schema:
+            type: boolean
+        - name: path_start
+          in: query
+          schema:
+            type: string
+      responses:
+        "200":
+          description: nats trigger list
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: "#/components/schemas/NatsTrigger"
+
+  
+  /w/{workspace}/nats_triggers/exists/{path}:
+    get:
+      summary: does nats trigger exists
+      operationId: existsNatsTrigger
+      tags:
+        - nats_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      responses:
+        "200":
+          description: nats trigger exists
+          content:
+            application/json:
+              schema:
+                type: boolean
+
+  /w/{workspace}/nats_triggers/setenabled/{path}:
+    post:
+      summary: set enabled nats trigger
+      operationId: setNatsTriggerEnabled
+      tags:
+        - nats_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      requestBody:
+        description: updated nats trigger enable
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                enabled:
+                  type: boolean
+              required:
+                - enabled
+      responses:
+        "200":
+          description: nats trigger enabled set
+          content:
+            text/plain:
+              schema:
+                type: string
+

  /groups/list:
    get:
@@ -8888,6 +9054,7 @@ paths:
                http_trigger,
                websocket_trigger,
                kafka_trigger,
+                nats_trigger,
              ]
      responses:
        "200":
@@ -8927,6 +9094,7 @@ paths:
                http_trigger,
                websocket_trigger,
                kafka_trigger,
+                nats_trigger,
              ]
      requestBody:
        description: acl to add
@@ -8977,6 +9145,7 @@ paths:
                http_trigger,
                websocket_trigger,
                kafka_trigger,
+                nats_trigger,
              ]
      requestBody:
        description: acl to add
@@ -12392,6 +12561,8 @@ components:
          type: number
        kafka_count:
          type: number
+        nats_count:
+          type: number

    WebsocketTrigger:
      type: object
@@ -12665,6 +12836,126 @@ components:
        - topics
        - is_flow

+    NatsTrigger:
+      type: object
+      properties:
+        path:
+          type: string
+        edited_by:
+          type: string
+        edited_at:
+          type: string
+          format: date-time
+        script_path:
+          type: string
+        nats_resource_path:
+          type: string
+        use_jetstream:
+          type: boolean
+        stream_name:
+          type: string
+        consumer_name:
+          type: string
+        subjects:
+          type: array
+          items:
+            type: string
+        is_flow:
+          type: boolean
+        extra_perms:
+          type: object
+          additionalProperties:
+            type: boolean
+        email:
+          type: string
+        workspace_id:
+          type: string
+        server_id:
+          type: string
+        last_server_ping:
+          type: string
+          format: date-time
+        error:
+          type: string
+        enabled:
+          type: boolean
+        
+      required:
+        - path
+        - edited_by
+        - edited_at
+        - script_path
+        - nats_resource_path
+        - use_jetstream
+        - subjects
+        - extra_perms
+        - is_flow
+        - email
+        - workspace_id
+        - enabled
+
+    NewNatsTrigger:
+      type: object
+      properties:
+        path:
+          type: string
+        script_path:
+          type: string
+        is_flow:
+          type: boolean
+        nats_resource_path:
+          type: string
+        use_jetstream:
+          type: boolean
+        stream_name:
+          type: string
+        consumer_name:
+          type: string
+        subjects:
+          type: array
+          items:
+            type: string
+        enabled:
+          type: boolean
+        
+      required:
+        - path
+        - script_path
+        - is_flow
+        - nats_resource_path
+        - use_jetstream
+        - subjects
+
+    EditNatsTrigger:
+      type: object
+      properties:
+        nats_resource_path:
+          type: string
+        use_jetstream:
+          type: boolean
+        stream_name:
+          type: string
+        consumer_name:
+          type: string
+        subjects:
+          type: array
+          items:
+            type: string
+        path:
+          type: string
+        script_path:
+          type: string
+        is_flow:
+          type: boolean
+              
+      required:
+        - path
+        - script_path
+        - nats_resource_path
+        - use_jetstream
+        - subjects
+        - is_flow
+
    Group:
      type: object
      properties:
@@ -13634,7 +13925,7 @@ components:

    CaptureTriggerKind:
      type: string
-      enum: [webhook, http, websocket, kafka, email]
+      enum: [webhook, http, websocket, kafka, email, nats]

    Capture:
      type: object
@@ -36,6 +36,8 @@ use windmill_queue::{PushArgs, PushArgsOwned};
 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},
@@ -84,6 +86,7 @@ pub enum TriggerKind {
    Websocket,
    Kafka,
    Email,
+    Nats,
 }

 impl fmt::Display for TriggerKind {
@@ -94,6 +97,7 @@ impl fmt::Display for TriggerKind {
            TriggerKind::Websocket => "websocket",
            TriggerKind::Kafka => "kafka",
            TriggerKind::Email => "email",
+            TriggerKind::Nats => "nats",
        };
        write!(f, "{}", s)
    }
@@ -123,6 +127,27 @@ pub struct KafkaTriggerConfig {
    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,
@@ -138,6 +163,8 @@ enum TriggerConfig {
    Websocket(WebsocketTriggerConfig),
    #[cfg(all(feature = "enterprise", feature = "kafka"))]
    Kafka(KafkaTriggerConfig),
+    #[cfg(all(feature = "enterprise", feature = "nats"))]
+    Nats(NatsTriggerConfig),
 }

 #[derive(Serialize, Deserialize)]
@@ -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 {
@@ -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;

@@ -274,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")]
        {
@@ -286,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
@@ -354,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(
@@ -414,7 +435,10 @@ 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()),
@@ -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
+}
@@ -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,
    }))
 }

@@ -252,7 +252,7 @@ async fn update_websocket_trigger(

    // important to update server_id, last_server_ping and error to NULL to stop current websocket listener
    sqlx::query!(
-        "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, initial_messages = $6, url_runnable_args = $7, edited_by = $8, email = $9, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL
+        "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
            WHERE workspace_id = $10 AND path = $11",
        ct.url,
        ct.script_path,
@@ -300,7 +300,7 @@ pub async fn set_enabled(

    // important to set server_id, last_server_ping and error to NULL to stop current websocket listener
    let one_o = sqlx::query_scalar!(
-        "UPDATE websocket_trigger SET enabled = $1, email = $2, edited_by = $3, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL
+        "UPDATE websocket_trigger SET enabled = $1, email = $2, edited_by = $3, edited_at = now(), server_id = NULL, error = NULL
        WHERE path = $4 AND workspace_id = $5 RETURNING 1",
        payload.enabled,
        &authed.email,
@@ -384,7 +384,7 @@ async fn listen_to_unlistened_websockets(
    match sqlx::query_as::<_, WebsocketTrigger>(
        r#"SELECT *
            FROM websocket_trigger
-            WHERE enabled IS TRUE AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')"#
+            WHERE enabled IS TRUE AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')"#
    )
    .fetch_all(db)
    .await
@@ -402,7 +402,7 @@ async fn listen_to_unlistened_websockets(

    match sqlx::query_as!(
        CaptureConfigForWebsocket,
-        r#"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 (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')"#
+        r#"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')"#
    )
    .fetch_all(db)
    .await
@@ -643,7 +643,7 @@ impl WebsocketTrigger {
        killpill_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> () {
        match sqlx::query_scalar!(
-            "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",
+            "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",
            *INSTANCE_NAME,
            self.workspace_id,
            self.path,
@@ -845,7 +845,7 @@ impl CaptureConfigForWebsocket {
        killpill_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> () {
        match sqlx::query_scalar!(
-            "UPDATE capture_config SET server_id = $1, last_server_ping = now() WHERE last_client_ping > NOW() - INTERVAL '10 seconds' AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'websocket' AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
+            "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",
            *INSTANCE_NAME,
            self.workspace_id,
            self.path,
@@ -1078,117 +1078,116 @@ async fn listen_to_websocket(
                        let (writer, mut reader) = ws_stream.split();
                        let mut last_ping = tokio::time::Instant::now();

-                        tokio::select! {
-                            biased;
-                            _ = killpill_rx.recv() => {
-                                return;
-                            }
-                            _ = async {
-                                match &ws {
-                                    WebsocketEnum::Trigger(ws_trigger) => {
-                                        if let Err(err) = ws_trigger.send_initial_messages(writer, &db).await {
+
+                        // send initial messages
+                        match &ws {
+                            WebsocketEnum::Trigger(ws_trigger) => {
+                                tokio::select! {
+                                    biased;
+                                    _ = killpill_rx.recv() => {
+                                        return;
+                                    },
+                                    _ = loop_ping(&db, &ws, Some("Sending initial messages...")) => {
+                                        return;
+                                    },
+                                    result = ws_trigger.send_initial_messages(writer, &db) => {
+                                        if let Err(err) = result {
                                            ws_trigger.disable_with_error(&db, format!("Error sending initial messages: {:?}", err)).await;
+                                            return
                                        } else {
                                            tracing::debug!("Initial messages sent successfully to websocket {}", url);
-                                            // if initial messages sent successfully, wait forever
-                                            futures::future::pending::<()>().await;
                                        }
-                                    },
-                                    WebsocketEnum::Capture(_) => {
-                                        futures::future::pending::<()>().await;
                                    }
                                }
-                            } => {
-                                // was disabled => exit
-                                return;
                            },
-                            _ = async {
-                                loop {
-                                    tokio::select! {
-                                        biased;
-                                        msg = reader.next() => {
-                                            if let Some(msg) = msg {
-                                                if last_ping.elapsed() > tokio::time::Duration::from_secs(5) {
-                                                    if let None = ws.update_ping(&db, None).await {
-                                                        return;
-                                                    }
-                                                    last_ping = tokio::time::Instant::now();
-                                                }
-                                                match msg {
-                                                    Ok(msg) => {
-                                                        match msg {
-                                                            tokio_tungstenite::tungstenite::Message::Text(text) => {
-                                                                tracing::debug!("Received text message from websocket {}: {}", url, text);
-                                                                let mut should_handle = true;
-                                                                for filter in &filters {
-                                                                    match filter {
-                                                                        Filter::JsonFilter(JsonFilter { key, value }) => {
-                                                                            let mut deserializer = serde_json::Deserializer::from_str(text.as_str());
-                                                                            should_handle = match is_value_superset(&mut deserializer, key, &value) {
-                                                                                Ok(filter_match) => {
-                                                                                    filter_match
-                                                                                },
-                                                                                Err(err) => {
-                                                                                    tracing::warn!("Error deserializing filter for websocket {}: {:?}", url, err);
-                                                                                    false
-                                                                                }
-                                                                            };
-                                                                        }
-                                                                    }
-                                                                    if !should_handle {
-                                                                        break;
-                                                                    }
-                                                                }
-                                                                if should_handle {
+                            _ => {}
+                        }

-                                                                    let args = HashMap::from([("msg".to_string(), to_raw_value(&text))]);
-                                                                    let extra = Some(HashMap::from([(
-                                                                        "wm_trigger".to_string(),
-                                                                        to_raw_value(&serde_json::json!({"kind": "websocket", "websocket": { "url": url }})),
-                                                                    )]));
-
-                                                                    let args = PushArgsOwned { args, extra };
-                                                                    match &ws {
-                                                                        WebsocketEnum::Trigger(ws_trigger) => {
-                                                                            ws_trigger.handle(&db, args).await;
-                                                                        },
-                                                                        WebsocketEnum::Capture(capture) => {
-                                                                            capture.handle(&db, args).await;
-                                                                        }
-                                                                    }
-                                                                }
-                                                            },
-                                                            a @ _ => {
-                                                                tracing::debug!("Received non text-message from websocket {}: {:?}", url, a);
-                                                            }
-                                                        }
-                                                    },
-                                                    Err(err) => {
-                                                        tracing::error!("Error reading from websocket {}: {:?}", url, err);
-                                                    }
-                                                }
-                                            } else {
-                                                tracing::error!("Websocket {} closed", url);
-                                                if let None = ws.update_ping(&db, Some("Websocket closed")).await {
-                                                    return;
-                                                }
-                                                tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
-                                                break;
-                                            }
-                                        },
-                                        _ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => {
-                                            tracing::debug!("Sending ping to websocket {}", url);
+                        loop {
+                            tokio::select! {
+                                biased;
+                                _ = killpill_rx.recv() => {
+                                    return;
+                                }
+                                msg = reader.next() => {
+                                    if let Some(msg) = msg {
+                                        if last_ping.elapsed() > tokio::time::Duration::from_secs(5) {
                                            if let None = ws.update_ping(&db, None).await {
                                                return;
                                            }
                                            last_ping = tokio::time::Instant::now();
-                                        },
+                                        }
+                                        match msg {
+                                            Ok(msg) => {
+                                                match msg {
+                                                    tokio_tungstenite::tungstenite::Message::Text(text) => {
+                                                        tracing::debug!("Received text message from websocket {}: {}", url, text);
+                                                        let mut should_handle = true;
+                                                        for filter in &filters {
+                                                            match filter {
+                                                                Filter::JsonFilter(JsonFilter { key, value }) => {
+                                                                    let mut deserializer = serde_json::Deserializer::from_str(text.as_str());
+                                                                    should_handle = match is_value_superset(&mut deserializer, key, &value) {
+                                                                        Ok(filter_match) => {
+                                                                            filter_match
+                                                                        },
+                                                                        Err(err) => {
+                                                                            tracing::warn!("Error deserializing filter for websocket {}: {:?}", url, err);
+                                                                            false
+                                                                        }
+                                                                    };
+                                                                }
+                                                            }
+                                                            if !should_handle {
+                                                                break;
+                                                            }
+                                                        }
+                                                        if should_handle {
+
+                                                            let args = HashMap::from([("msg".to_string(), to_raw_value(&text))]);
+                                                            let extra = Some(HashMap::from([(
+                                                                "wm_trigger".to_string(),
+                                                                to_raw_value(&serde_json::json!({"kind": "websocket", "websocket": { "url": url }})),
+                                                            )]));
+
+                                                            let args = PushArgsOwned { args, extra };
+                                                            match &ws {
+                                                                WebsocketEnum::Trigger(ws_trigger) => {
+                                                                    ws_trigger.handle(&db, args).await;
+                                                                },
+                                                                WebsocketEnum::Capture(capture) => {
+                                                                    capture.handle(&db, args).await;
+                                                                }
+                                                            }
+                                                        }
+                                                    },
+                                                    a @ _ => {
+                                                        tracing::debug!("Received non text-message from websocket {}: {:?}", url, a);
+                                                    }
+                                                }
+                                            },
+                                            Err(err) => {
+                                                tracing::error!("Error reading from websocket {}: {:?}", url, err);
+                                            }
+                                        }
+                                    } else {
+                                        tracing::error!("Websocket {} closed", url);
+                                        if let None = ws.update_ping(&db, Some("Websocket closed")).await {
+                                            return;
+                                        }
+                                        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+                                        break;
                                    }
-                                }
-                            } => {
-                                return;
+                                },
+                                _ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => {
+                                    tracing::debug!("Sending ping to websocket {}", url);
+                                    if let None = ws.update_ping(&db, None).await {
+                                        return;
+                                    }
+                                    last_ping = tokio::time::Instant::now();
+                                },
                            }
-                        };
+                        }
                    }
                    Err(err) => {
                        tracing::error!("Error connecting to websocket {}: {:?}", url, err);
@@ -1303,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(
@@ -1316,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)
@@ -510,7 +510,7 @@
 														class="rounded-full p-1 bg-surface-secondary duration-200 hover:bg-surface-hover ml-2"
 														aria-label="Clear"
 														on:click={() => {
-															value.splice(i, 1)
+															value = value.filter((_, index) => index !== i)
 															redraw += 1
 														}}
 													>
@@ -15,7 +15,8 @@
 		HttpTriggerService,
 		VariableService,
 		WebsocketTriggerService,
-		KafkaTriggerService
+		KafkaTriggerService,
+		NatsTriggerService
 	} from '$lib/gen'
 	import { superadmin, userStore, workspaceStore } from '$lib/stores'
 	import { createEventDispatcher, getContext } from 'svelte'
@@ -39,6 +40,7 @@
 		| 'http_trigger'
 		| 'websocket_trigger'
 		| 'kafka_trigger'
+		| 'nats_trigger'
 	let meta: Meta | undefined = undefined
 	export let fullNamePlaceholder: string | undefined = undefined
 	export let namePlaceholder = ''
@@ -232,6 +234,11 @@
 				workspace: $workspaceStore!,
 				path: path
 			})
+		} else if (kind == 'nats_trigger') {
+			return await NatsTriggerService.existsNatsTrigger({
+				workspace: $workspaceStore!,
+				path: path
+			})
 		} else {
 			return false
 		}
@@ -85,8 +85,6 @@
 	let itemPicker: ItemPicker | undefined = undefined
 	let variableEditor: VariableEditor | undefined = undefined

-	$: isValid = allTrue(inputCheck ?? {})
-
 	let resourceTypes: string[] | undefined = undefined

 	async function loadResourceTypes() {
@@ -95,7 +93,7 @@

 	loadResourceTypes()

-	$: schema && reorder()
+	$: schema && (reorder(), (hidden = {}))

 	function hasExtraKeys() {
 		return Object.keys(args ?? {}).some((x) => !keys.includes(x))
@@ -134,6 +132,28 @@
 	}

 	$: fields = items ?? keys.map((x) => ({ id: x, value: x }))
+
+	let hidden: Record<string, boolean> = {}
+
+	function handleHiddenFields(schema: Schema | any, args: Record<string, any>) {
+		for (const x of fields) {
+			if (schema?.properties[x.value]?.showExpr) {
+				if (computeShow(x.value, schema.properties[x.value].showExpr, args)) {
+					hidden[x.value] = false
+				} else if (!hidden[x.value]) {
+					hidden[x.value] = true
+					// remove arg (important: will not trigger a re-render)
+					delete args[x.value]
+					// make sure it's made valid
+					inputCheck[x.value] = true
+				}
+			}
+		}
+	}
+
+	$: handleHiddenFields(schema, args)
+
+	$: isValid = allTrue(inputCheck ?? {})
 </script>

 {#if showReset}
@@ -164,7 +184,7 @@
 						}}
 					>
 						{#if typeof args == 'object' && schema?.properties[argName]}
-							{#if computeShow(argName, schema?.properties[argName].showExpr, args)}
+							{#if !hidden[argName]}
 								{#if lightweightMode}
 									<LightweightArgInput
 										label={argName}
@@ -27,6 +27,7 @@
 		| 'http_trigger'
 		| 'websocket_trigger'
 		| 'kafka_trigger'
+		| 'nats_trigger'
 	let kind: Kind

 	let path: string = ''
@@ -13,11 +13,24 @@
 	export let id: string | undefined = undefined
 	export let active: boolean | undefined = false
 	export let exact = false
+	export let otherValues: string[] = []

 	export let disabled: boolean = false
 	const { selected, update, hashNavigation } = getContext<TabsContext>('Tabs')

-	$: isSelected = exact ? $selected == value : $selected?.startsWith(value)
+	function getIsSelectedFn(exact: boolean, otherValues: string[]) {
+		if (otherValues.length > 0) {
+			return (selected: string) => selected == value || otherValues.includes(selected)
+		} else if (exact) {
+			return (selected: string) => selected == value
+		} else {
+			return (selected: string) => selected?.startsWith(value)
+		}
+	}
+
+	$: isSelectedFn = getIsSelectedFn(exact, otherValues)
+
+	$: isSelected = isSelectedFn($selected)

 	const fontSizeClasses = {
 		xs: 'text-xs',
@@ -12,7 +12,9 @@
 		| 'cli'
 		| 'routes'
 		| 'websockets'
-		| 'scheduledPoll' = 'webhooks'
+		| 'scheduledPoll'
+		| 'kafka'
+		| 'nats' = 'webhooks'
 	export let flow_json: any | undefined = undefined
 	export let simplfiedPoll: boolean = false

@@ -52,6 +54,7 @@
 						<slot slot="routes" name="routes" />
 						<slot slot="websockets" name="websockets" />
 						<slot slot="kafka" name="kafka" />
+						<slot slot="nats" name="nats" />
 						<slot slot="emails" name="emails" />
 						<slot slot="schedules" name="schedules" />
 						<slot slot="cli" name="cli" />
@@ -20,7 +20,15 @@

 	const primaryScheduleStore = writable<ScheduleTrigger | undefined | false>(undefined)
 	const selectedTriggerStore = writable<
-		'webhooks' | 'emails' | 'schedules' | 'cli' | 'routes' | 'websockets' | 'scheduledPoll'
+		| 'webhooks'
+		| 'emails'
+		| 'schedules'
+		| 'cli'
+		| 'routes'
+		| 'websockets'
+		| 'scheduledPoll'
+		| 'kafka'
+		| 'nats'
 	>('webhooks')

 	const simplifiedPoll = writable(false)
@@ -55,6 +63,7 @@
 							<slot slot="routes" name="routes" />
 							<slot slot="websockets" name="websockets" />
 							<slot slot="kafka" name="kafka" />
+							<slot slot="nats" name="nats" />
 							<slot slot="emails" name="emails" />
 							<slot slot="schedules" name="schedules" />
 							<slot slot="cli" name="cli" />
@@ -100,6 +109,7 @@
 								<slot slot="script" name="script" />
 								<slot slot="websockets" name="websockets" />
 								<slot slot="kafka" name="kafka" />
+								<slot slot="nats" name="nats" />
 								<slot slot="emails" name="emails" />
 								<slot slot="schedules" name="schedules" />
 								<slot slot="cli" name="cli" />
@@ -1,9 +1,20 @@
 <script lang="ts">
 	import { Tabs, Tab } from '$lib/components/common'
-	import { CalendarCheck2, MailIcon, Route, Terminal, Webhook, Unplug } from 'lucide-svelte'
+	import {
+		CalendarCheck2,
+		MailIcon,
+		Route,
+		Terminal,
+		Webhook,
+		Unplug,
+		PlugZap
+	} from 'lucide-svelte'

 	import HighlightTheme from '../HighlightTheme.svelte'
 	import KafkaIcon from '../icons/KafkaIcon.svelte'
+	import NatsIcon from '../icons/NatsIcon.svelte'
+	import ToggleButtonGroup from '../common/toggleButton-v2/ToggleButtonGroup.svelte'
+	import ToggleButton from '../common/toggleButton-v2/ToggleButton.svelte'

 	export let triggerSelected:
 		| 'webhooks'
@@ -13,8 +24,17 @@
 		| 'routes'
 		| 'websockets'
 		| 'kafka'
+		| 'nats'
 		| 'scheduledPoll' = 'webhooks'
 	export let simplfiedPoll: boolean = false
+
+	export let eventStreamType: 'kafka' | 'nats' = 'kafka'
+
+	$: {
+		if (triggerSelected === 'kafka' || triggerSelected === 'nats') {
+			eventStreamType = triggerSelected
+		}
+	}
 </script>

 <HighlightTheme />
@@ -45,10 +65,10 @@
 				Websockets
 			</span>
 		</Tab>
-		<Tab value="kafka">
+		<Tab value="kafka" otherValues={['nats']}>
 			<span class="flex flex-row gap-2 items-center text-xs">
-				<KafkaIcon size={12} />
-				Kafka
+				<PlugZap size={12} />
+				Event streams
 			</span>
 		</Tab>
 		<Tab value="emails">
@@ -77,8 +97,18 @@
 				<slot name="schedules" />
 			{:else if triggerSelected === 'websockets'}
 				<slot name="websockets" />
-			{:else if triggerSelected === 'kafka'}
-				<slot name="kafka" />
+			{:else if triggerSelected === 'kafka' || triggerSelected === 'nats'}
+				<div class="m-1.5">
+					<ToggleButtonGroup bind:selected={eventStreamType}>
+						<ToggleButton value="kafka" label="Kafka" icon={KafkaIcon} />
+						<ToggleButton value="nats" label="NATS" icon={NatsIcon} />
+					</ToggleButtonGroup>
+				</div>
+				{#if eventStreamType === 'kafka'}
+					<slot name="kafka" />
+				{:else if eventStreamType === 'nats'}
+					<slot name="nats" />
+				{/if}
 			{:else if triggerSelected === 'cli'}
 				<slot name="cli" />
 			{/if}
@@ -33,7 +33,6 @@
 				pad: false
 			})
 			.toLowerCase()
-		console.log(encodedPrefix)
 		return `${encodedPrefix}@${emailDomain}`
 	}

@@ -1,16 +1,17 @@
 <script lang="ts">
-	import { Calendar, Mail, Webhook, Unplug } from 'lucide-svelte'
+	import { Calendar, Mail, Webhook, Unplug, PlugZap } from 'lucide-svelte'
 	import TriggerButton from './TriggerButton.svelte'

 	import Popover from '$lib/components/Popover.svelte'
 	import TriggerCount from './TriggerCount.svelte'
-	import { createEventDispatcher, onMount } from 'svelte'
+	import { createEventDispatcher, onMount, type ComponentType } from 'svelte'
 	import { Route } from 'lucide-svelte'
 	import { getContext } from 'svelte'
 	import { type TriggerContext } from '$lib/components/triggers'
 	import { FlowService, ScriptService } from '$lib/gen'
 	import { enterpriseLicense, workspaceStore } from '$lib/stores'
 	import KafkaIcon from '$lib/components/icons/KafkaIcon.svelte'
+	import NatsIcon from '$lib/components/icons/NatsIcon.svelte'

 	const { selectedTrigger, triggersCount } = getContext<TriggerContext>('TriggerContext')

@@ -25,8 +26,12 @@
 		| 'routes'
 		| 'websockets'
 		| 'kafka'
+		| 'nats'
 		| 'emails'
-	)[] = ['webhooks', 'schedules', 'routes', 'websockets', 'kafka', 'emails']
+		| 'eventStreams'
+	)[] = showOnlyWithCount
+		? ['webhooks', 'schedules', 'routes', 'websockets', 'kafka', 'nats', 'emails']
+		: ['webhooks', 'schedules', 'routes', 'websockets', 'eventStreams', 'emails']
 	const dispatch = createEventDispatcher()

 	onMount(() => {
@@ -49,29 +54,43 @@
 		}
 	}

-	const triggerTypeConfig = {
+	const triggerTypeConfig: {
+		[key: string]: { icon: ComponentType; countKey?: string }
+	} = {
 		webhooks: { icon: Webhook, countKey: 'webhook_count' },
 		schedules: { icon: Calendar, countKey: 'schedule_count' },
 		routes: { icon: Route, countKey: 'http_routes_count' },
 		websockets: { icon: Unplug, countKey: 'websocket_count' },
 		kafka: { icon: KafkaIcon, countKey: 'kafka_count' },
-		emails: { icon: Mail, countKey: 'email_count' }
+		nats: { icon: NatsIcon, countKey: 'nats_count' },
+		emails: { icon: Mail, countKey: 'email_count' },
+		eventStreams: { icon: PlugZap }
+	}
+
+	function camelCaseToWords(s: string) {
+		const result = s.replace(/([A-Z])/g, ' $1')
+		return result.charAt(0).toUpperCase() + result.slice(1).toLowerCase()
 	}
 </script>

 {#each triggersToDisplay as type}
 	{@const { icon, countKey } = triggerTypeConfig[type]}
-	{#if (!showOnlyWithCount || ($triggersCount?.[countKey] ?? 0) > 0) && !(type === 'kafka' && !$enterpriseLicense)}
+	{#if (!showOnlyWithCount || ((countKey && $triggersCount?.[countKey]) || 0) > 0) && !(type === 'kafka' && !$enterpriseLicense) && !(type === 'nats' && !$enterpriseLicense)}
 		<Popover>
-			<svelte:fragment slot="text">{type.charAt(0).toUpperCase() + type.slice(1)}</svelte:fragment>
+			<svelte:fragment slot="text">{camelCaseToWords(type)}</svelte:fragment>
 			<TriggerButton
 				on:click={() => {
-					$selectedTrigger = type
+					$selectedTrigger = type === 'eventStreams' ? 'kafka' : type
 					dispatch('select')
 				}}
-				selected={selected && $selectedTrigger === type}
+				selected={selected &&
+					($selectedTrigger === type ||
+						(type === 'eventStreams' &&
+							($selectedTrigger === 'kafka' || $selectedTrigger === 'nats')))}
 			>
-				<TriggerCount count={$triggersCount?.[countKey]} />
+				{#if countKey}
+					<TriggerCount count={$triggersCount?.[countKey]} />
+				{/if}
 				<svelte:component this={icon} size={12} />
 			</TriggerButton>
 		</Popover>
@@ -0,0 +1,20 @@
+<script lang="ts">
+	export let size: number = 16
+	export let color: string | undefined = undefined
+	let clazz = ''
+	export { clazz as class }
+</script>
+
+<svg
+	xmlns="http://www.w3.org/2000/svg"
+	width={size}
+	height={size}
+	viewBox="0 0 64 64"
+	class={clazz}
+	fill={color ?? 'currentColor'}
+	stroke-width="0"
+>
+	<path
+		d="M1.156 50H27.53l15.125 14V50h20.187V0H1.156zm9.937-38.312h11.563l22.25 20.75v-20.75h7.375V38.5H41.094L18.47 17.375V38.5h-7.375z"
+	/>
+</svg>
@@ -94,6 +94,8 @@ import KeycloakIcon from './KeycloakIcon.svelte'
 import ZitadelIcon from './ZitadelIcon.svelte'
 import SpotifyIcon from './SpotifyIcon.svelte'
 import XeroIcon from './XeroIcon.svelte'
+import KafkaIcon from './KafkaIcon.svelte'
+import NatsIcon from './NatsIcon.svelte'
 export const APP_TO_ICON_COMPONENT = {
 	postgresql: PostgresIcon,
 	mysql: Mysql,
@@ -192,7 +194,9 @@ export const APP_TO_ICON_COMPONENT = {
 	keycloak: KeycloakIcon,
 	zitadel: ZitadelIcon,
 	spotify: SpotifyIcon,
-	xero: XeroIcon
+	xero: XeroIcon,
+	kafka: KafkaIcon,
+	nats: NatsIcon
 } as const

 export {
@@ -285,5 +289,7 @@ export {
 	JumpCloudIcon,
 	KeycloakIcon,
 	ZitadelIcon,
-	XeroIcon
+	XeroIcon,
+	KafkaIcon,
+	NatsIcon
 }
@@ -7,9 +7,7 @@
 		workspaceStore,
 		isCriticalAlertsUIOpen,
 		enterpriseLicense,
-
 		devopsRole
-
 	} from '$lib/stores'
 	import { SIDEBAR_SHOW_SCHEDULES } from '$lib/consts'
 	import {
@@ -53,6 +51,7 @@
 	import { page } from '$app/stores'
 	import SideBarNotification from './SideBarNotification.svelte'
 	import KafkaIcon from '../icons/KafkaIcon.svelte'
+	import NatsIcon from '../icons/NatsIcon.svelte'

 	export let numUnacknowledgedCriticalAlerts = 0

@@ -108,6 +107,13 @@
 			icon: KafkaIcon,
 			disabled: $userStore?.operator || !$enterpriseLicense,
 			kind: 'kafka'
+		},
+		{
+			label: 'NATS' + ($enterpriseLicense ? '' : ' (EE)'),
+			href: '/nats_triggers',
+			icon: NatsIcon,
+			disabled: $userStore?.operator || !$enterpriseLicense,
+			kind: 'nats'
 		}
 	]

@@ -48,7 +48,7 @@ export type TriggerKind =
 	| 'websockets'
 	| 'scheduledPoll'
 	| 'kafka'
-
+	| 'nats'
 export function captureTriggerKindToTriggerKind(kind: CaptureTriggerKind): TriggerKind {
 	switch (kind) {
 		case 'webhook':
@@ -61,6 +61,8 @@ export function captureTriggerKindToTriggerKind(kind: CaptureTriggerKind): Trigg
 			return 'websockets'
 		case 'kafka':
 			return 'kafka'
+		case 'nats':
+			return 'nats'
 		default:
 			throw new Error(`Unknown CaptureTriggerKind: ${kind}`)
 	}
@@ -8,6 +8,7 @@
 	import { createEventDispatcher } from 'svelte'
 	import { captureTriggerKindToTriggerKind } from '../triggers'
 	import CaptureIcon from './CaptureIcon.svelte'
+	import NatsIcon from '../icons/NatsIcon.svelte'

 	let isOpen = false

@@ -76,6 +77,16 @@
 					<p class="text-xs text-secondary">Kafka</p>
 				</div>
 			</button>
+			<button
+				disabled={!$enterpriseLicense}
+				class="hover:bg-surface-hover p-2 transition-colors duration-150"
+				on:click={() => handleClick('nats')}
+			>
+				<div class="flex flex-row items-center gap-2">
+					<NatsIcon size={16} />
+					<p class="text-xs text-secondary">Nats</p>
+				</div>
+			</button>
 		</div>
 	</svelte:fragment>
 </Popover>
@@ -13,6 +13,7 @@
 	import type { ConnectionInfo } from '../common/alert/ConnectionIndicator.svelte'
 	import type { CaptureInfo } from './CaptureSection.svelte'
 	import CaptureTable from './CaptureTable.svelte'
+	import NatsTriggersConfigSection from './NatsTriggersConfigSection.svelte'

 	export let isFlow: boolean
 	export let path: string
@@ -131,15 +132,21 @@
 	let config: CaptureConfig | undefined
 	$: config = captureConfigs[captureType]

-	let cloudDisabled = (captureType === 'websocket' || captureType === 'kafka') && isCloudHosted()
+	let cloudDisabled =
+		(captureType === 'websocket' || captureType === 'kafka' || captureType === 'nats') &&
+		isCloudHosted()

 	function updateConnectionInfo(config: CaptureConfig | undefined, captureActive: boolean) {
-		if ((captureType === 'websocket' || captureType === 'kafka') && config && captureActive) {
+		if (
+			(captureType === 'websocket' || captureType === 'kafka' || captureType === 'nats') &&
+			config &&
+			captureActive
+		) {
 			const serverEnabled = getServerEnabled(config)
 			const connected = serverEnabled && !config.error
 			const message = connected
-				? `${capitalize(captureType)} is connected`
-				: `${capitalize(captureType)} is not connected${config.error ? ': ' + config.error : ''}`
+				? `Connected`
+				: `Not connected${config.error ? ': ' + config.error : ''}`
 			connectionInfo = {
 				connected,
 				message
@@ -251,5 +258,21 @@
 				handleCapture()
 			}}
 		/>
+	{:else if captureType === 'nats'}
+		<NatsTriggersConfigSection
+			headless={true}
+			bind:args
+			{path}
+			staticInputDisabled={false}
+			{showCapture}
+			{captureInfo}
+			bind:captureTable
+			on:applyArgs
+			on:updateSchema
+			on:addPreprocessor
+			on:captureToggle={() => {
+				handleCapture()
+			}}
+		/>
 	{/if}
 </div>
@@ -0,0 +1,27 @@
+<script lang="ts">
+	import { tick } from 'svelte'
+	import NatsTriggerEditorInner from './NatsTriggerEditorInner.svelte'
+
+	let open = false
+	export async function openEdit(ePath: string, isFlow: boolean) {
+		open = true
+		await tick()
+		drawer?.openEdit(ePath, isFlow)
+	}
+
+	export async function openNew(
+		is_flow: boolean,
+		initial_script_path?: string,
+		defaultValues?: Record<string, any>
+	) {
+		open = true
+		await tick()
+		drawer?.openNew(is_flow, initial_script_path, defaultValues)
+	}
+
+	let drawer: NatsTriggerEditorInner
+</script>
+
+{#if open}
+	<NatsTriggerEditorInner on:update bind:this={drawer} />
+{/if}
@@ -0,0 +1,253 @@
+<script lang="ts">
+	import { Alert, Button } from '$lib/components/common'
+	import Drawer from '$lib/components/common/drawer/Drawer.svelte'
+	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
+	import Path from '$lib/components/Path.svelte'
+	import Required from '$lib/components/Required.svelte'
+	import ScriptPicker from '$lib/components/ScriptPicker.svelte'
+	import { NatsTriggerService } from '$lib/gen'
+	import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
+	import { canWrite, emptyString, sendUserToast } from '$lib/utils'
+	import { createEventDispatcher } from 'svelte'
+	import Section from '$lib/components/Section.svelte'
+	import { Loader2, Save } from 'lucide-svelte'
+	import Label from '$lib/components/Label.svelte'
+	import Toggle from '../Toggle.svelte'
+	import NatsTriggersConfigSection from './NatsTriggersConfigSection.svelte'
+
+	let drawer: Drawer
+	let is_flow: boolean = false
+	let initialPath = ''
+	let edit = true
+	let itemKind: 'flow' | 'script' = 'script'
+	let script_path = ''
+	let initialScriptPath = ''
+	let fixedScriptPath = ''
+	let path: string = ''
+	let pathError = ''
+	let enabled = false
+	let dirtyPath = false
+	let can_write = true
+	let drawerLoading = true
+	let defaultValues: Record<string, any> | undefined = undefined
+	let args: Record<string, any> = {}
+
+	const dispatch = createEventDispatcher()
+
+	$: is_flow = itemKind === 'flow'
+
+	export async function openEdit(ePath: string, isFlow: boolean) {
+		drawerLoading = true
+		try {
+			drawer?.openDrawer()
+			initialPath = ePath
+			itemKind = isFlow ? 'flow' : 'script'
+			edit = true
+			dirtyPath = false
+			await loadTrigger()
+		} catch (err) {
+			sendUserToast(`Could not load nats trigger: ${err}`, true)
+		} finally {
+			drawerLoading = false
+		}
+	}
+
+	export async function openNew(
+		nis_flow: boolean,
+		fixedScriptPath_?: string,
+		nDefaultValues?: Record<string, any>
+	) {
+		drawerLoading = true
+		try {
+			drawer?.openDrawer()
+			is_flow = nis_flow
+			edit = false
+			itemKind = nis_flow ? 'flow' : 'script'
+			args.nats_resource_path = nDefaultValues?.nats_resource_path ?? ''
+			args.subjects = nDefaultValues?.subjects ?? ['']
+			args.use_jetstream = nDefaultValues?.use_jetstream ?? false
+			args.stream_name = args.use_jetstream ? nDefaultValues?.stream_name ?? '' : undefined
+			args.consumer_name = args.use_jetstream ? nDefaultValues?.consumer_name ?? '' : undefined
+			initialScriptPath = ''
+			fixedScriptPath = fixedScriptPath_ ?? ''
+			script_path = fixedScriptPath
+			path = ''
+			initialPath = ''
+			dirtyPath = false
+			defaultValues = nDefaultValues
+		} finally {
+			drawerLoading = false
+		}
+	}
+
+	async function loadTrigger(): Promise<void> {
+		const s = await NatsTriggerService.getNatsTrigger({
+			workspace: $workspaceStore!,
+			path: initialPath
+		})
+		script_path = s.script_path
+		initialScriptPath = s.script_path
+
+		is_flow = s.is_flow
+		path = s.path
+		args.nats_resource_path = s.nats_resource_path
+		args.stream_name = s.stream_name
+		args.consumer_name = s.consumer_name
+		args.subjects = s.subjects
+		args.use_jetstream = s.use_jetstream
+		enabled = s.enabled
+
+		can_write = canWrite(s.path, s.extra_perms, $userStore)
+	}
+
+	async function updateTrigger(): Promise<void> {
+		if (edit) {
+			await NatsTriggerService.updateNatsTrigger({
+				workspace: $workspaceStore!,
+				path: initialPath,
+				requestBody: {
+					path,
+					script_path,
+					is_flow,
+					nats_resource_path: args.nats_resource_path,
+					stream_name: args.stream_name,
+					consumer_name: args.consumer_name,
+					subjects: args.subjects,
+					use_jetstream: args.use_jetstream
+				}
+			})
+			sendUserToast(`Nats trigger ${path} updated`)
+		} else {
+			await NatsTriggerService.createNatsTrigger({
+				workspace: $workspaceStore!,
+				requestBody: {
+					path,
+					script_path,
+					is_flow,
+					enabled: true,
+					nats_resource_path: args.nats_resource_path,
+					stream_name: args.stream_name,
+					consumer_name: args.consumer_name,
+					subjects: args.subjects,
+					use_jetstream: args.use_jetstream
+				}
+			})
+			sendUserToast(`Nats trigger ${path} created`)
+		}
+		if (!$usedTriggerKinds.includes('nats')) {
+			$usedTriggerKinds = [...$usedTriggerKinds, 'nats']
+		}
+		dispatch('update')
+		drawer.closeDrawer()
+	}
+
+	function useDefaultValues() {
+		if (args.nats_resource_path && args.nats_resource_path != '') {
+			return false
+		}
+		if (!defaultValues) {
+			return false
+		}
+		return (
+			defaultValues.servers &&
+			defaultValues.servers.length > 0 &&
+			defaultValues.servers.some((broker: string) => broker.trim() !== '')
+		)
+	}
+
+	let isValid = false
+</script>
+
+<Drawer size="800px" bind:this={drawer}>
+	<DrawerContent
+		title={edit
+			? can_write
+				? `Edit NATS trigger ${initialPath}`
+				: `NATS trigger ${initialPath}`
+			: 'New NATS trigger'}
+		on:close={drawer.closeDrawer}
+	>
+		<svelte:fragment slot="actions">
+			{#if !drawerLoading}
+				{#if edit}
+					<div class="mr-8 center-center -mt-1">
+						<Toggle
+							disabled={!can_write}
+							checked={enabled}
+							options={{ right: 'enable', left: 'disable' }}
+							on:change={async (e) => {
+								await NatsTriggerService.setNatsTriggerEnabled({
+									path: initialPath,
+									workspace: $workspaceStore ?? '',
+									requestBody: { enabled: e.detail }
+								})
+								sendUserToast(`${e.detail ? 'enabled' : 'disabled'} NATS trigger ${initialPath}`)
+							}}
+						/>
+					</div>
+				{/if}
+				{#if can_write}
+					<Button
+						startIcon={{ icon: Save }}
+						disabled={pathError != '' || emptyString(script_path) || !can_write || !isValid}
+						on:click={updateTrigger}
+					>
+						Save
+					</Button>
+				{/if}
+			{/if}
+		</svelte:fragment>
+		{#if drawerLoading}
+			<Loader2 class="animate-spin" />
+		{:else}
+			<Alert title="Info" type="info">
+				{#if edit}
+					Changes can take up to 30 seconds to take effect.
+				{:else}
+					NATS consumers can take up to 30 seconds to start.
+				{/if}
+			</Alert>
+			<div class="flex flex-col gap-12 mt-6">
+				<div class="flex flex-col gap-4">
+					<Label label="Path">
+						<Path
+							bind:dirty={dirtyPath}
+							bind:error={pathError}
+							bind:path
+							{initialPath}
+							checkInitialPathExistence={!edit}
+							namePlaceholder="nats_trigger"
+							kind="nats_trigger"
+							disabled={!can_write}
+						/>
+					</Label>
+				</div>
+
+				<NatsTriggersConfigSection
+					{path}
+					bind:args
+					bind:isValid
+					defaultValues={useDefaultValues() ? defaultValues : undefined}
+				/>
+
+				<Section label="Runnable">
+					<p class="text-xs mb-1 text-tertiary">
+						Pick a script or flow to be triggered<Required required={true} />
+					</p>
+					<div class="flex flex-row mb-2">
+						<ScriptPicker
+							disabled={fixedScriptPath != '' || !can_write}
+							initialPath={fixedScriptPath || initialScriptPath}
+							kinds={['script']}
+							allowFlow={true}
+							bind:itemKind
+							bind:scriptPath={script_path}
+							allowRefresh={can_write}
+							allowEdit={!$userStore?.operator}
+						/>
+					</div>
+				</Section>
+			</div>
+		{/if}
+	</DrawerContent>
+</Drawer>
@@ -0,0 +1,287 @@
+<script lang="ts">
+	import ResourcePicker from '../ResourcePicker.svelte'
+	import { Boxes } from 'lucide-svelte'
+	import Section from '$lib/components/Section.svelte'
+	import Subsection from '$lib/components/Subsection.svelte'
+	import ToggleButton from '../common/toggleButton-v2/ToggleButton.svelte'
+	import ToggleButtonGroup from '../common/toggleButton-v2/ToggleButtonGroup.svelte'
+	import SchemaForm from '../SchemaForm.svelte'
+	import CaptureSection, { type CaptureInfo } from './CaptureSection.svelte'
+	import CaptureTable from './CaptureTable.svelte'
+	import { workspaceStore } from '$lib/stores'
+	export let defaultValues: Record<string, any> | undefined = undefined
+	export let headless: boolean = false
+	export let args: Record<string, any> = {}
+	export let staticInputDisabled: boolean = true
+	export let showCapture: boolean = false
+	export let captureInfo: CaptureInfo | undefined = undefined
+	export let captureTable: CaptureTable | undefined = undefined
+	export let isValid: boolean = false
+	export let path: string
+	let selected: 'resource' | 'static' = staticInputDisabled ? 'resource' : 'static'
+
+	const connnectionSchema = {
+		$schema: 'http://json-schema.org/draft-07/schema#',
+		type: 'object',
+		properties: {
+			servers: {
+				type: 'array',
+				items: {
+					type: 'string'
+				},
+				title: 'Servers',
+				default: ['']
+			},
+			require_tls: {
+				type: 'boolean',
+				title: 'Require TLS',
+				default: false
+			},
+			auth: {
+				type: 'object',
+				title: 'Authentication',
+				default: {
+					label: 'NO_AUTH'
+				},
+				oneOf: [
+					{
+						type: 'object',
+						title: 'NO_AUTH',
+						properties: {
+							label: {
+								enum: ['NO_AUTH'],
+								type: 'string'
+							}
+						}
+					},
+					{
+						type: 'object',
+						order: ['token'],
+						title: 'TOKEN',
+						required: ['token'],
+						properties: {
+							label: {
+								enum: ['TOKEN'],
+								type: 'string'
+							},
+							token: {
+								type: 'string',
+								password: true
+							}
+						}
+					},
+					{
+						type: 'object',
+						title: 'USER_PASSWORD',
+						required: ['user', 'password'],
+						properties: {
+							label: {
+								enum: ['USER_PASSWORD'],
+								type: 'string'
+							},
+							user: {
+								type: 'string'
+							},
+							password: {
+								type: 'string',
+								password: true
+							}
+						}
+					},
+					{
+						type: 'object',
+						order: ['nkey'],
+						title: 'NKEY',
+						required: ['nkey'],
+						properties: {
+							label: {
+								enum: ['NKEY'],
+								type: 'string'
+							},
+							seed: {
+								type: 'string',
+								password: true
+							}
+						}
+					},
+					{
+						type: 'object',
+						title: 'JWT',
+						required: ['jwt', 'seed'],
+						properties: {
+							label: {
+								enum: ['JWT'],
+								type: 'string'
+							},
+							jwt: {
+								type: 'string',
+								password: true
+							},
+							seed: {
+								type: 'string',
+								password: true
+							}
+						}
+					}
+				]
+			}
+		},
+		required: ['servers', 'auth', 'require_tls']
+	}
+
+	const argsSchema = {
+		$schema: 'http://json-schema.org/draft-07/schema#',
+		type: 'object',
+		properties: {
+			subjects: {
+				type: 'array',
+				items: {
+					type: 'string'
+				},
+				nullable: false,
+				title: 'Subjects',
+				default: [''],
+				description:
+					'The subjects to listen to. If not using JetStream, only one subject is supported.'
+			},
+			use_jetstream: {
+				type: 'boolean',
+				title: 'Use JetStream',
+				default: false,
+				description: 'Whether to use JetStream (durable push-consumer)'
+			},
+			stream_name: {
+				type: 'string',
+				title: 'Stream name',
+				pattern: '^[a-zA-Z0-9-_]+$',
+				customErrorMessage: 'Invalid stream name',
+				showExpr: 'fields.use_jetstream',
+				description:
+					'Required if using JetStream. If the stream already exists, it will be updated to include the specified subjects if not already present. Otherwise, it will be created.'
+			},
+			consumer_name: {
+				type: 'string',
+				title: 'Consumer name',
+				pattern: '^[a-zA-Z0-9-_]+$',
+				customErrorMessage: 'Invalid consumer name',
+				showExpr: 'fields.use_jetstream',
+				description:
+					'Required is using JetStream. If a consumer with the same name already exists, it will be overwritten. It is also used as the deliver subject.'
+			}
+		},
+		required: ['subjects', 'use_jetstream', 'stream_name', 'consumer_name']
+	}
+
+	let connectionValid = false
+	let otherArgsValid = false
+	let globalError = ''
+
+	$: globalError =
+		!args.use_jetstream && args.subjects && args.subjects.length > 1
+			? 'Only one subject is supported if not using JetStream.'
+			: ''
+
+	$: isValid =
+		(selected === 'resource'
+			? !!args.nats_resource_path
+			: connectionValid &&
+			  args.servers &&
+			  args.servers.length > 0 &&
+			  args.servers.every((b) => b.length > 0) &&
+			  args.require_tls !== undefined &&
+			  args.require_tls !== null) &&
+		otherArgsValid &&
+		args.subjects &&
+		args.subjects.length > 0 &&
+		args.subjects.every((b) => /^[a-zA-Z0-9-_.*>]+$/.test(b)) &&
+		globalError === ''
+
+	$: usingResource = !!args.nats_resource_path
+	$: usingResource && (selected = 'resource')
+
+	function setStreamAndConsumerNames() {
+		if (!args.stream_name) {
+			args.stream_name = `windmill_stream-${$workspaceStore}-${path.replaceAll('/', '__')}`
+		}
+		if (!args.consumer_name) {
+			args.consumer_name = `windmill_consumer-${$workspaceStore}-${path.replaceAll('/', '__')}`
+		}
+	}
+	$: useJetStream = args.use_jetstream
+	$: useJetStream && setStreamAndConsumerNames()
+</script>
+
+<div>
+	{#if showCapture && captureInfo}
+		<CaptureSection
+			captureType="nats"
+			disabled={!isValid}
+			{captureInfo}
+			on:captureToggle
+			on:applyArgs
+			on:updateSchema
+			on:addPreprocessor
+			bind:captureTable
+		/>
+	{/if}
+	<Section label="NATS" {headless}>
+		<div class="flex flex-col w-full gap-4">
+			<div class="block grow w-full">
+				<Subsection label="Connection">
+					<svelte:fragment slot="header">
+						{#if !staticInputDisabled}
+							<ToggleButtonGroup
+								bind:selected
+								class="h-full"
+								on:selected={(ev) => {
+									if (ev.detail === 'static') {
+										delete args.nats_resource_path
+										args.require_tls = false
+										args.servers = ['']
+										args.auth = {
+											label: 'NO_AUTH'
+										}
+									} else {
+										delete args.servers
+										delete args.auth
+										delete args.require_tls
+									}
+								}}
+							>
+								<ToggleButton value="static" label="Static" small={true} />
+								<ToggleButton value="resource" label="Resource" icon={Boxes} small={true} />
+							</ToggleButtonGroup>
+						{/if}
+					</svelte:fragment>
+
+					{#if selected === 'resource'}
+						<ResourcePicker
+							resourceType="nats"
+							bind:value={args.nats_resource_path}
+							{defaultValues}
+						/>
+					{:else}
+						<SchemaForm
+							schema={connnectionSchema}
+							bind:args
+							bind:isValid={connectionValid}
+							lightHeader={true}
+						/>
+					{/if}
+				</Subsection>
+			</div>
+
+			<div class="block grow w-full">
+				<Subsection headless={true}>
+					<SchemaForm
+						schema={argsSchema}
+						bind:args
+						bind:isValid={otherArgsValid}
+						lightHeader={true}
+					/>
+				</Subsection>
+				<span class="text-xs text-red-500">{globalError}</span>
+			</div>
+		</div>
+	</Section>
+</div>
@@ -0,0 +1,140 @@
+<script lang="ts">
+	import { enterpriseLicense, userStore, workspaceStore } from '$lib/stores'
+	import { NatsTriggerService, type NatsTrigger } from '$lib/gen'
+	import { canWrite } from '$lib/utils'
+	import Alert from '../common/alert/Alert.svelte'
+	import type { TriggerContext } from '../triggers'
+	import { getContext, onMount } from 'svelte'
+	import NatsTriggerEditor from './NatsTriggerEditor.svelte'
+	import { isCloudHosted } from '$lib/cloud'
+	import TriggersEditorSection from './TriggersEditorSection.svelte'
+	import Section from '$lib/components/Section.svelte'
+	import Skeleton from '$lib/components/common/skeleton/Skeleton.svelte'
+	import Description from '../Description.svelte'
+
+	export let isFlow: boolean
+	export let path: string
+	export let newItem: boolean = false
+	export let isEditor: boolean = false
+	export let canHavePreprocessor: boolean = false
+	export let hasPreprocessor: boolean = false
+
+	let natsTriggerEditor: NatsTriggerEditor
+
+	$: path && loadTriggers()
+
+	const { triggersCount, selectedTrigger, defaultValues } =
+		getContext<TriggerContext>('TriggerContext')
+
+	onMount(() => {
+		if (
+			defaultValues &&
+			$selectedTrigger === 'nats' &&
+			Object.keys($defaultValues ?? {}).length > 0
+		) {
+			natsTriggerEditor.openNew(isFlow, path, $defaultValues)
+			defaultValues.set(undefined)
+		}
+	})
+
+	let natsTriggers: (NatsTrigger & { canWrite: boolean })[] | undefined = undefined
+	export async function loadTriggers() {
+		try {
+			natsTriggers = (
+				await NatsTriggerService.listNatsTriggers({
+					workspace: $workspaceStore ?? '',
+					path,
+					isFlow
+				})
+			).map((x) => {
+				return { canWrite: canWrite(x.path, x.extra_perms!, $userStore), ...x }
+			})
+			$triggersCount = { ...($triggersCount ?? {}), nats_count: natsTriggers?.length }
+		} catch (e) {
+			console.error('impossible to load nats triggers', e)
+		}
+	}
+
+	let data = {
+		natsTriggers: [],
+		newItem
+	}
+
+	function saveTrigger(path: string, args?: Record<string, any>) {
+		natsTriggerEditor?.openNew(isFlow, path, args)
+	}
+</script>
+
+<NatsTriggerEditor
+	on:update={() => {
+		loadTriggers()
+	}}
+	bind:this={natsTriggerEditor}
+/>
+
+{#if !$enterpriseLicense}
+	<Alert title="EE Only" type="warning" size="xs">
+		Nats triggers are an enterprise only feature.
+	</Alert>
+{:else if isCloudHosted()}
+	<Alert title="Not compatible with multi-tenant cloud" type="warning" size="xs">
+		Nats triggers are disabled in the multi-tenant cloud.
+	</Alert>
+{:else}
+	<div class="flex flex-col gap-4">
+		<Description link="https://www.windmill.dev/docs/core_concepts/nats_triggers">
+			NATS triggers execute scripts and flows in response to messages published to NATS subjects.
+		</Description>
+		<TriggersEditorSection
+			on:saveTrigger={(e) => {
+				saveTrigger(path, e.detail.config)
+			}}
+			on:applyArgs
+			on:addPreprocessor
+			cloudDisabled={false}
+			triggerType="nats"
+			{isFlow}
+			{data}
+			{path}
+			{isEditor}
+			{canHavePreprocessor}
+			{hasPreprocessor}
+			{newItem}
+		/>
+
+		{#if !newItem}
+			{#if natsTriggers}
+				<Section label="NATS Triggers">
+					{#if natsTriggers.length == 0}
+						<div class="text-xs text-secondary text-center"> No nats triggers </div>
+					{:else}
+						<div class="flex flex-col divide-y pt-2">
+							{#each natsTriggers as natsTrigger (natsTrigger.path)}
+								<div class="grid grid-cols-5 text-2xs items-center py-2">
+									<div class="col-span-2 truncate">{natsTrigger.path}</div>
+									<div class="col-span-2 truncate">
+										{natsTrigger.nats_resource_path}
+									</div>
+									<div class="flex justify-end">
+										<button
+											on:click={() => natsTriggerEditor?.openEdit(natsTrigger.path, isFlow)}
+											class="px-2"
+										>
+											{#if natsTrigger.canWrite}
+												Edit
+											{:else}
+												View
+											{/if}
+										</button>
+									</div>
+								</div>
+							{/each}
+						</div>
+					{/if}
+				</Section>
+			{:else}
+				<Skeleton layout={[[8]]} />
+			{/if}
+		{/if}
+	</div>
+{/if}
@@ -13,6 +13,10 @@
 	import WebsocketTriggersPanel from './WebsocketTriggersPanel.svelte'
 	import ScheduledPollPanel from './ScheduledPollPanel.svelte'
 	import KafkaTriggersPanel from './KafkaTriggersPanel.svelte'
+	import NatsTriggersPanel from './NatsTriggersPanel.svelte'
+	import ToggleButtonGroup from '../common/toggleButton-v2/ToggleButtonGroup.svelte'
+	import ToggleButton from '../common/toggleButton-v2/ToggleButton.svelte'
+	import { KafkaIcon, NatsIcon } from '../icons'

 	export let noEditor: boolean
 	export let newItem = false
@@ -24,6 +28,14 @@
 	export let canHavePreprocessor: boolean = false
 	export let hasPreprocessor: boolean = false

+	let eventStreamType: 'kafka' | 'nats' = 'kafka'
+
+	$: {
+		if ($selectedTrigger === 'kafka' || $selectedTrigger === 'nats') {
+			eventStreamType = $selectedTrigger
+		}
+	}
+
 	const { selectedTrigger, simplifiedPoll } = getContext<TriggerContext>('TriggerContext')

 	const dispatch = createEventDispatcher()
@@ -40,7 +52,9 @@
 			<Tab value="schedules" selectedClass="text-primary text-sm font-semibold">Schedules</Tab>
 			<Tab value="routes" selectedClass="text-primary text-sm font-semibold">HTTP</Tab>
 			<Tab value="websockets" selectedClass="text-primary text-sm font-semibold">Websockets</Tab>
-			<Tab value="kafka" selectedClass="text-primary text-sm font-semibold">Kafka</Tab>
+			<Tab value="kafka" otherValues={['nats']} selectedClass="text-primary text-sm font-semibold"
+				>Event streams</Tab
+			>
 			<Tab value="emails" selectedClass="text-primary text-sm font-semibold">Email</Tab>
 			{#if isFlow}
 				<Tab value="scheduledPoll" selectedClass="text-primary text-sm font-semibold"
@@ -105,18 +119,35 @@
 							{hasPreprocessor}
 						/>
 					</div>
-				{:else if $selectedTrigger === 'kafka'}
-					<div class="p-4">
-						<KafkaTriggersPanel
-							on:applyArgs
-							on:addPreprocessor
-							{newItem}
-							path={currentPath}
-							{isFlow}
-							isEditor={true}
-							{canHavePreprocessor}
-							{hasPreprocessor}
-						/>
+				{:else if $selectedTrigger === 'kafka' || $selectedTrigger === 'nats'}
+					<div class="p-4 flex flex-col gap-2">
+						<ToggleButtonGroup bind:selected={eventStreamType}>
+							<ToggleButton value="kafka" label="Kafka" icon={KafkaIcon} />
+							<ToggleButton value="nats" label="NATS" icon={NatsIcon} />
+						</ToggleButtonGroup>
+						{#if eventStreamType === 'kafka'}
+							<KafkaTriggersPanel
+								on:applyArgs
+								on:addPreprocessor
+								{newItem}
+								path={currentPath}
+								{isFlow}
+								isEditor={true}
+								{canHavePreprocessor}
+								{hasPreprocessor}
+							/>
+						{:else if eventStreamType === 'nats'}
+							<NatsTriggersPanel
+								on:applyArgs
+								on:addPreprocessor
+								{newItem}
+								path={currentPath}
+								{isFlow}
+								isEditor={true}
+								{canHavePreprocessor}
+								{hasPreprocessor}
+							/>
+						{/if}
 					</div>
 				{:else if $selectedTrigger === 'schedules'}
 					<div class="p-4">
@@ -28,7 +28,8 @@
 		websocket: 'New websocket trigger',
 		webhook: 'Webhook',
 		kafka: 'New kafka trigger',
-		email: 'Email trigger'
+		email: 'Email trigger',
+		nats: 'NATS trigger'
 	}

 	const { captureOn } = getContext<TriggerContext>('TriggerContext')
@@ -7,6 +7,7 @@
 	import WebhooksConfigSection from './WebhooksConfigSection.svelte'
 	import EmailTriggerConfigSection from '../details/EmailTriggerConfigSection.svelte'
 	import KafkaTriggersConfigSection from './KafkaTriggersConfigSection.svelte'
+	import NatsTriggersConfigSection from './NatsTriggersConfigSection.svelte'

 	export let triggerType: CaptureTriggerKind = 'webhook'
 	export let cloudDisabled: boolean = false
@@ -58,5 +59,7 @@
 		/>
 	{:else if triggerType === 'kafka'}
 		<KafkaTriggersConfigSection headless={true} bind:args staticInputDisabled={false} />
+	{:else if triggerType === 'nats'}
+		<NatsTriggersConfigSection headless={true} bind:args staticInputDisabled={false} {path} />
 	{/if}
 </div>
@@ -564,7 +564,7 @@ export async function main(approver?: string) {
 export const BUN_PREPROCESSOR_MODULE_CODE = `
 export async function preprocessor(
 	wm_trigger: {
-		kind: 'http' | 'email' | 'webhook' | 'websocket' | 'kafka',
+		kind: 'http' | 'email' | 'webhook' | 'websocket' | 'kafka' | 'nats',
 		http?: {
 			route: string // The route path, e.g. "/users/:id"
 			path: string // The actual path called, e.g. "/users/123"
@@ -580,6 +580,14 @@ export async function preprocessor(
 			brokers: string[]
 			topic: string
 			group_id: string
+		},
+		nats?: {
+			servers: string[]
+			subject: string
+			headers?: Record<string, string[]>
+			status?: number
+			description?: string
+			length: number
 		}
 	},
 	/* your other args */ 
@@ -593,7 +601,7 @@ export async function preprocessor(
 const DENO_PREPROCESSOR_MODULE_CODE = `
 export async function preprocessor(
 	wm_trigger: {
-		kind: 'http' | 'email' | 'webhook' | 'websocket' | 'kafka',
+		kind: 'http' | 'email' | 'webhook' | 'websocket' | 'kafka' | 'nats',
 		http?: {
 			route: string // The route path, e.g. "/users/:id"
 			path: string // The actual path called, e.g. "/users/123"
@@ -609,6 +617,14 @@ export async function preprocessor(
 			brokers: string[]
 			topic: string
 			group_id: string
+		},
+		nats?: {
+			servers: string[]
+			subject: string
+			headers?: Record<string, string[]>
+			status?: number
+			description?: string
+			length: number
 		}
 	},
 	/* your other args */ 
@@ -664,11 +680,20 @@ class Kafka(TypedDict):
 	brokers: list[str]
 	group_id: str

+class Nats(TypedDict):
+	servers: list[str]
+	subject: str
+	headers: dict[str, list[str]] | None
+	status: int | None
+	description: str | None
+	length: int
+
 class WmTrigger(TypedDict):
-	kind: Literal["http", "email", "webhook", "websocket", "kafka"]
+	kind: Literal["http", "email", "webhook", "websocket", "kafka", "nats"]
 	http: Http | None
 	websocket: Websocket | None
 	kafka: Kafka | None
+	nats: Nats | None

 def preprocessor(
 	wm_trigger: WmTrigger,
@@ -29,9 +29,7 @@
 		defaultScripts,
 		hubBaseUrlStore,
 		usedTriggerKinds,
-
 		devopsRole
-
 	} from '$lib/stores'
 	import CenteredModal from '$lib/components/CenteredModal.svelte'
 	import { afterNavigate, beforeNavigate } from '$app/navigation'
@@ -193,11 +191,10 @@

 	async function loadUsedTriggerKinds() {
 		let usedKinds: string[] = []
-		const { http_routes_used, websocket_used, kafka_used } = await WorkspaceService.getUsedTriggers(
-			{
+		const { http_routes_used, websocket_used, kafka_used, nats_used } =
+			await WorkspaceService.getUsedTriggers({
 				workspace: $workspaceStore ?? ''
-			}
-		)
+			})
 		if (http_routes_used) {
 			usedKinds.push('http')
 		}
@@ -207,6 +204,9 @@
 		if (kafka_used) {
 			usedKinds.push('kafka')
 		}
+		if (nats_used) {
+			usedKinds.push('nats')
+		}
 		$usedTriggerKinds = usedKinds
 	}

@@ -62,6 +62,7 @@
 	import TriggersBadge from '$lib/components/graph/renderers/triggers/TriggersBadge.svelte'
 	import WebsocketTriggersPanel from '$lib/components/triggers/WebsocketTriggersPanel.svelte'
 	import KafkaTriggersPanel from '$lib/components/triggers/KafkaTriggersPanel.svelte'
+	import NatsTriggersPanel from '$lib/components/triggers/NatsTriggersPanel.svelte'

 	let flow: Flow | undefined
 	let can_write = false
@@ -547,6 +548,12 @@
 			</div>
 		</svelte:fragment>

+		<svelte:fragment slot="nats">
+			<div class="p-2">
+				<NatsTriggersPanel path={flow.path ?? ''} isFlow />
+			</div>
+		</svelte:fragment>
+
 		<svelte:fragment slot="emails">
 			<div class="p-2">
 				<EmailTriggerPanel
@@ -58,7 +58,8 @@
 						...triggers[i],
 						error: newTrigger.error,
 						last_server_ping: newTrigger.last_server_ping,
-						enabled: newTrigger.enabled
+						enabled: newTrigger.enabled,
+						server_id: newTrigger.server_id
 					}
 				}
 			}
@@ -260,9 +261,10 @@
 			<div class="text-center text-sm text-tertiary mt-2"> No kafka triggers </div>
 		{:else if items?.length}
 			<div class="border rounded-md divide-y">
-				{#each items.slice(0, nbDisplayed) as { path, edited_by, edited_at, script_path, is_flow, kafka_resource_path, topics, extra_perms, canWrite, marked, error, last_server_ping, enabled } (path)}
+				{#each items.slice(0, nbDisplayed) as { path, edited_by, edited_at, script_path, is_flow, kafka_resource_path, topics, extra_perms, canWrite, marked, server_id, error, last_server_ping, enabled } (path)}
 					{@const href = `${is_flow ? '/flows/get' : '/scripts/get'}/${script_path}`}
 					{@const ping = last_server_ping ? new Date(last_server_ping) : undefined}
+					{@const pinging = ping && ping.getTime() > new Date().getTime() - 15 * 1000}

 					<div
 						class="hover:bg-surface-hover w-full items-center px-4 py-2 gap-4 first-of-type:!border-t-0
@@ -298,7 +300,7 @@
 							</div>

 							<div class="w-10">
-								{#if (enabled && (!ping || ping.getTime() < new Date().getTime() - 15 * 1000 || error)) || (!enabled && error)}
+								{#if (enabled && (!pinging || error)) || (!enabled && error) || (enabled && !server_id)}
 									<Popover notClickable>
 										<span class="flex h-4 w-4">
 											<Circle
@@ -309,7 +311,11 @@
 										</span>
 										<div slot="text">
 											{#if enabled}
-												Consumer is not connected{error ? ': ' + error : ''}
+												{#if !server_id}
+													Consumer is starting...
+												{:else}
+													Consumer is not connected{error ? ': ' + error : ''}
+												{/if}
 											{:else}
 												Consumer was disabled because of an error: {error}
 											{/if}
@@ -0,0 +1,5 @@
+export function load() {
+	return {
+		stuff: { title: 'NATS triggers' }
+	}
+}
@@ -0,0 +1,428 @@
+<script lang="ts">
+	import { NatsTriggerService, type NatsTrigger } from '$lib/gen'
+	import {
+		canWrite,
+		displayDate,
+		getLocalSetting,
+		sendUserToast,
+		storeLocalSetting
+	} from '$lib/utils'
+	import { base } from '$app/paths'
+	import CenteredPage from '$lib/components/CenteredPage.svelte'
+	import { Alert, Button, Skeleton } from '$lib/components/common'
+	import Dropdown from '$lib/components/DropdownV2.svelte'
+	import PageHeader from '$lib/components/PageHeader.svelte'
+	import SharedBadge from '$lib/components/SharedBadge.svelte'
+	import ShareModal from '$lib/components/ShareModal.svelte'
+	import Toggle from '$lib/components/Toggle.svelte'
+	import { userStore, workspaceStore } from '$lib/stores'
+	import { Code, Eye, Pen, Plus, Share, Trash, Circle } from 'lucide-svelte'
+	import { goto } from '$lib/navigation'
+	import SearchItems from '$lib/components/SearchItems.svelte'
+	import NoItemFound from '$lib/components/home/NoItemFound.svelte'
+	import RowIcon from '$lib/components/common/table/RowIcon.svelte'
+	import ListFilters from '$lib/components/home/ListFilters.svelte'
+	import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
+	import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
+	import { setQuery } from '$lib/navigation'
+	import { onDestroy, onMount } from 'svelte'
+	import NatsTriggerEditor from '$lib/components/triggers/NatsTriggerEditor.svelte'
+	import Popover from '$lib/components/Popover.svelte'
+	import { isCloudHosted } from '$lib/cloud'
+	import NatsIcon from '$lib/components/icons/NatsIcon.svelte'
+
+	type TriggerW = NatsTrigger & { canWrite: boolean }
+
+	let triggers: TriggerW[] = []
+	let shareModal: ShareModal
+	let loading = true
+
+	async function loadTriggers(): Promise<void> {
+		triggers = (await NatsTriggerService.listNatsTriggers({ workspace: $workspaceStore! })).map(
+			(x) => {
+				return { canWrite: canWrite(x.path, x.extra_perms!, $userStore), ...x }
+			}
+		)
+		loading = false
+	}
+
+	let interval = setInterval(async () => {
+		try {
+			const newTriggers = await NatsTriggerService.listNatsTriggers({
+				workspace: $workspaceStore!
+			})
+			for (let i = 0; i < triggers.length; i++) {
+				const newTrigger = newTriggers.find((x) => x.path === triggers[i].path)
+				if (newTrigger) {
+					triggers[i] = {
+						...triggers[i],
+						error: newTrigger.error,
+						last_server_ping: newTrigger.last_server_ping,
+						enabled: newTrigger.enabled,
+						server_id: newTrigger.server_id
+					}
+				}
+			}
+		} catch (err) {
+			console.error(err)
+		}
+	}, 5000)
+
+	onDestroy(() => {
+		clearInterval(interval)
+	})
+
+	async function setTriggerEnabled(path: string, enabled: boolean): Promise<void> {
+		try {
+			await NatsTriggerService.setNatsTriggerEnabled({
+				path,
+				workspace: $workspaceStore!,
+				requestBody: { enabled }
+			})
+		} catch (err) {
+			sendUserToast(
+				`Cannot ` + (enabled ? 'enable' : 'disable') + ` nats trigger: ${err.body}`,
+				true
+			)
+		} finally {
+			loadTriggers()
+		}
+	}
+
+	$: {
+		if ($workspaceStore && $userStore) {
+			loadTriggers()
+		}
+	}
+	let natsTriggerEditor: NatsTriggerEditor
+
+	let filteredItems: (TriggerW & { marked?: any })[] | undefined = []
+	let items: typeof filteredItems | undefined = []
+	let preFilteredItems: typeof filteredItems | undefined = []
+	let filter = ''
+	let ownerFilter: string | undefined = undefined
+	let nbDisplayed = 15
+
+	const TRIGGER_PATH_KIND_FILTER_SETTING = 'filter_path_of'
+	const FILTER_USER_FOLDER_SETTING_NAME = 'user_and_folders_only'
+	let selectedFilterKind =
+		(getLocalSetting(TRIGGER_PATH_KIND_FILTER_SETTING) as 'trigger' | 'script_flow') ?? 'trigger'
+	let filterUserFolders = getLocalSetting(FILTER_USER_FOLDER_SETTING_NAME) == 'true'
+
+	$: storeLocalSetting(TRIGGER_PATH_KIND_FILTER_SETTING, selectedFilterKind)
+	$: storeLocalSetting(FILTER_USER_FOLDER_SETTING_NAME, filterUserFolders ? 'true' : undefined)
+
+	function filterItemsPathsBaseOnUserFilters(
+		item: TriggerW,
+		selectedFilterKind: 'trigger' | 'script_flow',
+		filterUserFolders: boolean
+	) {
+		if ($workspaceStore == 'admins') return true
+		if (filterUserFolders) {
+			if (selectedFilterKind === 'trigger') {
+				return (
+					!item.path.startsWith('u/') || item.path.startsWith('u/' + $userStore?.username + '/')
+				)
+			} else {
+				return (
+					!item.script_path.startsWith('u/') ||
+					item.script_path.startsWith('u/' + $userStore?.username + '/')
+				)
+			}
+		} else {
+			return true
+		}
+	}
+
+	$: preFilteredItems =
+		ownerFilter != undefined
+			? selectedFilterKind === 'trigger'
+				? triggers?.filter(
+						(x) =>
+							x.path.startsWith(ownerFilter + '/') &&
+							filterItemsPathsBaseOnUserFilters(x, selectedFilterKind, filterUserFolders)
+				  )
+				: triggers?.filter(
+						(x) =>
+							x.script_path.startsWith(ownerFilter + '/') &&
+							filterItemsPathsBaseOnUserFilters(x, selectedFilterKind, filterUserFolders)
+				  )
+			: triggers?.filter((x) =>
+					filterItemsPathsBaseOnUserFilters(x, selectedFilterKind, filterUserFolders)
+			  )
+
+	$: if ($workspaceStore) {
+		ownerFilter = undefined
+	}
+
+	$: owners =
+		selectedFilterKind === 'trigger'
+			? Array.from(
+					new Set(filteredItems?.map((x) => x.path.split('/').slice(0, 2).join('/')) ?? [])
+			  ).sort()
+			: Array.from(
+					new Set(filteredItems?.map((x) => x.script_path.split('/').slice(0, 2).join('/')) ?? [])
+			  ).sort()
+
+	$: items = filter !== '' ? filteredItems : preFilteredItems
+
+	function updateQueryFilters(selectedFilterKind, filterUserFolders) {
+		setQuery(
+			new URL(window.location.href),
+			TRIGGER_PATH_KIND_FILTER_SETTING,
+			selectedFilterKind
+		).then(() => {
+			setQuery(
+				new URL(window.location.href),
+				FILTER_USER_FOLDER_SETTING_NAME,
+				String(filterUserFolders)
+			)
+		})
+	}
+
+	function loadQueryFilters() {
+		let url = new URL(window.location.href)
+		let queryFilterKind = url.searchParams.get(TRIGGER_PATH_KIND_FILTER_SETTING)
+		let queryFilterUserFolders = url.searchParams.get(FILTER_USER_FOLDER_SETTING_NAME)
+		if (queryFilterKind) {
+			selectedFilterKind = queryFilterKind as 'trigger' | 'script_flow'
+		}
+		if (queryFilterUserFolders) {
+			filterUserFolders = queryFilterUserFolders == 'true'
+		}
+	}
+
+	onMount(() => {
+		loadQueryFilters()
+	})
+
+	$: updateQueryFilters(selectedFilterKind, filterUserFolders)
+</script>
+
+<NatsTriggerEditor on:update={loadTriggers} bind:this={natsTriggerEditor} />
+
+<SearchItems
+	{filter}
+	items={preFilteredItems}
+	bind:filteredItems
+	f={(x) => (x.summary ?? '') + ' ' + x.path + ' (' + x.script_path + ')'}
+/>
+
+<CenteredPage>
+	<PageHeader
+		title="NATS triggers"
+		tooltip="Windmill can consume nats events and trigger scripts or flows based on them."
+	>
+		<Button size="md" startIcon={{ icon: Plus }} on:click={() => natsTriggerEditor.openNew(false)}>
+			New&nbsp;NATS trigger
+		</Button>
+	</PageHeader>
+
+	{#if isCloudHosted()}
+		<Alert title="Not compatible with multi-tenant cloud" type="warning">
+			NATS triggers are disabled in the multi-tenant cloud.
+		</Alert>
+		<div class="py-4" />
+	{/if}
+	<div class="w-full h-full flex flex-col">
+		<div class="w-full pb-4 pt-6">
+			<input
+				type="text"
+				placeholder="Search nats triggers"
+				bind:value={filter}
+				class="search-item"
+			/>
+			<div class="flex flex-row items-center gap-2 mt-6">
+				<div class="text-sm shrink-0"> Filter by path of </div>
+				<ToggleButtonGroup bind:selected={selectedFilterKind}>
+					<ToggleButton small value="trigger" label="Nats trigger" icon={NatsIcon} />
+					<ToggleButton small value="script_flow" label="Script/Flow" icon={Code} />
+				</ToggleButtonGroup>
+			</div>
+			<ListFilters syncQuery bind:selectedFilter={ownerFilter} filters={owners} />
+
+			<div class="flex flex-row items-center justify-end gap-4">
+				{#if $userStore?.is_super_admin && $userStore.username.includes('@')}
+					<Toggle size="xs" bind:checked={filterUserFolders} options={{ right: 'Only f/*' }} />
+				{:else if $userStore?.is_admin || $userStore?.is_super_admin}
+					<Toggle
+						size="xs"
+						bind:checked={filterUserFolders}
+						options={{ right: `Only u/${$userStore.username} and f/*` }}
+					/>
+				{/if}
+			</div>
+		</div>
+		{#if loading}
+			{#each new Array(6) as _}
+				<Skeleton layout={[[6], 0.4]} />
+			{/each}
+		{:else if !triggers?.length}
+			<div class="text-center text-sm text-tertiary mt-2"> No nats triggers </div>
+		{:else if items?.length}
+			<div class="border rounded-md divide-y">
+				{#each items.slice(0, nbDisplayed) as { path, edited_by, edited_at, script_path, is_flow, nats_resource_path, subjects, extra_perms, canWrite, marked, server_id, error, last_server_ping, enabled } (path)}
+					{@const href = `${is_flow ? '/flows/get' : '/scripts/get'}/${script_path}`}
+					{@const ping = last_server_ping ? new Date(last_server_ping) : undefined}
+					{@const pinging = ping && ping.getTime() > new Date().getTime() - 15 * 1000}
+
+					<div
+						class="hover:bg-surface-hover w-full items-center px-4 py-2 gap-4 first-of-type:!border-t-0
+				first-of-type:rounded-t-md last-of-type:rounded-b-md flex flex-col"
+					>
+						<div class="w-full flex gap-5 items-center">
+							<RowIcon kind={is_flow ? 'flow' : 'script'} />
+
+							<a
+								href="#{path}"
+								on:click={() => natsTriggerEditor?.openEdit(path, is_flow)}
+								class="min-w-0 grow hover:underline decoration-gray-400"
+							>
+								<div class="text-primary flex-wrap text-left text-md font-semibold mb-1 truncate">
+									{#if marked}
+										<span class="text-xs">
+											{@html marked}
+										</span>
+									{:else}
+										{nats_resource_path} - {subjects.join(', ')}
+									{/if}
+								</div>
+								<div class="text-secondary text-xs truncate text-left font-light">
+									{path}
+								</div>
+								<div class="text-secondary text-xs truncate text-left font-light">
+									runnable: {script_path}
+								</div>
+							</a>
+
+							<div class="hidden lg:flex flex-row gap-1 items-center">
+								<SharedBadge {canWrite} extraPerms={extra_perms} />
+							</div>
+
+							<div class="w-10">
+								{#if (enabled && (!pinging || error)) || (!enabled && error) || (enabled && !server_id)}
+									<Popover notClickable>
+										<span class="flex h-4 w-4">
+											<Circle
+												class="text-red-600 animate-ping absolute inline-flex fill-current"
+												size={12}
+											/>
+											<Circle class="text-red-600 relative inline-flex fill-current" size={12} />
+										</span>
+										<div slot="text">
+											{#if enabled}
+												{#if !server_id}
+													Consumer is starting...
+												{:else}
+													Consumer is not connected{error ? ': ' + error : ''}
+												{/if}
+											{:else}
+												Consumer was disabled because of an error: {error}
+											{/if}
+										</div>
+									</Popover>
+								{:else if enabled}
+									<Popover notClickable>
+										<span class="flex h-4 w-4">
+											<Circle class="text-green-600 relative inline-flex fill-current" size={12} />
+										</span>
+										<div slot="text"> Consumer is connected </div>
+									</Popover>
+								{/if}
+							</div>
+
+							<Toggle
+								checked={enabled}
+								disabled={!canWrite}
+								on:change={(e) => {
+									setTriggerEnabled(path, e.detail)
+								}}
+							/>
+
+							<div class="flex gap-2 items-center justify-end">
+								<Button
+									on:click={() => natsTriggerEditor?.openEdit(path, is_flow)}
+									size="xs"
+									startIcon={canWrite
+										? { icon: Pen }
+										: {
+												icon: Eye
+										  }}
+									color="dark"
+								>
+									{canWrite ? 'Edit' : 'View'}
+								</Button>
+								<Dropdown
+									items={[
+										{
+											displayName: `View ${is_flow ? 'Flow' : 'Script'}`,
+											icon: Eye,
+											action: () => {
+												goto(href)
+											}
+										},
+										{
+											displayName: 'Delete',
+											type: 'delete',
+											icon: Trash,
+											disabled: !canWrite,
+											action: async () => {
+												await NatsTriggerService.deleteNatsTrigger({
+													workspace: $workspaceStore ?? '',
+													path
+												})
+												loadTriggers()
+											}
+										},
+										{
+											displayName: canWrite ? 'Edit' : 'View',
+											icon: canWrite ? Pen : Eye,
+											action: () => {
+												natsTriggerEditor?.openEdit(path, is_flow)
+											}
+										},
+										{
+											displayName: 'Audit logs',
+											icon: Eye,
+											href: `${base}/audit_logs?resource=${path}`
+										},
+										{
+											displayName: canWrite ? 'Share' : 'See Permissions',
+											icon: Share,
+											action: () => {
+												shareModal.openDrawer(path, 'nats_trigger')
+											}
+										}
+									]}
+								/>
+							</div>
+						</div>
+						<div class="w-full flex justify-between items-baseline">
+							<div
+								class="flex flex-wrap text-[0.7em] text-tertiary gap-1 items-center justify-end truncate pr-2"
+								><div class="truncate">edited by {edited_by}</div><div class="truncate"
+									>the {displayDate(edited_at)}</div
+								></div
+							></div
+						>
+					</div>
+				{/each}
+			</div>
+		{:else}
+			<NoItemFound />
+		{/if}
+	</div>
+	{#if items && items?.length > 15 && nbDisplayed < items.length}
+		<span class="text-xs"
+			>{nbDisplayed} items out of {items.length}
+			<button class="ml-4" on:click={() => (nbDisplayed += 30)}>load 30 more</button></span
+		>
+	{/if}
+</CenteredPage>
+
+<ShareModal
+	bind:this={shareModal}
+	on:change={() => {
+		loadTriggers()
+	}}
+/>
@@ -85,6 +85,7 @@
 	import TriggersBadge from '$lib/components/graph/renderers/triggers/TriggersBadge.svelte'
 	import WebsocketTriggersPanel from '$lib/components/triggers/WebsocketTriggersPanel.svelte'
 	import KafkaTriggersPanel from '$lib/components/triggers/KafkaTriggersPanel.svelte'
+	import NatsTriggersPanel from '$lib/components/triggers/NatsTriggersPanel.svelte'

 	let script: Script | undefined
 	let topHash: string | undefined
@@ -727,6 +728,11 @@
 					<KafkaTriggersPanel path={script.path ?? ''} isFlow={false} />
 				</div>
 			</svelte:fragment>
+			<svelte:fragment slot="nats">
+				<div class="p-2">
+					<NatsTriggersPanel path={script.path ?? ''} isFlow={false} />
+				</div>
+			</svelte:fragment>
 			<svelte:fragment slot="emails">
 				<div class="p-2">
 					<EmailTriggerPanel
@@ -57,7 +57,8 @@
 						...triggers[i],
 						error: newTrigger.error,
 						last_server_ping: newTrigger.last_server_ping,
-						enabled: newTrigger.enabled
+						enabled: newTrigger.enabled,
+						server_id: newTrigger.server_id
 					}
 				}
 			}
@@ -258,9 +259,10 @@
 			<div class="text-center text-sm text-tertiary mt-2"> No websocket triggers </div>
 		{:else if items?.length}
 			<div class="border rounded-md divide-y">
-				{#each items.slice(0, nbDisplayed) as { path, edited_by, edited_at, script_path, url, is_flow, extra_perms, canWrite, marked, error, last_server_ping, enabled } (path)}
+				{#each items.slice(0, nbDisplayed) as { path, edited_by, edited_at, script_path, url, is_flow, extra_perms, canWrite, marked, error, last_server_ping, server_id, enabled } (path)}
 					{@const href = `${is_flow ? '/flows/get' : '/scripts/get'}/${script_path}`}
 					{@const ping = last_server_ping ? new Date(last_server_ping) : undefined}
+					{@const pinging = ping && ping.getTime() > new Date().getTime() - 15 * 1000}

 					<div
 						class="hover:bg-surface-hover w-full items-center px-4 py-2 gap-4 first-of-type:!border-t-0
@@ -300,7 +302,7 @@
 							</div>

 							<div class="w-10">
-								{#if (enabled && (!ping || ping.getTime() < new Date().getTime() - 15 * 1000 || error)) || (!enabled && error)}
+								{#if (enabled && (!pinging || error)) || (!enabled && error) || (enabled && !server_id)}
 									<Popover notClickable>
 										<span class="flex h-4 w-4">
 											<Circle
@@ -311,7 +313,11 @@
 										</span>
 										<div slot="text">
 											{#if enabled}
-												Websocket is not connected{error ? ': ' + error : ''}
+												{#if !server_id}
+													Websocket is starting...
+												{:else}
+													Websocket is not connected{error ? ': ' + error : ''}
+												{/if}
 											{:else}
 												Websocket was disabled because of an error: {error}
 											{/if}
@@ -322,7 +328,9 @@
 										<span class="flex h-4 w-4">
 											<Circle class="text-green-600 relative inline-flex fill-current" size={12} />
 										</span>
-										<div slot="text"> Websocket is connected </div>
+										<div slot="text">
+											Websocket is connected{!server_id ? ' (shutting down...)' : ''}
+										</div>
 									</Popover>
 								{/if}
 							</div>