mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(trigger): add cron trigger engine with lamport coordination
This commit is contained in:
matevip
parent
67279bb9c9
commit
54b12f5270
@ -0,0 +1,52 @@
|
||||
package vip.mate.trigger.dispatch;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
import vip.mate.workflow.compiler.WorkflowParser;
|
||||
import vip.mate.workflow.model.WorkflowEntity;
|
||||
import vip.mate.workflow.model.WorkflowRevisionEntity;
|
||||
import vip.mate.workflow.repository.WorkflowMapper;
|
||||
import vip.mate.workflow.repository.WorkflowRevisionMapper;
|
||||
|
||||
/**
|
||||
* Production binding for {@link WorkflowGraphLoader}. Looks up
|
||||
* {@code mate_workflow.latest_revision_id} and parses the corresponding
|
||||
* {@code mate_workflow_revision.graph_json}. Returns
|
||||
* {@link Loaded#missing()} when either lookup fails or the workflow is
|
||||
* disabled — triggers should not fire workflows that the user already
|
||||
* paused or removed.
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class DefaultWorkflowGraphLoader implements WorkflowGraphLoader {
|
||||
|
||||
private final WorkflowMapper workflowMapper;
|
||||
private final WorkflowRevisionMapper revisionMapper;
|
||||
private final WorkflowParser parser;
|
||||
|
||||
public DefaultWorkflowGraphLoader(WorkflowMapper workflowMapper,
|
||||
WorkflowRevisionMapper revisionMapper,
|
||||
WorkflowParser parser) {
|
||||
this.workflowMapper = workflowMapper;
|
||||
this.revisionMapper = revisionMapper;
|
||||
this.parser = parser;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Loaded load(long workflowId) {
|
||||
WorkflowEntity workflow = workflowMapper.selectById(workflowId);
|
||||
if (workflow == null || Boolean.FALSE.equals(workflow.getEnabled())
|
||||
|| workflow.getLatestRevisionId() == null) {
|
||||
return Loaded.missing();
|
||||
}
|
||||
WorkflowRevisionEntity revision = revisionMapper.selectById(workflow.getLatestRevisionId());
|
||||
if (revision == null) return Loaded.missing();
|
||||
try {
|
||||
return new Loaded(parser.parse(revision.getGraphJson()), revision.getId());
|
||||
} catch (Exception e) {
|
||||
log.warn("Trigger graph load: revision {} failed to parse: {}",
|
||||
revision.getId(), e.getMessage());
|
||||
return Loaded.missing();
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,92 @@
|
||||
package vip.mate.trigger.dispatch;
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
import vip.mate.trigger.model.TriggerEntity;
|
||||
import vip.mate.workflow.compiler.PebbleSubsetEvaluator;
|
||||
import vip.mate.workflow.runtime.WorkflowRunRequest;
|
||||
import vip.mate.workflow.runtime.WorkflowRunResult;
|
||||
import vip.mate.workflow.runtime.WorkflowRunner;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Translates a fired trigger into a workflow run. Renders the trigger's
|
||||
* {@code payloadTemplate} as JSON via Pebble, parses the result into the
|
||||
* input map, and asks the runner to execute the latest revision of the
|
||||
* target workflow. Logs and swallows failures so a bad trigger never takes
|
||||
* the scheduler thread down.
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class TriggerDispatcher {
|
||||
|
||||
private static final TypeReference<Map<String, Object>> MAP_REF = new TypeReference<>() {};
|
||||
|
||||
private final WorkflowGraphLoader graphLoader;
|
||||
private final WorkflowRunner runner;
|
||||
private final PebbleSubsetEvaluator pebble;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
public TriggerDispatcher(WorkflowGraphLoader graphLoader,
|
||||
WorkflowRunner runner,
|
||||
PebbleSubsetEvaluator pebble,
|
||||
ObjectMapper objectMapper) {
|
||||
this.graphLoader = graphLoader;
|
||||
this.runner = runner;
|
||||
this.pebble = pebble;
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
/**
|
||||
* Dispatch a single fire of {@code trigger}. {@code event} is the
|
||||
* source-event context (cron tick metadata, channel message, etc.) — its
|
||||
* top-level fields are exposed to the payload template under
|
||||
* {@code event.*}. Returns {@code null} when the trigger is not
|
||||
* dispatchable (no published revision, unsupported target type) so the
|
||||
* caller can record the skip without erroring.
|
||||
*/
|
||||
public WorkflowRunResult dispatch(TriggerEntity trigger, Map<String, Object> event) {
|
||||
if (!"workflow".equalsIgnoreCase(trigger.getTargetType())) {
|
||||
log.warn("Trigger {} target_type {} not supported in v0; skipping fire",
|
||||
trigger.getId(), trigger.getTargetType());
|
||||
return null;
|
||||
}
|
||||
WorkflowGraphLoader.Loaded loaded = graphLoader.load(trigger.getTargetId());
|
||||
if (loaded.graph() == null) {
|
||||
log.info("Trigger {} dispatch skipped: no published revision for workflow {}",
|
||||
trigger.getId(), trigger.getTargetId());
|
||||
return null;
|
||||
}
|
||||
|
||||
Map<String, Object> inputs = renderInputs(trigger, event);
|
||||
WorkflowRunRequest req = new WorkflowRunRequest(
|
||||
trigger.getTargetId(),
|
||||
loaded.revisionId(),
|
||||
trigger.getWorkspaceId(),
|
||||
"trigger:" + trigger.getId(),
|
||||
inputs);
|
||||
return runner.run(loaded.graph(), req);
|
||||
}
|
||||
|
||||
private Map<String, Object> renderInputs(TriggerEntity trigger, Map<String, Object> event) {
|
||||
if (trigger.getPayloadTemplate() == null || trigger.getPayloadTemplate().isBlank()) {
|
||||
return event == null ? Map.of() : event;
|
||||
}
|
||||
try {
|
||||
var compiled = pebble.parseTemplate(trigger.getPayloadTemplate());
|
||||
String rendered = pebble.evaluateAsString(compiled,
|
||||
Map.of("event", event == null ? Map.of() : event,
|
||||
"trigger", Map.of(
|
||||
"id", trigger.getId(),
|
||||
"name", trigger.getName() == null ? "" : trigger.getName())));
|
||||
return objectMapper.readValue(rendered, MAP_REF);
|
||||
} catch (Exception e) {
|
||||
log.warn("Trigger {} payload template render failed; falling back to raw event: {}",
|
||||
trigger.getId(), e.getMessage());
|
||||
return event == null ? Map.of() : event;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,23 @@
|
||||
package vip.mate.trigger.dispatch;
|
||||
|
||||
import vip.mate.workflow.compiler.ir.WorkflowGraph;
|
||||
|
||||
/**
|
||||
* SPI for "given a workflow id, load the published WorkflowGraph the trigger
|
||||
* should fire". Production binding reads {@code mate_workflow.latest_revision_id}
|
||||
* and parses {@code mate_workflow_revision.graph_json}; tests stub this so a
|
||||
* fire path can be exercised without standing up the publish pipeline.
|
||||
*/
|
||||
public interface WorkflowGraphLoader {
|
||||
|
||||
/**
|
||||
* Result of a graph load. {@code graph == null} indicates the workflow
|
||||
* has no published revision yet (or was deleted) and the fire should
|
||||
* be skipped instead of erroring.
|
||||
*/
|
||||
record Loaded(WorkflowGraph graph, Long revisionId) {
|
||||
public static Loaded missing() { return new Loaded(null, null); }
|
||||
}
|
||||
|
||||
Loaded load(long workflowId);
|
||||
}
|
||||
@ -0,0 +1,216 @@
|
||||
package vip.mate.trigger.scheduler;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import net.javacrumbs.shedlock.core.LockConfiguration;
|
||||
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.concurrent.ThreadPoolTaskScheduler;
|
||||
import org.springframework.scheduling.support.CronTrigger;
|
||||
import org.springframework.stereotype.Component;
|
||||
import vip.mate.trigger.dispatch.TriggerDispatcher;
|
||||
import vip.mate.trigger.model.TriggerEntity;
|
||||
import vip.mate.trigger.repository.TriggerMapper;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.ZoneId;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.TimeZone;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
|
||||
/**
|
||||
* Maintains the in-memory map of cron-pattern triggers active on this node
|
||||
* and fires them through {@link TriggerDispatcher}. Coordination across
|
||||
* nodes uses ShedLock (per-trigger lock keyed by id) so simultaneous fires
|
||||
* collapse into one. Each scheduled task captures the trigger's
|
||||
* {@code patternVersion} at register time; on fire the live row's version
|
||||
* is re-read and the local task self-cancels when it has fallen behind a
|
||||
* newer cron expression — no need to chase a stale {@link ScheduledFuture}.
|
||||
*
|
||||
* <p>Only the {@code cron} pattern type registers here. Other pattern
|
||||
* flavours (channel_message, workflow_completion, ...) drive triggers
|
||||
* through their own ingestion pipeline and do not occupy a scheduler tick.
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class TriggerScheduler {
|
||||
|
||||
private static final String PATTERN_CRON = "cron";
|
||||
|
||||
private final TriggerMapper triggerMapper;
|
||||
private final TriggerDispatcher dispatcher;
|
||||
private final LockProvider lockProvider;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
private final ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
|
||||
private final Map<Long, Registration> registrations = new ConcurrentHashMap<>();
|
||||
|
||||
public TriggerScheduler(TriggerMapper triggerMapper,
|
||||
TriggerDispatcher dispatcher,
|
||||
LockProvider lockProvider,
|
||||
ObjectMapper objectMapper) {
|
||||
this.triggerMapper = triggerMapper;
|
||||
this.dispatcher = dispatcher;
|
||||
this.lockProvider = lockProvider;
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
@PostConstruct
|
||||
void initScheduler() {
|
||||
scheduler.setPoolSize(4);
|
||||
scheduler.setThreadNamePrefix("trigger-tick-");
|
||||
scheduler.setDaemon(true);
|
||||
scheduler.initialize();
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
void shutdownScheduler() {
|
||||
scheduler.shutdown();
|
||||
registrations.clear();
|
||||
}
|
||||
|
||||
/** Boot-time registration sweep; runs after Flyway and bean wiring complete. */
|
||||
@EventListener(ApplicationReadyEvent.class)
|
||||
void registerEnabledTriggersOnStartup() {
|
||||
var enabled = triggerMapper.selectList(new LambdaQueryWrapper<TriggerEntity>()
|
||||
.eq(TriggerEntity::getEnabled, true)
|
||||
.eq(TriggerEntity::getDeleted, 0));
|
||||
int loaded = 0;
|
||||
for (TriggerEntity t : enabled) {
|
||||
if (PATTERN_CRON.equalsIgnoreCase(t.getPatternType())) {
|
||||
if (registerInternal(t)) loaded++;
|
||||
}
|
||||
}
|
||||
log.info("[TriggerScheduler] Registered {} cron triggers at startup", loaded);
|
||||
}
|
||||
|
||||
/** Register or replace a single trigger (called from {@code TriggerService} on save). */
|
||||
public boolean register(TriggerEntity trigger) {
|
||||
if (trigger == null || !PATTERN_CRON.equalsIgnoreCase(trigger.getPatternType())) {
|
||||
return false;
|
||||
}
|
||||
return registerInternal(trigger);
|
||||
}
|
||||
|
||||
/** Cancel any active schedule for {@code triggerId}. Idempotent. */
|
||||
public void unregister(long triggerId) {
|
||||
Registration r = registrations.remove(triggerId);
|
||||
if (r != null) {
|
||||
r.future.cancel(false);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether {@code triggerId} currently occupies an active scheduled task on
|
||||
* this node. Visible because monitoring / health endpoints surface the
|
||||
* same fact, and the alternative would be exposing the raw registration
|
||||
* map.
|
||||
*/
|
||||
public boolean isRegistered(long triggerId) {
|
||||
return registrations.containsKey(triggerId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Manually drive the lamport + dispatch path the cron tick would otherwise
|
||||
* call. Used by integration tests; production code should never call this
|
||||
* directly — the scheduler owns its own tick.
|
||||
*/
|
||||
public void fireForTest(long triggerId, long capturedVersion) {
|
||||
fireWithCoordination(triggerId, capturedVersion);
|
||||
}
|
||||
|
||||
private boolean registerInternal(TriggerEntity trigger) {
|
||||
unregister(trigger.getId());
|
||||
ParsedCron parsed = parseCron(trigger);
|
||||
if (parsed == null) return false;
|
||||
|
||||
long capturedVersion = trigger.getPatternVersion() == null ? 1L : trigger.getPatternVersion();
|
||||
Runnable task = () -> fireWithCoordination(trigger.getId(), capturedVersion);
|
||||
ScheduledFuture<?> future = scheduler.schedule(task,
|
||||
new CronTrigger(parsed.expression, parsed.timeZone));
|
||||
registrations.put(trigger.getId(), new Registration(future, capturedVersion));
|
||||
log.info("[TriggerScheduler] Registered trigger {} cron='{}' tz={} version={}",
|
||||
trigger.getId(), parsed.expression, parsed.timeZone.getID(), capturedVersion);
|
||||
return true;
|
||||
}
|
||||
|
||||
private void fireWithCoordination(long triggerId, long capturedVersion) {
|
||||
// Per-fire lamport check: a newer expression in the DB invalidates
|
||||
// this scheduled task. Drop the fire and unregister so the next
|
||||
// registration cycle picks up the new schedule.
|
||||
TriggerEntity live = triggerMapper.selectById(triggerId);
|
||||
if (live == null || Boolean.FALSE.equals(live.getEnabled())) {
|
||||
unregister(triggerId);
|
||||
return;
|
||||
}
|
||||
long liveVersion = live.getPatternVersion() == null ? 1L : live.getPatternVersion();
|
||||
if (liveVersion != capturedVersion) {
|
||||
log.info("[TriggerScheduler] trigger {} self-cancelling (version changed {} -> {})",
|
||||
triggerId, capturedVersion, liveVersion);
|
||||
unregister(triggerId);
|
||||
return;
|
||||
}
|
||||
if (live.getMaxFires() != null && live.getMaxFires() > 0
|
||||
&& live.getFireCount() != null && live.getFireCount() >= live.getMaxFires()) {
|
||||
log.info("[TriggerScheduler] trigger {} reached max_fires={}, unregistering",
|
||||
triggerId, live.getMaxFires());
|
||||
unregister(triggerId);
|
||||
return;
|
||||
}
|
||||
|
||||
// Cross-node coordination: at-most-one node fires per tick.
|
||||
Optional<SimpleLock> lock = lockProvider.lock(new LockConfiguration(
|
||||
Instant.now(),
|
||||
"trigger-fire-" + triggerId,
|
||||
Duration.ofSeconds(60),
|
||||
Duration.ofSeconds(5)));
|
||||
if (lock.isEmpty()) {
|
||||
return; // peer is firing
|
||||
}
|
||||
try {
|
||||
dispatcher.dispatch(live, Map.of("firedAt", Instant.now().toString()));
|
||||
// Bump fire bookkeeping post-dispatch so a slow workflow does not
|
||||
// delay subsequent ticks; this row update is best-effort.
|
||||
live.setFireCount((live.getFireCount() == null ? 0L : live.getFireCount()) + 1);
|
||||
live.setLastFiredAt(LocalDateTime.now());
|
||||
triggerMapper.updateById(live);
|
||||
} catch (Exception e) {
|
||||
log.error("[TriggerScheduler] trigger {} fire failed: {}", triggerId, e.getMessage(), e);
|
||||
} finally {
|
||||
lock.get().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private record ParsedCron(String expression, TimeZone timeZone) {}
|
||||
|
||||
private ParsedCron parseCron(TriggerEntity trigger) {
|
||||
try {
|
||||
JsonNode node = objectMapper.readTree(
|
||||
trigger.getPatternJson() == null ? "{}" : trigger.getPatternJson());
|
||||
String expr = node.path("cron").asText("");
|
||||
if (expr.isBlank()) {
|
||||
log.warn("[TriggerScheduler] trigger {} missing 'cron' in pattern_json; skipping",
|
||||
trigger.getId());
|
||||
return null;
|
||||
}
|
||||
String tz = node.path("timezone").asText("UTC");
|
||||
return new ParsedCron(expr, TimeZone.getTimeZone(ZoneId.of(tz)));
|
||||
} catch (Exception e) {
|
||||
log.warn("[TriggerScheduler] trigger {} pattern_json parse failed: {}",
|
||||
trigger.getId(), e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private record Registration(ScheduledFuture<?> future, long capturedVersion) {}
|
||||
}
|
||||
@ -0,0 +1,101 @@
|
||||
package vip.mate.trigger.service;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import vip.mate.trigger.model.TriggerEntity;
|
||||
import vip.mate.trigger.repository.TriggerMapper;
|
||||
import vip.mate.trigger.scheduler.TriggerScheduler;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* CRUD facade for {@code mate_trigger} that keeps the in-memory cron
|
||||
* registration in sync with the persisted row. Pattern_version is the
|
||||
* lamport counter the scheduler uses to invalidate stale schedules across
|
||||
* a multi-node deployment — every change to {@code patternJson},
|
||||
* {@code patternType}, or the disabled→enabled transition bumps it.
|
||||
*
|
||||
* <p>The service intentionally does not wrap reads in transactions; only
|
||||
* mutating paths are {@code @Transactional} so the scheduler hand-off
|
||||
* (which reads the row again under its own connection) sees committed
|
||||
* data.
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class TriggerService {
|
||||
|
||||
private final TriggerMapper triggerMapper;
|
||||
private final TriggerScheduler scheduler;
|
||||
|
||||
public List<TriggerEntity> listByWorkspace(long workspaceId) {
|
||||
return triggerMapper.selectList(new LambdaQueryWrapper<TriggerEntity>()
|
||||
.eq(TriggerEntity::getWorkspaceId, workspaceId)
|
||||
.orderByDesc(TriggerEntity::getCreateTime));
|
||||
}
|
||||
|
||||
public TriggerEntity get(long id) {
|
||||
return triggerMapper.selectById(id);
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public TriggerEntity create(TriggerEntity trigger) {
|
||||
ensureDefaults(trigger);
|
||||
trigger.setPatternVersion(1L);
|
||||
trigger.setFireCount(0L);
|
||||
triggerMapper.insert(trigger);
|
||||
if (Boolean.TRUE.equals(trigger.getEnabled())) {
|
||||
scheduler.register(trigger);
|
||||
}
|
||||
return trigger;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public TriggerEntity update(TriggerEntity updated) {
|
||||
TriggerEntity existing = triggerMapper.selectById(updated.getId());
|
||||
if (existing == null) {
|
||||
throw new IllegalArgumentException("trigger not found: " + updated.getId());
|
||||
}
|
||||
|
||||
boolean patternChanged = !Objects.equals(existing.getPatternJson(), updated.getPatternJson())
|
||||
|| !Objects.equals(existing.getPatternType(), updated.getPatternType());
|
||||
boolean enableTransition = !Objects.equals(existing.getEnabled(), updated.getEnabled());
|
||||
|
||||
if (patternChanged || enableTransition) {
|
||||
long bumped = (existing.getPatternVersion() == null ? 1L : existing.getPatternVersion()) + 1L;
|
||||
updated.setPatternVersion(bumped);
|
||||
} else {
|
||||
updated.setPatternVersion(existing.getPatternVersion());
|
||||
}
|
||||
// Preserve fireCount / lastFiredAt — those are scheduler-owned.
|
||||
updated.setFireCount(existing.getFireCount());
|
||||
updated.setLastFiredAt(existing.getLastFiredAt());
|
||||
|
||||
triggerMapper.updateById(updated);
|
||||
|
||||
if (Boolean.TRUE.equals(updated.getEnabled())) {
|
||||
scheduler.register(updated);
|
||||
} else {
|
||||
scheduler.unregister(updated.getId());
|
||||
}
|
||||
return updated;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void delete(long id) {
|
||||
scheduler.unregister(id);
|
||||
triggerMapper.deleteById(id);
|
||||
}
|
||||
|
||||
private static void ensureDefaults(TriggerEntity t) {
|
||||
if (t.getRateLimitPerMin() == null) t.setRateLimitPerMin(60);
|
||||
if (t.getDedupWindowSecs() == null) t.setDedupWindowSecs(60);
|
||||
if (t.getBotSelfFilter() == null) t.setBotSelfFilter(true);
|
||||
if (t.getEnabled() == null) t.setEnabled(true);
|
||||
if (t.getMaxFires() == null) t.setMaxFires(0L);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user