mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(trigger): add event ingest pipeline with dedup / rate limit / bot-self
This commit is contained in:
matevip
parent
54b12f5270
commit
ce08b311d3
@ -0,0 +1,22 @@
|
||||
package vip.mate.trigger.ingest;
|
||||
|
||||
/**
|
||||
* Drops events whose sender id matches a registered bot identity for the
|
||||
* workspace. The intent: MateClaw's own outbound channel messages would
|
||||
* otherwise loop back through the channel webhook, fire a trigger, and
|
||||
* dispatch a fresh workflow run — a recipe for a runaway echo loop on any
|
||||
* channel where the bot account can read its own posts.
|
||||
*
|
||||
* <p>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);
|
||||
}
|
||||
@ -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;
|
||||
}
|
||||
}
|
||||
@ -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:
|
||||
*
|
||||
* <ul>
|
||||
* <li>{@code workspaceId} — scopes which triggers can fire on this event.</li>
|
||||
* <li>{@code patternType} — matched against {@code mate_trigger.pattern_type};
|
||||
* the ingest looks up only triggers whose pattern type equals this.</li>
|
||||
* <li>{@code eventId} — stable upstream identifier used as the dedup key
|
||||
* when present; the ingest falls back to a content hash when blank.</li>
|
||||
* <li>{@code senderId} — the upstream actor; used by the bot-self filter
|
||||
* to drop events that originate from MateClaw's own outbound traffic.</li>
|
||||
* <li>{@code data} — free-form payload exposed to the trigger's payload
|
||||
* template under {@code event.*}.</li>
|
||||
* </ul>
|
||||
*/
|
||||
public record TriggerEventEnvelope(
|
||||
long workspaceId,
|
||||
String patternType,
|
||||
String eventId,
|
||||
String senderId,
|
||||
Map<String, Object> data
|
||||
) {
|
||||
public TriggerEventEnvelope {
|
||||
data = data == null ? Map.of() : Map.copyOf(data);
|
||||
}
|
||||
}
|
||||
@ -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:
|
||||
*
|
||||
* <ol>
|
||||
* <li>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.</li>
|
||||
* <li>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.</li>
|
||||
* <li>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.</li>
|
||||
* <li>Sliding-window rate limit — per-trigger 60s cap; an over-cap event
|
||||
* is logged and dropped without dispatching.</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p>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<IngestResult> ingest(TriggerEventEnvelope envelope) {
|
||||
if (envelope.patternType() == null || envelope.patternType().isBlank()) {
|
||||
return List.of();
|
||||
}
|
||||
List<TriggerEntity> candidates = triggerMapper.selectList(new LambdaQueryWrapper<TriggerEntity>()
|
||||
.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<IngestResult> 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<TriggerEventEntity>()
|
||||
.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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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<Long, Deque<Instant>> 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.
|
||||
*
|
||||
* <p>{@code limitPerMin <= 0} disables the limiter for that trigger.
|
||||
*/
|
||||
public boolean tryAcquire(long triggerId, int limitPerMin, Instant now) {
|
||||
if (limitPerMin <= 0) return true;
|
||||
Deque<Instant> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user