feat：会话存档分表存储和查询

fs-qw-api-msg/src/main/resources/application.yml

@@ -1,5 +1,15 @@
 server:
   port: 8667
+# 企微会话存档
+qw:
+  msg-audit:
+    sdk-lib-path: C:/finance-sdk/libcrypto-3-x64.dll,libcurl-x64.dll,libssl-3-x64.dll,WeWorkFinanceSdk.dll
+    # 企微语音 amr 转 mp3 时调用，需在服务器安装 ffmpeg 或填写可执行文件绝对路径
+    ffmpeg-path: C:\ffmpeg.exe
+    voice-amr-to-mp3: true
+    # 解密后 raw_json 中文乱码（如「娴嬭瘯」应为「测试」）时改为 true；Linux 上若一直正常请保持 false
+    plain-json-fix-gbk-misread: false
+
 # Spring配置
 spring:
   profiles:

fs-service/src/main/java/com/fs/qw/mapper/QwMsgAuditMessageMapper.java

@@ -5,7 +5,6 @@ import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.fs.qw.dto.QwMsgAuditConversationDTO;
 import com.fs.qw.domain.audit.QwMsgAuditMessage;
 import org.apache.ibatis.annotations.Param;
-import org.apache.ibatis.annotations.Select;
 
 /**
  * 企微会话存档结构化消息Mapper接口
@@ -18,9 +17,10 @@ public interface QwMsgAuditMessageMapper extends BaseMapper<QwMsgAuditMessage>{
      * 查询企微会话存档结构化消息
      *
      * @param id 企微会话存档结构化消息主键
+     * @param shard
      * @return 企微会话存档结构化消息
      */
-    QwMsgAuditMessage selectQwMsgAuditMessageById(Long id);
+    QwMsgAuditMessage selectQwMsgAuditMessageById(@Param("id") Long id, @Param("shard") int shard);
 
     /**
      * 查询企微会话存档结构化消息列表
@@ -50,19 +50,20 @@ public interface QwMsgAuditMessageMapper extends BaseMapper<QwMsgAuditMessage>{
      * 删除企微会话存档结构化消息
      *
      * @param id 企微会话存档结构化消息主键
+     * @param shard
      * @return 结果
      */
-    int deleteQwMsgAuditMessageById(Long id);
+    int deleteQwMsgAuditMessageById(@Param("id") Long id, @Param("shard") int shard);
 
     /**
      * 批量删除企微会话存档结构化消息
      *
      * @param ids 需要删除的数据主键集合
+     * @param shard
      * @return 结果
      */
-    int deleteQwMsgAuditMessageByIds(Long[] ids);
+    int deleteQwMsgAuditMessageByIdsInShard(@Param("ids") Long[] ids, @Param("shard") int shard);
 
-    @Select("select * from qw_msg_audit_message where corp_id = #{corpId} and seq = #{seq} limit 1")
     QwMsgAuditMessage selectByCorpIdAndSeq(@Param("corpId") String corpId, @Param("seq") Long seq);
 
     /**
@@ -73,9 +74,6 @@ public interface QwMsgAuditMessageMapper extends BaseMapper<QwMsgAuditMessage>{
     /**
      * 查询指定企业下指定消息类型中 media_oss_url 为空的记录（待下载上传）
      */
-    @Select("select * from qw_msg_audit_message where corp_id = #{corpId} and msg_type = #{msgType} and media_sdkfileid is not null and media_oss_url is null limit #{limit}")
     List<QwMsgAuditMessage> selectMediaPendingOssUpload(@Param("corpId") String corpId, @Param("msgType") String msgType, @Param("limit") int limit);
 
-//    int batchInsertMessage(QwMsgAuditMessage record);
-
 }

fs-service/src/main/java/com/fs/qw/mapper/QwMsgAuditRawMapper.java

@@ -4,7 +4,6 @@ import java.util.List;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.fs.qw.domain.audit.QwMsgAuditRaw;
 import org.apache.ibatis.annotations.Param;
