连龙刚andClaude
968a5f2e42
联调(新疆信创测试应用 1600134460,租户sa 67群)发现并修复4类问题: 1.群消息死循环:group_msg_get_simple 的 IsFinished=2 表"还有更旧可拉",占位消息群恒为2; 原 oldestSeq=max(seq) 游标方向反,反复拉同一批致2217次/200s超时。 改 min(seq) 向旧翻页 + 游标未推进即停兜底 + 过滤 IsPlaceMsg=1 占位消息(119次/15.7s) 2.get_group_info 请求体字段名 GroupId→GroupIdList(腾讯要求 GroupIdList:["id"] 字符串数组) 3.运行库缺列致 upserted=0:init.sql 仅新建库执行,新增 SchemaUpgradeBootstrap 启动幂等ALTER(6条DDL) 4.maxGroups 防护:break 由 for 外移入 for 内,精确限制避免单页海量群绕过上限 联调验证(租户sa 67群):群组67/67、群成员241/13、群消息119次/imported=3,死循环消除 Co-Authored-By: Claude <noreply@anthropic.com>
495 lines
20 KiB
Java
495 lines
20 KiB
Java
package com.imutil.service.impl;
|
||
|
||
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
|
||
import com.imutil.common.BizException;
|
||
import com.imutil.common.MsgKeys;
|
||
import com.imutil.entity.GroupMapping;
|
||
import com.imutil.entity.ImMessage;
|
||
import com.imutil.entity.MigrateTask;
|
||
import com.imutil.entity.SourceApp;
|
||
import com.imutil.entity.Tenant;
|
||
import com.imutil.entity.UserMapping;
|
||
import com.imutil.mapper.GroupMappingMapper;
|
||
import com.imutil.mapper.ImMessageMapper;
|
||
import com.imutil.mapper.MigrateTaskMapper;
|
||
import com.imutil.mapper.SourceAppMapper;
|
||
import com.imutil.mapper.TenantMapper;
|
||
import com.imutil.mapper.UserMappingMapper;
|
||
import com.imutil.service.SyncService;
|
||
import com.imutil.tencent.TencentImClient;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.noear.snack4.ONode;
|
||
import org.noear.solon.annotation.Component;
|
||
import org.noear.solon.annotation.Inject;
|
||
|
||
import java.time.Instant;
|
||
import java.time.OffsetDateTime;
|
||
import java.time.ZoneId;
|
||
import java.util.List;
|
||
|
||
/**
|
||
* 数据同步服务实现（T17）
|
||
* <p>
|
||
* 只读拉取腾讯 IM 数据到本地表，供后台按租户查看。与 MigrateService（带写回副作用）互补。
|
||
* <ul>
|
||
* <li>群组：get_appid_group_list 全量 Next 分页 + 逐群 get_group_info → upsert group_mapping</li>
|
||
* <li>群成员(→用户)：遍历该租户已同步群，取 MemberList → upsert user_mapping（腾讯无全量用户接口，靠群成员反推）</li>
|
||
* <li>群消息：遍历该租户已同步群，getGroupMsg + IsFinished 滚动全量 → 写 im_message(source=SYNC)</li>
|
||
* </ul>
|
||
* 数据源密钥：sourceAppId=null 走主应用全局密钥（需指定归属租户）；非空走 source_app 密钥（租户取其关联值）。
|
||
* 幂等可重跑：群/用户按 imGroupId/imUserId 唯一索引 upsert；消息按 msgKey 查重跳过。
|
||
*
|
||
* @author imutil
|
||
*/
|
||
@Slf4j
|
||
@Component
|
||
public class SyncServiceImpl implements SyncService {
|
||
|
||
@Inject
|
||
private SourceAppMapper sourceAppMapper;
|
||
|
||
@Inject
|
||
private TenantMapper tenantMapper;
|
||
|
||
@Inject
|
||
private GroupMappingMapper groupMappingMapper;
|
||
|
||
@Inject
|
||
private UserMappingMapper userMappingMapper;
|
||
|
||
@Inject
|
||
private ImMessageMapper imMessageMapper;
|
||
|
||
@Inject
|
||
private MigrateTaskMapper migrateTaskMapper;
|
||
|
||
@Inject
|
||
private TencentImClient tencentImClient;
|
||
|
||
/** 主应用全局密钥（sourceAppId=null 时用） */
|
||
@Inject("${imutil.tencent.sdkAppId:0}")
|
||
private long configSdkAppId;
|
||
|
||
@Inject("${imutil.tencent.secretKey:}")
|
||
private String configSecretKey;
|
||
|
||
/** 拉群列表每页数量（get_appid_group_list Limit） */
|
||
@Inject("${imutil.sync.groupListPageSize:100}")
|
||
private int groupListPageSize;
|
||
|
||
/** 拉群消息每页条数（getGroupMsg reqMsgNumber） */
|
||
@Inject("${imutil.sync.groupMsgPerPage:100}")
|
||
private int groupMsgPerPage;
|
||
|
||
/** 单群消息分页上限（防异常死循环） */
|
||
@Inject("${imutil.sync.groupMsgMaxPages:500}")
|
||
private int groupMsgMaxPages;
|
||
|
||
/** 单次同步群组数量上限（同步阻塞 HTTP，避免海量群卡死；如需全量调大） */
|
||
@Inject("${imutil.sync.maxGroups:500}")
|
||
private int maxGroups;
|
||
|
||
/** 数据源密钥解析结果 */
|
||
private static class SyncCtx {
|
||
long sdkAppId;
|
||
String secretKey;
|
||
String tenantId;
|
||
/** migrate_task.source_app_id；主应用记 0（该字段 NOT NULL，0 表示主应用） */
|
||
Long taskSourceAppId;
|
||
}
|
||
|
||
/**
|
||
* 解析数据源密钥 + 归属租户
|
||
*/
|
||
private SyncCtx resolveCtx(Long sourceAppId, String tenantId) {
|
||
SyncCtx ctx = new SyncCtx();
|
||
if (sourceAppId == null) {
|
||
if (tenantId == null || tenantId.isEmpty()) {
|
||
throw new BizException("主应用同步需指定归属租户");
|
||
}
|
||
Tenant t = tenantMapper.selectById(tenantId);
|
||
if (t == null) {
|
||
throw new BizException("租户不存在: " + tenantId);
|
||
}
|
||
ctx.sdkAppId = configSdkAppId;
|
||
ctx.secretKey = configSecretKey;
|
||
ctx.tenantId = tenantId;
|
||
ctx.taskSourceAppId = 0L;
|
||
} else {
|
||
SourceApp app = sourceAppMapper.selectById(sourceAppId);
|
||
if (app == null) {
|
||
throw new BizException("老应用不存在");
|
||
}
|
||
ctx.sdkAppId = Long.parseLong(app.getSdkAppId());
|
||
ctx.secretKey = app.getSecretKey();
|
||
ctx.tenantId = app.getTenantId();
|
||
ctx.taskSourceAppId = sourceAppId;
|
||
}
|
||
if (ctx.sdkAppId <= 0 || ctx.secretKey == null || ctx.secretKey.isEmpty()) {
|
||
throw new BizException("数据源密钥未配置");
|
||
}
|
||
return ctx;
|
||
}
|
||
|
||
private MigrateTask startTask(SyncCtx ctx, String objectType) {
|
||
MigrateTask task = new MigrateTask();
|
||
task.setSourceAppId(ctx.taskSourceAppId);
|
||
task.setObjectType(objectType);
|
||
task.setStatus(1);
|
||
task.setStartedAt(OffsetDateTime.now());
|
||
migrateTaskMapper.insert(task);
|
||
return task;
|
||
}
|
||
|
||
// ==================== 同步群组 ====================
|
||
|
||
@Override
|
||
public MigrateTask syncGroups(Long sourceAppId, String tenantId) {
|
||
SyncCtx ctx = resolveCtx(sourceAppId, tenantId);
|
||
MigrateTask task = startTask(ctx, "GROUP_SYNC");
|
||
try {
|
||
int total = 0;
|
||
int upserted = 0;
|
||
String next = "";
|
||
int guard = 0;
|
||
// Next 游标分页拉全量群，直到返回 Next 为空
|
||
while (guard++ < 100000) {
|
||
String resp = tencentImClient.getAppidGroupList(groupListPageSize, next, ctx.sdkAppId, ctx.secretKey);
|
||
ONode root = ONode.ofJson(resp);
|
||
if (!"OK".equalsIgnoreCase(root.get("ActionStatus").getString())) {
|
||
throw new BizException("拉群组列表失败: " + safeErr(root));
|
||
}
|
||
ONode groupIds = root.get("GroupIdList");
|
||
int n = groupIds.size();
|
||
for (int i = 0; i < n; i++) {
|
||
// 实测 GroupIdList 元素为对象 {GroupId:"..."}，兼容字符串形态
|
||
ONode gidNode = groupIds.get(i);
|
||
String groupId = gidNode.isObject() ? gidNode.get("GroupId").getString() : gidNode.getString();
|
||
if (groupId == null || groupId.isEmpty()) {
|
||
continue;
|
||
}
|
||
total++;
|
||
try {
|
||
if (upsertGroupFromRemote(ctx, groupId)) {
|
||
upserted++;
|
||
}
|
||
} catch (Exception ge) {
|
||
log.warn("同步单群失败 {} : {}", groupId, ge.toString());
|
||
}
|
||
// 达上限停止（for 内精确限制，避免单页海量群卡死同步 HTTP）
|
||
if (total >= maxGroups) {
|
||
log.warn("群组同步达上限 {}，停止（调大 imutil.sync.maxGroups 可继续）", maxGroups);
|
||
break;
|
||
}
|
||
}
|
||
task.setTotalCount((long) total);
|
||
task.setProcessedCount((long) upserted);
|
||
task.setPosCursor("page:" + guard);
|
||
migrateTaskMapper.updateById(task);
|
||
|
||
// 达上限停止（由 for 内触发并告警，此处仅跳出 while）
|
||
if (total >= maxGroups) {
|
||
break;
|
||
}
|
||
// Next 游标推进；为空或未变化则结束（防死循环）
|
||
String respNext = root.get("Next").getString();
|
||
if (respNext == null || respNext.isEmpty() || respNext.equals(next)) {
|
||
break;
|
||
}
|
||
next = respNext;
|
||
}
|
||
task.setStatus(3);
|
||
task.setFinishedAt(OffsetDateTime.now());
|
||
task.setVerifyResult(String.format("{\"total\":%d,\"upserted\":%d}", total, upserted));
|
||
migrateTaskMapper.updateById(task);
|
||
log.info("群组同步完成 tenant={} total={} upserted={}", ctx.tenantId, total, upserted);
|
||
return task;
|
||
} catch (Exception e) {
|
||
failTask(task, e);
|
||
log.error("群组同步失败 tenant={}", ctx.tenantId, e);
|
||
throw e;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 拉单群资料 → upsert group_mapping
|
||
*
|
||
* @return 是否成功处理（已记录为 true）
|
||
*/
|
||
private boolean upsertGroupFromRemote(SyncCtx ctx, String groupId) {
|
||
String resp = tencentImClient.getGroupInfo(groupId, ctx.sdkAppId, ctx.secretKey);
|
||
ONode root = ONode.ofJson(resp);
|
||
if (!"OK".equalsIgnoreCase(root.get("ActionStatus").getString())) {
|
||
return false;
|
||
}
|
||
ONode groupInfo = root.get("GroupInfo");
|
||
if (groupInfo.size() == 0) {
|
||
return false;
|
||
}
|
||
ONode g = groupInfo.get(0);
|
||
String name = g.get("Name").getString();
|
||
String type = g.get("Type").getString();
|
||
String owner = g.get("Owner_Account").getString();
|
||
long memberNum = g.get("MemberNum").getLong();
|
||
OffsetDateTime now = OffsetDateTime.now();
|
||
|
||
GroupMapping exist = groupMappingMapper.selectOne(Wrappers.<GroupMapping>lambdaQuery()
|
||
.eq(GroupMapping::getImGroupId, groupId).last("LIMIT 1"));
|
||
if (exist != null) {
|
||
exist.setTenantId(ctx.tenantId);
|
||
exist.setBizGroupId(groupId);
|
||
exist.setGroupType(type != null && !type.isEmpty() ? type : "Public");
|
||
exist.setName(name);
|
||
exist.setOwnerAccount(owner);
|
||
exist.setMemberCount((int) memberNum);
|
||
exist.setLastSyncedAt(now);
|
||
groupMappingMapper.updateById(exist);
|
||
} else {
|
||
GroupMapping gm = new GroupMapping();
|
||
gm.setTenantId(ctx.tenantId);
|
||
gm.setBizGroupId(groupId);
|
||
gm.setImGroupId(groupId);
|
||
gm.setGroupType(type != null && !type.isEmpty() ? type : "Public");
|
||
gm.setName(name);
|
||
gm.setOwnerAccount(owner);
|
||
gm.setMemberCount((int) memberNum);
|
||
gm.setLastSyncedAt(now);
|
||
groupMappingMapper.insert(gm);
|
||
}
|
||
return true;
|
||
}
|
||
|
||
// ==================== 同步群成员 → 用户 ====================
|
||
|
||
@Override
|
||
public MigrateTask syncGroupMembers(Long sourceAppId, String tenantId) {
|
||
SyncCtx ctx = resolveCtx(sourceAppId, tenantId);
|
||
MigrateTask task = startTask(ctx, "USER_SYNC");
|
||
try {
|
||
List<GroupMapping> groups = groupMappingMapper.selectList(Wrappers.<GroupMapping>lambdaQuery()
|
||
.eq(GroupMapping::getTenantId, ctx.tenantId));
|
||
task.setTotalCount((long) groups.size());
|
||
migrateTaskMapper.updateById(task);
|
||
|
||
int memberSeen = 0;
|
||
int upserted = 0;
|
||
for (int gi = 0; gi < groups.size(); gi++) {
|
||
String groupId = groups.get(gi).getImGroupId();
|
||
try {
|
||
String resp = tencentImClient.getGroupInfo(groupId, ctx.sdkAppId, ctx.secretKey);
|
||
ONode root = ONode.ofJson(resp);
|
||
if (!"OK".equalsIgnoreCase(root.get("ActionStatus").getString())) {
|
||
continue;
|
||
}
|
||
ONode groupInfo = root.get("GroupInfo");
|
||
if (groupInfo.size() == 0) {
|
||
continue;
|
||
}
|
||
ONode memberList = groupInfo.get(0).get("MemberList");
|
||
int mc = memberList.size();
|
||
for (int j = 0; j < mc; j++) {
|
||
ONode mem = memberList.get(j);
|
||
String accountId = mem.get("Member_Account").getString();
|
||
if (accountId == null || accountId.isEmpty()) {
|
||
continue;
|
||
}
|
||
memberSeen++;
|
||
String nick = mem.get("Name").getString();
|
||
if (upsertUser(ctx, accountId, nick)) {
|
||
upserted++;
|
||
}
|
||
}
|
||
} catch (Exception ge) {
|
||
log.warn("同步群成员失败 {} : {}", groupId, ge.getMessage());
|
||
}
|
||
task.setProcessedCount((long) (gi + 1));
|
||
task.setPosCursor("group:" + (gi + 1));
|
||
migrateTaskMapper.updateById(task);
|
||
}
|
||
task.setStatus(3);
|
||
task.setFinishedAt(OffsetDateTime.now());
|
||
task.setVerifyResult(String.format("{\"groups\":%d,\"memberSeen\":%d,\"upserted\":%d}",
|
||
groups.size(), memberSeen, upserted));
|
||
migrateTaskMapper.updateById(task);
|
||
log.info("群成员同步完成 tenant={} groups={} memberSeen={} upserted={}",
|
||
ctx.tenantId, groups.size(), memberSeen, upserted);
|
||
return task;
|
||
} catch (Exception e) {
|
||
failTask(task, e);
|
||
log.error("群成员同步失败 tenant={}", ctx.tenantId, e);
|
||
throw e;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* upsert user_mapping（按 im_user_id 唯一索引）
|
||
*
|
||
* @return 是否为本次新增
|
||
*/
|
||
private boolean upsertUser(SyncCtx ctx, String imUserId, String nick) {
|
||
UserMapping exist = userMappingMapper.selectOne(Wrappers.<UserMapping>lambdaQuery()
|
||
.eq(UserMapping::getImUserId, imUserId).last("LIMIT 1"));
|
||
OffsetDateTime now = OffsetDateTime.now();
|
||
if (exist != null) {
|
||
exist.setNick(nick);
|
||
exist.setLastSyncedAt(now);
|
||
if (exist.getStatus() == null) {
|
||
exist.setStatus(1);
|
||
}
|
||
userMappingMapper.updateById(exist);
|
||
return false;
|
||
}
|
||
UserMapping um = new UserMapping();
|
||
um.setTenantId(ctx.tenantId);
|
||
// 同步场景无业务侧 ID，用 IM ID 兜底（biz_user_id 仅做本地标识）
|
||
um.setBizUserId(imUserId);
|
||
um.setImUserId(imUserId);
|
||
um.setIsDefault(false);
|
||
um.setIsGlobal(false);
|
||
um.setStatus(1);
|
||
um.setNick(nick);
|
||
um.setLastSyncedAt(now);
|
||
userMappingMapper.insert(um);
|
||
return true;
|
||
}
|
||
|
||
// ==================== 同步群消息 ====================
|
||
|
||
@Override
|
||
public MigrateTask syncGroupMessages(Long sourceAppId, String tenantId) {
|
||
SyncCtx ctx = resolveCtx(sourceAppId, tenantId);
|
||
MigrateTask task = startTask(ctx, "GMSG_SYNC");
|
||
try {
|
||
List<GroupMapping> groups = groupMappingMapper.selectList(Wrappers.<GroupMapping>lambdaQuery()
|
||
.eq(GroupMapping::getTenantId, ctx.tenantId));
|
||
task.setTotalCount((long) groups.size());
|
||
migrateTaskMapper.updateById(task);
|
||
|
||
int imported = 0;
|
||
for (int gi = 0; gi < groups.size(); gi++) {
|
||
String groupId = groups.get(gi).getImGroupId();
|
||
try {
|
||
long reqSeq = 0;
|
||
int pages = 0;
|
||
// IsFinished 滚动全量：reqSeq 取本页最旧 seq 继续向前
|
||
while (pages++ < groupMsgMaxPages) {
|
||
String resp = tencentImClient.getGroupMsgAs(groupId, groupMsgPerPage, reqSeq, ctx.sdkAppId, ctx.secretKey);
|
||
ONode root = ONode.ofJson(resp);
|
||
if (!"OK".equalsIgnoreCase(root.get("ActionStatus").getString())) {
|
||
break;
|
||
}
|
||
ONode rspList = root.get("RspMsgList");
|
||
int n = rspList.size();
|
||
long oldestSeq = Long.MAX_VALUE; // 本页最小 seq（最旧）
|
||
for (int i = 0; i < n; i++) {
|
||
ONode m = rspList.get(i);
|
||
long seq = m.get("MsgSeq").getLong();
|
||
if (seq > 0 && seq < oldestSeq) {
|
||
oldestSeq = seq;
|
||
}
|
||
ImMessage msg = parseGroupMsgForSync(m, ctx.tenantId, groupId);
|
||
if (msg == null) {
|
||
continue;
|
||
}
|
||
// 幂等：本地已存在跳过
|
||
long cnt = imMessageMapper.selectCount(Wrappers.<ImMessage>lambdaQuery()
|
||
.eq(ImMessage::getMsgKey, msg.getMsgKey()));
|
||
if (cnt > 0) {
|
||
continue;
|
||
}
|
||
imMessageMapper.insert(msg);
|
||
imported++;
|
||
}
|
||
boolean finished = root.get("IsFinished").getLong() == 1;
|
||
// 终止：已拉完 / 空页 / 本页无有效 seq
|
||
if (finished || n == 0 || oldestSeq == Long.MAX_VALUE) {
|
||
break;
|
||
}
|
||
// 游标未向更旧方向推进则停（占位消息群 IsFinished 恒为 2，靠此兜底防死循环）
|
||
if (reqSeq > 0 && oldestSeq >= reqSeq) {
|
||
break;
|
||
}
|
||
reqSeq = oldestSeq;
|
||
}
|
||
} catch (Exception ge) {
|
||
log.warn("同步群消息失败 {} : {}", groupId, ge.getMessage());
|
||
}
|
||
task.setProcessedCount((long) imported);
|
||
task.setPosCursor("group:" + (gi + 1));
|
||
migrateTaskMapper.updateById(task);
|
||
}
|
||
task.setStatus(3);
|
||
task.setFinishedAt(OffsetDateTime.now());
|
||
task.setVerifyResult(String.format("{\"groups\":%d,\"imported\":%d}", groups.size(), imported));
|
||
migrateTaskMapper.updateById(task);
|
||
log.info("群消息同步完成 tenant={} groups={} imported={}", ctx.tenantId, groups.size(), imported);
|
||
return task;
|
||
} catch (Exception e) {
|
||
failTask(task, e);
|
||
log.error("群消息同步失败 tenant={}", ctx.tenantId, e);
|
||
throw e;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 解析腾讯群消息节点为 ImMessage（source=SYNC，dist_status=1 不进分发）
|
||
*/
|
||
private ImMessage parseGroupMsgForSync(ONode m, String tenantId, String groupId) {
|
||
// 过滤占位/系统消息（IsPlaceMsg=1：From_Account 与 MsgBody 均空，非真实消息，不入库）
|
||
if (m.get("IsPlaceMsg").getLong() == 1) {
|
||
return null;
|
||
}
|
||
long msgSeq = m.get("MsgSeq").getLong();
|
||
long msgRandom = m.get("MsgRandom").getLong();
|
||
long msgTs = m.get("MsgTimeStamp").getLong();
|
||
String from = m.get("From_Account").getString();
|
||
ONode body = m.get("MsgBody");
|
||
ImMessage msg = new ImMessage();
|
||
msg.setMsgKey(MsgKeys.build(from, groupId, msgSeq, msgRandom));
|
||
msg.setTenantId(tenantId);
|
||
msg.setMsgTime(toTime(msgTs));
|
||
msg.setConvType(2);
|
||
msg.setConvId(groupId);
|
||
msg.setFromAccount(from);
|
||
msg.setGroupId(groupId);
|
||
msg.setMsgType(firstBodyType(body));
|
||
msg.setMsgBody(body.toString());
|
||
msg.setSource("SYNC");
|
||
msg.setIsCrossTenant(false);
|
||
msg.setDistStatus(1);
|
||
return msg;
|
||
}
|
||
|
||
// ==================== 工具方法 ====================
|
||
|
||
private void failTask(MigrateTask task, Exception e) {
|
||
task.setStatus(4);
|
||
task.setErrorMsg(e.getClass().getSimpleName() + ": " + e.getMessage());
|
||
task.setFinishedAt(OffsetDateTime.now());
|
||
migrateTaskMapper.updateById(task);
|
||
}
|
||
|
||
private String safeErr(ONode root) {
|
||
try {
|
||
String info = root.get("ErrorInfo").getString();
|
||
return info == null ? "" : info;
|
||
} catch (Exception e) {
|
||
return "";
|
||
}
|
||
}
|
||
|
||
private String firstBodyType(ONode body) {
|
||
if (body == null || !body.isArray() || body.size() == 0) {
|
||
return null;
|
||
}
|
||
return body.get(0).get("MsgType").getString();
|
||
}
|
||
|
||
private OffsetDateTime toTime(long epochSecond) {
|
||
if (epochSecond <= 0) {
|
||
return OffsetDateTime.now();
|
||
}
|
||
return OffsetDateTime.ofInstant(Instant.ofEpochSecond(epochSecond), ZoneId.systemDefault());
|
||
}
|
||
}
|