diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventIngestService.java b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventIngestService.java index dc726961..db25c2ee 100644 --- a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventIngestService.java +++ b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventIngestService.java @@ -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) { diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerPatternMatcher.java b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerPatternMatcher.java new file mode 100644 index 00000000..cbdbc84c --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerPatternMatcher.java @@ -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. + * + *

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. + * + *

v0 supports four pattern shapes: + *

+ * + *

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 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 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 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; + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/scheduler/TriggerScheduler.java b/mateclaw-server/src/main/java/vip/mate/trigger/scheduler/TriggerScheduler.java index 8ed6beba..8ba37e05 100644 --- a/mateclaw-server/src/main/java/vip/mate/trigger/scheduler/TriggerScheduler.java +++ b/mateclaw-server/src/main/java/vip/mate/trigger/scheduler/TriggerScheduler.java @@ -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}. + * + *

Reasons this exists: + *

+ * + *

Convergence rules: + *

+ */ + @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() .eq(TriggerEntity::getEnabled, true) .eq(TriggerEntity::getDeleted, 0)); - int loaded = 0; + java.util.Set 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). */ diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/service/TriggerService.java b/mateclaw-server/src/main/java/vip/mate/trigger/service/TriggerService.java index 49b04f2c..aaf62478 100644 --- a/mateclaw-server/src/main/java/vip/mate/trigger/service/TriggerService.java +++ b/mateclaw-server/src/main/java/vip/mate/trigger/service/TriggerService.java @@ -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 {