-import org.apache.ibatis.annotations.Select;
 
 /**
  * 企微会话存档原始明文Mapper接口
@@ -17,9 +16,10 @@ public interface QwMsgAuditRawMapper extends BaseMapper<QwMsgAuditRaw>{
      * 查询企微会话存档原始明文
      *
      * @param id 企微会话存档原始明文主键
+     * @param shard
      * @return 企微会话存档原始明文
      */
-    QwMsgAuditRaw selectQwMsgAuditRawById(Long id);
+    QwMsgAuditRaw selectQwMsgAuditRawById(@Param("id") Long id, @Param("shard") int shard);
 
     /**
      * 查询企微会话存档原始明文列表
@@ -49,21 +49,20 @@ public interface QwMsgAuditRawMapper extends BaseMapper<QwMsgAuditRaw>{
      * 删除企微会话存档原始明文
      *
      * @param id 企微会话存档原始明文主键
+     * @param shard
      * @return 结果
      */
-    int deleteQwMsgAuditRawById(Long id);
+    int deleteQwMsgAuditRawById(@Param("id") Long id, @Param("shard") int shard);
 
     /**
      * 批量删除企微会话存档原始明文
      *
      * @param ids 需要删除的数据主键集合
+     * @param shard
      * @return 结果
      */
-    int deleteQwMsgAuditRawByIds(Long[] ids);
+    int deleteQwMsgAuditRawByIdsInShard(@Param("ids") Long[] ids, @Param("shard") int shard);
 
-    @Select("select * from qw_msg_audit_raw where corp_id = #{corpId} and seq = #{seq} limit 1")
     QwMsgAuditRaw selectByCorpIdAndSeq(@Param("corpId") String corpId, @Param("seq") Long seq);
 
-//    int batchInsertRaw(QwMsgAuditRaw record);
-
 }

fs-service/src/main/java/com/fs/qw/service/impl/QwMsgAuditMessageServiceImpl.java

@@ -2,13 +2,8 @@ package com.fs.qw.service.impl;
 
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Locale;
-import java.util.Map;
 
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONArray;
 import com.fs.common.utils.DateUtils;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.fs.qw.domain.QwExternalContact;
@@ -17,7 +12,7 @@ import com.fs.qw.vo.QwMsgAuditConversationVO;
 import com.fs.qw.domain.audit.QwMsgAuditMessage;
 import com.fs.qw.mapper.QwExternalContactMapper;
 import com.fs.qw.mapper.QwMsgAuditMessageMapper;
-import com.fs.qw.mapper.QwUserMapper;
+import com.fs.qw.shardingConfig.QwMsgAuditMessageSharding;
 import com.fs.qw.service.IQwMsgAuditMessageService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -35,9 +30,6 @@ public class QwMsgAuditMessageServiceImpl extends ServiceImpl<QwMsgAuditMessageM
     @Autowired
     private QwExternalContactMapper qwExternalContactMapper;
 
