diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/DefaultWorkflowGraphLoader.java b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/DefaultWorkflowGraphLoader.java new file mode 100644 index 00000000..9a4bbe6f --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/DefaultWorkflowGraphLoader.java @@ -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(); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/TriggerDispatcher.java b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/TriggerDispatcher.java new file mode 100644 index 00000000..14f44109 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/TriggerDispatcher.java @@ -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_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 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 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 renderInputs(TriggerEntity trigger, Map 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; + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowGraphLoader.java b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowGraphLoader.java new file mode 100644 index 00000000..5f73f7e3 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowGraphLoader.java @@ -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); +} 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 new file mode 100644 index 00000000..8ed6beba --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/trigger/scheduler/TriggerScheduler.java @@ -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}. + * + *

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 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() + .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 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) {} +} 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 new file mode 100644 index 00000000..49b04f2c --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/trigger/service/TriggerService.java @@ -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. + * + *

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 listByWorkspace(long workspaceId) { + return triggerMapper.selectList(new LambdaQueryWrapper() + .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); + } +}