mateclaw/mateclaw-server/src/main/java/vip/mate/channel/AsyncTaskMediaDispatcher.java

package vip.mate.channel;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import vip.mate.channel.model.ChannelSessionEntity;
import vip.mate.workspace.conversation.model.MessageContentPart;

import java.util.List;
import java.util.Set;

/**
 * Forward async-task results (image / video / music / 3D) generated by tool
 * pipelines to the IM channel that originated the conversation.
 * <p>
 * Without this dispatcher, completion bytes only land in
 * {@code mate_message} + a Web SSE broadcast — IM users (WeCom / DingTalk /
 * Feishu / Telegram / etc.) see nothing arrive in their chat client because
 * the tool pipeline doesn't know about channel adapters. This dispatcher
 * closes that loop: look up the conversation's bound channel session, get
 * the live adapter from {@link ChannelManager}, and call
 * {@link ChannelAdapter#sendContentParts} so the same bytes ride the
 * channel-native attachment protocol.
 * <p>
 * Web / webchat conversations are intentionally skipped because their SSE
 * stream already carries the result; double-dispatching would render the
 * image twice.
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class AsyncTaskMediaDispatcher {

    private final ChannelSessionStore channelSessionStore;
    private final ChannelManager channelManager;

    /**
     * Channel types that handle their own UX via SSE (no IM forward needed).
     * Everything not in this set is treated as an IM channel and gets the
     * generated parts pushed via the adapter.
     */
    private static final Set<String> WEB_CHANNEL_TYPES = Set.of("web", "webchat");

    /**
     * Forward generated content parts to the IM channel bound to this
     * conversation, if any.
     * <p>
     * Best-effort: missing session, missing adapter, or adapter exception
     * are all logged at debug/warn and never propagate. The caller has
     * already persisted the message to {@code mate_message} and broadcast
     * to Web SSE before invoking this — IM forwarding is additive.
     *
     * @param conversationId the conversation id used by the agent (e.g.
     *                       {@code wecom:XuZhanFu}, {@code dingtalk:cid_xxx},
     *                       {@code conv_xxx} for Web)
     * @param parts          assistant content parts to dispatch (typically a
     *                       single image / video / audio / file part)
     */
    public void forwardToImIfBound(String conversationId, List<MessageContentPart> parts) {
        if (conversationId == null || conversationId.isBlank() || parts == null || parts.isEmpty()) {
            return;
        }

        ChannelSessionEntity session = channelSessionStore.getSession(conversationId);
        if (session == null) {
            // Common case for Web-only conversations — the session is never
            // populated because Web doesn't write to ChannelSessionStore.
            log.debug("[async-forward] No channel session for conv={}, skipping IM forward",
                    conversationId);
            return;
        }

        String channelType = session.getChannelType();
        if (channelType == null || WEB_CHANNEL_TYPES.contains(channelType)) {
            log.debug("[async-forward] conv={} is web-class ({}), skipping IM forward",
                    conversationId, channelType);
            return;
        }

        Long channelId = session.getChannelId();
        if (channelId == null) {
            log.debug("[async-forward] conv={} session has no channelId, skipping",
                    conversationId);
            return;
        }

        ChannelAdapter adapter = channelManager.getAdapter(channelId).orElse(null);
        if (adapter == null) {
            log.warn("[async-forward] conv={} channelId={} has no live adapter (channel disabled?), skipping",
                    conversationId, channelId);
            return;
        }

        String targetId = session.getTargetId();
        if (targetId == null || targetId.isBlank()) {
            log.warn("[async-forward] conv={} session has no targetId, skipping",
                    conversationId);
            return;
        }

        try {
            adapter.sendContentParts(targetId, parts);
            log.info("[async-forward] Dispatched {} part(s) to {} adapter (conv={}, target={})",
                    parts.size(), channelType, conversationId, targetId);
        } catch (UnsupportedOperationException uoe) {
            // Adapter doesn't override sendContentParts — fall through to
            // text-only fallback. Most adapters that handle media (wecom /
            // feishu / dingtalk) override; the rest will stay text-only
            // until they implement the part dispatcher.
            log.info("[async-forward] {} adapter does not implement sendContentParts, skipping (conv={})",
                    channelType, conversationId);
        } catch (Exception e) {
            log.warn("[async-forward] Failed to dispatch to {} adapter for conv={}: {}",
                    channelType, conversationId, e.getMessage());
        }
    }
}