-    @Autowired
-    private QwUserMapper qwUserMapper;
-
     /**
      * 查询企微会话存档结构化消息
      *
@@ -47,7 +39,16 @@ public class QwMsgAuditMessageServiceImpl extends ServiceImpl<QwMsgAuditMessageM
     @Override
     public QwMsgAuditMessage selectQwMsgAuditMessageById(Long id)
     {
-        return baseMapper.selectQwMsgAuditMessageById(id);
+        if (id == null) {
+            return null;
+        }
+        for (int s = 0; s < QwMsgAuditMessageSharding.SHARD_COUNT; s++) {
+            QwMsgAuditMessage row = baseMapper.selectQwMsgAuditMessageById(id, s);
+            if (row != null) {
+                return row;
+            }
+        }
+        return null;
     }
 
     /**
@@ -59,6 +60,9 @@ public class QwMsgAuditMessageServiceImpl extends ServiceImpl<QwMsgAuditMessageM
     @Override
     public List<QwMsgAuditMessage> selectQwMsgAuditMessageList(QwMsgAuditMessage qwMsgAuditMessage)
     {
+        if (qwMsgAuditMessage == null || !StringUtils.hasText(qwMsgAuditMessage.getCorpId())) {
+            throw new IllegalArgumentException("corpId不能为空");
+        }
         List<QwMsgAuditMessage> list = baseMapper.selectQwMsgAuditMessageList(qwMsgAuditMessage);
 //        fillContactNameAndQwUserName(list);
         return list;
@@ -301,6 +305,9 @@ public class QwMsgAuditMessageServiceImpl extends ServiceImpl<QwMsgAuditMessageM
     @Override
     public int insertQwMsgAuditMessage(QwMsgAuditMessage qwMsgAuditMessage)
     {
+        if (qwMsgAuditMessage == null || !StringUtils.hasText(qwMsgAuditMessage.getCorpId())) {
+            throw new IllegalArgumentException("corpId 不能为空");
+        }
         qwMsgAuditMessage.setCreateTime(DateUtils.getNowDate());
         return baseMapper.insertQwMsgAuditMessage(qwMsgAuditMessage);
     }
@@ -314,6 +321,9 @@ public class QwMsgAuditMessageServiceImpl extends ServiceImpl<QwMsgAuditMessageM
     @Override
     public int updateQwMsgAuditMessage(QwMsgAuditMessage qwMsgAuditMessage)
     {
+        if (qwMsgAuditMessage == null || !StringUtils.hasText(qwMsgAuditMessage.getCorpId())) {
+            throw new IllegalArgumentException("corpId 不能为空");
+        }
         return baseMapper.updateQwMsgAuditMessage(qwMsgAuditMessage);
     }
 
@@ -326,7 +336,14 @@ public class QwMsgAuditMessageServiceImpl extends ServiceImpl<QwMsgAuditMessageM
     @Override
     public int deleteQwMsgAuditMessageByIds(Long[] ids)
     {
-        return baseMapper.deleteQwMsgAuditMessageByIds(ids);
+        if (ids == null || ids.length == 0) {
+            return 0;
+        }
+        int total = 0;
+        for (int s = 0; s < QwMsgAuditMessageSharding.SHARD_COUNT; s++) {
+            total += baseMapper.deleteQwMsgAuditMessageByIdsInShard(ids, s);
+        }
+        return total;
     }
 
     /**
@@ -338,6 +355,13 @@ public class QwMsgAuditMessageServiceImpl extends ServiceImpl<QwMsgAuditMessageM
     @Override
     public int deleteQwMsgAuditMessageById(Long id)
     {
-        return baseMapper.deleteQwMsgAuditMessageById(id);
+        if (id == null) {
+            return 0;
+        }
+        int total = 0;
+        for (int s = 0; s < QwMsgAuditMessageSharding.SHARD_COUNT; s++) {
+            total += baseMapper.deleteQwMsgAuditMessageById(id, s);
+        }
+        return total;
     }
 }

fs-service/src/main/java/com/fs/qw/shardingConfig/QwMsgAuditMessageSharding.java

@@ -0,0 +1 @@
+package com.fs.qw.shardingConfig;

import org.springframework.util.StringUtils;
import java.nio.charset.StandardCharsets;
import java.util.zip.CRC32;

/**
 * 会话存档表分片：qw_msg_audit_message_0.._11、qw_msg_audit_raw_0.._11；
 */

public final class QwMsgAuditMessageSharding {

    public static final int SHARD_COUNT = 12;

    public static final String TABLE_PREFIX = "qw_msg_audit_message_";

    public static final String RAW_TABLE_PREFIX = "qw_msg_audit_raw_";

    private QwMsgAuditMessageSharding() {
    }

    /**

     * @return 0 .. SHARD_COUNT-1，corpId 为空时返回 0（避免 NPE，调用方应校验 corpId）

     */
    public static int shardIndex(String corpId) {

        if (!StringUtils.hasText(corpId)) {
            return 0;
        }
        CRC32 crc = new CRC32();
        crc.update(corpId.getBytes(StandardCharsets.UTF_8));
        long v = crc.getValue();
        return (int) (v % SHARD_COUNT);
    }

