mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(trigger): match pattern_json before fan-out + periodic sync
This commit is contained in:
matevip
parent
05a2caa47c
commit
e09c62b529
@ -55,18 +55,21 @@ public class TriggerEventIngestService {
|
||||
private final TriggerDispatcher dispatcher;
|
||||
private final BotSelfFilter botSelfFilter;
|
||||
private final ObjectMapper objectMapper;
|
||||
private final TriggerPatternMatcher patternMatcher;
|
||||
private final TriggerRateLimiter rateLimiter = new TriggerRateLimiter();
|
||||
|
||||
public TriggerEventIngestService(TriggerMapper triggerMapper,
|
||||
TriggerEventMapper eventMapper,
|
||||
TriggerDispatcher dispatcher,
|
||||
BotSelfFilter botSelfFilter,
|
||||
ObjectMapper objectMapper) {
|
||||
ObjectMapper objectMapper,
|
||||
TriggerPatternMatcher patternMatcher) {
|
||||
this.triggerMapper = triggerMapper;
|
||||
this.eventMapper = eventMapper;
|
||||
this.dispatcher = dispatcher;
|
||||
this.botSelfFilter = botSelfFilter;
|
||||
this.objectMapper = objectMapper;
|
||||
this.patternMatcher = patternMatcher;
|
||||
}
|
||||
|
||||
/**
|
||||
@ -92,6 +95,14 @@ public class TriggerEventIngestService {
|
||||
}
|
||||
|
||||
private IngestResult processSingle(TriggerEntity trigger, TriggerEventEnvelope envelope) {
|
||||
// Pattern matching is the first gate — without it, every channel
|
||||
// event would broadcast to every channel-message trigger in the
|
||||
// workspace, which is exactly the storm hazard the design forbade.
|
||||
// Run it before all the other filters so a non-matching trigger
|
||||
// doesn't even allocate a dedup row.
|
||||
if (!patternMatcher.matches(trigger, envelope)) {
|
||||
return IngestResult.dropped(trigger.getId(), Reason.PATTERN_MISMATCH);
|
||||
}
|
||||
if (Boolean.TRUE.equals(trigger.getBotSelfFilter())
|
||||
&& botSelfFilter.isBotSelf(envelope.workspaceId(), envelope.senderId())) {
|
||||
return IngestResult.dropped(trigger.getId(), Reason.BOT_SELF);
|
||||
@ -163,7 +174,7 @@ public class TriggerEventIngestService {
|
||||
}
|
||||
|
||||
public enum Reason {
|
||||
BOT_SELF, DUPLICATE, RATE_LIMITED, EXHAUSTED, DISPATCH_ERROR
|
||||
PATTERN_MISMATCH, BOT_SELF, DUPLICATE, RATE_LIMITED, EXHAUSTED, DISPATCH_ERROR
|
||||
}
|
||||
|
||||
public record IngestResult(long triggerId, boolean fired, Reason droppedReason) {
|
||||
|
||||
@ -0,0 +1,165 @@
|
||||
package vip.mate.trigger.ingest;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
import vip.mate.trigger.model.TriggerEntity;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Decides whether a trigger's stored {@code pattern_json} actually matches
|
||||
* an inbound envelope, beyond the coarse {@code (workspaceId, patternType)}
|
||||
* filter the SQL query already does.
|
||||
*
|
||||
* <p>Without this layer the ingest service broadcasts every event to every
|
||||
* trigger in the same workspace that happens to share a {@code patternType},
|
||||
* which is the event-storm hazard the design has warned about — one channel
|
||||
* message would fire every channel-message trigger regardless of intent.
|
||||
*
|
||||
* <p>v0 supports four pattern shapes:
|
||||
* <ul>
|
||||
* <li><b>cron</b> — never matches an inbound envelope. Cron triggers run
|
||||
* through the scheduler, not the ingest pipeline.</li>
|
||||
* <li><b>channel_message</b> — optional {@code channelType} narrows by
|
||||
* which adapter the envelope came from; optional {@code senderEquals}
|
||||
* narrows to a specific sender id.</li>
|
||||
* <li><b>agent_lifecycle</b> — optional {@code agentId} narrows to a
|
||||
* specific agent's lifecycle events; optional {@code phase} narrows
|
||||
* to {@code spawned} / {@code terminated} / {@code crashed}.</li>
|
||||
* <li><b>content_match</b> — required {@code substring} must appear in
|
||||
* the envelope's {@code data.content} field (case-insensitive); this
|
||||
* is the explicit pattern that the design always intended to require
|
||||
* payload-level evaluation.</li>
|
||||
* <li><b>workflow_completion</b> — optional {@code sourceWorkflowId}
|
||||
* narrows to a specific upstream workflow; optional {@code stateFilter}
|
||||
* narrows to {@code completed} / {@code failed} / {@code any}.</li>
|
||||
* <li><b>webhook</b> — opaque pass-through. v0 doesn't filter further.</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>Unknown pattern types fail closed (no match) so a typo'd or future
|
||||
* pattern type can't silently fire every workspace trigger.
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class TriggerPatternMatcher {
|
||||
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
public TriggerPatternMatcher(ObjectMapper objectMapper) {
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
public boolean matches(TriggerEntity trigger, TriggerEventEnvelope envelope) {
|
||||
String type = trigger.getPatternType();
|
||||
if (type == null) return false;
|
||||
JsonNode pattern = parsePattern(trigger);
|
||||
return switch (type) {
|
||||
case "cron" -> false; // scheduler-driven, not ingested
|
||||
case "channel_message" -> matchesChannelMessage(pattern, envelope);
|
||||
case "agent_lifecycle" -> matchesAgentLifecycle(pattern, envelope);
|
||||
case "content_match" -> matchesContent(pattern, envelope);
|
||||
case "workflow_completion" -> matchesWorkflowCompletion(pattern, envelope);
|
||||
case "webhook" -> true; // pass-through; secret check happens at the HTTP boundary
|
||||
default -> {
|
||||
log.warn("Trigger {} uses unknown patternType '{}' — failing closed",
|
||||
trigger.getId(), type);
|
||||
yield false;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private JsonNode parsePattern(TriggerEntity trigger) {
|
||||
String json = trigger.getPatternJson();
|
||||
if (json == null || json.isBlank()) return objectMapper.nullNode();
|
||||
try {
|
||||
return objectMapper.readTree(json);
|
||||
} catch (Exception e) {
|
||||
// A trigger with malformed pattern_json should never have been
|
||||
// accepted at create / update time; fail closed at fire time.
|
||||
log.warn("Trigger {} pattern_json parse failed: {}", trigger.getId(), e.getMessage());
|
||||
return objectMapper.nullNode();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean matchesChannelMessage(JsonNode pattern, TriggerEventEnvelope envelope) {
|
||||
if (envelope == null) return false;
|
||||
// channelType lives in envelope.data ("channelType" key) — the upstream
|
||||
// ChannelWebhookController stuffs it there. envelope itself is generic
|
||||
// and doesn't have a typed channel field.
|
||||
String wantChannel = textOrNull(pattern, "channelType");
|
||||
if (wantChannel != null) {
|
||||
Object actual = envelope.data() == null ? null : envelope.data().get("channelType");
|
||||
if (!(actual instanceof String s) || !wantChannel.equalsIgnoreCase(s)) return false;
|
||||
}
|
||||
String wantSender = textOrNull(pattern, "senderEquals");
|
||||
if (wantSender != null && !wantSender.equals(envelope.senderId())) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private boolean matchesAgentLifecycle(JsonNode pattern, TriggerEventEnvelope envelope) {
|
||||
Map<String, Object> data = envelope.data();
|
||||
if (data == null) return false;
|
||||
Long wantAgent = longOrNull(pattern, "agentId");
|
||||
if (wantAgent != null) {
|
||||
Object actual = data.get("agentId");
|
||||
if (!(actual instanceof Number n) || n.longValue() != wantAgent) return false;
|
||||
}
|
||||
String wantPhase = textOrNull(pattern, "phase");
|
||||
if (wantPhase != null) {
|
||||
Object phase = data.get("phase");
|
||||
if (!(phase instanceof String s) || !wantPhase.equalsIgnoreCase(s)) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private boolean matchesContent(JsonNode pattern, TriggerEventEnvelope envelope) {
|
||||
String needle = textOrNull(pattern, "substring");
|
||||
if (needle == null || needle.isBlank()) {
|
||||
// content_match without a substring is a misconfiguration — refuse
|
||||
// to fire blanket-on-every-event rather than acting as a wildcard.
|
||||
return false;
|
||||
}
|
||||
Map<String, Object> data = envelope.data();
|
||||
if (data == null) return false;
|
||||
Object content = data.get("content");
|
||||
if (!(content instanceof String s)) return false;
|
||||
return s.toLowerCase().contains(needle.toLowerCase());
|
||||
}
|
||||
|
||||
private boolean matchesWorkflowCompletion(JsonNode pattern, TriggerEventEnvelope envelope) {
|
||||
Map<String, Object> data = envelope.data();
|
||||
if (data == null) return false;
|
||||
Long wantSource = longOrNull(pattern, "sourceWorkflowId");
|
||||
if (wantSource != null) {
|
||||
Object actual = data.get("sourceWorkflowId");
|
||||
if (!(actual instanceof Number n) || n.longValue() != wantSource) return false;
|
||||
}
|
||||
String wantState = textOrNull(pattern, "stateFilter");
|
||||
if (wantState != null && !"any".equalsIgnoreCase(wantState)) {
|
||||
Object state = data.get("state");
|
||||
if (!(state instanceof String s) || !wantState.equalsIgnoreCase(s)) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private static String textOrNull(JsonNode node, String key) {
|
||||
if (node == null || !node.hasNonNull(key)) return null;
|
||||
String s = node.get(key).asText(null);
|
||||
return (s == null || s.isBlank()) ? null : s;
|
||||
}
|
||||
|
||||
private static Long longOrNull(JsonNode node, String key) {
|
||||
if (node == null || !node.hasNonNull(key)) return null;
|
||||
JsonNode v = node.get(key);
|
||||
if (v.isNumber()) return v.asLong();
|
||||
try {
|
||||
return Long.parseLong(v.asText());
|
||||
} catch (NumberFormatException e) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -11,6 +11,7 @@ import net.javacrumbs.shedlock.core.LockProvider;
|
||||
import net.javacrumbs.shedlock.core.SimpleLock;
|
||||
import org.springframework.boot.context.event.ApplicationReadyEvent;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
import org.springframework.scheduling.support.CronTrigger;
|
||||
import org.springframework.stereotype.Component;
|
||||
@ -82,16 +83,67 @@ public class TriggerScheduler {
|
||||
/** Boot-time registration sweep; runs after Flyway and bean wiring complete. */
|
||||
@EventListener(ApplicationReadyEvent.class)
|
||||
void registerEnabledTriggersOnStartup() {
|
||||
syncFromDatabase();
|
||||
}
|
||||
|
||||
/**
|
||||
* Periodic sweep that converges this node's local registrations with
|
||||
* the canonical state in {@code mate_trigger}.
|
||||
*
|
||||
* <p>Reasons this exists:
|
||||
* <ul>
|
||||
* <li>Multi-instance: when node A creates / updates / disables a
|
||||
* cron trigger, node B never gets the local-only register call.
|
||||
* The fire-time {@code patternVersion} guard self-cancels stale
|
||||
* schedules but does NOT register newly-created or newly-enabled
|
||||
* triggers — only this sweep does.</li>
|
||||
* <li>Recovery from missed events: if a register / unregister call
|
||||
* races with a node restart, the in-memory map can drift from
|
||||
* the row state. Refreshing every minute caps the divergence.</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>Convergence rules:
|
||||
* <ul>
|
||||
* <li>Row enabled + cron type + not registered locally → register.</li>
|
||||
* <li>Row enabled but local {@code capturedVersion} differs from
|
||||
* row's {@code pattern_version} → re-register (the schedule
|
||||
* carries the new expression).</li>
|
||||
* <li>Local registration exists for a row that's now disabled,
|
||||
* deleted, or no longer cron-typed → unregister.</li>
|
||||
* </ul>
|
||||
*/
|
||||
@Scheduled(fixedDelayString = "${mateclaw.workflow.trigger.sync-interval-ms:60000}",
|
||||
initialDelayString = "${mateclaw.workflow.trigger.sync-initial-delay-ms:60000}")
|
||||
public void syncFromDatabase() {
|
||||
var enabled = triggerMapper.selectList(new LambdaQueryWrapper<TriggerEntity>()
|
||||
.eq(TriggerEntity::getEnabled, true)
|
||||
.eq(TriggerEntity::getDeleted, 0));
|
||||
int loaded = 0;
|
||||
java.util.Set<Long> seenIds = new java.util.HashSet<>();
|
||||
int registered = 0, refreshed = 0, removed = 0;
|
||||
for (TriggerEntity t : enabled) {
|
||||
if (PATTERN_CRON.equalsIgnoreCase(t.getPatternType())) {
|
||||
if (registerInternal(t)) loaded++;
|
||||
if (!PATTERN_CRON.equalsIgnoreCase(t.getPatternType())) continue;
|
||||
seenIds.add(t.getId());
|
||||
Registration current = registrations.get(t.getId());
|
||||
long liveVersion = t.getPatternVersion() == null ? 1L : t.getPatternVersion();
|
||||
if (current == null) {
|
||||
if (registerInternal(t)) registered++;
|
||||
} else if (current.capturedVersion != liveVersion) {
|
||||
if (registerInternal(t)) refreshed++;
|
||||
}
|
||||
}
|
||||
log.info("[TriggerScheduler] Registered {} cron triggers at startup", loaded);
|
||||
// Drop registrations whose row was disabled / deleted / changed type
|
||||
// since the last sweep. Snapshot the keys first to avoid concurrent
|
||||
// modification on the underlying map.
|
||||
for (Long localId : new java.util.ArrayList<>(registrations.keySet())) {
|
||||
if (!seenIds.contains(localId)) {
|
||||
unregister(localId);
|
||||
removed++;
|
||||
}
|
||||
}
|
||||
if (registered + refreshed + removed > 0) {
|
||||
log.info("[TriggerScheduler] sync: registered={} refreshed={} removed={} active={}",
|
||||
registered, refreshed, removed, registrations.size());
|
||||
}
|
||||
}
|
||||
|
||||
/** Register or replace a single trigger (called from {@code TriggerService} on save). */
|
||||
|
||||
@ -61,11 +61,25 @@ public class TriggerService {
|
||||
throw new IllegalArgumentException("trigger not found: " + updated.getId());
|
||||
}
|
||||
|
||||
// Bump pattern_version whenever ANY field that changes the
|
||||
// schedule's behavior, payload rendering, or rate decisions
|
||||
// changes. This is the lamport other instances rely on at fire
|
||||
// time to decide whether their captured registration is stale —
|
||||
// missing a field here means a peer fires the new payload with
|
||||
// the old throttling settings (or vice versa) until it next
|
||||
// self-cancels for some other reason.
|
||||
boolean patternChanged = !Objects.equals(existing.getPatternJson(), updated.getPatternJson())
|
||||
|| !Objects.equals(existing.getPatternType(), updated.getPatternType());
|
||||
boolean payloadChanged = !Objects.equals(existing.getPayloadTemplate(), updated.getPayloadTemplate());
|
||||
boolean targetChanged = !Objects.equals(existing.getTargetType(), updated.getTargetType())
|
||||
|| !Objects.equals(existing.getTargetId(), updated.getTargetId());
|
||||
boolean fireConfigChanged = !Objects.equals(existing.getRateLimitPerMin(), updated.getRateLimitPerMin())
|
||||
|| !Objects.equals(existing.getDedupWindowSecs(), updated.getDedupWindowSecs())
|
||||
|| !Objects.equals(existing.getMaxFires(), updated.getMaxFires())
|
||||
|| !Objects.equals(existing.getBotSelfFilter(), updated.getBotSelfFilter());
|
||||
boolean enableTransition = !Objects.equals(existing.getEnabled(), updated.getEnabled());
|
||||
|
||||
if (patternChanged || enableTransition) {
|
||||
if (patternChanged || payloadChanged || targetChanged || fireConfigChanged || enableTransition) {
|
||||
long bumped = (existing.getPatternVersion() == null ? 1L : existing.getPatternVersion()) + 1L;
|
||||
updated.setPatternVersion(bumped);
|
||||
} else {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user