v0 keeps the bot identity registry in-memory; production will likely
+ * wire this to {@code mate_channel.bot_identity} once that schema lands.
+ * The interface lets tests inject a deterministic resolver.
+ */
+public interface BotSelfFilter {
+
+ /**
+ * Whether {@code senderId} matches a known bot identity in
+ * {@code workspaceId}. Returning {@code true} causes the ingest pipeline
+ * to drop the event silently.
+ */
+ boolean isBotSelf(long workspaceId, String senderId);
+}
diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/NoopBotSelfFilter.java b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/NoopBotSelfFilter.java
new file mode 100644
index 00000000..a73cb0eb
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/NoopBotSelfFilter.java
@@ -0,0 +1,18 @@
+package vip.mate.trigger.ingest;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * Default {@link BotSelfFilter} binding — never identifies a sender as a
+ * bot. Acts as the v0 placeholder until the channel-side bot identity
+ * registry is wired through; channels that already know their own bot id
+ * may also call the filter directly to skip ingest before it begins.
+ */
+@Component
+public class NoopBotSelfFilter implements BotSelfFilter {
+
+ @Override
+ public boolean isBotSelf(long workspaceId, String senderId) {
+ return false;
+ }
+}
diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventEnvelope.java b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventEnvelope.java
new file mode 100644
index 00000000..d5ed478a
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventEnvelope.java
@@ -0,0 +1,33 @@
+package vip.mate.trigger.ingest;
+
+import java.util.Map;
+
+/**
+ * Generic event envelope used by upstream sources (channel webhooks,
+ * agent-lifecycle hooks, workflow-completion hooks, ad-hoc REST callers)
+ * to feed the trigger pipeline. The pipeline owns dedup / rate-limit /
+ * bot-self filtering; sources only need to fill this record:
+ *
+ *
data
+) {
+ public TriggerEventEnvelope {
+ data = data == null ? Map.of() : Map.copyOf(data);
+ }
+}
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
new file mode 100644
index 00000000..dc726961
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventIngestService.java
@@ -0,0 +1,177 @@
+package vip.mate.trigger.ingest;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import vip.mate.trigger.dispatch.TriggerDispatcher;
+import vip.mate.trigger.model.TriggerEntity;
+import vip.mate.trigger.model.TriggerEventEntity;
+import vip.mate.trigger.repository.TriggerEventMapper;
+import vip.mate.trigger.repository.TriggerMapper;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+
+/**
+ * Single ingress for every event-driven trigger. Runs the four-stage filter
+ * the design committee picked for v0:
+ *
+ *
+ * - Look up enabled triggers in the workspace whose {@code patternType}
+ * matches the envelope. Triggers in disabled workspaces, soft-deleted
+ * triggers, and triggers exhausted on {@code max_fires} are skipped.
+ * - Bot-self filter — drop events whose sender matches a registered
+ * bot identity, even if the trigger config has it disabled, because
+ * a runaway echo from our own outbound traffic is the worst-case
+ * failure and not worth a per-trigger opt-out.
+ * - Dedup window — insert a {@code mate_trigger_event} row keyed on
+ * {@code (trigger_id, dedup_key)} where the dedup key is the envelope
+ * eventId or a SHA-256 of the payload data when the upstream channel
+ * did not provide a stable id. A duplicate-key error short-circuits
+ * the dispatch silently.
+ * - Sliding-window rate limit — per-trigger 60s cap; an over-cap event
+ * is logged and dropped without dispatching.
+ *
+ *
+ * Each accepted event is then handed to {@link TriggerDispatcher} which
+ * runs the workflow synchronously. v0 does not queue dispatches; if the
+ * sender's webhook holds the connection open, the trigger runs in the
+ * caller's thread.
+ */
+@Slf4j
+@Service
+public class TriggerEventIngestService {
+
+ private final TriggerMapper triggerMapper;
+ private final TriggerEventMapper eventMapper;
+ private final TriggerDispatcher dispatcher;
+ private final BotSelfFilter botSelfFilter;
+ private final ObjectMapper objectMapper;
+ private final TriggerRateLimiter rateLimiter = new TriggerRateLimiter();
+
+ public TriggerEventIngestService(TriggerMapper triggerMapper,
+ TriggerEventMapper eventMapper,
+ TriggerDispatcher dispatcher,
+ BotSelfFilter botSelfFilter,
+ ObjectMapper objectMapper) {
+ this.triggerMapper = triggerMapper;
+ this.eventMapper = eventMapper;
+ this.dispatcher = dispatcher;
+ this.botSelfFilter = botSelfFilter;
+ this.objectMapper = objectMapper;
+ }
+
+ /**
+ * Process one envelope through the pipeline. Returns a result per
+ * candidate trigger so callers can surface a partial-accept summary.
+ */
+ public List ingest(TriggerEventEnvelope envelope) {
+ if (envelope.patternType() == null || envelope.patternType().isBlank()) {
+ return List.of();
+ }
+ List candidates = triggerMapper.selectList(new LambdaQueryWrapper()
+ .eq(TriggerEntity::getWorkspaceId, envelope.workspaceId())
+ .eq(TriggerEntity::getPatternType, envelope.patternType())
+ .eq(TriggerEntity::getEnabled, true)
+ .eq(TriggerEntity::getDeleted, 0));
+ if (candidates.isEmpty()) return List.of();
+
+ List results = new ArrayList<>(candidates.size());
+ for (TriggerEntity trigger : candidates) {
+ results.add(processSingle(trigger, envelope));
+ }
+ return results;
+ }
+
+ private IngestResult processSingle(TriggerEntity trigger, TriggerEventEnvelope envelope) {
+ if (Boolean.TRUE.equals(trigger.getBotSelfFilter())
+ && botSelfFilter.isBotSelf(envelope.workspaceId(), envelope.senderId())) {
+ return IngestResult.dropped(trigger.getId(), Reason.BOT_SELF);
+ }
+ if (trigger.getMaxFires() != null && trigger.getMaxFires() > 0
+ && trigger.getFireCount() != null && trigger.getFireCount() >= trigger.getMaxFires()) {
+ return IngestResult.dropped(trigger.getId(), Reason.EXHAUSTED);
+ }
+ if (!recordDedupRow(trigger, envelope)) {
+ return IngestResult.dropped(trigger.getId(), Reason.DUPLICATE);
+ }
+ int limit = trigger.getRateLimitPerMin() == null ? 0 : trigger.getRateLimitPerMin();
+ if (!rateLimiter.tryAcquire(trigger.getId(), limit, Instant.now())) {
+ return IngestResult.dropped(trigger.getId(), Reason.RATE_LIMITED);
+ }
+ try {
+ dispatcher.dispatch(trigger, envelope.data());
+ } catch (Exception e) {
+ log.error("Trigger {} dispatch threw on event ingest: {}",
+ trigger.getId(), e.getMessage(), e);
+ return IngestResult.dropped(trigger.getId(), Reason.DISPATCH_ERROR);
+ }
+ return IngestResult.fired(trigger.getId());
+ }
+
+ private boolean recordDedupRow(TriggerEntity trigger, TriggerEventEnvelope envelope) {
+ TriggerEventEntity row = new TriggerEventEntity();
+ row.setTriggerId(trigger.getId());
+ row.setDedupKey(resolveDedupKey(envelope));
+ int windowSecs = trigger.getDedupWindowSecs() == null ? 60 : trigger.getDedupWindowSecs();
+ Instant now = Instant.now();
+ row.setReceivedAt(LocalDateTime.ofInstant(now, ZoneOffset.systemDefault()));
+ row.setExpiresAt(LocalDateTime.ofInstant(now.plusSeconds(windowSecs),
+ ZoneOffset.systemDefault()));
+ try {
+ eventMapper.insert(row);
+ return true;
+ } catch (DuplicateKeyException e) {
+ // Within the dedup window — silently drop.
+ return false;
+ }
+ }
+
+ private String resolveDedupKey(TriggerEventEnvelope envelope) {
+ if (envelope.eventId() != null && !envelope.eventId().isBlank()) {
+ return truncate(envelope.eventId());
+ }
+ try {
+ byte[] body = objectMapper.writeValueAsBytes(envelope.data());
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ return "sha256:" + HexFormat.of().formatHex(digest.digest(body));
+ } catch (Exception e) {
+ // Fall back to a per-call random so we never hard-fail ingest.
+ return "rand:" + java.util.UUID.randomUUID();
+ }
+ }
+
+ private static String truncate(String s) {
+ if (s == null) return null;
+ // Column is VARCHAR(128) — keep some headroom for trigger-prefixed keys.
+ return s.length() <= 120 ? s : s.substring(0, 120);
+ }
+
+ /** Cleanup tick for expired dedup rows. Run from a scheduler in production. */
+ public int sweepExpired() {
+ return eventMapper.delete(new LambdaQueryWrapper()
+ .lt(TriggerEventEntity::getExpiresAt,
+ LocalDateTime.ofInstant(Instant.now(), ZoneOffset.systemDefault())));
+ }
+
+ public enum Reason {
+ BOT_SELF, DUPLICATE, RATE_LIMITED, EXHAUSTED, DISPATCH_ERROR
+ }
+
+ public record IngestResult(long triggerId, boolean fired, Reason droppedReason) {
+ public static IngestResult fired(long triggerId) {
+ return new IngestResult(triggerId, true, null);
+ }
+ public static IngestResult dropped(long triggerId, Reason r) {
+ return new IngestResult(triggerId, false, r);
+ }
+ }
+}
diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerRateLimiter.java b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerRateLimiter.java
new file mode 100644
index 00000000..e76212aa
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerRateLimiter.java
@@ -0,0 +1,53 @@
+package vip.mate.trigger.ingest;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Per-trigger sliding-window rate limiter. Each {@code triggerId} keeps a
+ * 60-second window of fire timestamps; an event is allowed iff fewer than
+ * the trigger's {@code rate_limit_per_min} entries already live in the
+ * window. The window is local to this node — for a multi-node deployment
+ * the cap is a per-node bound, not a global one. v0 accepts that trade
+ * because the alternative (DB-backed counters) costs a round-trip on every
+ * event and event volumes are well below the cap in practice.
+ */
+public class TriggerRateLimiter {
+
+ private final Map> windows = new ConcurrentHashMap<>();
+ private final Duration windowSize;
+
+ public TriggerRateLimiter() {
+ this(Duration.ofMinutes(1));
+ }
+
+ TriggerRateLimiter(Duration windowSize) {
+ this.windowSize = windowSize;
+ }
+
+ /**
+ * Try to admit an event for {@code triggerId} at {@code now}. Returns
+ * {@code true} when the event fits under {@code limitPerMin}; {@code false}
+ * when the window is full. The window is purged of expired entries first
+ * so a long-idle trigger reverts to full capacity.
+ *
+ * {@code limitPerMin <= 0} disables the limiter for that trigger.
+ */
+ public boolean tryAcquire(long triggerId, int limitPerMin, Instant now) {
+ if (limitPerMin <= 0) return true;
+ Deque window = windows.computeIfAbsent(triggerId, k -> new ArrayDeque<>());
+ Instant cutoff = now.minus(windowSize);
+ synchronized (window) {
+ while (!window.isEmpty() && !window.peekFirst().isAfter(cutoff)) {
+ window.pollFirst();
+ }
+ if (window.size() >= limitPerMin) return false;
+ window.addLast(now);
+ return true;
+ }
+ }
+}