    public static String physicalTableName(int shard) {

        if (shard < 0 || shard >= SHARD_COUNT) {
            throw new IllegalArgumentException("shard out of range: " + shard);
        }
        return TABLE_PREFIX + shard;
    }

    public static String physicalTableName(String corpId) {
        return physicalTableName(shardIndex(corpId));
    }

    public static String rawPhysicalTableName(int shard) {
        if (shard < 0 || shard >= SHARD_COUNT) {
            throw new IllegalArgumentException("shard out of range: " + shard);
        }
        return RAW_TABLE_PREFIX + shard;
    }

    public static boolean isValidShard(int shard) {
        return shard >= 0 && shard < SHARD_COUNT;
    }

}

fs-service/src/main/resources/mapper/qw/QwMsgAuditMessageMapper.xml

@@ -4,6 +4,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
 "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.fs.qw.mapper.QwMsgAuditMessageMapper">
 
+    <!-- 分表：上下文需含 corpId -->
+    <sql id="physicalTableByCorpId">qw_msg_audit_message_${@com.fs.qw.shardingConfig.QwMsgAuditMessageSharding@shardIndex(corpId)}</sql>
+
     <resultMap type="QwMsgAuditMessage" id="QwMsgAuditMessageResult">
         <result property="id"    column="id"    />
         <result property="corpId"    column="corp_id"    />
@@ -31,11 +34,12 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
     </resultMap>
 
     <sql id="selectQwMsgAuditMessageVo">
-        select id, corp_id, seq, msg_id, msg_time, msg_type, from_user, to_list, room_id, conversation_key, from_user_role, text_content, media_sdkfileid, media_md5sum, media_size, media_play_length, media_width, media_height, media_file_name, media_file_ext, media_oss_url, raw_id, create_time from qw_msg_audit_message
+        select id, corp_id, seq, msg_id, msg_time, msg_type, from_user, to_list, room_id, conversation_key, from_user_role, text_content, media_sdkfileid, media_md5sum, media_size, media_play_length, media_width, media_height, media_file_name, media_file_ext, media_oss_url, raw_id, create_time from
     </sql>
 
     <select id="selectQwMsgAuditMessageList" parameterType="QwMsgAuditMessage" resultMap="QwMsgAuditMessageResult">
         <include refid="selectQwMsgAuditMessageVo"/>
+        <include refid="physicalTableByCorpId"/>
         <where>
             <if test="corpId != null  and corpId != ''"> and corp_id = #{corpId}</if>
             <if test="seq != null "> and seq = #{seq}</if>
@@ -61,13 +65,15 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
             order by msg_time desc
     </select>
 
-    <select id="selectQwMsgAuditMessageById" parameterType="Long" resultMap="QwMsgAuditMessageResult">
+    <select id="selectQwMsgAuditMessageById" resultMap="QwMsgAuditMessageResult">
         <include refid="selectQwMsgAuditMessageVo"/>
+        qw_msg_audit_message_${shard}
         where id = #{id}
     </select>
 
     <insert id="insertQwMsgAuditMessage" parameterType="QwMsgAuditMessage" useGeneratedKeys="true" keyProperty="id">
-        insert into qw_msg_audit_message
+        insert into
+        <include refid="physicalTableByCorpId"/>
         <trim prefix="(" suffix=")" suffixOverrides=",">
             <if test="corpId != null and corpId != ''">corp_id,</if>
             <if test="seq != null">seq,</if>
@@ -119,7 +125,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
     </insert>
 
     <update id="updateQwMsgAuditMessage" parameterType="QwMsgAuditMessage">
-        update qw_msg_audit_message
+        update
+        <include refid="physicalTableByCorpId"/>
         <trim prefix="SET" suffixOverrides=",">
             <if test="corpId != null and corpId != ''">corp_id = #{corpId},</if>
             <if test="seq != null">seq = #{seq},</if>
