diff --git a/backend/mci-channel-wecom/pom.xml b/backend/mci-channel-wecom/pom.xml index 0a97a30..ccbcffe 100644 --- a/backend/mci-channel-wecom/pom.xml +++ b/backend/mci-channel-wecom/pom.xml @@ -23,11 +23,21 @@ org.springframework spring-context + + org.springframework.boot + spring-boot-autoconfigure + provided + org.springframework.data spring-data-redis provided + + com.fasterxml.jackson.core + jackson-databind + provided + com.github.binarywang weixin-java-cp diff --git a/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/FinanceSdkAutoConfiguration.java b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/FinanceSdkAutoConfiguration.java new file mode 100644 index 0000000..a899371 --- /dev/null +++ b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/FinanceSdkAutoConfiguration.java @@ -0,0 +1,29 @@ +package com.sino.mci.channel.wecom.archive; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * 企微会话存档 Finance SDK 客户端自动配置。 + * + *

默认注入 {@link MockFinanceSdkClient}；设置 {@code sino.channel.wecom.finance-sdk.native=true} + * 可切换为 {@link NativeFinanceSdkClient}（需自行完成 JNI 接入）。

+ */ +@Configuration +public class FinanceSdkAutoConfiguration { + + @Bean + @ConditionalOnMissingBean(FinanceSdkClient.class) + @ConditionalOnProperty(name = "sino.channel.wecom.finance-sdk.native", havingValue = "true") + public FinanceSdkClient nativeFinanceSdkClient() { + return new NativeFinanceSdkClient(); + } + + @Bean + @ConditionalOnMissingBean(FinanceSdkClient.class) + public FinanceSdkClient mockFinanceSdkClient() { + return new MockFinanceSdkClient(); + } +} diff --git a/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/FinanceSdkClient.java b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/FinanceSdkClient.java new file mode 100644 index 0000000..20c4a81 --- /dev/null +++ b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/FinanceSdkClient.java @@ -0,0 +1,51 @@ +package com.sino.mci.channel.wecom.archive; + +import com.sino.mci.channel.wecom.archive.model.FinanceMessage; + +import java.util.List; + +/** + * 企业微信会话存档 Finance SDK 客户端接口。 + * + *

对应官方 C/JNI SDK 的封装：

+ * + * + *

实现说明：

+ * + */ +public interface FinanceSdkClient { + + /** + * 初始化 SDK 会话。 + * + * @param corpId 企业 ID + * @param secret 会话存档 Secret + * @return 是否初始化成功 + */ + boolean init(String corpId, String secret); + + /** + * 按 seq 游标分页拉取加密聊天数据。 + * + * @param seq 起始 seq（游标） + * @param limit 本次拉取条数，建议 100-1000 + * @return 加密消息列表；无新消息时返回空列表 + */ + List getChatData(long seq, int limit); + + /** + * 解密单条加密聊天消息。 + * + * @param encryptKey RSA 解密后的消息密钥（encrypt_random_key 解密结果） + * @param encryptChatMsg 加密消息体（encrypt_chat_msg） + * @return 解密后的明文 JSON 字符串 + */ + String decryptData(String encryptKey, String encryptChatMsg); +} diff --git a/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/MockFinanceSdkClient.java b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/MockFinanceSdkClient.java new file mode 100644 index 0000000..b176092 --- /dev/null +++ b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/MockFinanceSdkClient.java @@ -0,0 +1,77 @@ +package com.sino.mci.channel.wecom.archive; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sino.mci.channel.wecom.archive.model.FinanceMessage; +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; + +/** + * 企业微信会话存档 Finance SDK 模拟客户端。 + * + *

默认实现，不依赖 JNI/.so 库，返回固定结构的模拟数据，用于开发、联调与单元测试。

+ */ +@Slf4j +public class MockFinanceSdkClient implements FinanceSdkClient { + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private final AtomicLong seqGenerator = new AtomicLong(1000); + + @Override + public boolean init(String corpId, String secret) { + log.info("[MockFinanceSdk] init corpId={}, secret=***", corpId); + return true; + } + + @Override + public List getChatData(long seq, int limit) { + log.info("[MockFinanceSdk] getChatData seq={}, limit={}", seq, limit); + List messages = new ArrayList<>(); + int count = Math.min(limit, 2); + for (int i = 0; i < count; i++) { + long currentSeq = seq + i; + String msgId = "mock-msg-" + currentSeq; + messages.add(FinanceMessage.builder() + .seq(currentSeq) + .msgId(msgId) + .publicKeyVer(1) + .encryptRandomKey("mock-random-key-" + currentSeq) + .encryptChatMsg("mock-chat-msg-" + currentSeq) + .build()); + } + return messages; + } + + @Override + public String decryptData(String encryptKey, String encryptChatMsg) { + log.info("[MockFinanceSdk] decryptData encryptKey={}, msg={}", encryptKey, encryptChatMsg); + try { + // 模拟解密：直接返回一个结构化的 JSON 明文 + Map payload = Map.of( + "msgid", extractSeq(encryptChatMsg), + "action", "send", + "from", "MockUser", + "tolist", List.of("ExternalUser"), + "msgtype", "text", + "text", Map.of("content", "模拟会话存档消息：" + encryptChatMsg), + "msgtime", System.currentTimeMillis() + ); + return OBJECT_MAPPER.writeValueAsString(payload); + } catch (Exception e) { + log.warn("[MockFinanceSdk] decryptData failed", e); + return null; + } + } + + private String extractSeq(String encryptChatMsg) { + if (encryptChatMsg == null) { + return "mock-msg"; + } + int idx = encryptChatMsg.lastIndexOf('-'); + return idx > 0 ? "mock-msg-" + encryptChatMsg.substring(idx + 1) : encryptChatMsg; + } +} diff --git a/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/NativeFinanceSdkClient.java b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/NativeFinanceSdkClient.java new file mode 100644 index 0000000..ce95adf --- /dev/null +++ b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/NativeFinanceSdkClient.java @@ -0,0 +1,42 @@ +package com.sino.mci.channel.wecom.archive; + +import com.sino.mci.channel.wecom.archive.model.FinanceMessage; +import lombok.extern.slf4j.Slf4j; + +import java.util.List; + +/** + * 企业微信会话存档 Finance SDK 原生 JNI 客户端（预留实现）。 + * + *

该实现基于官方 C 库 {@code com.tencent.wework.Finance}，通过 JNI 调用：

+ *
    + *
  • {@code Finance.NewSdk()} / {@code Finance.DestroySdk(long)}
    • + *
  • {@code Finance.GetChatData(long, long, long, String, String, int, long)}
    • + *
  • {@code Finance.DecryptData(long, String, String, long)}
    • + *
+ * + *

注意：官方 SDK 为 C 库 + JNI，不提供 Maven 坐标，也不应把 {@code .so/.dll} 二进制提交到仓库。 + * 本期仅保留骨架；接入真实 JNI 时，请在运行环境放置对应平台的动态库，并在启动参数中配置 + * {@code -Djava.library.path} 或通过 {@link System#loadLibrary(String)} 加载。

+ */ +@Slf4j +public class NativeFinanceSdkClient implements FinanceSdkClient { + + @Override + public boolean init(String corpId, String secret) { + throw new UnsupportedOperationException( + "NativeFinanceSdkClient 尚未接入 JNI；请先放置官方 .so/.dll 并完成 JNI 绑定"); + } + + @Override + public List getChatData(long seq, int limit) { + throw new UnsupportedOperationException( + "NativeFinanceSdkClient 尚未接入 JNI；请先放置官方 .so/.dll 并完成 JNI 绑定"); + } + + @Override + public String decryptData(String encryptKey, String encryptChatMsg) { + throw new UnsupportedOperationException( + "NativeFinanceSdkClient 尚未接入 JNI；请先放置官方 .so/.dll 并完成 JNI 绑定"); + } +} diff --git a/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/model/FinanceMessage.java b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/model/FinanceMessage.java new file mode 100644 index 0000000..da6e1ed --- /dev/null +++ b/backend/mci-channel-wecom/src/main/java/com/sino/mci/channel/wecom/archive/model/FinanceMessage.java @@ -0,0 +1,41 @@ +package com.sino.mci.channel.wecom.archive.model; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * 企业微信会话存档单条消息（GetChatData 返回的加密消息结构）。 + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class FinanceMessage { + + /** + * 消息序号，单调递增，用于断点续拉。 + */ + private long seq; + + /** + * 消息唯一 ID。 + */ + private String msgId; + + /** + * 公钥版本。 + */ + private int publicKeyVer; + + /** + * 经过 RSA 加密的随机密钥，需用企业私钥解密后得到消息密钥。 + */ + private String encryptRandomKey; + + /** + * 经过 AES 加密的消息体，需用消息密钥解密。 + */ + private String encryptChatMsg; +} diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/controller/WeComCallbackController.java b/backend/mci-server/src/main/java/com/sino/mci/channel/controller/WeComCallbackController.java index 145a719..12bcf32 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/channel/controller/WeComCallbackController.java +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/controller/WeComCallbackController.java @@ -1,16 +1,35 @@ package com.sino.mci.channel.controller; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.sino.mci.channel.entity.ChannelSubject; +import com.sino.mci.channel.messaging.RabbitConfig; +import com.sino.mci.channel.messaging.SessionArchivePullCommand; +import com.sino.mci.channel.repository.ChannelSubjectMapper; import jakarta.servlet.http.HttpServletRequest; +import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.web.bind.annotation.*; +import org.w3c.dom.Document; +import org.w3c.dom.Element; +import org.w3c.dom.Node; +import org.w3c.dom.NodeList; +import org.xml.sax.InputSource; +import javax.xml.parsers.DocumentBuilder; +import javax.xml.parsers.DocumentBuilderFactory; +import java.io.StringReader; import java.util.Enumeration; @Slf4j @RestController @RequestMapping("/callback/wecom") +@RequiredArgsConstructor public class WeComCallbackController { + private final ChannelSubjectMapper channelSubjectMapper; + private final RabbitTemplate rabbitTemplate; + @GetMapping public String verifyUrl(HttpServletRequest request) { log.info("WeCom callback URL verification: {}", request.getQueryString()); @@ -26,7 +45,107 @@ public class WeComCallbackController { String name = headerNames.nextElement(); log.debug("Header {}: {}", name, request.getHeader(name)); } - // Phase 2: only accept and acknowledge; event parsing and session archive pull will be implemented later + + try { + CallbackInfo info = parseCallbackXml(body); + log.info("WeCom callback parsed: event={}, corpId={}", info.getEvent(), info.getCorpId()); + + if ("msgaudit_notify".equalsIgnoreCase(info.getEvent())) { + handleMsgAuditNotify(info); + } + } catch (Exception e) { + log.warn("WeCom callback parse failed, still ack", e); + } + return "success"; } + + private void handleMsgAuditNotify(CallbackInfo info) { + String corpId = info.getCorpId(); + if (corpId == null || corpId.isBlank()) { + log.warn("[WeComCallback] msgaudit_notify 缺少 corpId"); + return; + } + + ChannelSubject subject = channelSubjectMapper.selectOne( + new LambdaQueryWrapper() + .eq(ChannelSubject::getCorpId, corpId) + .eq(ChannelSubject::getStatus, 1) + .last("LIMIT 1")); + if (subject == null) { + log.warn("[WeComCallback] 未找到 corpId={} 的渠道主体", corpId); + return; + } + if (subject.getSessionArchiveEnabled() == null || subject.getSessionArchiveEnabled() != 1) { + log.warn("[WeComCallback] 渠道主体未开启会话存档 corpId={}", corpId); + return; + } + + SessionArchivePullCommand command = new SessionArchivePullCommand(); + command.setChannelSubjectId(subject.getId()); + command.setTenantCode(subject.getTenantCode()); + command.setCorpId(subject.getCorpId()); + command.setSecret(subject.getCorpSecret()); + command.setPrivateKey(subject.getSessionArchivePrivateKey()); + + rabbitTemplate.convertAndSend(RabbitConfig.SESSION_ARCHIVE_EXCHANGE, + RabbitConfig.SESSION_ARCHIVE_ROUTING_KEY, command); + log.info("[WeComCallback] 已发送会话存档拉取命令 subjectId={}, corpId={}", subject.getId(), corpId); + } + + private CallbackInfo parseCallbackXml(String body) { + CallbackInfo info = new CallbackInfo(); + try { + DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance(); + factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true); + DocumentBuilder builder = factory.newDocumentBuilder(); + Document document = builder.parse(new InputSource(new StringReader(body))); + Element root = document.getDocumentElement(); + info.setCorpId(getTagValue(root, "ToUserName")); + info.setEvent(getTagValue(root, "Event")); + info.setRawBody(body); + } catch (Exception e) { + log.warn("XML parse failed: {}", e.getMessage()); + } + return info; + } + + private String getTagValue(Element parent, String tagName) { + NodeList nodeList = parent.getElementsByTagName(tagName); + if (nodeList.getLength() == 0) { + return null; + } + Node node = nodeList.item(0); + return node.getTextContent(); + } + + private static class CallbackInfo { + private String corpId; + private String event; + private String rawBody; + + public String getCorpId() { + return corpId; + } + + public void setCorpId(String corpId) { + this.corpId = corpId; + } + + public String getEvent() { + return event; + } + + public void setEvent(String event) { + this.event = event; + } + + public String getRawBody() { + return rawBody; + } + + public void setRawBody(String rawBody) { + this.rawBody = rawBody; + } + } } diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/messaging/RabbitConfig.java b/backend/mci-server/src/main/java/com/sino/mci/channel/messaging/RabbitConfig.java new file mode 100644 index 0000000..ba31324 --- /dev/null +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/messaging/RabbitConfig.java @@ -0,0 +1,44 @@ +package com.sino.mci.channel.messaging; + +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.TopicExchange; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * 企微会话存档拉取任务 RabbitMQ 配置。 + */ +@Configuration +public class RabbitConfig { + + public static final String SESSION_ARCHIVE_QUEUE = "mci.wecom.session_archive.pull"; + public static final String SESSION_ARCHIVE_EXCHANGE = "mci.wecom"; + public static final String SESSION_ARCHIVE_ROUTING_KEY = "session_archive.pull"; + + @Bean + public Queue sessionArchiveQueue() { + return new Queue(SESSION_ARCHIVE_QUEUE, true); + } + + @Bean + public TopicExchange sessionArchiveExchange() { + return new TopicExchange(SESSION_ARCHIVE_EXCHANGE, true, false); + } + + @Bean + public Binding sessionArchiveBinding(Queue sessionArchiveQueue, TopicExchange sessionArchiveExchange) { + return BindingBuilder.bind(sessionArchiveQueue).to(sessionArchiveExchange).with(SESSION_ARCHIVE_ROUTING_KEY); + } + + @Bean + public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { + RabbitTemplate template = new RabbitTemplate(connectionFactory); + template.setMessageConverter(new Jackson2JsonMessageConverter()); + return template; + } +} diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/messaging/SessionArchivePullCommand.java b/backend/mci-server/src/main/java/com/sino/mci/channel/messaging/SessionArchivePullCommand.java new file mode 100644 index 0000000..7100707 --- /dev/null +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/messaging/SessionArchivePullCommand.java @@ -0,0 +1,24 @@ +package com.sino.mci.channel.messaging; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.io.Serializable; + +/** + * 触发企微会话存档拉取任务的命令消息。 + */ +@Data +@NoArgsConstructor +@AllArgsConstructor +public class SessionArchivePullCommand implements Serializable { + + private static final long serialVersionUID = 1L; + + private Long channelSubjectId; + private String tenantCode; + private String corpId; + private String secret; + private String privateKey; +} diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/service/SessionArchivePullService.java b/backend/mci-server/src/main/java/com/sino/mci/channel/service/SessionArchivePullService.java new file mode 100644 index 0000000..f826fcd --- /dev/null +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/service/SessionArchivePullService.java @@ -0,0 +1,181 @@ +package com.sino.mci.channel.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sino.mci.channel.messaging.SessionArchivePullCommand; +import com.sino.mci.channel.wecom.archive.FinanceSdkClient; +import com.sino.mci.channel.wecom.archive.model.FinanceMessage; +import com.sino.mci.inbound.entity.MessageEvent; +import com.sino.mci.inbound.repository.MessageEventMapper; +import com.sino.mci.message.entity.MessageMedia; +import com.sino.mci.message.repository.MessageMediaMapper; +import com.sino.mci.shared.util.JsonUtils; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.util.StringUtils; + +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** + * 企微会话存档消息拉取与落库服务。 + * + *

消费 RabbitMQ 命令，按渠道主体维度使用 Redis 缓存 seq 游标，调用 Finance SDK 拉取、解密， + * 并将消息写入 {@code mci_message_event}，媒体文件写入 {@code mci_message_media}。

+ */ +@Slf4j +@Service +@RequiredArgsConstructor +public class SessionArchivePullService { + + private static final String SEQ_KEY_PREFIX = "wecom:session_archive:seq:"; + private static final int DEFAULT_LIMIT = 100; + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private final FinanceSdkClient financeSdkClient; + private final StringRedisTemplate redisTemplate; + private final MessageEventMapper messageEventMapper; + private final MessageMediaMapper messageMediaMapper; + + /** + * 消费会话存档拉取命令。 + */ + @RabbitListener(queues = "${sino.channel.wecom.session-archive.queue:mci.wecom.session_archive.pull}") + @Transactional + public void onPullCommand(SessionArchivePullCommand command) { + log.info("[SessionArchive] 开始拉取会话存档 subjectId={}, corpId={}", + command.getChannelSubjectId(), command.getCorpId()); + + if (!StringUtils.hasText(command.getCorpId()) || !StringUtils.hasText(command.getSecret())) { + log.warn("[SessionArchive] 缺少 corpId 或 secret，跳过拉取"); + return; + } + + if (!financeSdkClient.init(command.getCorpId(), command.getSecret())) { + log.warn("[SessionArchive] Finance SDK 初始化失败 corpId={}", command.getCorpId()); + return; + } + + long currentSeq = readCurrentSeq(command.getChannelSubjectId()); + List messages = financeSdkClient.getChatData(currentSeq, DEFAULT_LIMIT); + if (messages == null || messages.isEmpty()) { + log.info("[SessionArchive] 无新消息 subjectId={}, seq={}", command.getChannelSubjectId(), currentSeq); + return; + } + + long maxSeq = currentSeq; + for (FinanceMessage msg : messages) { + String decryptedJson = decryptMessage(command, msg); + if (!StringUtils.hasText(decryptedJson)) { + log.warn("[SessionArchive] 解密失败 seq={}, msgId={}", msg.getSeq(), msg.getMsgId()); + continue; + } + saveMessageEvent(command, msg, decryptedJson); + saveMediaIfNeeded(command, msg, decryptedJson); + maxSeq = Math.max(maxSeq, msg.getSeq() + 1); + } + + updateSeq(command.getChannelSubjectId(), maxSeq); + log.info("[SessionArchive] 拉取完成 subjectId={}, messages={}, nextSeq={}", + command.getChannelSubjectId(), messages.size(), maxSeq); + } + + private long readCurrentSeq(Long subjectId) { + String key = seqKey(subjectId); + String value = redisTemplate.opsForValue().get(key); + if (!StringUtils.hasText(value)) { + return 0L; + } + try { + return Long.parseLong(value); + } catch (NumberFormatException e) { + log.warn("[SessionArchive] seq 缓存格式异常，重置为 0 key={}", key); + return 0L; + } + } + + private void updateSeq(Long subjectId, long seq) { + redisTemplate.opsForValue().set(seqKey(subjectId), String.valueOf(seq)); + } + + private String seqKey(Long subjectId) { + return SEQ_KEY_PREFIX + subjectId; + } + + private String decryptMessage(SessionArchivePullCommand command, FinanceMessage msg) { + // 模拟 RSA 解密 encrypt_random_key；真实环境应使用 command.getPrivateKey() 解密 + String encryptKey = msg.getEncryptRandomKey(); + return financeSdkClient.decryptData(encryptKey, msg.getEncryptChatMsg()); + } + + private void saveMessageEvent(SessionArchivePullCommand command, FinanceMessage msg, String decryptedJson) { + MessageEvent event = new MessageEvent(); + event.setEventId(msg.getMsgId()); + event.setTenantCode(command.getTenantCode()); + event.setChannelType("wecom"); + event.setChannelSubjectId(command.getChannelSubjectId()); + event.setDirection(1); + event.setEventType("message"); + event.setMessageSource("session_archive"); + event.setContent(decryptedJson); + event.setRawPayload(JsonUtils.toJson(Map.of( + "seq", msg.getSeq(), + "publicKeyVer", msg.getPublicKeyVer(), + "encryptRandomKey", msg.getEncryptRandomKey(), + "encryptChatMsg", msg.getEncryptChatMsg() + ))); + event.setIsProcessed(0); + event.setStatus(1); + messageEventMapper.insert(event); + } + + private void saveMediaIfNeeded(SessionArchivePullCommand command, FinanceMessage msg, String decryptedJson) { + try { + Map payload = OBJECT_MAPPER.readValue(decryptedJson, Map.class); + String msgType = Optional.ofNullable(payload.get("msgtype")).map(Object::toString).orElse(""); + if (!isMediaType(msgType)) { + return; + } + @SuppressWarnings("unchecked") + Map mediaBody = (Map) payload.get(msgType); + if (mediaBody == null) { + return; + } + MessageMedia media = new MessageMedia(); + media.setTenantCode(command.getTenantCode()); + media.setMsgid(msg.getMsgId()); + media.setSdkfileid(Optional.ofNullable(mediaBody.get("sdkfileid")).map(Object::toString).orElse(null)); + media.setMediaType(msgType); + media.setFileName(Optional.ofNullable(mediaBody.get("filename")).map(Object::toString).orElse(null)); + media.setFileSize(parseLong(mediaBody.get("filesize"))); + media.setMd5sum(Optional.ofNullable(mediaBody.get("md5sum")).map(Object::toString).orElse(null)); + media.setStatus(0); + messageMediaMapper.insert(media); + } catch (Exception e) { + log.warn("[SessionArchive] 解析媒体消息失败 msgId={}", msg.getMsgId(), e); + } + } + + private boolean isMediaType(String msgType) { + return "image".equals(msgType) || "voice".equals(msgType) + || "video".equals(msgType) || "file".equals(msgType) || "emotion".equals(msgType); + } + + private Long parseLong(Object value) { + if (value == null) { + return null; + } + if (value instanceof Number n) { + return n.longValue(); + } + try { + return Long.parseLong(value.toString()); + } catch (NumberFormatException e) { + return null; + } + } +} diff --git a/backend/mci-server/src/test/java/com/sino/mci/channel/controller/WeComSessionArchiveTest.java b/backend/mci-server/src/test/java/com/sino/mci/channel/controller/WeComSessionArchiveTest.java new file mode 100644 index 0000000..6103928 --- /dev/null +++ b/backend/mci-server/src/test/java/com/sino/mci/channel/controller/WeComSessionArchiveTest.java @@ -0,0 +1,226 @@ +package com.sino.mci.channel.controller; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sino.mci.channel.entity.ChannelSubject; +import com.sino.mci.channel.messaging.RabbitConfig; +import com.sino.mci.channel.messaging.SessionArchivePullCommand; +import com.sino.mci.channel.repository.ChannelSubjectMapper; +import com.sino.mci.channel.service.SessionArchivePullService; +import com.sino.mci.channel.wecom.archive.FinanceSdkClient; +import com.sino.mci.channel.wecom.archive.model.FinanceMessage; +import com.sino.mci.inbound.entity.MessageEvent; +import com.sino.mci.inbound.repository.MessageEventMapper; +import com.sino.mci.message.entity.MessageMedia; +import com.sino.mci.message.repository.MessageMediaMapper; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; +import org.springframework.context.annotation.Bean; +import org.springframework.http.MediaType; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.transaction.annotation.Transactional; + +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import static com.sino.mci.test.AuthTestHelper.createTenantAndLogin; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +@SpringBootTest +@AutoConfigureMockMvc +@TestPropertySource(properties = { + "spring.rabbitmq.listener.simple.auto-startup=false", + "spring.main.allow-bean-definition-overriding=true", + "sino.channel.wecom.session-archive.listener-enabled=false" +}) +@Transactional +class WeComSessionArchiveTest { + + private static final String CORP_ID = "ww-session-archive"; + + @Autowired + private MockMvc mockMvc; + + @Autowired + private ChannelSubjectMapper channelSubjectMapper; + + @Autowired + private MessageEventMapper messageEventMapper; + + @Autowired + private MessageMediaMapper messageMediaMapper; + + @Autowired + private SessionArchivePullService sessionArchivePullService; + + @Autowired + private FinanceSdkClient financeSdkClient; + + @Autowired + private RabbitTemplate rabbitTemplate; + + private final ObjectMapper objectMapper = new ObjectMapper(); + + @Test + void shouldReceiveMsgAuditNotifyAndPublishPullCommand() throws Exception { + ChannelSubject subject = createSessionArchiveSubject(); + + String xml = """ + + + + 1700000000 + + + + """.formatted(CORP_ID); + + mockMvc.perform(post("/msg-platform/callback/wecom") + .contextPath("/msg-platform") + .contentType(MediaType.APPLICATION_XML) + .content(xml)) + .andExpect(status().isOk()) + .andExpect(content().string("success")); + + ArgumentCaptor captor = ArgumentCaptor.forClass(SessionArchivePullCommand.class); + verify(rabbitTemplate).convertAndSend(eq(RabbitConfig.SESSION_ARCHIVE_EXCHANGE), + eq(RabbitConfig.SESSION_ARCHIVE_ROUTING_KEY), captor.capture()); + SessionArchivePullCommand command = captor.getValue(); + assertThat(command.getChannelSubjectId()).isEqualTo(subject.getId()); + assertThat(command.getCorpId()).isEqualTo(CORP_ID); + assertThat(command.getTenantCode()).isEqualTo(subject.getTenantCode()); + } + + @Test + void shouldPullAndSaveMessageEvent() throws Exception { + ChannelSubject subject = createSessionArchiveSubject(); + when(financeSdkClient.init(anyString(), anyString())).thenReturn(true); + when(financeSdkClient.getChatData(anyLong(), anyInt())).thenReturn(List.of( + FinanceMessage.builder() + .seq(100L) + .msgId("msg-100") + .publicKeyVer(1) + .encryptRandomKey("random-100") + .encryptChatMsg("chat-100") + .build() + )); + when(financeSdkClient.decryptData(anyString(), anyString())).thenReturn( + objectMapper.writeValueAsString(Map.of( + "msgid", "msg-100", + "msgtype", "text", + "text", Map.of("content", "hello") + )) + ); + + SessionArchivePullCommand command = new SessionArchivePullCommand(); + command.setChannelSubjectId(subject.getId()); + command.setTenantCode(subject.getTenantCode()); + command.setCorpId(subject.getCorpId()); + command.setSecret(subject.getCorpSecret()); + command.setPrivateKey(subject.getSessionArchivePrivateKey()); + + sessionArchivePullService.onPullCommand(command); + + MessageEvent event = messageEventMapper.selectOne( + new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper() + .eq(MessageEvent::getEventId, "msg-100")); + assertThat(event).isNotNull(); + assertThat(event.getMessageSource()).isEqualTo("session_archive"); + assertThat(event.getChannelSubjectId()).isEqualTo(subject.getId()); + assertThat(event.getChannelType()).isEqualTo("wecom"); + assertThat(event.getContent()).contains("hello"); + } + + @Test + void shouldSaveMediaRecordForImageMessage() throws Exception { + ChannelSubject subject = createSessionArchiveSubject(); + when(financeSdkClient.init(anyString(), anyString())).thenReturn(true); + when(financeSdkClient.getChatData(anyLong(), anyInt())).thenReturn(List.of( + FinanceMessage.builder() + .seq(200L) + .msgId("msg-200") + .publicKeyVer(1) + .encryptRandomKey("random-200") + .encryptChatMsg("chat-200") + .build() + )); + when(financeSdkClient.decryptData(anyString(), anyString())).thenReturn( + objectMapper.writeValueAsString(Map.of( + "msgid", "msg-200", + "msgtype", "image", + "image", Map.of( + "sdkfileid", "sdkfileid-200", + "filesize", 1024, + "md5sum", "abc123" + ) + )) + ); + + SessionArchivePullCommand command = new SessionArchivePullCommand(); + command.setChannelSubjectId(subject.getId()); + command.setTenantCode(subject.getTenantCode()); + command.setCorpId(subject.getCorpId()); + command.setSecret(subject.getCorpSecret()); + command.setPrivateKey(subject.getSessionArchivePrivateKey()); + + sessionArchivePullService.onPullCommand(command); + + MessageMedia media = messageMediaMapper.selectOne( + new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper() + .eq(MessageMedia::getMsgid, "msg-200")); + assertThat(media).isNotNull(); + assertThat(media.getSdkfileid()).isEqualTo("sdkfileid-200"); + assertThat(media.getMediaType()).isEqualTo("image"); + assertThat(media.getStatus()).isEqualTo(0); + assertThat(media.getFileSize()).isEqualTo(1024L); + } + + private ChannelSubject createSessionArchiveSubject() throws Exception { + String tenantCode = "sa-" + UUID.randomUUID().toString().substring(0, 8); + createTenantAndLogin(mockMvc, tenantCode, "会话存档租户"); + + ChannelSubject subject = new ChannelSubject(); + subject.setTenantCode(tenantCode); + subject.setChannelType("wecom"); + subject.setSubjectCode("archive"); + subject.setSubjectName("会话存档主体"); + subject.setCorpId(CORP_ID); + subject.setCorpSecret("secret-" + UUID.randomUUID()); + subject.setSessionArchiveEnabled(1); + subject.setSessionArchivePrivateKey("private-key"); + subject.setStatus(1); + channelSubjectMapper.insert(subject); + return subject; + } + + @TestConfiguration + static class MockConfig { + + @Bean + public FinanceSdkClient financeSdkClient() { + return mock(FinanceSdkClient.class); + } + + @Bean + public RabbitTemplate rabbitTemplate() { + return mock(RabbitTemplate.class); + } + } +}