mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(channel,trigger): wire channel_message + content_match event sources
This commit is contained in:
matevip
parent
69719eb35e
commit
dae74cf26c
@ -1,6 +1,8 @@
|
||||
package vip.mate.channel;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.stereotype.Component;
|
||||
import reactor.core.publisher.Flux;
|
||||
import vip.mate.agent.AgentService;
|
||||
@ -8,6 +10,7 @@ import vip.mate.agent.context.ChatOrigin;
|
||||
import vip.mate.approval.ApprovalWorkflowService;
|
||||
import vip.mate.approval.ResolveOutcome;
|
||||
import vip.mate.approval.PendingApproval;
|
||||
import vip.mate.channel.event.ChannelMessageReceivedEvent;
|
||||
import vip.mate.channel.model.ChannelEntity;
|
||||
import vip.mate.channel.notification.ApprovalNotificationService;
|
||||
import vip.mate.channel.service.ChannelService;
|
||||
@ -59,6 +62,11 @@ public class ChannelMessageRouter {
|
||||
private final ChatStreamTracker streamTracker;
|
||||
private final ChannelChatOriginFactory chatOriginFactory;
|
||||
private final ChannelErrorClassifier errorClassifier;
|
||||
/** Field-injected (rather than constructor) to avoid a signature
|
||||
* change that would ripple through every test that constructs the
|
||||
* router directly. Spring's stock publisher is always available. */
|
||||
@Autowired(required = false)
|
||||
private ApplicationEventPublisher events;
|
||||
|
||||
/** 队列条目：封装消息及其路由上下文 */
|
||||
private record QueueEntry(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) {}
|
||||
@ -189,6 +197,14 @@ public class ChannelMessageRouter {
|
||||
* @param channelEntity 渠道配置（含关联 agentId）
|
||||
*/
|
||||
public void enqueue(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) {
|
||||
// Fan out to the trigger pipeline FIRST — channel_message and
|
||||
// content_match triggers fire on every received message regardless
|
||||
// of whether the channel has an agent attached. If we returned
|
||||
// early on a missing agent below without publishing, the workflow
|
||||
// side would silently lose every channel-event that doesn't also
|
||||
// route to a chat agent.
|
||||
publishChannelEvent(message, adapter, channelEntity);
|
||||
|
||||
Long agentId = channelEntity.getAgentId();
|
||||
if (agentId == null) {
|
||||
log.warn("Channel {} has no associated agent, ignoring message from {}",
|
||||
@ -231,6 +247,43 @@ public class ChannelMessageRouter {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Publish a {@link ChannelMessageReceivedEvent} so the trigger module's
|
||||
* bridge can fan the message out to channel_message + content_match
|
||||
* triggers. Best-effort — a publish failure must never block the
|
||||
* primary chat-routing path. {@code messageId} is used as the dedup
|
||||
* key downstream so repeated webhook deliveries can't double-fire
|
||||
* the same trigger.
|
||||
*/
|
||||
private void publishChannelEvent(ChannelMessage message, ChannelAdapter adapter,
|
||||
ChannelEntity channelEntity) {
|
||||
if (events == null || message == null || adapter == null || channelEntity == null) return;
|
||||
try {
|
||||
long ws = channelEntity.getWorkspaceId() == null ? 0L : channelEntity.getWorkspaceId();
|
||||
String channelType = adapter.getChannelType();
|
||||
// messageId may be null for adapters that don't surface one;
|
||||
// fall back to a sender+timestamp composite so the dedup key
|
||||
// is at least deterministic-ish per webhook delivery.
|
||||
String messageId = message.getMessageId();
|
||||
if (messageId == null || messageId.isBlank()) {
|
||||
messageId = channelType + ":" + message.getSenderId() + ":"
|
||||
+ (message.getTimestamp() == null ? System.currentTimeMillis()
|
||||
: message.getTimestamp());
|
||||
}
|
||||
events.publishEvent(new ChannelMessageReceivedEvent(
|
||||
ws,
|
||||
channelType,
|
||||
messageId,
|
||||
message.getSenderId(),
|
||||
message.getSenderName(),
|
||||
message.getChatId(),
|
||||
message.getContent()));
|
||||
} catch (Exception e) {
|
||||
log.warn("[ChannelMessageRouter] event publish failed for sender {}: {}",
|
||||
message.getSenderId(), e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 防抖到期：将合并后的消息真正放入渠道队列
|
||||
*/
|
||||
|
||||
@ -0,0 +1,26 @@
|
||||
package vip.mate.channel.event;
|
||||
|
||||
/**
|
||||
* Spring application event fired when a channel adapter accepts an
|
||||
* inbound message and hands it off to {@code ChannelMessageRouter}.
|
||||
* The trigger module subscribes via {@code @EventListener} and forwards
|
||||
* the payload through {@code TriggerEventIngestService} so triggers of
|
||||
* pattern type {@code channel_message} or {@code content_match} can fan
|
||||
* out to workflows. Going through the event bus instead of injecting
|
||||
* the trigger service directly into the channel module keeps the two
|
||||
* worlds decoupled and dodges the construction cycle.
|
||||
*
|
||||
* <p>{@code messageId} doubles as the dedup key — repeated webhook
|
||||
* deliveries of the same message can't double-fire downstream triggers
|
||||
* because the {@code mate_trigger_event} unique constraint catches the
|
||||
* second insert.
|
||||
*/
|
||||
public record ChannelMessageReceivedEvent(
|
||||
long workspaceId,
|
||||
String channelType,
|
||||
String messageId,
|
||||
String senderId,
|
||||
String senderName,
|
||||
String chatId,
|
||||
String content
|
||||
) {}
|
||||
@ -0,0 +1,70 @@
|
||||
package vip.mate.trigger.dispatch;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
import vip.mate.channel.event.ChannelMessageReceivedEvent;
|
||||
import vip.mate.trigger.ingest.TriggerEventEnvelope;
|
||||
import vip.mate.trigger.ingest.TriggerEventIngestService;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Bridges {@link ChannelMessageReceivedEvent} from the channel module
|
||||
* into two trigger pattern types — {@code channel_message} (matches by
|
||||
* {@code channelType} / {@code senderEquals}) and {@code content_match}
|
||||
* (matches by substring inside the message content). The same envelope
|
||||
* fans out to both since the matcher's per-pattern key on the SQL
|
||||
* candidate query selects which triggers actually run.
|
||||
*
|
||||
* <p>Lives in the trigger module so the channel runtime stays free of
|
||||
* trigger / ingest dependencies. Failures inside ingest are logged and
|
||||
* swallowed — a bad downstream trigger MUST NOT corrupt the primary
|
||||
* chat-routing path that just published the event.
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class ChannelMessageEventBridge {
|
||||
|
||||
private final TriggerEventIngestService ingestService;
|
||||
|
||||
@EventListener
|
||||
public void onChannelMessage(ChannelMessageReceivedEvent event) {
|
||||
if (event == null) return;
|
||||
try {
|
||||
Map<String, Object> data = new HashMap<>();
|
||||
data.put("channelType", event.channelType());
|
||||
data.put("senderId", event.senderId());
|
||||
if (event.senderName() != null) data.put("senderName", event.senderName());
|
||||
if (event.chatId() != null) data.put("chatId", event.chatId());
|
||||
// The matcher's content_match pattern reads `data.content`,
|
||||
// so we put the message body there even when it's blank.
|
||||
data.put("content", event.content() == null ? "" : event.content());
|
||||
|
||||
// Fan to channel_message pattern triggers.
|
||||
ingestService.ingest(new TriggerEventEnvelope(
|
||||
event.workspaceId(),
|
||||
"channel_message",
|
||||
event.messageId(),
|
||||
event.senderId(),
|
||||
data));
|
||||
// And to content_match triggers, which live under a different
|
||||
// patternType but read the same envelope shape. Two separate
|
||||
// ingests instead of one because the SQL candidate query
|
||||
// filters on patternType — a single dispatch with one
|
||||
// patternType cannot reach the other set.
|
||||
ingestService.ingest(new TriggerEventEnvelope(
|
||||
event.workspaceId(),
|
||||
"content_match",
|
||||
event.messageId(),
|
||||
event.senderId(),
|
||||
data));
|
||||
} catch (Exception e) {
|
||||
log.warn("[ChannelMessageBridge] forwarding message {} from {} failed: {}",
|
||||
event.messageId(), event.senderId(), e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -58,8 +58,13 @@ const selectStep = inject<(data: StepNodeData | null) => void>(
|
||||
'selectStepCallback',
|
||||
() => {}
|
||||
)
|
||||
function handleClick() {
|
||||
function handleClick(event: MouseEvent) {
|
||||
selectStep(props.data as StepNodeData)
|
||||
const root = event.currentTarget as HTMLElement | null
|
||||
const canvasId = root?.closest('.workflow-canvas')?.getAttribute('data-canvas-id') ?? undefined
|
||||
window.dispatchEvent(new CustomEvent('mateclaw:workflow-step-select', {
|
||||
detail: { canvasId, data: props.data as StepNodeData },
|
||||
}))
|
||||
}
|
||||
|
||||
// Position values come from the parent canvas (LR or TB orientation);
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
<template>
|
||||
<div class="workflow-canvas" :class="{ fullscreen }">
|
||||
<div class="workflow-canvas" :class="{ fullscreen }" :data-canvas-id="canvasId">
|
||||
<div class="canvas-toolbar">
|
||||
<div class="canvas-toolbar-group">
|
||||
<button class="canvas-btn" :class="{ active: direction === 'LR' }" @click="direction = 'LR'">
|
||||
@ -75,6 +75,10 @@
|
||||
zoomable
|
||||
/>
|
||||
</VueFlow>
|
||||
|
||||
<div v-if="$slots.panel" class="canvas-panel-slot">
|
||||
<slot name="panel" />
|
||||
</div>
|
||||
</div>
|
||||
</template>
|
||||
|
||||
@ -155,7 +159,13 @@ function selectStep(data: StepNodeData | null) {
|
||||
provide('selectStepCallback', selectStep)
|
||||
|
||||
const flow = useVueFlow(props.canvasId)
|
||||
function handleDomStepSelect(event: Event) {
|
||||
const detail = (event as CustomEvent<{ canvasId?: string; data?: StepNodeData }>).detail
|
||||
if (!detail?.data || detail.canvasId !== props.canvasId) return
|
||||
selectStep(detail.data)
|
||||
}
|
||||
onMounted(() => {
|
||||
window.addEventListener('mateclaw:workflow-step-select', handleDomStepSelect as EventListener)
|
||||
flow.onNodeClick((evt) => {
|
||||
const data = (evt?.node as Node<StepNodeData> | undefined)?.data
|
||||
if (data) selectStep(data)
|
||||
@ -198,6 +208,7 @@ watch(fullscreen, (on) => {
|
||||
}
|
||||
})
|
||||
onBeforeUnmount(() => {
|
||||
window.removeEventListener('mateclaw:workflow-step-select', handleDomStepSelect as EventListener)
|
||||
document.removeEventListener('keydown', handleEsc)
|
||||
document.body.style.overflow = ''
|
||||
})
|
||||
@ -322,4 +333,40 @@ onBeforeUnmount(() => {
|
||||
height: 100%;
|
||||
min-height: 280px;
|
||||
}
|
||||
.canvas-flow :deep(.vue-flow__edge-path) {
|
||||
stroke: var(--mc-edge, rgba(74, 85, 104, 0.78));
|
||||
stroke-width: 2;
|
||||
}
|
||||
.canvas-flow :deep(.vue-flow__edge-textbg) {
|
||||
fill: var(--mc-bg-elevated, #ffffff);
|
||||
}
|
||||
.canvas-flow :deep(.vue-flow__edge-text) {
|
||||
fill: var(--mc-text-secondary, #666666);
|
||||
font-size: 11px;
|
||||
}
|
||||
.canvas-panel-slot {
|
||||
position: absolute;
|
||||
top: 56px;
|
||||
right: 14px;
|
||||
bottom: 14px;
|
||||
width: min(340px, calc(100% - 28px));
|
||||
z-index: 12;
|
||||
pointer-events: none;
|
||||
}
|
||||
.canvas-panel-slot :deep(.step-panel) {
|
||||
height: 100%;
|
||||
max-height: none;
|
||||
pointer-events: auto;
|
||||
box-shadow: 0 18px 40px rgba(0, 0, 0, 0.18);
|
||||
}
|
||||
@media (max-width: 760px) {
|
||||
.canvas-panel-slot {
|
||||
top: auto;
|
||||
left: 10px;
|
||||
right: 10px;
|
||||
bottom: 10px;
|
||||
width: auto;
|
||||
height: min(56vh, 420px);
|
||||
}
|
||||
}
|
||||
</style>
|
||||
|
||||
@ -124,6 +124,15 @@ export function buildGraph(json: string): { nodes: Node<StepNodeData>[]; edges:
|
||||
// mergePoint: the node id whose output the next non-fan_out step should consume
|
||||
// fanGroup: the ids of fan_out steps in the most recent open group, ready to be joined by a collect
|
||||
const edges: Edge[] = []
|
||||
const pushEdge = (source: string, target: string, extra: Partial<Edge> = {}) => {
|
||||
edges.push({
|
||||
id: `edge-${edges.length + 1}`,
|
||||
source,
|
||||
target,
|
||||
animated: false,
|
||||
...extra,
|
||||
})
|
||||
}
|
||||
let mergePoint: string | null = null
|
||||
let fanGroup: string[] = []
|
||||
let prevModeWasFanOut = false
|
||||
@ -135,12 +144,7 @@ export function buildGraph(json: string): { nodes: Node<StepNodeData>[]; edges:
|
||||
if (!prevModeWasFanOut) fanGroup = []
|
||||
fanGroup.push(node.id)
|
||||
if (mergePoint) {
|
||||
edges.push({
|
||||
id: `e-${mergePoint}->${node.id}`,
|
||||
source: mergePoint,
|
||||
target: node.id,
|
||||
animated: false,
|
||||
})
|
||||
pushEdge(mergePoint, node.id)
|
||||
}
|
||||
prevModeWasFanOut = true
|
||||
} else if (mode === 'collect') {
|
||||
@ -150,12 +154,7 @@ export function buildGraph(json: string): { nodes: Node<StepNodeData>[]; edges:
|
||||
// canvas surfaces the orphan visually.
|
||||
const sources = fanGroup.length ? fanGroup : (mergePoint ? [mergePoint] : [])
|
||||
for (const src of sources) {
|
||||
edges.push({
|
||||
id: `e-${src}->${node.id}`,
|
||||
source: src,
|
||||
target: node.id,
|
||||
animated: false,
|
||||
})
|
||||
pushEdge(src, node.id)
|
||||
}
|
||||
fanGroup = []
|
||||
mergePoint = node.id
|
||||
@ -164,16 +163,11 @@ export function buildGraph(json: string): { nodes: Node<StepNodeData>[]; edges:
|
||||
// Sequential / conditional / await_approval / dispatch_channel / write_memory
|
||||
// all attach to the current merge point and become the next merge point.
|
||||
if (mergePoint) {
|
||||
const edge: Edge = {
|
||||
id: `e-${mergePoint}->${node.id}`,
|
||||
source: mergePoint,
|
||||
target: node.id,
|
||||
animated: false,
|
||||
}
|
||||
const edge: Partial<Edge> = {}
|
||||
if (mode === 'conditional' && node.data?.expression) {
|
||||
edge.label = `if ${node.data.expression}`
|
||||
}
|
||||
edges.push(edge)
|
||||
pushEdge(mergePoint, node.id, edge)
|
||||
}
|
||||
mergePoint = node.id
|
||||
fanGroup = []
|
||||
|
||||
@ -65,15 +65,17 @@
|
||||
v-model="canvasModel"
|
||||
:canvas-id="`wf-${selected.id}`"
|
||||
@select-step="onCanvasSelect"
|
||||
/>
|
||||
<StepPropertyPanel
|
||||
v-if="canvasSelection"
|
||||
:step="selectedStep"
|
||||
:index="canvasSelection.index"
|
||||
@patch="onStepPatch"
|
||||
@duplicate="onStepDuplicate"
|
||||
@delete="onStepDelete"
|
||||
/>
|
||||
>
|
||||
<template v-if="canvasSelection" #panel>
|
||||
<StepPropertyPanel
|
||||
:step="selectedStep"
|
||||
:index="canvasSelection.index"
|
||||
@patch="onStepPatch"
|
||||
@duplicate="onStepDuplicate"
|
||||
@delete="onStepDelete"
|
||||
/>
|
||||
</template>
|
||||
</WorkflowCanvas>
|
||||
</div>
|
||||
|
||||
<div v-else class="json-pane">
|
||||
|
||||
Loading…
Reference in New Issue
Block a user