@@ -147,22 +154,35 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         where id = #{id}
     </update>
 
-    <delete id="deleteQwMsgAuditMessageById" parameterType="Long">
-        delete from qw_msg_audit_message where id = #{id}
+    <delete id="deleteQwMsgAuditMessageById">
+        delete from qw_msg_audit_message_${shard} where id = #{id}
     </delete>
 
-    <delete id="deleteQwMsgAuditMessageByIds" parameterType="String">
-        delete from qw_msg_audit_message where id in
-        <foreach item="id" collection="array" open="(" separator="," close=")">
+    <delete id="deleteQwMsgAuditMessageByIdsInShard">
+        delete from qw_msg_audit_message_${shard} where id in
+        <foreach item="id" collection="ids" open="(" separator="," close=")">
             #{id}
         </foreach>
     </delete>
 
+    <select id="selectByCorpIdAndSeq" resultMap="QwMsgAuditMessageResult">
+        <include refid="selectQwMsgAuditMessageVo"/>
+        <include refid="physicalTableByCorpId"/>
+        where corp_id = #{corpId} and seq = #{seq} limit 1
+    </select>
+
+    <select id="selectMediaPendingOssUpload" resultMap="QwMsgAuditMessageResult">
+        <include refid="selectQwMsgAuditMessageVo"/>
+        <include refid="physicalTableByCorpId"/>
+        where corp_id = #{corpId} and msg_type = #{msgType} and media_sdkfileid is not null and media_oss_url is null limit #{limit}
+    </select>
+
     <!-- 单聊会话列表 -->
     <select id="selectSingleConversationList" resultType="com.fs.qw.dto.QwMsgAuditConversationDTO">
         SELECT m.conversation_key AS conversationKey,
                MAX(m.msg_time)    AS lastMsgTime
-        FROM qw_msg_audit_message m
+        FROM
+        qw_msg_audit_message_${@com.fs.qw.shardingConfig.QwMsgAuditMessageSharding@shardIndex(corpId)} m
         WHERE m.corp_id = #{corpId}
           AND (m.room_id IS NULL OR m.room_id = '')
           AND m.conversation_key IS NOT NULL
@@ -179,17 +199,4 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         ORDER BY lastMsgTime DESC
     </select>
 
