From 7c3c68bcfe6d31e468cdcc85100cc559543798ec Mon Sep 17 00:00:00 2001 From: matevip Date: Fri, 8 May 2026 15:04:26 +0800 Subject: [PATCH] feat(workflow): add dispatch_channel adapter for multi-channel delivery --- .../workflow/runtime/ChannelDispatcher.java | 29 +++++++ .../runtime/DefaultChannelDispatcher.java | 53 ++++++++++++ .../mode/DispatchChannelStepAdapter.java | 86 +++++++++++++++++++ 3 files changed, 168 insertions(+) create mode 100644 mateclaw-server/src/main/java/vip/mate/workflow/runtime/ChannelDispatcher.java create mode 100644 mateclaw-server/src/main/java/vip/mate/workflow/runtime/DefaultChannelDispatcher.java create mode 100644 mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/DispatchChannelStepAdapter.java diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/ChannelDispatcher.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/ChannelDispatcher.java new file mode 100644 index 00000000..914af03e --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/ChannelDispatcher.java @@ -0,0 +1,29 @@ +package vip.mate.workflow.runtime; + +/** + * SPI for "deliver this rendered content to a target on this channel". Kept + * thin so unit tests can stub channel side effects without booting the full + * channel adapter graph; production binding lives in + * {@link DefaultChannelDispatcher} and delegates to {@code ChannelManager}. + */ +public interface ChannelDispatcher { + + /** + * Send {@code content} to {@code targetId} on the channel identified by + * {@code channelType} (e.g. {@code "feishu"}, {@code "dingtalk"}). Returns + * an {@link DispatchResult} so the step adapter can build a per-channel + * report; throwing is reserved for programmer errors. + */ + DispatchResult dispatch(long workspaceId, String channelType, String targetId, String content); + + /** + * Per-channel dispatch outcome. {@code success=false} entries are turned + * into a step failure by the calling adapter; the message field surfaces + * to the run-step row's error column. + */ + record DispatchResult(boolean success, String message) { + public static DispatchResult ok() { return new DispatchResult(true, null); } + public static DispatchResult ok(String message) { return new DispatchResult(true, message); } + public static DispatchResult fail(String message) { return new DispatchResult(false, message); } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/DefaultChannelDispatcher.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/DefaultChannelDispatcher.java new file mode 100644 index 00000000..bccbbaeb --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/DefaultChannelDispatcher.java @@ -0,0 +1,53 @@ +package vip.mate.workflow.runtime; + +import org.springframework.stereotype.Component; +import vip.mate.channel.ChannelAdapter; +import vip.mate.channel.ChannelManager; + +import java.util.Optional; + +/** + * Production binding for {@link ChannelDispatcher}. Looks the channel up by + * type via {@link ChannelManager#getAdapterByType} and either calls + * {@code proactiveSend} when the adapter supports it or {@code sendMessage} + * otherwise. A missing adapter or one that's not running is reported back + * as a failed dispatch — the step adapter decides whether that fails the + * step or merely records a partial result. + */ +@Component +public class DefaultChannelDispatcher implements ChannelDispatcher { + + private final ChannelManager channelManager; + + public DefaultChannelDispatcher(ChannelManager channelManager) { + this.channelManager = channelManager; + } + + @Override + public DispatchResult dispatch(long workspaceId, String channelType, String targetId, String content) { + if (channelType == null || channelType.isBlank()) { + return DispatchResult.fail("channelType is required"); + } + Optional adapterOpt = channelManager.getAdapterByType(channelType); + if (adapterOpt.isEmpty()) { + return DispatchResult.fail("no active adapter for channel type '" + channelType + "'"); + } + ChannelAdapter adapter = adapterOpt.get(); + if (!adapter.isRunning()) { + return DispatchResult.fail("channel '" + channelType + "' adapter is not running"); + } + if (targetId == null || targetId.isBlank()) { + return DispatchResult.fail("missing targetId for channel '" + channelType + "'"); + } + try { + if (adapter.supportsProactiveSend()) { + adapter.proactiveSend(targetId, content); + } else { + adapter.sendMessage(targetId, content); + } + return DispatchResult.ok(); + } catch (Exception e) { + return DispatchResult.fail("dispatch to '" + channelType + "' failed: " + e.getMessage()); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/DispatchChannelStepAdapter.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/DispatchChannelStepAdapter.java new file mode 100644 index 00000000..0a85485b --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/DispatchChannelStepAdapter.java @@ -0,0 +1,86 @@ +package vip.mate.workflow.runtime.mode; + +import org.springframework.stereotype.Component; +import vip.mate.workflow.compiler.PebbleSubsetEvaluator; +import vip.mate.workflow.compiler.ir.StepMode; +import vip.mate.workflow.compiler.ir.WorkflowStep; +import vip.mate.workflow.runtime.ChannelDispatcher; +import vip.mate.workflow.runtime.PayloadStore; +import vip.mate.workflow.runtime.StepAdapter; +import vip.mate.workflow.runtime.StepResult; +import vip.mate.workflow.runtime.WorkflowRunContext; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * {@code dispatch_channel} — render the content template, then deliver the + * rendered text to every configured channel via {@link ChannelDispatcher}. + * Targets are looked up in the step's {@code targets} map keyed by channel + * type. The step fails iff any channel fails to deliver; partial successes + * are still flagged failed because step state is binary in v0 and silent + * delivery loss would be worse than an explicit error. + * + *

The rendered content payload is also written through to + * {@code mate_workflow_payload} so the run-step row's {@code output_ref} + * points at exactly what was sent. + */ +@Component +public class DispatchChannelStepAdapter implements StepAdapter { + + private final PebbleSubsetEvaluator pebble; + private final PayloadStore payloadStore; + private final ChannelDispatcher dispatcher; + + public DispatchChannelStepAdapter(PebbleSubsetEvaluator pebble, + PayloadStore payloadStore, + ChannelDispatcher dispatcher) { + this.pebble = pebble; + this.payloadStore = payloadStore; + this.dispatcher = dispatcher; + } + + @Override + public String typeName() { return "dispatch_channel"; } + + @Override + public StepResult execute(WorkflowStep step, WorkflowRunContext context) { + if (!(step.mode() instanceof StepMode.DispatchChannel cfg)) { + return StepResult.failed("dispatch_channel adapter received non-dispatch mode: " + + step.mode().typeName()); + } + + String rendered; + try { + var compiled = pebble.parseTemplate(cfg.content()); + rendered = pebble.evaluateAsString(compiled, context.templateContext()); + } catch (Exception e) { + return StepResult.failed("dispatch_channel content render failed for step '" + + step.name() + "': " + e.getMessage()); + } + + Map targets = cfg.targets() == null ? Map.of() : cfg.targets(); + List failures = new ArrayList<>(); + List delivered = new ArrayList<>(); + for (String channel : cfg.channels()) { + String target = targets.get(channel); + ChannelDispatcher.DispatchResult result = + dispatcher.dispatch(context.workspaceId(), channel, target, rendered); + if (result.success()) { + delivered.add(channel); + } else { + failures.add(channel + ": " + result.message()); + } + } + + String payloadUri = payloadStore.storeString(context.workspaceId(), rendered, "text/plain"); + + if (!failures.isEmpty()) { + return StepResult.failed("dispatch_channel partial / total failure: " + + String.join("; ", failures)); + } + return StepResult.succeeded(payloadUri, "text", rendered, + "delivered to " + String.join(", ", delivered)); + } +}