feat: kafka triggers (#4713)

* feat: kafka triggers * sqlx * fix build * improve error messages on windows * nit * fix build * missing action * nit * maybe fix ssl on windows * nit * update ee ref --------- Co-authored-by: Ruben Fiszel <ruben@rubenfiszel.com> Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
2024-11-18 17:03:50 +01:00
parent 6f4a6d8a72
commit 2a3c0d55fd
51 changed files with 1736 additions and 92 deletions
--- a/.github/workflows/build-publish-rh-image.yml
+++ b/.github/workflows/build-publish-rh-image.yml
@@ -3,8 +3,7 @@ env:
  IMAGE_NAME: ${{ github.repository }}

 name: Build and publish windmill for RHEL9
-on:
-  workflow_dispatch
+on: workflow_dispatch

 permissions: write-all

@@ -65,7 +64,7 @@ jobs:
          platforms: linux/amd64
          push: true
          build-args: |
-            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core
+            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core,kafka
          secrets: |
            rh_username=${{ secrets.RH_USERNAME }}
            rh_password=${{ secrets.RH_PASSWORD }}
@@ -74,7 +73,7 @@ jobs:
          labels: |
            ${{ steps.meta-ee-public.outputs.labels }}-amd64
            org.opencontainers.image.licenses=Windmill-Enterprise-License
-      
+
      - name: Build and push publicly ee arm64
        uses: depot/build-push-action@v1
        with:
@@ -82,7 +81,7 @@ jobs:
          platforms: linux/arm64
          push: true
          build-args: |
-            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core
+            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core,kafka
          secrets: |
            rh_username=${{ secrets.RH_USERNAME }}
            rh_password=${{ secrets.RH_PASSWORD }}
@@ -108,7 +107,7 @@ jobs:
        run: |
          mv "${{ steps.extract-ee-amd64.outputs.destination }}/windmill" "${{ steps.extract-ee-amd64.outputs.destination }}/windmill-ee-amd64-rhel9"
          mv "${{ steps.extract-ee-arm64.outputs.destination }}/windmill" "${{ steps.extract-ee-arm64.outputs.destination }}/windmill-ee-arm64-rhel9"
-      
+
      - uses: actions/upload-artifact@v4
        with:
          name: RHEL9-amd64 build
--- a/.github/workflows/build-staging-image.yml
+++ b/.github/workflows/build-staging-image.yml
@@ -62,7 +62,7 @@ jobs:
          platforms: linux/amd64,linux/arm64
          push: true
          build-args: |
-            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core
+            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core,kafka
          tags: |
            ${{ steps.meta-ee-public.outputs.tags }}
          labels: |
--- a/.github/workflows/build_windows_worker_.yml
+++ b/.github/workflows/build_windows_worker_.yml
@@ -45,7 +45,7 @@ jobs:
          $env:OPENSSL_DIR="${Env:VCPKG_INSTALLATION_ROOT}\installed\x64-windows-static"
          mkdir frontend/build && cd backend
          New-Item -Path . -Name "windmill-api/openapi-deref.yaml" -ItemType "File" -Force
-          cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core
+          cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka

      - name: Rename binary with corresponding architecture
        run: |
--- a/.github/workflows/docker-image.yml
+++ b/.github/workflows/docker-image.yml
@@ -1,10 +1,8 @@
 env:
  REGISTRY: ghcr.io
-  IMAGE_NAME:
-    ${{ github.event_name != 'pull_request' && github.repository ||
+  IMAGE_NAME: ${{ github.event_name != 'pull_request' && github.repository ||
    'windmill-labs/windmill-test' }}
-  DEV_SHA:
-    ${{ github.event_name != 'pull_request' && 'dev' || format('pr-{0}',
+  DEV_SHA: ${{ github.event_name != 'pull_request' && 'dev' || format('pr-{0}',
    github.event.number) }}

 name: Build windmill:main
@@ -140,7 +138,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
+            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka
          tags: |
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee:${{ env.DEV_SHA }}
            ${{ steps.meta-ee-public.outputs.tags }}
@@ -202,7 +200,7 @@ jobs:
          platforms: linux/amd64
          push: true
          build-args: |
-            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core
+            features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka
            PYTHON_IMAGE=python:3.12.2-slim-bookworm
          tags: |
            ${{ steps.meta-ee-public-py312.outputs.tags }}
--- a/.github/workflows/publish_windows_worker.yml
+++ b/.github/workflows/publish_windows_worker.yml
@@ -47,7 +47,7 @@ jobs:
          $env:OPENSSL_DIR="${Env:VCPKG_INSTALLATION_ROOT}\installed\x64-windows-static"
          mkdir frontend/build && cd backend
          New-Item -Path . -Name "windmill-api/openapi-deref.yaml" -ItemType "File" -Force
-          cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core
+          cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka

      - name: Rename binary with corresponding architecture
        run: |
--- a/backend/.sqlx/query-2139f1fb1877294bbf55d786000c5c32f582a3911efcf88e437aa90d7d5a49b5.json
+++ b/backend/.sqlx/query-2139f1fb1877294bbf55d786000c5c32f582a3911efcf88e437aa90d7d5a49b5.json
@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COUNT(*) FROM kafka_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": "2139f1fb1877294bbf55d786000c5c32f582a3911efcf88e437aa90d7d5a49b5"
+}
--- a/backend/.sqlx/query-3997dcf2c11817e59bf50fd896381d870b847a8ba07e6197212fdf85ec901b09.json
+++ b/backend/.sqlx/query-3997dcf2c11817e59bf50fd896381d870b847a8ba07e6197212fdf85ec901b09.json
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM kafka_trigger WHERE workspace_id = $1 AND path = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "3997dcf2c11817e59bf50fd896381d870b847a8ba07e6197212fdf85ec901b09"
+}
--- a/backend/.sqlx/query-3e64c894c89ef82c4527180b82c4e82c1b7060ba0ca288dc6a7c7afcd76212e8.json
+++ b/backend/.sqlx/query-3e64c894c89ef82c4527180b82c4e82c1b7060ba0ca288dc6a7c7afcd76212e8.json
@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE kafka_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": "3e64c894c89ef82c4527180b82c4e82c1b7060ba0ca288dc6a7c7afcd76212e8"
+}
--- a/backend/.sqlx/query-5962733746c81480abe9ab3a6ccf5664130ae7500279055c08db42d67b2822f6.json
+++ b/backend/.sqlx/query-5962733746c81480abe9ab3a6ccf5664130ae7500279055c08db42d67b2822f6.json
@@ -0,0 +1,23 @@
+{
+  "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",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "VarcharArray",
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5962733746c81480abe9ab3a6ccf5664130ae7500279055c08db42d67b2822f6"
+}
--- a/backend/.sqlx/query-6d418b5cd7a4df54cfe3ec06e2a957c87f2307a218115ca749ea0e71eca2002e.json
+++ b/backend/.sqlx/query-6d418b5cd7a4df54cfe3ec06e2a957c87f2307a218115ca749ea0e71eca2002e.json
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT 1 FROM kafka_trigger WHERE path = $1 AND workspace_id = $2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "6d418b5cd7a4df54cfe3ec06e2a957c87f2307a218115ca749ea0e71eca2002e"
+}
--- a/backend/.sqlx/query-6e70ebf078ac04a2933d2f83791973e8fc108d9f32be8a8501391052d76e191e.json
+++ b/backend/.sqlx/query-6e70ebf078ac04a2933d2f83791973e8fc108d9f32be8a8501391052d76e191e.json
@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE kafka_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": "6e70ebf078ac04a2933d2f83791973e8fc108d9f32be8a8501391052d76e191e"
+}
--- a/backend/.sqlx/query-6eb076afcf11dba0dfe35ee108c64814f95dbe5376f0539195d24a8ca96ca3e2.json
+++ b/backend/.sqlx/query-6eb076afcf11dba0dfe35ee108c64814f95dbe5376f0539195d24a8ca96ca3e2.json
@@ -0,0 +1,26 @@
+{
+  "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",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "6eb076afcf11dba0dfe35ee108c64814f95dbe5376f0539195d24a8ca96ca3e2"
+}
--- a/backend/.sqlx/query-cc8a71abdeccf0787695af32aa21fe73aba65cfdc8329c02b4d257de4d2d168a.json
+++ b/backend/.sqlx/query-cc8a71abdeccf0787695af32aa21fe73aba65cfdc8329c02b4d257de4d2d168a.json
@@ -0,0 +1,34 @@
+{
+  "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!\"",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "websocket_used!",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 1,
+        "name": "http_routes_used!",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 2,
+        "name": "kafka_used!",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null,
+      null,
+      null
+    ]
+  },
+  "hash": "cc8a71abdeccf0787695af32aa21fe73aba65cfdc8329c02b4d257de4d2d168a"
+}
--- a/backend/.sqlx/query-d6615719bf8db4b333ed55c9a3c160ad1962668fc3305d8e80ae91ef73614a80.json
+++ b/backend/.sqlx/query-d6615719bf8db4b333ed55c9a3c160ad1962668fc3305d8e80ae91ef73614a80.json
@@ -0,0 +1,24 @@
+{
+  "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"
+}
--- a/backend/.sqlx/query-db24f1f6ef1e26b7de7926bd3f32d1fa673c4970c25ff9ad98f0fdf199801b53.json
+++ b/backend/.sqlx/query-db24f1f6ef1e26b7de7926bd3f32d1fa673c4970c25ff9ad98f0fdf199801b53.json
@@ -1,28 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "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!\"",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "websocket_used!",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 1,
-        "name": "http_routes_used!",
-        "type_info": "Bool"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": [
-      null,
-      null
-    ]
-  },
-  "hash": "db24f1f6ef1e26b7de7926bd3f32d1fa673c4970c25ff9ad98f0fdf199801b53"
-}
--- a/backend/Cargo.lock
+++ b/backend/Cargo.lock
@@ -1378,7 +1378,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2593a3b8b938bd68373196c9832f516be11fa487ef4ae745eb282e6a56a7244"
 dependencies = [
 "once_cell",
- "proc-macro-crate",
+ "proc-macro-crate 3.2.0",
 "proc-macro2",
 "quote",
 "syn 2.0.87",
@@ -5313,7 +5313,7 @@ dependencies = [
 "darling 0.20.10",
 "heck 0.5.0",
 "num-bigint",
- "proc-macro-crate",
+ "proc-macro-crate 3.2.0",
 "proc-macro-error2",
 "proc-macro2",
 "quote",
@@ -5556,6 +5556,27 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "num_enum"
+version = "0.5.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1f646caf906c20226733ed5b1374287eb97e3c2a5c227ce668c1f2ce20ae57c9"
+dependencies = [
+ "num_enum_derive",
+]
+
+[[package]]
+name = "num_enum_derive"
+version = "0.5.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dcbff9bc912032c62bf65ef1d5aea88983b420f4f839db1e9b0c281a25c9c799"
+dependencies = [
+ "proc-macro-crate 1.3.1",
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "number_prefix"
 version = "0.4.0"
@@ -5736,6 +5757,15 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"

+[[package]]
+name = "openssl-src"
+version = "300.4.0+3.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a709e02f2b4aca747929cca5ed248880847c650233cf8b8cdc48f40aaf4898a6"
+dependencies = [
+ "cc",
+]
+
 [[package]]
 name = "openssl-sys"
 version = "0.9.104"
@@ -5744,6 +5774,7 @@ checksum = "45abf306cbf99debc8195b66b7346498d7b10c210de50418b5ccd7ceba08c741"
 dependencies = [
 "cc",
 "libc",
+ "openssl-src",
 "pkg-config",
 "vcpkg",
 ]
@@ -6239,6 +6270,16 @@ dependencies = [
 "elliptic-curve",
 ]

+[[package]]
+name = "proc-macro-crate"
+version = "1.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f4c021e1093a56626774e81216a4ce732a735e5bad4868a03f3ed65ca0c3919"
+dependencies = [
+ "once_cell",
+ "toml_edit 0.19.15",
+]
+
 [[package]]
 name = "proc-macro-crate"
 version = "3.2.0"
@@ -6711,6 +6752,38 @@ dependencies = [
 "crossbeam-utils",
 ]

+[[package]]
+name = "rdkafka"
+version = "0.36.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1beea247b9a7600a81d4cc33f659ce1a77e1988323d7d2809c7ed1c21f4c316d"
+dependencies = [
+ "futures-channel",
+ "futures-util",
+ "libc",
+ "log",
+ "rdkafka-sys",
+ "serde",
+ "serde_derive",
+ "serde_json",
+ "slab",
+ "tokio",
+]
+
+[[package]]
+name = "rdkafka-sys"
+version = "4.7.0+2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "55e0d2f9ba6253f6ec72385e453294f8618e9e15c2c6aba2a5c01ccf9622d615"
+dependencies = [
+ "cmake",
+ "libc",
+ "libz-sys",
+ "num_enum",
+ "openssl-sys",
+ "pkg-config",
+]
+
 [[package]]
 name = "reborrow"
 version = "0.5.5"
@@ -10695,6 +10768,7 @@ dependencies = [
 "prometheus",
 "quick_cache",
 "rand 0.8.5",
+ "rdkafka",
 "regex",
 "reqwest 0.12.9",
 "rsa 0.7.2",
--- a/backend/Cargo.toml
+++ b/backend/Cargo.toml
@@ -62,6 +62,7 @@ jemalloc = ["windmill-common/jemalloc", "dep:tikv-jemallocator", "dep:tikv-jemal
 tantivy = ["dep:windmill-indexer", "windmill-api/tantivy"]
 sqlx = ["windmill-worker/sqlx"]
 deno_core = ["windmill-worker/deno_core", "dep:deno_core"]
+kafka = ["windmill-api/kafka"]

 [dependencies]
 anyhow.workspace = true
@@ -266,6 +267,7 @@ tokio-native-tls = "^0"
 openssl = "=0.10"
 mail-parser = "^0"
 matchit = "=0.7.3"
+rdkafka = { version = "0.36.2", features = ["cmake-build", "ssl-vendored"] }

 datafusion = "39.0.0"
 object_store = { version = "0.10.0", features = ["aws", "azure"] }
--- a/backend/ee-repo-ref.txt
+++ b/backend/ee-repo-ref.txt
@@ -1 +1 @@
-51dcbf93b0d127af9f33fa346cc63fcd2475d4fa
+7cdaac656fde3f3e267dfb9e94dda7f3843aa44e
--- a/backend/migrations/20241112104946_kafka_triggers.down.sql
+++ b/backend/migrations/20241112104946_kafka_triggers.down.sql
@@ -0,0 +1,2 @@
+-- Add down migration script here
+DROP TABLE kafka_trigger;
--- a/backend/migrations/20241112104946_kafka_triggers.up.sql
+++ b/backend/migrations/20241112104946_kafka_triggers.up.sql
@@ -0,0 +1,69 @@
+-- Add up migration script here
+-- Add up migration script here
+
+CREATE TABLE kafka_trigger (
+    path VARCHAR(255) NOT NULL,
+    kafka_resource_path VARCHAR(255) NOT NULL,
+    topics VARCHAR(255)[] NOT NULL,
+    group_id VARCHAR(255) 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)
+);
+
+GRANT ALL ON kafka_trigger TO windmill_user;
+GRANT ALL ON kafka_trigger TO windmill_admin;
+
+ALTER TABLE kafka_trigger ENABLE ROW LEVEL SECURITY;
+
+CREATE POLICY admin_policy ON kafka_trigger FOR ALL TO windmill_admin USING (true);
+
+CREATE POLICY see_folder_extra_perms_user_select ON kafka_trigger FOR SELECT TO windmill_user
+USING (SPLIT_PART(kafka_trigger.path, '/', 1) = 'f' AND SPLIT_PART(kafka_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_read'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_insert ON kafka_trigger FOR INSERT TO windmill_user
+WITH CHECK (SPLIT_PART(kafka_trigger.path, '/', 1) = 'f' AND SPLIT_PART(kafka_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_update ON kafka_trigger FOR UPDATE TO windmill_user
+USING (SPLIT_PART(kafka_trigger.path, '/', 1) = 'f' AND SPLIT_PART(kafka_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_delete ON kafka_trigger FOR DELETE TO windmill_user
+USING (SPLIT_PART(kafka_trigger.path, '/', 1) = 'f' AND SPLIT_PART(kafka_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+
+CREATE POLICY see_own ON kafka_trigger FOR ALL TO windmill_user
+USING (SPLIT_PART(kafka_trigger.path, '/', 1) = 'u' AND SPLIT_PART(kafka_trigger.path, '/', 2) = current_setting('session.user'));
+CREATE POLICY see_member ON kafka_trigger FOR ALL TO windmill_user
+USING (SPLIT_PART(kafka_trigger.path, '/', 1) = 'g' AND SPLIT_PART(kafka_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[]));
+
+CREATE POLICY see_extra_perms_user_select ON kafka_trigger FOR SELECT TO windmill_user
+USING (extra_perms ? CONCAT('u/', current_setting('session.user')));
+CREATE POLICY see_extra_perms_user_insert ON kafka_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 kafka_trigger FOR UPDATE TO windmill_user
+USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+CREATE POLICY see_extra_perms_user_delete ON kafka_trigger FOR DELETE TO windmill_user
+USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+
+CREATE POLICY see_extra_perms_groups_select ON kafka_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 kafka_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 kafka_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 kafka_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));
--- a/backend/windmill-api/Cargo.toml
+++ b/backend/windmill-api/Cargo.toml
@@ -19,6 +19,7 @@ parquet = ["dep:datafusion", "dep:object_store", "dep:url", "windmill-common/par
 prometheus = ["windmill-common/prometheus", "windmill-queue/prometheus", "dep:prometheus"]
 openidconnect = ["dep:openidconnect"]
 tantivy = ["dep:windmill-indexer"]
+kafka = ["dep:rdkafka"]

 [dependencies]
 windmill-queue.workspace = true
@@ -96,6 +97,7 @@ url = { workspace = true, optional = true}
 jsonwebtoken = { workspace = true }
 matchit.workspace = true
 tokio-tungstenite.workspace = true
+rdkafka = { workspace = true, optional = true }

 pin-project.workspace = true
 http.workspace = true
--- a/backend/windmill-api/openapi.yaml
+++ b/backend/windmill-api/openapi.yaml
@@ -2332,9 +2332,12 @@ paths:
                    type: boolean
                  websocket_used:
                    type: boolean
+                  kafka_used:
+                    type: boolean
                required:
                  - http_routes_used
                  - websocket_used
+                  - kafka_used

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

+  /w/{workspace}/kafka_triggers/create:
+    post:
+      summary: create kafka trigger
+      operationId: createKafkaTrigger
+      tags:
+        - kafka_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      requestBody:
+        description: new kafka trigger
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/NewKafkaTrigger"
+      responses:
+        "201":
+          description: kafka trigger created
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /w/{workspace}/kafka_triggers/update/{path}:
+    post:
+      summary: update kafka trigger
+      operationId: updateKafkaTrigger
+      tags:
+        - kafka_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      requestBody:
+        description: updated trigger
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/EditKafkaTrigger"
+      responses:
+        "200":
+          description: kafka trigger updated
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /w/{workspace}/kafka_triggers/delete/{path}:
+    delete:
+      summary: delete kafka trigger
+      operationId: deleteKafkaTrigger
+      tags:
+        -  kafka_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      responses:
+        "200":
+          description: kafka trigger deleted
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /w/{workspace}/kafka_triggers/get/{path}:
+    get:
+      summary: get kafka trigger
+      operationId: getKafkaTrigger
+      tags:
+        - kafka_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      responses:
+        "200":
+          description: kafka trigger deleted
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/KafkaTrigger"
+
+  
+  /w/{workspace}/kafka_triggers/list:
+    get:
+      summary: list kafka triggers
+      operationId: listKafkaTriggers
+      tags:
+        - kafka_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: kafka trigger list
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: "#/components/schemas/KafkaTrigger"
+
+  
+  /w/{workspace}/kafka_triggers/exists/{path}:
+    get:
+      summary: does kafka trigger exists
+      operationId: existsKafkaTrigger
+      tags:
+        - kafka_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      responses:
+        "200":
+          description: kafka trigger exists
+          content:
+            application/json:
+              schema:
+                type: boolean
+
+  /w/{workspace}/kafka_triggers/setenabled/{path}:
+    post:
+      summary: set enabled kafka trigger
+      operationId: setKafkaTriggerEnabled
+      tags:
+        - kafka_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      requestBody:
+        description: updated kafka trigger enable
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                enabled:
+                  type: boolean
+              required:
+                - enabled
+      responses:
+        "200":
+          description: kafka trigger enabled set
+          content:
+            text/plain:
+              schema:
+                type: string
+

  /groups/list:
    get:
@@ -8554,6 +8720,7 @@ paths:
                raw_app,
                http_trigger,
                websocket_trigger,
+                kafka_trigger,
              ]
      responses:
        "200":
@@ -8592,6 +8759,7 @@ paths:
                raw_app,
                http_trigger,
                websocket_trigger,
+                kafka_trigger,
              ]
      requestBody:
        description: acl to add
@@ -8641,6 +8809,7 @@ paths:
                raw_app,
                http_trigger,
                websocket_trigger,
+                kafka_trigger,
              ]
      requestBody:
        description: acl to add
@@ -11932,6 +12101,8 @@ components:
          type: number
        websocket_count:
          type: number
+        kafka_count:
+          type: number

    WebsocketTrigger:
      type: object
@@ -12103,6 +12274,114 @@ components:
          required: 
            - runnable_result

+    KafkaTrigger:
+      type: object
+      properties:
+        path:
+          type: string
+        edited_by:
+          type: string
+        edited_at:
+          type: string
+          format: date-time
+        script_path:
+          type: string
+        kafka_resource_path:
+          type: string
+        group_id:
+          type: string
+        topics:
+          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
+        - kafka_resource_path
+        - group_id
+        - topics
+        - extra_perms
+        - is_flow
+        - email
+        - workspace_id
+        - enabled
+
+    NewKafkaTrigger:
+      type: object
+      properties:
+        path:
+          type: string
+        script_path:
+          type: string
+        is_flow:
+          type: boolean
+        kafka_resource_path:
+          type: string
+        group_id:
+          type: string
+        topics:
+          type: array
+          items:
+            type: string
+        enabled:
+          type: boolean
+        
+      required:
+        - path
+        - script_path
+        - is_flow
+        - kafka_resource_path
+        - group_id
+        - topics
+
+    EditKafkaTrigger:
+      type: object
+      properties:
+        kafka_resource_path:
+          type: string
+        group_id:
+          type: string
+        topics:
+          type: array
+          items:
+            type: string
+        path:
+          type: string
+        script_path:
+          type: string
+        is_flow:
+          type: boolean
+              
+      required:
+        - path
+        - script_path
+        - kafka_resource_path
+        - group_id
+        - topics
+        - is_flow
+
    Group:
      type: object
      properties:
--- a/backend/windmill-api/src/apps.rs
+++ b/backend/windmill-api/src/apps.rs
@@ -8,11 +8,6 @@ use std::collections::HashMap;
 * LICENSE-AGPL for a copy of the license.
 */

-#[cfg(feature = "parquet")]
-use crate::{job_helpers_ee::{
-    get_random_file_name, get_s3_resource, get_workspace_s3_resource, upload_file_internal,
-    UploadFileResponse,
-}, users::fetch_api_authed_from_permissioned_as};
 use crate::{
    db::{ApiAuthed, DB},
    resources::get_resource_value_interpolated_internal,
@@ -22,6 +17,14 @@ use crate::{
    webhook_util::{WebhookMessage, WebhookShared},
    HTTP_CLIENT,
 };
+#[cfg(feature = "parquet")]
+use crate::{
+    job_helpers_ee::{
+        get_random_file_name, get_s3_resource, get_workspace_s3_resource, upload_file_internal,
+        UploadFileResponse,
+    },
+    users::fetch_api_authed_from_permissioned_as,
+};
 use axum::{
    extract::{Extension, Json, Path, Query},
    response::IntoResponse,
@@ -1288,7 +1291,6 @@ async fn upload_s3_file_from_app() -> Result<()> {
    ));
 }

-
 #[cfg(feature = "parquet")]
 #[derive(Debug, Deserialize, Clone)]
 struct UploadFileToS3Query {
@@ -1353,9 +1355,14 @@ async fn upload_s3_file_from_app(
        let (username, permissioned_as, email) =
            get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?;

-        let on_behalf_authed =
-            fetch_api_authed_from_permissioned_as(permissioned_as, email, &w_id, &db, username)
-                .await?;
+        let on_behalf_authed = fetch_api_authed_from_permissioned_as(
+            permissioned_as,
+            email,
+            &w_id,
+            &db,
+            Some(username),
+        )
+        .await?;

        if let Some(file_key) = query.file_key {
            // file key is provided => requires workspace, user or list policy and must match the regex
--- a/backend/windmill-api/src/granular_acls.rs
+++ b/backend/windmill-api/src/granular_acls.rs
@@ -23,7 +23,7 @@ use windmill_common::{
    utils::{not_found_if_none, StripPath},
 };

-const KINDS: [&str; 10] = [
+const KINDS: [&str; 12] = [
    "script",
    "group_",
    "resource",
@@ -34,6 +34,8 @@ const KINDS: [&str; 10] = [
    "app",
    "raw_app",
    "http_trigger",
+    "websocket_trigger",
+    "kafka_trigger",
 ];

 pub fn workspaced_service() -> Router {
--- a/backend/windmill-api/src/http_triggers.rs
+++ b/backend/windmill-api/src/http_triggers.rs
@@ -507,7 +507,7 @@ async fn get_http_route_trigger(
        trigger.email.clone(),
        &trigger.workspace_id,
        &db,
-        username_override.unwrap_or("anonymous".to_string()),
+        Some(username_override.unwrap_or("anonymous".to_string())),
    )
    .await?;

--- a/backend/windmill-api/src/kafka_triggers_ee.rs
+++ b/backend/windmill-api/src/kafka_triggers_ee.rs
@@ -0,0 +1,14 @@
+use crate::db::DB;
+use axum::Router;
+
+pub fn workspaced_service() -> Router {
+    Router::new()
+}
+
+pub async fn start_kafka_consumers(
+    _db: DB,
+    _rsmq: Option<rsmq_async::MultiplexedRsmq>,
+    mut _killpill_rx: tokio::sync::broadcast::Receiver<()>,
+) -> () {
+    // implementation is not open source
+}
--- a/backend/windmill-api/src/lib.rs
+++ b/backend/windmill-api/src/lib.rs
@@ -68,6 +68,8 @@ mod ai;
 mod job_helpers_ee;
 pub mod job_metrics;
 pub mod jobs;
+#[cfg(all(feature = "enterprise", feature = "kafka"))]
+mod kafka_triggers_ee;
 pub mod oauth2_ee;
 mod oidc_ee;
 mod raw_apps;
@@ -235,6 +237,9 @@ pub async fn run_server(
        }
    }

+    // #[cfg(feature = "kafka")]
+    // start_listening().await;
+
    let job_helpers_service = {
        #[cfg(feature = "parquet")]
        {
@@ -247,9 +252,27 @@ pub async fn run_server(
        }
    };

+    let kafka_triggers_service = {
+        #[cfg(all(feature = "enterprise", feature = "kafka"))]
+        {
+            kafka_triggers_ee::workspaced_service()
+        }
+
+        #[cfg(not(all(feature = "enterprise", feature = "kafka")))]
+        {
+            Router::new()
+        }
+    };
+
    if !*CLOUD_HOSTED {
        let ws_killpill_rx = rx.resubscribe();
-        websocket_triggers::start_websockets(db.clone(), rsmq, ws_killpill_rx).await;
+        websocket_triggers::start_websockets(db.clone(), rsmq.clone(), ws_killpill_rx).await;
+
+        #[cfg(all(feature = "enterprise", feature = "kafka"))]
+        {
+            let kafka_killpill_rx = rx.resubscribe();
+            kafka_triggers_ee::start_kafka_consumers(db.clone(), rsmq, kafka_killpill_rx).await;
+        }
    }

    // build our application with a route
@@ -296,7 +319,8 @@ pub async fn run_server(
                        .nest(
                            "/websocket_triggers",
                            websocket_triggers::workspaced_service(),
-                        ),
+                        )
+                        .nest("/kafka_triggers", kafka_triggers_service),
                )
                .nest("/workspaces", workspaces::global_service())
                .nest(
@@ -321,10 +345,7 @@ pub async fn run_server(
                    "/srch/w/:workspace_id/index",
                    indexer_ee::workspaced_service(),
                )
-                .nest(
-                    "/srch/index",
-                    indexer_ee::global_service(),
-                )
+                .nest("/srch/index", indexer_ee::global_service())
                .nest("/oidc", oidc_ee::global_service())
                .nest(
                    "/saml",
--- a/backend/windmill-api/src/triggers.rs
+++ b/backend/windmill-api/src/triggers.rs
@@ -18,6 +18,7 @@ pub struct TriggersCount {
    webhook_count: i64,
    email_count: i64,
    websocket_count: i64,
+    kafka_count: i64,
 }
 pub(crate) async fn get_triggers_count_internal(
    db: &DB,
@@ -64,6 +65,16 @@ pub(crate) async fn get_triggers_count_internal(
    .await?
    .unwrap_or(0);

+    let kafka_count = sqlx::query_scalar!(
+        "SELECT COUNT(*) FROM kafka_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[]",
@@ -105,6 +116,7 @@ pub(crate) async fn get_triggers_count_internal(
        webhook_count,
        email_count,
        websocket_count,
+        kafka_count,
    }))
 }

--- a/backend/windmill-api/src/users.rs
+++ b/backend/windmill-api/src/users.rs
@@ -746,7 +746,7 @@ pub async fn fetch_api_authed(
    email: String,
    w_id: &str,
    db: &DB,
-    username_override: String,
+    username_override: Option<String>,
 ) -> error::Result<ApiAuthed> {
    let permissioned_as = username_to_permissioned_as(username.as_str());
    fetch_api_authed_from_permissioned_as(permissioned_as, email, w_id, db, username_override).await
@@ -757,7 +757,7 @@ pub async fn fetch_api_authed_from_permissioned_as(
    email: String,
    w_id: &str,
    db: &DB,
-    username_override: String,
+    username_override: Option<String>,
 ) -> error::Result<ApiAuthed> {
    let authed =
        fetch_authed_from_permissioned_as(permissioned_as, email.clone(), w_id, db).await?;
@@ -769,7 +769,7 @@ pub async fn fetch_api_authed_from_permissioned_as(
        groups: authed.groups,
        folders: authed.folders,
        scopes: authed.scopes,
-        username_override: Some(username_override),
+        username_override: username_override,
    })
 }

--- a/backend/windmill-api/src/websocket_triggers.rs
+++ b/backend/windmill-api/src/websocket_triggers.rs
@@ -527,7 +527,7 @@ async fn wait_runnable_result(
        ws_trigger.email.clone(),
        &ws_trigger.workspace_id,
        &db,
-        username_override,
+        Some(username_override),
    )
    .await?;

@@ -773,7 +773,9 @@ async fn listen_to_websocket(
    rsmq: Option<rsmq_async::MultiplexedRsmq>,
    mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
 ) -> () {
-    update_ping(&db, &ws_trigger, Some("Connecting...")).await;
+    if let None = update_ping(&db, &ws_trigger, Some("Connecting...")).await {
+        return;
+    }

    let url = ws_trigger.url.as_str();

@@ -970,7 +972,7 @@ async fn run_job(
        trigger.email.clone(),
        &trigger.workspace_id,
        db,
-        "anonymous".to_string(),
+        Some("anonymous".to_string()),
    )
    .await?;

--- a/backend/windmill-api/src/workspaces.rs
+++ b/backend/windmill-api/src/workspaces.rs
@@ -1296,6 +1296,7 @@ async fn set_encryption_key(
 struct UsedTriggers {
    pub websocket_used: bool,
    pub http_routes_used: bool,
+    pub kafka_used: bool,
 }

 async fn get_used_triggers(
@@ -1306,7 +1307,10 @@ async fn get_used_triggers(
    let mut tx = user_db.begin(&authed).await?;
    let websocket_used = sqlx::query_as!(
        UsedTriggers,
-        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!""#,
+        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!""#,
        w_id,
    )
    .fetch_one(&mut *tx)
--- a/frontend/src/lib/components/Path.svelte
+++ b/frontend/src/lib/components/Path.svelte
@@ -14,7 +14,8 @@
 		ScriptService,
 		HttpTriggerService,
 		VariableService,
-		WebsocketTriggerService
+		WebsocketTriggerService,
+		KafkaTriggerService
 	} from '$lib/gen'
 	import { superadmin, userStore, workspaceStore } from '$lib/stores'
 	import { createEventDispatcher, getContext } from 'svelte'
@@ -37,6 +38,7 @@
 		| 'raw_app'
 		| 'http_trigger'
 		| 'websocket_trigger'
+		| 'kafka_trigger'
 	let meta: Meta | undefined = undefined
 	export let fullNamePlaceholder: string | undefined = undefined
 	export let namePlaceholder = ''
@@ -225,6 +227,11 @@
 				workspace: $workspaceStore!,
 				path: path
 			})
+		} else if (kind == 'kafka_trigger') {
+			return await KafkaTriggerService.existsKafkaTrigger({
+				workspace: $workspaceStore!,
+				path: path
+			})
 		} else {
 			return false
 		}
--- a/frontend/src/lib/components/ShareModal.svelte
+++ b/frontend/src/lib/components/ShareModal.svelte
@@ -26,6 +26,7 @@
 		| 'raw_app'
 		| 'http_trigger'
 		| 'websocket_trigger'
+		| 'kafka_trigger'
 	let kind: Kind

 	let path: string = ''
--- a/frontend/src/lib/components/details/DetailPageDetailPanel.svelte
+++ b/frontend/src/lib/components/details/DetailPageDetailPanel.svelte
@@ -51,6 +51,7 @@
 						<slot slot="webhooks" name="webhooks" />
 						<slot slot="routes" name="routes" />
 						<slot slot="websockets" name="websockets" />
+						<slot slot="kafka" name="kafka" />
 						<slot slot="emails" name="emails" />
 						<slot slot="schedules" name="schedules" />
 						<slot slot="cli" name="cli" />
--- a/frontend/src/lib/components/details/DetailPageLayout.svelte
+++ b/frontend/src/lib/components/details/DetailPageLayout.svelte
@@ -52,6 +52,7 @@
 							<slot slot="webhooks" name="webhooks" />
 							<slot slot="routes" name="routes" />
 							<slot slot="websockets" name="websockets" />
+							<slot slot="kafka" name="kafka" />
 							<slot slot="emails" name="emails" />
 							<slot slot="schedules" name="schedules" />
 							<slot slot="cli" name="cli" />
@@ -94,6 +95,7 @@
 								<slot slot="routes" name="routes" />
 								<slot slot="script" name="script" />
 								<slot slot="websockets" name="websockets" />
+								<slot slot="kafka" name="kafka" />
 								<slot slot="emails" name="emails" />
 								<slot slot="schedules" name="schedules" />
 								<slot slot="cli" name="cli" />
--- a/frontend/src/lib/components/details/DetailPageTriggerPanel.svelte
+++ b/frontend/src/lib/components/details/DetailPageTriggerPanel.svelte
@@ -3,6 +3,7 @@
 	import { CalendarCheck2, MailIcon, Route, Terminal, Webhook, Unplug } from 'lucide-svelte'

 	import HighlightTheme from '../HighlightTheme.svelte'
+	import KafkaIcon from '../icons/KafkaIcon.svelte'

 	export let triggerSelected:
 		| 'webhooks'
@@ -11,6 +12,7 @@
 		| 'cli'
 		| 'routes'
 		| 'websockets'
+		| 'kafka'
 		| 'scheduledPoll' = 'webhooks'
 	export let simplfiedPoll: boolean = false
 </script>
@@ -43,6 +45,12 @@
 				Websockets
 			</span>
 		</Tab>
+		<Tab value="kafka">
+			<span class="flex flex-row gap-2 items-center text-xs">
+				<KafkaIcon size={12} />
+				Kafka
+			</span>
+		</Tab>
 		<Tab value="emails">
 			<span class="flex flex-row gap-2 items-center text-xs">
 				<MailIcon size={12} />
@@ -69,6 +77,8 @@
 				<slot name="schedules" />
 			{:else if triggerSelected === 'websockets'}
 				<slot name="websockets" />
+			{:else if triggerSelected === 'kafka'}
+				<slot name="kafka" />
 			{:else if triggerSelected === 'cli'}
 				<slot name="cli" />
 			{/if}
--- a/frontend/src/lib/components/graph/renderers/triggers/TriggersBadge.svelte
+++ b/frontend/src/lib/components/graph/renderers/triggers/TriggersBadge.svelte
@@ -10,6 +10,7 @@
 	import { type TriggerContext } from '$lib/components/triggers'
 	import { FlowService, ScriptService } from '$lib/gen'
 	import { workspaceStore } from '$lib/stores'
+	import KafkaIcon from '$lib/components/icons/KafkaIcon.svelte'

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

@@ -18,8 +19,14 @@
 	export let isFlow: boolean
 	export let selected: boolean
 	export let showOnlyWithCount: boolean
-	export let triggersToDisplay: ('webhooks' | 'schedules' | 'routes' | 'websockets' | 'emails')[] =
-		['webhooks', 'schedules', 'routes', 'websockets', 'emails']
+	export let triggersToDisplay: (
+		| 'webhooks'
+		| 'schedules'
+		| 'routes'
+		| 'websockets'
+		| 'kafka'
+		| 'emails'
+	)[] = ['webhooks', 'schedules', 'routes', 'websockets', 'kafka', 'emails']
 	const dispatch = createEventDispatcher()

 	onMount(() => {
@@ -47,6 +54,7 @@
 		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' }
 	}
 </script>
--- a/frontend/src/lib/components/icons/KafkaIcon.svelte
+++ b/frontend/src/lib/components/icons/KafkaIcon.svelte
--- a/frontend/src/lib/components/sidebar/SidebarContent.svelte
+++ b/frontend/src/lib/components/sidebar/SidebarContent.svelte
@@ -49,6 +49,7 @@
 	import { type Changelog, changelogs } from './changelogs'
 	import { page } from '$app/stores'
 	import SideBarNotification from './SideBarNotification.svelte'
+	import KafkaIcon from '../icons/KafkaIcon.svelte'

 	export let numUnacknowledgedCriticalAlerts = 0

@@ -97,6 +98,13 @@
 			icon: Unplug,
 			disabled: $userStore?.operator,
 			kind: 'ws'
+		},
+		{
+			label: 'Kafka',
+			href: '/kafka_triggers',
+			icon: KafkaIcon,
+			disabled: $userStore?.operator,
+			kind: 'kafka'
 		}
 	]

@@ -205,10 +213,10 @@
 										action: () => {
 											isCriticalAlertsUIOpen.set(true)
 										},
-							icon: AlertCircle,
-							notificationCount: numUnacknowledgedCriticalAlerts
-						}
-					]
+										icon: AlertCircle,
+										notificationCount: numUnacknowledgedCriticalAlerts
+									}
+							  ]
 							: [])
 					]
 			  }
--- a/frontend/src/lib/components/triggers.ts
+++ b/frontend/src/lib/components/triggers.ts
@@ -11,7 +11,14 @@ export type ScheduleTrigger = {

 export type TriggerContext = {
 	selectedTrigger: Writable<
-		'webhooks' | 'emails' | 'schedules' | 'cli' | 'routes' | 'websockets' | 'scheduledPoll'
+		| 'webhooks'
+		| 'emails'
+		| 'schedules'
+		| 'cli'
+		| 'routes'
+		| 'websockets'
+		| 'scheduledPoll'
+		| 'kafka'
 	>
 	primarySchedule: Writable<ScheduleTrigger | undefined | false>
 	triggersCount: Writable<TriggersCount | undefined>
--- a/frontend/src/lib/components/triggers/KafkaTriggerEditor.svelte
+++ b/frontend/src/lib/components/triggers/KafkaTriggerEditor.svelte
@@ -0,0 +1,23 @@
+<script lang="ts">
+	import { tick } from 'svelte'
+	import KafkaTriggerEditorInner from './KafkaTriggerEditorInner.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) {
+		open = true
+		await tick()
+		drawer?.openNew(is_flow, initial_script_path)
+	}
+
+	let drawer: KafkaTriggerEditorInner
+</script>
+
+{#if open}
+	<KafkaTriggerEditorInner on:update bind:this={drawer} />
+{/if}
--- a/frontend/src/lib/components/triggers/KafkaTriggerEditorInner.svelte
+++ b/frontend/src/lib/components/triggers/KafkaTriggerEditorInner.svelte
@@ -0,0 +1,310 @@
+<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 { KafkaTriggerService } 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, X, Plus } from 'lucide-svelte'
+	import Label from '$lib/components/Label.svelte'
+	import Toggle from '../Toggle.svelte'
+	import { fade } from 'svelte/transition'
+	import ResourcePicker from '../ResourcePicker.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 kafka_resource_path = ''
+	let group_id = ''
+	let topics: string[] = ['']
+	let dirtyGroupId = false
+	let enabled = false
+	let dirtyPath = false
+	let can_write = true
+	let drawerLoading = true
+
+	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
+			dirtyGroupId = false
+			await loadTrigger()
+		} catch (err) {
+			sendUserToast(`Could not load kafka trigger: ${err}`, true)
+		} finally {
+			drawerLoading = false
+		}
+	}
+
+	export async function openNew(nis_flow: boolean, fixedScriptPath_?: string) {
+		drawerLoading = true
+		try {
+			drawer?.openDrawer()
+			is_flow = nis_flow
+			edit = false
+			itemKind = nis_flow ? 'flow' : 'script'
+			kafka_resource_path = ''
+			group_id = ''
+			topics = ['']
+			dirtyGroupId = false
+			initialScriptPath = ''
+			fixedScriptPath = fixedScriptPath_ ?? ''
+			script_path = fixedScriptPath
+			path = ''
+			initialPath = ''
+			dirtyPath = false
+		} finally {
+			drawerLoading = false
+		}
+	}
+
+	async function loadTrigger(): Promise<void> {
+		const s = await KafkaTriggerService.getKafkaTrigger({
+			workspace: $workspaceStore!,
+			path: initialPath
+		})
+		script_path = s.script_path
+		initialScriptPath = s.script_path
+
+		is_flow = s.is_flow
+		path = s.path
+		kafka_resource_path = s.kafka_resource_path
+		group_id = s.group_id
+		topics = s.topics
+		enabled = s.enabled
+
+		can_write = canWrite(s.path, s.extra_perms, $userStore)
+	}
+
+	async function updateTrigger(): Promise<void> {
+		if (edit) {
+			await KafkaTriggerService.updateKafkaTrigger({
+				workspace: $workspaceStore!,
+				path: initialPath,
+				requestBody: {
+					path,
+					script_path,
+					is_flow,
+					kafka_resource_path,
+					group_id,
+					topics
+				}
+			})
+			sendUserToast(`Kafka trigger ${path} updated`)
+		} else {
+			await KafkaTriggerService.createKafkaTrigger({
+				workspace: $workspaceStore!,
+				requestBody: {
+					path,
+					script_path,
+					is_flow,
+					enabled: true,
+					kafka_resource_path,
+					group_id,
+					topics
+				}
+			})
+			sendUserToast(`Kafka trigger ${path} created`)
+		}
+		if (!$usedTriggerKinds.includes('kafka')) {
+			$usedTriggerKinds = [...$usedTriggerKinds, 'kafka']
+		}
+		dispatch('update')
+		drawer.closeDrawer()
+	}
+
+	$: topicsError = topics.some((b) => /[^[a-zA-Z0-9-_.]/.test(b)) ? 'Invalid topics' : ''
+	$: groupIdError = /[^a-zA-Z0-9-_.]/.test(group_id) ? 'Invalid group ID' : ''
+
+	$: !dirtyGroupId &&
+		path &&
+		(group_id = `windmill_consumer-${$workspaceStore}-${path.replaceAll('/', '__')}`)
+</script>
+
+<Drawer size="800px" bind:this={drawer}>
+	<DrawerContent
+		title={edit
+			? can_write
+				? `Edit kafka trigger ${initialPath}`
+				: `Kafka trigger ${initialPath}`
+			: 'New kafka trigger'}
+		on:close={drawer.closeDrawer}
+	>
+		<svelte:fragment slot="actions">
+			{#if !drawerLoading && can_write}
+				{#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 KafkaTriggerService.setKafkaTriggerEnabled({
+									path: initialPath,
+									workspace: $workspaceStore ?? '',
+									requestBody: { enabled: e.detail }
+								})
+								sendUserToast(`${e.detail ? 'enabled' : 'disabled'} kafka trigger ${initialPath}`)
+							}}
+						/>
+					</div>
+				{/if}
+				<Button
+					startIcon={{ icon: Save }}
+					disabled={pathError != '' ||
+						emptyString(script_path) ||
+						emptyString(kafka_resource_path) ||
+						topics.length < 1 ||
+						topics.some((t) => emptyString(t)) ||
+						topicsError != '' ||
+						emptyString(group_id) ||
+						groupIdError != '' ||
+						!can_write}
+					on:click={updateTrigger}
+				>
+					Save
+				</Button>
+			{/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}
+					Kafka 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="kafka_trigger"
+							kind="kafka_trigger"
+							disabled={!can_write}
+						/>
+					</Label>
+				</div>
+
+				<Section label="Kafka">
+					<div class="flex flex-col w-full gap-4">
+						<div class="block grow w-full">
+							<div class="text-secondary text-sm mb-1">
+								Resource
+								<Required required={true} />
+							</div>
+							<ResourcePicker resourceType="kafka" bind:value={kafka_resource_path} />
+						</div>
+						<label class="block grow w-full">
+							<div class="text-secondary text-sm flex items-center gap-1 w-full justify-between">
+								<div>
+									Topics
+									<Required required={true} />
+								</div>
+							</div>
+
+							<div class="flex flex-col gap-4 mt-1">
+								{#each topics as v, i}
+									<div class="flex w-full gap-2 items-center">
+										<input type="text" bind:value={v} />
+
+										<button
+											transition:fade|local={{ duration: 100 }}
+											class="rounded-full p-1 bg-surface-secondary duration-200 hover:bg-surface-hover"
+											aria-label="Clear"
+											on:click={() => {
+												topics = topics.filter((_, index) => index !== i)
+											}}
+											class:hidden={topics.length === 1}
+										>
+											<X size={14} />
+										</button>
+									</div>
+								{/each}
+
+								<div class="flex items-baseline">
+									<Button
+										variant="border"
+										color="light"
+										size="xs"
+										on:click={() => {
+											if (topics == undefined || !Array.isArray(topics)) {
+												topics = []
+											}
+											topics = topics.concat('')
+										}}
+										startIcon={{ icon: Plus }}
+									>
+										Add topic
+									</Button>
+								</div>
+							</div>
+							<div class="text-red-600 dark:text-red-400 text-2xs mt-1.5">
+								{topicsError}
+							</div>
+						</label>
+
+						<label class="block grow w-full">
+							<div class="text-secondary text-sm flex items-center gap-1 w-full justify-between">
+								<div>
+									Group ID
+									<Required required={true} />
+								</div>
+							</div>
+							<div class="mt-1">
+								<input type="text" bind:value={group_id} on:focus={() => (dirtyGroupId = true)} />
+							</div>
+
+							<div class="text-red-600 dark:text-red-400 text-2xs mt-1.5">
+								{groupIdError}
+							</div>
+						</label>
+					</div>
+				</Section>
+
+				<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
+						/>
+					</div>
+				</Section>
+			</div>
+		{/if}
+	</DrawerContent>
+</Drawer>
--- a/frontend/src/lib/components/triggers/KafkaTriggersPanel.svelte
+++ b/frontend/src/lib/components/triggers/KafkaTriggersPanel.svelte
@@ -0,0 +1,108 @@
+<script lang="ts">
+	import { Button } from '../common'
+	import { userStore, workspaceStore } from '$lib/stores'
+	import { KafkaTriggerService, type KafkaTrigger } from '$lib/gen'
+	import { UnplugIcon } from 'lucide-svelte'
+
+	import Skeleton from '../common/skeleton/Skeleton.svelte'
+	import { canWrite } from '$lib/utils'
+	import Alert from '../common/alert/Alert.svelte'
+	import type { TriggerContext } from '../triggers'
+	import { getContext } from 'svelte'
+	import KafkaTriggerEditor from './KafkaTriggerEditor.svelte'
+	import { isCloudHosted } from '$lib/cloud'
+
+	export let isFlow: boolean
+	export let path: string
+	export let newItem: boolean = false
+
+	let kafkaTriggerEditor: KafkaTriggerEditor
+
+	$: path && loadTriggers()
+
+	const { triggersCount } = getContext<TriggerContext>('TriggerContext')
+
+	let kafkaTriggers: (KafkaTrigger & { canWrite: boolean })[] | undefined = undefined
+	export async function loadTriggers() {
+		try {
+			kafkaTriggers = (
+				await KafkaTriggerService.listKafkaTriggers({
+					workspace: $workspaceStore ?? '',
+					path,
+					isFlow
+				})
+			).map((x) => {
+				return { canWrite: canWrite(x.path, x.extra_perms!, $userStore), ...x }
+			})
+			$triggersCount = { ...($triggersCount ?? {}), kafka_count: kafkaTriggers?.length }
+		} catch (e) {
+			console.error('impossible to load kafka triggers', e)
+		}
+	}
+</script>
+
+<KafkaTriggerEditor
+	on:update={() => {
+		loadTriggers()
+	}}
+	bind:this={kafkaTriggerEditor}
+/>
+
+<div class="flex flex-col gap-4">
+	{#if !newItem}
+		{#if isCloudHosted()}
+			<Alert title="Not compatible with multi-tenant cloud" type="warning">
+				Kafka triggers are disabled in the multi-tenant cloud.
+			</Alert>
+		{:else if $userStore?.is_admin || $userStore?.is_super_admin}
+			<Button
+				on:click={() => kafkaTriggerEditor?.openNew(isFlow, path)}
+				variant="border"
+				color="light"
+				size="xs"
+				startIcon={{ icon: UnplugIcon }}
+			>
+				New Kafka Trigger
+			</Button>
+		{:else}
+			<Alert title="Only workspace admins can create routes" type="warning" size="xs" />
+		{/if}
+	{/if}
+
+	{#if kafkaTriggers}
+		{#if kafkaTriggers.length == 0}
+			<div class="text-xs text-secondary"> No kafka triggers </div>
+		{:else}
+			<div class="flex flex-col divide-y pt-2">
+				{#each kafkaTriggers as kafkaTrigger (kafkaTrigger.path)}
+					<div class="grid grid-cols-5 text-2xs items-center py-2">
+						<div class="col-span-2 truncate">{kafkaTrigger.path}</div>
+						<div class="col-span-2 truncate">
+							{kafkaTrigger.kafka_resource_path}
+						</div>
+						<div class="flex justify-end">
+							<button
+								on:click={() => kafkaTriggerEditor?.openEdit(kafkaTrigger.path, isFlow)}
+								class="px-2"
+							>
+								{#if kafkaTrigger.canWrite}
+									Edit
+								{:else}
+									View
+								{/if}
+							</button>
+						</div>
+					</div>
+				{/each}
+			</div>
+		{/if}
+	{:else}
+		<Skeleton layout={[[8]]} />
+	{/if}
+
+	{#if newItem}
+		<Alert title="Triggers disabled" type="warning" size="xs">
+			Deploy the {isFlow ? 'flow' : 'script'} to add kafka triggers.
+		</Alert>
+	{/if}
+</div>
--- a/frontend/src/lib/components/triggers/TriggersEditor.svelte
+++ b/frontend/src/lib/components/triggers/TriggersEditor.svelte
@@ -12,6 +12,7 @@
 	import type { TriggerContext } from '$lib/components/triggers'
 	import WebsocketTriggersPanel from './WebsocketTriggersPanel.svelte'
 	import ScheduledPollPanel from './ScheduledPollPanel.svelte'
+	import KafkaTriggersPanel from './KafkaTriggersPanel.svelte'

 	export let noEditor: boolean
 	export let newItem = false
@@ -31,6 +32,7 @@
 				<Tab value="schedules" selectedClass="text-primary text-sm font-semibold">Schedules</Tab>
 				<Tab value="routes" selectedClass="text-primary text-sm font-semibold">Routes</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="emails" selectedClass="text-primary text-sm font-semibold">Email</Tab>
 				{#if isFlow}
 					<Tab value="scheduledPoll" selectedClass="text-primary text-sm font-semibold"
@@ -75,6 +77,12 @@
 						</div>
 					{/if}

+					{#if $selectedTrigger === 'kafka'}
+						<div class="p-4">
+							<KafkaTriggersPanel {newItem} path={currentPath} {isFlow} />
+						</div>
+					{/if}
+
 					{#if $selectedTrigger === 'schedules'}
 						<div class="p-4">
 							<RunPageSchedules
--- a/frontend/src/lib/components/triggers/WebsocketTriggerEditorInner.svelte
+++ b/frontend/src/lib/components/triggers/WebsocketTriggerEditorInner.svelte
@@ -198,7 +198,7 @@
 					url_runnable_args
 				}
 			})
-			sendUserToast(`Route ${path} updated`)
+			sendUserToast(`Websocket trigger ${path} updated`)
 		} else {
 			await WebsocketTriggerService.createWebsocketTrigger({
 				workspace: $workspaceStore!,
@@ -213,7 +213,7 @@
 					url_runnable_args
 				}
 			})
-			sendUserToast(`Route ${path} created`)
+			sendUserToast(`Websocket trigger ${path} created`)
 		}
 		if (!$usedTriggerKinds.includes('ws')) {
 			$usedTriggerKinds = [...$usedTriggerKinds, 'ws']
--- a/frontend/src/lib/script_helpers.ts
+++ b/frontend/src/lib/script_helpers.ts
@@ -514,17 +514,22 @@ export async function main(approver?: string) {
 const BUN_PREPROCESSOR_MODULE_CODE = `
 export async function preprocessor(
 	wm_trigger: {
-		kind: 'http' | 'email' | 'webhook' | 'websocket',
+		kind: 'http' | 'email' | 'webhook' | 'websocket' | 'kafka',
 		http?: {
 			route: string // The route path, e.g. "/users/:id"
 			path: string // The actual path called, e.g. "/users/123"
 			method: string
-			params: Record<string, string>
-			query: Record<string, string>
+			params: Record<string, string> // path parameters
+			query: Record<string, string> // query parameters
 			headers: Record<string, string>
 		},
 		websocket?: {
 			url: string // The websocket url
+		},
+		kafka?: {
+			brokers: string[]
+			topic: string
+			group_id: string
 		}
 	},
 	/* your other args */ 
@@ -538,17 +543,22 @@ export async function preprocessor(
 const DENO_PREPROCESSOR_MODULE_CODE = `
 export async function preprocessor(
 	wm_trigger: {
-		kind: 'http' | 'email' | 'wehbook' | 'websocket',
+		kind: 'http' | 'email' | 'webhook' | 'websocket' | 'kafka',
 		http?: {
 			route: string // The route path, e.g. "/users/:id"
 			path: string // The actual path called, e.g. "/users/123"
 			method: string
-			params: Record<string, string>
-			query: Record<string, string>
+			params: Record<string, string> // path parameters
+			query: Record<string, string> // query parameters
 			headers: Record<string, string>
 		},
 		websocket?: {
 			url: string // The websocket url
+		},
+		kafka?: {
+			brokers: string[]
+			topic: string
+			group_id: string
 		}
 	},
 	/* your other args */ 
@@ -599,10 +609,16 @@ class Http(TypedDict):
 class Websocket(TypedDict):
 	url: str # The websocket url

+class Kafka(TypedDict):
+	topic: str
+	brokers: list[str]
+	group_id: str
+
 class WmTrigger(TypedDict):
-    kind: Literal["http", "email", "webhook", "websocket"]
-    http: Http | None
-    websocket: Websocket | None
+	kind: Literal["http", "email", "webhook", "websocket", "kafka"]
+	http: Http | None
+	websocket: Websocket | None
+	kafka: Kafka | None

 def preprocessor(
 	wm_trigger: WmTrigger,
--- a/frontend/src/routes/(root)/(logged)/+layout.svelte
+++ b/frontend/src/routes/(root)/(logged)/+layout.svelte
@@ -190,15 +190,20 @@

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

--- a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte
@@ -61,6 +61,7 @@
 	import { writable } from 'svelte/store'
 	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'

 	let flow: Flow | undefined
 	let can_write = false
@@ -539,6 +540,13 @@
 				<WebsocketTriggersPanel path={flow.path ?? ''} isFlow />
 			</div>
 		</svelte:fragment>
+
+		<svelte:fragment slot="kafka">
+			<div class="p-2">
+				<KafkaTriggersPanel path={flow.path ?? ''} isFlow />
+			</div>
+		</svelte:fragment>
+
 		<svelte:fragment slot="emails">
 			<div class="p-2">
 				<EmailTriggerPanel
--- a/frontend/src/routes/(root)/(logged)/kafka_triggers/+page.js
+++ b/frontend/src/routes/(root)/(logged)/kafka_triggers/+page.js
@@ -0,0 +1,5 @@
+export function load() {
+	return {
+		stuff: { title: 'Kafka triggers' }
+	}
+}
--- a/frontend/src/routes/(root)/(logged)/kafka_triggers/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/kafka_triggers/+page.svelte
@@ -0,0 +1,422 @@
+<script lang="ts">
+	import { KafkaTriggerService, type KafkaTrigger } 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 KafkaTriggerEditor from '$lib/components/triggers/KafkaTriggerEditor.svelte'
+	import Popover from '$lib/components/Popover.svelte'
+	import { isCloudHosted } from '$lib/cloud'
+	import KafkaIcon from '$lib/components/icons/KafkaIcon.svelte'
+
+	type TriggerW = KafkaTrigger & { canWrite: boolean }
+
+	let triggers: TriggerW[] = []
+	let shareModal: ShareModal
+	let loading = true
+
+	async function loadTriggers(): Promise<void> {
+		triggers = (await KafkaTriggerService.listKafkaTriggers({ workspace: $workspaceStore! })).map(
+			(x) => {
+				return { canWrite: canWrite(x.path, x.extra_perms!, $userStore), ...x }
+			}
+		)
+		loading = false
+	}
+
+	let interval = setInterval(async () => {
+		try {
+			const newTriggers = await KafkaTriggerService.listKafkaTriggers({
+				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
+					}
+				}
+			}
+		} catch (err) {
+			console.error(err)
+		}
+	}, 5000)
+
+	onDestroy(() => {
+		clearInterval(interval)
+	})
+
+	async function setTriggerEnabled(path: string, enabled: boolean): Promise<void> {
+		try {
+			await KafkaTriggerService.setKafkaTriggerEnabled({
+				path,
+				workspace: $workspaceStore!,
+				requestBody: { enabled }
+			})
+		} catch (err) {
+			sendUserToast(
+				`Cannot ` + (enabled ? 'enable' : 'disable') + ` kafka trigger: ${err.body}`,
+				true
+			)
+		} finally {
+			loadTriggers()
+		}
+	}
+
+	$: {
+		if ($workspaceStore && $userStore) {
+			loadTriggers()
+		}
+	}
+	let kafkaTriggerEditor: KafkaTriggerEditor
+
+	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>
+
+<KafkaTriggerEditor on:update={loadTriggers} bind:this={kafkaTriggerEditor} />
+
+<SearchItems
+	{filter}
+	items={preFilteredItems}
+	bind:filteredItems
+	f={(x) => (x.summary ?? '') + ' ' + x.path + ' (' + x.script_path + ')'}
+/>
+
+<CenteredPage>
+	<PageHeader
+		title="Kafka triggers"
+		tooltip="Windmill can consume kafka events and trigger scripts or flows based on them."
+	>
+		<Button size="md" startIcon={{ icon: Plus }} on:click={() => kafkaTriggerEditor.openNew(false)}>
+			New&nbsp;kafka trigger
+		</Button>
+	</PageHeader>
+
+	{#if isCloudHosted()}
+		<Alert title="Not compatible with multi-tenant cloud" type="warning">
+			Kafka 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 kafka 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="Kafka Trigger" icon={KafkaIcon} />
+					<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 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)}
+					{@const href = `${is_flow ? '/flows/get' : '/scripts/get'}/${script_path}`}
+					{@const ping = last_server_ping ? new Date(last_server_ping) : undefined}
+
+					<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={() => kafkaTriggerEditor?.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}
+										{kafka_resource_path} - {topics.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 && (!ping || ping.getTime() < new Date().getTime() - 15 * 1000 || error)) || (!enabled && error)}
+									<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}
+												Consumer is not connected{error ? ': ' + error : ''}
+											{: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={() => kafkaTriggerEditor?.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 KafkaTriggerService.deleteKafkaTrigger({
+													workspace: $workspaceStore ?? '',
+													path
+												})
+												loadTriggers()
+											}
+										},
+										{
+											displayName: canWrite ? 'Edit' : 'View',
+											icon: canWrite ? Pen : Eye,
+											action: () => {
+												kafkaTriggerEditor?.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, 'kafka_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()
+	}}
+/>
--- a/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte
@@ -84,6 +84,7 @@
 	import { writable } from 'svelte/store'
 	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'

 	let script: Script | undefined
 	let topHash: string | undefined
@@ -721,6 +722,11 @@
 					<WebsocketTriggersPanel path={script.path ?? ''} isFlow={false} />
 				</div>
 			</svelte:fragment>
+			<svelte:fragment slot="kafka">
+				<div class="p-2">
+					<KafkaTriggersPanel path={script.path ?? ''} isFlow={false} />
+				</div>
+			</svelte:fragment>
 			<svelte:fragment slot="emails">
 				<div class="p-2">
 					<EmailTriggerPanel