-<!--    <insert id="batchInsertMessage" parameterType="com.fs.qw.domain.audit.QwMsgAuditMessage" useGeneratedKeys="true" keyProperty="id">-->
-<!--        insert into qw_msg_audit_message-->
-<!--        (corp_id, seq, msg_id, msg_time, msg_type, text_content,-->
-<!--         media_sdkfileid, media_md5sum, media_size, media_play_length,-->
-<!--         media_width, media_height, media_file_name, media_file_ext,-->
-<!--         raw_id, create_time)-->
-<!--        values-->
-<!--            (#{corpId}, #{seq}, #{msgId}, #{msgTime}, #{msgType}, #{textContent},-->
-<!--             #{mediaSdkfileid}, #{mediaMd5sum}, #{mediaSize}, #{mediaPlayLength},-->
-<!--             #{mediaWidth}, #{mediaHeight}, #{mediaFileName}, #{mediaFileExt},-->
-<!--             #{rawId}, #{createTime})-->
-<!--    </insert>-->
-
 </mapper>

fs-service/src/main/resources/mapper/qw/QwMsgAuditRawMapper.xml

@@ -4,6 +4,9 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
 "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.fs.qw.mapper.QwMsgAuditRawMapper">
 
+    <!-- 与 qw_msg_audit_message 同分片规则：MOD(CRC32(corp_id), 12) -->
+    <sql id="physicalRawTableByCorpId">qw_msg_audit_raw_${@com.fs.qw.shardingConfig.QwMsgAuditMessageSharding@shardIndex(corpId)}</sql>
+
     <resultMap type="QwMsgAuditRaw" id="QwMsgAuditRawResult">
         <result property="id"    column="id"    />
         <result property="corpId"    column="corp_id"    />
@@ -20,11 +23,12 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
     </resultMap>
 
     <sql id="selectQwMsgAuditRawVo">
-        select id, corp_id, seq, msg_id, action, from_user, to_list, room_id, msg_time, msg_type, raw_json, create_time from qw_msg_audit_raw
+        select id, corp_id, seq, msg_id, action, from_user, to_list, room_id, msg_time, msg_type, raw_json, create_time from
     </sql>
 
     <select id="selectQwMsgAuditRawList" parameterType="QwMsgAuditRaw" resultMap="QwMsgAuditRawResult">
         <include refid="selectQwMsgAuditRawVo"/>
+        <include refid="physicalRawTableByCorpId"/>
         <where>
             <if test="corpId != null  and corpId != ''"> and corp_id = #{corpId}</if>
             <if test="seq != null "> and seq = #{seq}</if>
@@ -39,13 +43,21 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         </where>
     </select>
 
-    <select id="selectQwMsgAuditRawById" parameterType="Long" resultMap="QwMsgAuditRawResult">
+    <select id="selectQwMsgAuditRawById" resultMap="QwMsgAuditRawResult">
         <include refid="selectQwMsgAuditRawVo"/>
+        qw_msg_audit_raw_${shard}
         where id = #{id}
     </select>
 
+    <select id="selectByCorpIdAndSeq" resultMap="QwMsgAuditRawResult">
+        <include refid="selectQwMsgAuditRawVo"/>
+        <include refid="physicalRawTableByCorpId"/>
+        where corp_id = #{corpId} and seq = #{seq} limit 1
+    </select>
+
     <insert id="insertQwMsgAuditRaw" parameterType="QwMsgAuditRaw" useGeneratedKeys="true" keyProperty="id">
-        insert into qw_msg_audit_raw
+        insert into
+        <include refid="physicalRawTableByCorpId"/>
         <trim prefix="(" suffix=")" suffixOverrides=",">
             <if test="corpId != null and corpId != ''">corp_id,</if>
             <if test="seq != null">seq,</if>
@@ -75,7 +87,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
     </insert>
 
     <update id="updateQwMsgAuditRaw" parameterType="QwMsgAuditRaw">
-        update qw_msg_audit_raw
+        update
+        <include refid="physicalRawTableByCorpId"/>
         <trim prefix="SET" suffixOverrides=",">
             <if test="corpId != null and corpId != ''">corp_id = #{corpId},</if>
             <if test="seq != null">seq = #{seq},</if>
@@ -92,22 +105,15 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         where id = #{id}
     </update>
 
-    <delete id="deleteQwMsgAuditRawById" parameterType="Long">
-        delete from qw_msg_audit_raw where id = #{id}
+    <delete id="deleteQwMsgAuditRawById">
+        delete from qw_msg_audit_raw_${shard} where id = #{id}
     </delete>
 
-    <delete id="deleteQwMsgAuditRawByIds" parameterType="String">
-        delete from qw_msg_audit_raw where id in
-        <foreach item="id" collection="array" open="(" separator="," close=")">
+    <delete id="deleteQwMsgAuditRawByIdsInShard">
+        delete from qw_msg_audit_raw_${shard} where id in
+        <foreach item="id" collection="ids" open="(" separator="," close=")">
             #{id}
         </foreach>
     </delete>
 
-<!--    <insert id="batchInsertRaw" parameterType="com.fs.qw.domain.audit.QwMsgAuditRaw" useGeneratedKeys="true" keyProperty="id">-->
-<!--        insert into qw_msg_audit_raw-->
-<!--        (corp_id, seq, msg_id, action, from_user, to_list, room_id, msg_time, msg_type, raw_json, create_time)-->
-<!--        values-->
-<!--            (#{corpId}, #{seq}, #{msgId}, #{action}, #{fromUser}, #{toList}, #{roomId}, #{msgTime}, #{msgType}, #{rawJson}, #{createTime})-->
-<!--    </insert>-->
-
 </mapper>