starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-22 00:01:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add OR logic support to kafka/websocket trigger filters (#8580)

Browse Source 
* feat: add OR logic support to kafka/websocket trigger filters

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* chore: update ee-repo-ref for OR logic filter support

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: add filter_logic to OpenAPI spec/save utils, fix websocket derive, show capture group ID

- Add filter_logic field to all 6 Kafka/WebSocket OpenAPI schemas so it
  is included in the generated frontend client types
- Include filter_logic in save request bodies (kafka/utils.ts, websocket/utils.ts)
- Fix misplaced #[derive(FromRow)] on WebsocketConfig (was on the default fn)
- Show copyable "Test group ID" in Kafka capture UI
- Remove capture event-loss warning for Kafka (uses separate consumer group)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* update sqlx

* update ee ref

* chore: regenerate system prompts for filter_logic schema changes

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: remove banned $bindable(default_value) pattern in TriggerFilters

Use $bindable() without default and $derived with ?? for the effective
value, per CLAUDE.md rules.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: make filterLogic prop required in TriggerFilters

All callers always pass it, no need for optional + derived fallback.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* chore: update ee-repo-ref to 5ee1382dfb23b6a1516e3c7586058cec8240fdf2

This commit updates the EE repository reference after PR #498 was merged in windmill-ee-private.

Previous ee-repo-ref: bbd674991c07bff1cb2f3744e71fda10df53f09d

New ee-repo-ref: 5ee1382dfb23b6a1516e3c7586058cec8240fdf2

Automated by sync-ee-ref workflow.

* fix: reset filterLogic to 'and' in openNew for kafka/websocket editors

Prevents stale OR logic from carrying over when creating a new trigger
after editing one with OR filters.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: hugocasa <hugo@casademont.ch>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
2026-03-30 19:32:24 +00:00
committed by GitHub
parent 8e973c892d
commit 3876902a7b
24 changed files with 238 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-68c19cb0e18b94870bbe81f9aab92ba37da67cd2a56834c9d1378eab7551284d.json
Generated
+30
View File
@@ -0,0 +1,30 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE kafka_trigger\n SET\n kafka_resource_path = $1,\n group_id = $2,\n topics = $3,\n filters = $4,\n filter_logic = $5,\n auto_offset_reset = $6,\n auto_commit = $7,\n script_path = $8,\n path = $9,\n is_flow = $10,\n edited_by = $11,\n permissioned_as = $12,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $15,\n error_handler_args = $16,\n retry = $17\n WHERE\n workspace_id = $13 AND path = $14\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"VarcharArray",
"JsonbArray",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Varchar",
"Text",
"Text",
"Varchar",
"Jsonb",
"Jsonb"
]
},
"nullable": []
},
"hash": "68c19cb0e18b94870bbe81f9aab92ba37da67cd2a56834c9d1378eab7551284d"
}
backend/.sqlx/query-942c0abb55c910862fd45d3fa56a4eb6729f1a658101bda2d0b0fca96b3cfee5.json → backend/.sqlx/query-6948eb5aabf82f2f4a08dd4410eb472080ecab3ed652912397245e5216ae0389.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO websocket_trigger (\n workspace_id,\n path,\n url,\n script_path,\n is_flow,\n mode,\n filters,\n initial_messages,\n url_runnable_args,\n edited_by,\n can_return_message,\n can_return_error_result,\n permissioned_as,\n edited_at,\n error_handler_path,\n error_handler_args,\n retry\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, now(), $14, $15, $16\n )\n ",
"query": "\n INSERT INTO websocket_trigger (\n workspace_id,\n path,\n url,\n script_path,\n is_flow,\n mode,\n filters,\n filter_logic,\n initial_messages,\n url_runnable_args,\n edited_by,\n can_return_message,\n can_return_error_result,\n permissioned_as,\n edited_at,\n error_handler_path,\n error_handler_args,\n retry\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, now(), $15, $16, $17\n )\n ",
"describe": {
"columns": [],
"parameters": {
@@ -23,6 +23,7 @@
}
},
"JsonbArray",
"Varchar",
"JsonbArray",
"Jsonb",
"Varchar",
@@ -36,5 +37,5 @@
},
"nullable": []
},
"hash": "942c0abb55c910862fd45d3fa56a4eb6729f1a658101bda2d0b0fca96b3cfee5"
"hash": "6948eb5aabf82f2f4a08dd4410eb472080ecab3ed652912397245e5216ae0389"
}
backend/.sqlx/query-a0a545fda5f3ebea0113d5daaf13358c964d9fb0f41bf2a1c834305b4d2398f2.json → backend/.sqlx/query-6a8f4ed9946bb2a3c5e90695c90b70aa2e83fcb5aa0c953febdd9bac2d95bbec.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO kafka_trigger (\n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\n filters,\n auto_offset_reset,\n auto_commit,\n script_path,\n is_flow,\n mode,\n edited_by,\n permissioned_as,\n edited_at,\n error_handler_path,\n error_handler_args,\n retry\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, now(), $14, $15, $16\n )\n ",
"query": "\n INSERT INTO kafka_trigger (\n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\n filters,\n filter_logic,\n auto_offset_reset,\n auto_commit,\n script_path,\n is_flow,\n mode,\n edited_by,\n permissioned_as,\n edited_at,\n error_handler_path,\n error_handler_args,\n retry\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, now(), $15, $16, $17\n )\n ",
"describe": {
"columns": [],
"parameters": {
@@ -12,6 +12,7 @@
"VarcharArray",
"JsonbArray",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Bool",
@@ -36,5 +37,5 @@
},
"nullable": []
},
"hash": "a0a545fda5f3ebea0113d5daaf13358c964d9fb0f41bf2a1c834305b4d2398f2"
"hash": "6a8f4ed9946bb2a3c5e90695c90b70aa2e83fcb5aa0c953febdd9bac2d95bbec"
}
backend/.sqlx/query-a37cfc632dd37cf37c06743239b5ebc784e5da5ee25d47af187a75220d8fded7.json
Generated
-29
View File
@@ -1,29 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE kafka_trigger\n SET\n kafka_resource_path = $1,\n group_id = $2,\n topics = $3,\n filters = $4,\n auto_offset_reset = $5,\n auto_commit = $6,\n script_path = $7,\n path = $8,\n is_flow = $9,\n edited_by = $10,\n permissioned_as = $11,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $14,\n error_handler_args = $15,\n retry = $16\n WHERE\n workspace_id = $12 AND path = $13\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"VarcharArray",
"JsonbArray",
"Varchar",
"Bool",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Varchar",
"Text",
"Text",
"Varchar",
"Jsonb",
"Jsonb"
]
},
"nullable": []
},
"hash": "a37cfc632dd37cf37c06743239b5ebc784e5da5ee25d47af187a75220d8fded7"
}
backend/.sqlx/query-c7aed7fe3b6774477d403bc3e7fcbce7cdbdd1feb553718cbde60bb8ccff4733.json
Generated
-29
View File
@@ -1,29 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE\n websocket_trigger\n SET\n url = $1,\n script_path = $2,\n path = $3,\n is_flow = $4,\n filters = $5,\n initial_messages = $6,\n url_runnable_args = $7,\n edited_by = $8,\n permissioned_as = $9,\n can_return_message = $10,\n can_return_error_result = $11,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $14,\n error_handler_args = $15,\n retry = $16\n WHERE\n workspace_id = $12 AND path = $13\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Bool",
"JsonbArray",
"JsonbArray",
"Jsonb",
"Varchar",
"Varchar",
"Bool",
"Bool",
"Text",
"Text",
"Varchar",
"Jsonb",
"Jsonb"
]
},
"nullable": []
},
"hash": "c7aed7fe3b6774477d403bc3e7fcbce7cdbdd1feb553718cbde60bb8ccff4733"
}
backend/.sqlx/query-e3d4f89ce36337af15d237b543eaca47771b480ff194884f9c947dcaf71d6cf9.json
Generated
+30
View File
@@ -0,0 +1,30 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE\n websocket_trigger\n SET\n url = $1,\n script_path = $2,\n path = $3,\n is_flow = $4,\n filters = $5,\n filter_logic = $6,\n initial_messages = $7,\n url_runnable_args = $8,\n edited_by = $9,\n permissioned_as = $10,\n can_return_message = $11,\n can_return_error_result = $12,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $15,\n error_handler_args = $16,\n retry = $17\n WHERE\n workspace_id = $13 AND path = $14\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Bool",
"JsonbArray",
"Varchar",
"JsonbArray",
"Jsonb",
"Varchar",
"Varchar",
"Bool",
"Bool",
"Text",
"Text",
"Varchar",
"Jsonb",
"Jsonb"
]
},
"nullable": []
},
"hash": "e3d4f89ce36337af15d237b543eaca47771b480ff194884f9c947dcaf71d6cf9"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
780857855e231c9d71f02fefd8253c254542ef32
5ee1382dfb23b6a1516e3c7586058cec8240fdf2
backend/migrations/20260328000000_trigger_filter_logic.down.sql
+2
View File
@@ -0,0 +1,2 @@
ALTER TABLE kafka_trigger DROP COLUMN filter_logic;
ALTER TABLE websocket_trigger DROP COLUMN filter_logic;
backend/migrations/20260328000000_trigger_filter_logic.up.sql
+2
View File
@@ -0,0 +1,2 @@
ALTER TABLE kafka_trigger ADD COLUMN filter_logic VARCHAR(3) NOT NULL DEFAULT 'and';
ALTER TABLE websocket_trigger ADD COLUMN filter_logic VARCHAR(3) NOT NULL DEFAULT 'and';
backend/windmill-api/openapi.yaml
+42
View File
@@ -21814,6 +21814,13 @@ components:
required:
- key
- value
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
initial_messages:
type: array
nullable: true
@@ -21875,6 +21882,13 @@ components:
required:
- key
- value
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
initial_messages:
type: array
nullable: true
@@ -21943,6 +21957,13 @@ components:
required:
- key
- value
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
initial_messages:
type: array
nullable: true
@@ -22819,6 +22840,13 @@ components:
required:
- key
- value
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
auto_offset_reset:
type: string
enum:
@@ -22890,6 +22918,13 @@ components:
required:
- key
- value
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
auto_offset_reset:
type: string
enum:
@@ -22953,6 +22988,13 @@ components:
required:
- key
- value
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
auto_offset_reset:
type: string
enum:
backend/windmill-trigger-websocket/src/handler.rs
+16 -11
View File
@@ -36,6 +36,7 @@ impl TriggerCrud for WebsocketTrigger {
const ADDITIONAL_SELECT_FIELDS: &[&'static str] = &[
"url",
"filters",
"filter_logic",
"initial_messages",
"url_runnable_args",
"can_return_message",
@@ -103,6 +104,7 @@ impl TriggerCrud for WebsocketTrigger {
is_flow,
mode,
filters,
filter_logic,
initial_messages,
url_runnable_args,
edited_by,
@@ -114,7 +116,7 @@ impl TriggerCrud for WebsocketTrigger {
error_handler_args,
retry
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, now(), $14, $15, $16
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, now(), $15, $16, $17
)
"#,
w_id,
@@ -124,6 +126,7 @@ impl TriggerCrud for WebsocketTrigger {
trigger.base.is_flow,
trigger.base.mode() as _,
&filters as _,
trigger.config.filter_logic,
&initial_messages as _,
trigger
.config
@@ -178,26 +181,28 @@ impl TriggerCrud for WebsocketTrigger {
path = $3,
is_flow = $4,
filters = $5,
initial_messages = $6,
url_runnable_args = $7,
edited_by = $8,
permissioned_as = $9,
can_return_message = $10,
can_return_error_result = $11,
filter_logic = $6,
initial_messages = $7,
url_runnable_args = $8,
edited_by = $9,
permissioned_as = $10,
can_return_message = $11,
can_return_error_result = $12,
edited_at = now(),
server_id = NULL,
error = NULL,
error_handler_path = $14,
error_handler_args = $15,
retry = $16
error_handler_path = $15,
error_handler_args = $16,
retry = $17
WHERE
workspace_id = $12 AND path = $13
workspace_id = $13 AND path = $14
",
trigger.config.url,
trigger.base.script_path,
trigger.base.path,
trigger.base.is_flow,
filters.as_slice() as &[SqlxJson<Box<RawValue>>],
trigger.config.filter_logic,
initial_messages.as_slice() as &[SqlxJson<Box<RawValue>>],
trigger
.config
backend/windmill-trigger-websocket/src/lib.rs
+12 -4
View File
@@ -1,12 +1,9 @@
use std::collections::HashMap;
use windmill_api_auth::ApiAuthed;
use windmill_trigger::trigger_helpers::{
trigger_runnable_and_wait_for_raw_result_with_error_ctx, TriggerJobArgs,
};
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sqlx::{types::Json as SqlxJson, FromRow};
use windmill_api_auth::ApiAuthed;
use windmill_common::{
error::{Error, Result},
jobs::JobTriggerKind,
@@ -15,6 +12,9 @@ use windmill_common::{
DB,
};
use windmill_queue::PushArgsOwned;
use windmill_trigger::trigger_helpers::{
trigger_runnable_and_wait_for_raw_result_with_error_ctx, TriggerJobArgs,
};
pub mod handler;
pub mod listener;
@@ -30,11 +30,17 @@ impl TriggerJobArgs for WebsocketTrigger {
}
}
fn default_filter_logic() -> String {
"and".to_string()
}
#[derive(Debug, Clone, FromRow, Serialize, Deserialize)]
pub struct WebsocketConfig {
pub url: String,
#[serde(default)]
pub filters: Vec<SqlxJson<Box<RawValue>>>,
#[serde(default = "default_filter_logic")]
pub filter_logic: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub initial_messages: Option<Vec<SqlxJson<Box<RawValue>>>>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -49,6 +55,8 @@ pub struct WebsocketConfig {
pub struct WebsocketConfigRequest {
url: String,
filters: Vec<serde_json::Value>,
#[serde(default = "default_filter_logic")]
filter_logic: String,
initial_messages: Option<Vec<serde_json::Value>>,
url_runnable_args: Option<serde_json::Value>,
can_return_message: bool,
backend/windmill-trigger-websocket/src/listener.rs
+3 -21
View File
@@ -18,7 +18,7 @@ use windmill_common::{
DB,
};
use windmill_queue::PushArgsOwned;
use windmill_trigger::filter::{is_value_superset, Filter, JsonFilter};
use windmill_trigger::filter::{check_filters, Filter};
use windmill_trigger::listener::ListeningTrigger;
use windmill_trigger::trigger_helpers::{
trigger_runnable, trigger_runnable_and_wait_for_raw_result,
@@ -267,26 +267,8 @@ impl Listener for WebsocketTrigger {
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;
}
}
let use_or = listening_trigger.trigger_config.filter_logic == "or";
let should_handle = check_filters(&text, &filters, use_or);
if should_handle {
let trigger_info = HashMap::from([
("url".to_string(), to_raw_value(&listening_trigger.trigger_config.url)),
backend/windmill-trigger/src/filter.rs
+21
View File
@@ -80,6 +80,27 @@ where
deserializer.deserialize_map(SupersetVisitor { key, value_to_check })
}
pub fn check_filters(text: &str, filters: &[Filter], use_or_logic: bool) -> bool {
if filters.is_empty() {
return true;
}
let check = |filter: &Filter| -> bool {
match filter {
Filter::JsonFilter(JsonFilter { key, value }) => {
let mut deserializer = serde_json::Deserializer::from_str(text);
is_value_superset(&mut deserializer, key, value).unwrap_or(false)
}
}
};
if use_or_logic {
filters.iter().any(check)
} else {
filters.iter().all(check)
}
}
#[cfg(test)]
mod tests {
use super::*;
cli/src/guidance/skills.ts
+14
View File
@@ -5820,6 +5820,13 @@ properties:
key:
type: string
value: {}
filter_logic:
type: string
enum:
- and
- or
description: Logic to apply when evaluating filters. 'and' requires all filters
to match, 'or' requires any filter to match.
auto_offset_reset:
type: string
enum:
@@ -6347,6 +6354,13 @@ properties:
value: {}
description: Array of key-value filters to match incoming messages (only matching
messages trigger the script)
filter_logic:
type: string
enum:
- and
- or
description: Logic to apply when evaluating filters. 'and' requires all filters
to match, 'or' requires any filter to match.
initial_messages:
type: array
items:
frontend/src/lib/components/triggers/CaptureWrapper.svelte
+1 -1
View File
@@ -260,7 +260,7 @@
{hasPreprocessor}
{isFlow}
{captureLoading}
{triggerDeployed}
groupId={args?.group_id}
on:applyArgs
on:updateSchema
on:addPreprocessor
frontend/src/lib/components/triggers/TriggerFilters.svelte
+24 -2
View File
@@ -1,23 +1,45 @@
<script lang="ts">
import { Button } from '$lib/components/common'
import Section from '$lib/components/Section.svelte'
import Select from '$lib/components/select/Select.svelte'
import { Plus, X } from 'lucide-svelte'
import { fade } from 'svelte/transition'
import JsonEditor from '$lib/components/JsonEditor.svelte'
interface Props {
filters: { key: string; value: any }[]
filterLogic: 'and' | 'or'
disabled?: boolean
}
let { filters = $bindable([]), disabled = false }: Props = $props()
let {
filters = $bindable([]),
filterLogic = $bindable(),
disabled = false
}: Props = $props()
const filterLogicItems = [
{ label: 'all criteria (AND)', value: 'and' as const },
{ label: 'any criterion (OR)', value: 'or' as const }
]
let description = $derived(
filterLogic === 'or'
? 'Filters will limit the execution of the trigger to only messages that match any criterion.'
: 'Filters will limit the execution of the trigger to only messages that match all criteria.'
)
</script>
<Section label="Filters">
<p class="text-xs mb-1 text-primary">
Filters will limit the execution of the trigger to only messages that match all criteria.<br />
{description}<br />
The JSON filter checks if the value at the key is equal or a superset of the filter value.
</p>
{#if filters.length > 0}
<div class="mt-2 mb-1 max-w-xs">
<Select items={filterLogicItems} bind:value={filterLogic} {disabled} size="sm" />
</div>
{/if}
<div class="flex flex-col gap-4 mt-1">
{#each filters as v, i (i)}
<div class="flex w-full gap-2 items-center">
frontend/src/lib/components/triggers/kafka/KafkaCapture.svelte
+8 -3
View File
@@ -1,5 +1,6 @@
<script lang="ts">
import { fade } from 'svelte/transition'
import { Url } from '$lib/components/common'
import type { CaptureInfo } from '../CaptureSection.svelte'
import CaptureSection from '../CaptureSection.svelte'
@@ -9,7 +10,7 @@
hasPreprocessor?: boolean
isFlow?: boolean
captureLoading?: boolean
triggerDeployed?: boolean
groupId?: string
}
let {
@@ -18,8 +19,10 @@
hasPreprocessor = false,
isFlow = false,
captureLoading = false,
triggerDeployed = false
groupId = undefined
}: Props = $props()
let captureGroupId = $derived(groupId ? `${groupId}_wm_capture` : undefined)
</script>
{#if captureInfo}
@@ -35,7 +38,6 @@
on:testWithArgs
{hasPreprocessor}
{isFlow}
displayAlert={triggerDeployed}
>
{#snippet description()}
{#if captureInfo.active}
@@ -54,5 +56,8 @@
</p>
{/if}
{/snippet}
{#if captureGroupId}
<Url label="Test group ID" url={captureGroupId} />
{/if}
</CaptureSection>
{/if}
frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte
+5 -1
View File
@@ -98,6 +98,7 @@
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let filters: { key: string; value: any }[] = $state([])
let filterLogic = $state<'and' | 'or'>('and')
let suspendedJobsModal = $state<TriggerSuspendedJobsModal | null>(null)
let originalConfig = $state<Record<string, any> | undefined>(undefined)
@@ -190,6 +191,7 @@
error_handler_args = nDefaultValues?.error_handler_args ?? {}
retry = nDefaultValues?.retry ?? undefined
filters = nDefaultValues?.filters ?? []
filterLogic = 'and'
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
mode = nDefaultValues?.mode ?? 'enabled'
originalConfig = undefined
@@ -219,6 +221,7 @@
error_handler_args = cfg?.error_handler_args ?? {}
retry = cfg?.retry
filters = cfg?.filters ?? []
filterLogic = cfg?.filter_logic ?? 'and'
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
permissionedAs = cfg?.permissioned_as
selectedPermissionedAs = undefined
@@ -247,6 +250,7 @@
group_id: kafkaCfg.group_id,
topics: kafkaCfg.topics,
filters,
filter_logic: filterLogic,
auto_offset_reset: autoOffsetReset,
auto_commit: autoCommit,
mode,
@@ -577,7 +581,7 @@
</Label>
{/if}
<TriggerFilters bind:filters disabled={!can_write} />
<TriggerFilters bind:filters bind:filterLogic disabled={!can_write} />
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
frontend/src/lib/components/triggers/kafka/utils.ts
+1
View File
@@ -24,6 +24,7 @@ export async function saveKafkaTriggerFromCfg(
group_id: cfg.group_id,
topics: cfg.topics,
filters: cfg.filters ?? [],
filter_logic: cfg.filter_logic ?? 'and',
auto_offset_reset: cfg.auto_offset_reset ?? 'latest',
auto_commit: cfg.auto_commit ?? true,
...errorHandlerAndRetries,
frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte
+5 -1
View File
@@ -96,6 +96,7 @@
key: string
value: any
}[] = $state([])
let filterLogic = $state<'and' | 'or'>('and')
let initial_messages: WebsocketTriggerInitialMessage[] = $state([])
let url_runnable_args: Record<string, any> | undefined = $state({})
let can_return_message = $state(false)
@@ -203,6 +204,7 @@
path = defaultValues?.path ?? ''
initialPath = ''
filters = []
filterLogic = 'and'
initial_messages = []
url_runnable_args = defaultValues?.url_runnable_args ?? {}
dirtyPath = false
@@ -228,6 +230,7 @@
path = cfg?.path
url = cfg?.url
filters = cfg?.filters
filterLogic = cfg?.filter_logic ?? 'and'
initial_messages = cfg?.initial_messages ?? []
url_runnable_args = cfg?.url_runnable_args
can_return_message = cfg?.can_return_message
@@ -250,6 +253,7 @@
path,
url,
filters,
filter_logic: filterLogic,
initial_messages,
url_runnable_args,
can_return_message,
@@ -713,7 +717,7 @@
/>
{/snippet}
<div class="flex flex-col gap-6">
<TriggerFilters bind:filters disabled={!can_write} />
<TriggerFilters bind:filters bind:filterLogic disabled={!can_write} />
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="error_handler" label="Error Handler" />
frontend/src/lib/components/triggers/websocket/utils.ts
+1
View File
@@ -25,6 +25,7 @@ export async function saveWebsocketTriggerFromCfg(
is_flow: triggerCfg.is_flow,
url: triggerCfg.url,
filters: triggerCfg.filters,
filter_logic: triggerCfg.filter_logic ?? 'and',
initial_messages: triggerCfg.initial_messages,
url_runnable_args: triggerCfg.url_runnable_args,
can_return_message: triggerCfg.can_return_message,
system_prompts/auto-generated/schemas/kafka_trigger.schema.yaml
+7
View File
@@ -28,6 +28,13 @@ properties:
key:
type: string
value: {}
filter_logic:
type: string
enum:
- and
- or
description: Logic to apply when evaluating filters. 'and' requires all filters
to match, 'or' requires any filter to match.
auto_offset_reset:
type: string
enum:
system_prompts/auto-generated/schemas/websocket_trigger.schema.yaml
+7
View File
@@ -23,6 +23,13 @@ properties:
value: {}
description: Array of key-value filters to match incoming messages (only matching
messages trigger the script)
filter_logic:
type: string
enum:
- and
- or
description: Logic to apply when evaluating filters. 'and' requires all filters
to match, 'or' requires any filter to match.
initial_messages:
type: array
items: