feat(t17): 数据同步——只读拉取群组/群成员/群消息到本地表，按租户查看
主体（T17 数据同步，/admin/sync 三个独立按钮）： - 同步群组：get_appid_group_list 全量 Next 分页 + 逐群 get_group_info → group_mapping - 同步群成员(→用户)：遍历该租户群取 MemberList → user_mapping（腾讯无全量用户API，靠群成员反推） - 同步群消息：遍历该租户群 getGroupMsg + IsFinished 滚动全量 → im_message(source=SYNC) - DB: group_mapping 加 name/owner_account/member_count/last_synced_at；user_mapping 加 nick/last_synced_at （init.sql 建表 + ADD COLUMN IF NOT EXISTS 老库升级补丁，幂等） - TencentImClient: 新增 getAppidGroupList(limit,next[,sdkAppId,secretKey]) - 已知限制：C2C单聊无全量会话API；超大群成员需换 get_group_member_info 分页 附带收尾此前未提交的改动： - 回调字段名修正(FromAccount→From_Account 等腾讯标准字段) + pickMsgRandom/Time/Type 兼容字段差异 - 租户识别重构：前缀经 TenantService.getByPrefixCode 反查 tenantId（主键雪花化与前缀解耦） - FreeMarker java.time ?string 坑修复(usage/queue Controller 预格式化) + 消息记录"全部"状态修复 - 新增项目 CLAUDE.md + .claude/memory 基建(gitignore 含密钥记忆，不进 git) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
连龙刚
@@ -0,0 +1,468 @@
|
||||
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;
|
||||
|
||||
/** 数据源密钥解析结果 */
|
||||
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++) {
|
||||
String groupId = groupIds.get(i).getString();
|
||||
if (groupId == null || groupId.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
total++;
|
||||
try {
|
||||
if (upsertGroupFromRemote(ctx, groupId)) {
|
||||
upserted++;
|
||||
}
|
||||
} catch (Exception ge) {
|
||||
log.warn("同步单群失败 {} : {}", groupId, ge.getMessage());
|
||||
}
|
||||
}
|
||||
task.setTotalCount((long) total);
|
||||
task.setProcessedCount((long) upserted);
|
||||
task.setPosCursor("page:" + guard);
|
||||
migrateTaskMapper.updateById(task);
|
||||
|
||||
next = root.get("Next").getString();
|
||||
if (next == null || next.isEmpty()) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
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 = 0;
|
||||
for (int i = 0; i < n; i++) {
|
||||
ONode m = rspList.get(i);
|
||||
long seq = m.get("MsgSeq").getLong();
|
||||
if (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;
|
||||
if (finished || n == 0 || oldestSeq <= 0) {
|
||||
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) {
|
||||
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());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user