From f401e8d0652a49098c6a99f2b5ec9526af998ca0 Mon Sep 17 00:00:00 2001 From: yixiong Date: Thu, 9 Jul 2026 12:57:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(t14-v1234):=20=E6=B6=88=E6=81=AF=E8=BF=81?= =?UTF-8?q?=E7=A7=BB=E5=A2=9E=E5=BC=BA=E2=80=94=E2=80=94C2C=E5=85=A8?= =?UTF-8?q?=E9=87=8F=E5=88=86=E9=A1=B5+=E7=BE=A4=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E8=BF=81=E7=A7=BB+429=E9=99=90=E9=80=9F=E9=87=8D=E8=AF=95+?= =?UTF-8?q?=E5=9B=9E=E6=BA=AF=E5=A4=A9=E6=95=B0=E5=8F=AF=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit V1: migrateMessages 7天窗滚动 + Complete分页拉全量历史 V2: migrateGroupMessages + getGroupMsgAs/importGroupMsg + /migrate/groupmsg V3: importMsg/importGroupMsg 捕获429重试 + app.yml imutil.migrate(msgLookbackDays/importMaxRetry/importRetryMs) Co-Authored-By: Claude --- .../imutil/controller/AdminController.java | 27 ++++ .../com/imutil/service/MigrateService.java | 7 + .../service/impl/MigrateServiceImpl.java | 133 ++++++++++++++++-- .../com/imutil/tencent/TencentImClient.java | 87 +++++++++++- src/main/resources/app.yml | 6 + src/main/resources/templates/migrate.ftl | 22 +++ 6 files changed, 265 insertions(+), 17 deletions(-) diff --git a/src/main/java/com/imutil/controller/AdminController.java b/src/main/java/com/imutil/controller/AdminController.java index e533777..431b75f 100644 --- a/src/main/java/com/imutil/controller/AdminController.java +++ b/src/main/java/com/imutil/controller/AdminController.java @@ -373,6 +373,33 @@ public class AdminController { ctx.redirect(basePath + "/admin/migrate?msg=gdone&migrated=" + task.getProcessedCount()); } + /** + * 群消息迁移：上传/粘贴群 ID 清单 → 逐群拉历史消息 → 加前缀 → importGroupMsg + */ + @Post + @Mapping("/migrate/groupmsg") + public void migrateGroupMsg(@Param Long sourceAppId, + @Param(defaultValue = "") String userText, + Context ctx) throws Throwable { + UploadedFile file = ctx.file("file"); + List groupIds = new ArrayList<>(); + if (file != null) { + parseUserIds(new String(file.getContentAsBytes(), StandardCharsets.UTF_8), groupIds); + } else if (!userText.isEmpty()) { + parseUserIds(userText, groupIds); + } + if (sourceAppId == null) { + ctx.redirect(basePath + "/admin/migrate?msg=noapp"); + return; + } + if (groupIds.isEmpty()) { + ctx.redirect(basePath + "/admin/migrate?msg=empty"); + return; + } + MigrateTask task = migrateService.migrateGroupMessages(sourceAppId, groupIds); + ctx.redirect(basePath + "/admin/migrate?msg=gmdone&imported=" + task.getProcessedCount()); + } + /** * 单聊消息迁移：上传会话清单（每行 from,to）或粘贴 → getRoamMsg 拉 → 加前缀 → importMsg */ diff --git a/src/main/java/com/imutil/service/MigrateService.java b/src/main/java/com/imutil/service/MigrateService.java index 7d9542e..5464c6f 100644 --- a/src/main/java/com/imutil/service/MigrateService.java +++ b/src/main/java/com/imutil/service/MigrateService.java @@ -49,6 +49,13 @@ public interface MigrateService { */ MigrateTask migrateMessages(Long sourceAppId, List c2cPairs); + /** + * 群消息迁移：逐群 getGroupMsgAs（老密钥）分页拉历史 → 加前缀 from → importGroupMsg（主密钥）→ 校验。 + * + * @param groupIds 老应用群 ID 清单 + */ + MigrateTask migrateGroupMessages(Long sourceAppId, List groupIds); + /** * 关系链迁移：逐用户 friendGetList（老密钥）拉好友 → 加前缀 → importFriend（主密钥）→ 校验。 *

diff --git a/src/main/java/com/imutil/service/impl/MigrateServiceImpl.java b/src/main/java/com/imutil/service/impl/MigrateServiceImpl.java index 0362a64..37d7fc4 100644 --- a/src/main/java/com/imutil/service/impl/MigrateServiceImpl.java +++ b/src/main/java/com/imutil/service/impl/MigrateServiceImpl.java @@ -52,6 +52,10 @@ public class MigrateServiceImpl implements MigrateService { @Inject private TencentImClient tencentImClient; + /** 消息迁移回溯天数，0=全量（T14 V3） */ + @Inject("${imutil.migrate.msgLookbackDays:7}") + private int msgLookbackDays; + @Override public MigrateTask migrateUsers(Long sourceAppId, List oldUserIds) { SourceApp app = sourceAppMapper.selectById(sourceAppId); @@ -281,20 +285,42 @@ public class MigrateServiceImpl implements MigrateService { String newFrom = prefix + "_" + oldFrom; String newTo = prefix + "_" + oldTo; try { - // v1：每会话拉最近 7 天内 100 条（腾讯 MaxTimeInterval 有限制；全量历史分页作为遗留） - long maxInterval = 604800L; // 7 天 - long minTime = nowSec - maxInterval; - String resp = tencentImClient.getRoamMsg(oldFrom, oldTo, 100, minTime, maxInterval); - ONode node = ONode.ofJson(resp); - ONode msgList = node.get("MsgList"); - int n = msgList.size(); - for (int i = 0; i < n; i++) { - ONode msg = msgList.get(i); - long ts = msg.get("MsgTimeStamp").getLong(); - String msgBodyJson = msg.get("MsgBody").toString(); - if (tencentImClient.importMsg(newFrom, newTo, ts, msgBodyJson)) { - imported++; + // 分页全量拉取：7 天窗滚动 + 窗内 Complete 分页（腾讯 MaxTimeInterval ≤ 604800） + // V3: 回溯天数可配置（0=全量，受套餐云端保存时长限制） + long minTimeBound = (msgLookbackDays <= 0) ? 0 : nowSec - msgLookbackDays * 86400L; + long segEnd = nowSec; + while (segEnd > minTimeBound) { + long segStart = Math.max(minTimeBound, segEnd - 604800L); + long fetchMaxTime = segEnd; + // 窗内 Complete 分页（LastMsgTime 滚动至 Complete=1 或无消息） + int guard = 0; + while (guard++ < 1000) { + long interval = Math.min(604800L, fetchMaxTime - segStart); + if (interval <= 0) { + break; + } + String resp = tencentImClient.getRoamMsgAs(oldFrom, oldTo, 100, segStart, interval, oldSdkAppId, oldSecret); + ONode node = ONode.ofJson(resp); + ONode msgList = node.get("MsgList"); + int n = msgList.size(); + for (int i = 0; i < n; i++) { + ONode msg = msgList.get(i); + long ts = msg.get("MsgTimeStamp").getLong(); + if (tencentImClient.importMsg(newFrom, newTo, ts, msg.get("MsgBody").toString())) { + imported++; + } + } + boolean complete = node.get("Complete").getLong() == 1; + if (complete || n == 0) { + break; + } + long nextLast = node.get("LastMsgTime").getLong(); + if (nextLast <= 0 || nextLast >= fetchMaxTime) { + break; // 防死循环：LastMsgTime 必须向前（减小） + } + fetchMaxTime = nextLast; } + segEnd = segStart; // 下一个 7 天窗 } } catch (Exception ce) { log.warn("单聊消息迁移会话失败 {}->{} : {}", oldFrom, oldTo, ce.getMessage()); @@ -320,6 +346,87 @@ public class MigrateServiceImpl implements MigrateService { } } + @Override + public MigrateTask migrateGroupMessages(Long sourceAppId, List groupIds) { + SourceApp app = sourceAppMapper.selectById(sourceAppId); + if (app == null) { + throw new BizException("老应用不存在"); + } + Tenant tenant = tenantMapper.selectById(app.getTenantId()); + if (tenant == null) { + throw new BizException("关联租户不存在: " + app.getTenantId()); + } + String prefix = tenant.getPrefixCode(); + long oldSdkAppId = Long.parseLong(app.getSdkAppId()); + String oldSecret = app.getSecretKey(); + + MigrateTask task = new MigrateTask(); + task.setSourceAppId(sourceAppId); + task.setObjectType("GROUP_MSG"); + task.setStatus(1); + int groups = groupIds == null ? 0 : groupIds.size(); + task.setTotalCount((long) groups); + task.setStartedAt(OffsetDateTime.now()); + migrateTaskMapper.insert(task); + + try { + int imported = 0; + for (int gi = 0; gi < groups; gi++) { + String oldGroupId = groupIds.get(gi).trim(); + if (oldGroupId.isEmpty()) { + continue; + } + String newGroupId = prefix + "_" + oldGroupId; + try { + // 分页拉取：ReqMsgSeq 滚动至 IsFinished=1 或无消息 + long reqSeq = 0; + int guard = 0; + while (guard++ < 10000) { + String resp = tencentImClient.getGroupMsgAs(oldGroupId, 100, reqSeq, oldSdkAppId, oldSecret); + ONode node = ONode.ofJson(resp); + ONode rspList = node.get("RspMsgList"); + int n = rspList.size(); + long oldestSeq = 0; + for (int i = 0; i < n; i++) { + ONode msg = rspList.get(i); + long seq = msg.get("MsgSeq").getLong(); + String from = msg.get("From_Account").getString(); + long ts = msg.get("MsgTimeStamp").getLong(); + String newFrom = (from != null && !from.isEmpty()) ? prefix + "_" + from : from; + if (tencentImClient.importGroupMsg(newGroupId, newFrom, ts, msg.get("MsgBody").toString())) { + imported++; + } + oldestSeq = seq; + } + boolean finished = node.get("IsFinished").getLong() == 1; + if (finished || n == 0 || oldestSeq <= 0) { + break; + } + reqSeq = oldestSeq; // 滚动到更旧 + } + } catch (Exception ge) { + log.warn("群消息迁移单群失败 {} : {}", oldGroupId, 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, imported)); + migrateTaskMapper.updateById(task); + log.info("群消息迁移完成 sourceApp={} groups={} imported={}", sourceAppId, groups, imported); + return task; + } catch (Exception e) { + task.setStatus(4); + task.setErrorMsg(e.getClass().getSimpleName() + ": " + e.getMessage()); + task.setFinishedAt(OffsetDateTime.now()); + migrateTaskMapper.updateById(task); + log.error("群消息迁移失败 sourceApp={}", sourceAppId, e); + throw e; + } + } + @Override public MigrateTask migrateRoster(Long sourceAppId, List oldUserIds) { SourceApp app = sourceAppMapper.selectById(sourceAppId); diff --git a/src/main/java/com/imutil/tencent/TencentImClient.java b/src/main/java/com/imutil/tencent/TencentImClient.java index 560de4e..0299fc0 100644 --- a/src/main/java/com/imutil/tencent/TencentImClient.java +++ b/src/main/java/com/imutil/tencent/TencentImClient.java @@ -1,5 +1,6 @@ package com.imutil.tencent; +import com.imutil.common.BizException; import com.imutil.common.Httpx; import com.imutil.common.Jsons; import com.imutil.common.LocalCache; @@ -60,6 +61,15 @@ public class TencentImClient { @Inject private RateLimiter rateLimiter; + // ===== 数据迁移：消息导入限速重试（T14 V3）===== + /** 消息导入 429 限流重试次数 */ + @Inject("${imutil.migrate.importMaxRetry:3}") + private int migrateImportMaxRetry; + + /** 消息导入重试间隔（毫秒） */ + @Inject("${imutil.migrate.importRetryMs:1000}") + private long migrateImportRetryMs; + @Inject private TenantService tenantService; @@ -283,6 +293,38 @@ public class TencentImClient { return callApi("group_open_http_svc/group_msg_get_simple", Jsons.stringify(body)); } + /** + * 查单聊历史漫游消息（老应用拉取，用老密钥）— 数据迁移用 + *

+ * 与 {@link #getRoamMsg} 同语义，但用指定老应用密钥调 callApiAs（迁移拉取消耗老应用配额）。 + */ + public String getRoamMsgAs(String fromAccount, String toAccount, int maxCnt, long minTime, long maxInterval, + long srcSdkAppId, String srcSecretKey) { + Map body = new HashMap<>(); + body.put("From_Account", fromAccount); + body.put("To_Account", toAccount); + body.put("MaxCnt", maxCnt); + body.put("MinTime", minTime); + body.put("MaxTimeInterval", maxInterval); + return callApiAs("openim_admin/get_roam_msg", Jsons.stringify(body), srcSdkAppId, srcSecretKey); + } + + /** + * 查群历史消息（老应用拉取，用老密钥，支持 ReqMsgSeq 分页）— 数据迁移用 + * + * @param reqMsgSeq 分页游标（上次最旧消息 seq；≤0 表示从最新开始） + */ + public String getGroupMsgAs(String groupId, int reqMsgNumber, long reqMsgSeq, + long srcSdkAppId, String srcSecretKey) { + Map body = new HashMap<>(); + body.put("GroupId", groupId); + body.put("ReqMsgNumber", reqMsgNumber); + if (reqMsgSeq > 0) { + body.put("ReqMsgSeq", reqMsgSeq); + } + return callApiAs("group_open_http_svc/group_msg_get_simple", Jsons.stringify(body), srcSdkAppId, srcSecretKey); + } + // ==================== 群迁移相关（T14 S3） ==================== /** @@ -384,8 +426,20 @@ public class TencentImClient { + "\",\"To_Account\":\"" + toAccount + "\",\"MsgTimeStamp\":" + msgTimeStamp + ",\"MsgBody\":" + (msgBodyJson == null ? "[]" : msgBodyJson) + "}"; - String resp = callApi("openim_http_svc/import_msg", body); - return isOk(resp); + // V3: 消息导入密集调用易触主应用限流，捕获 429 限速重试；耗尽或非 429 返回 false 跳过该条 + for (int attempt = 0; attempt <= migrateImportMaxRetry; attempt++) { + try { + String resp = callApi("openim_http_svc/import_msg", body); + return isOk(resp); + } catch (BizException e) { + if (e.getCode() != 429 || attempt == migrateImportMaxRetry) { + log.warn("导入单聊消息失败 from={} to={} ts={} : {}", fromAccount, toAccount, msgTimeStamp, e.getMessage()); + return false; + } + sleepRetry(); + } + } + return false; } /** @@ -400,8 +454,33 @@ public class TencentImClient { + "\",\"From_Account\":\"" + (fromAccount == null ? "" : fromAccount) + "\",\"MsgTimeStamp\":" + msgTimeStamp + ",\"SyncFromOldSystem\":1,\"MsgBody\":" + (msgBodyJson == null ? "[]" : msgBodyJson) + "}"; - String resp = callApi("group_open_http_svc/import_group_msg", body); - return isOk(resp); + // V3: 消息导入密集调用易触主应用限流，捕获 429 限速重试；耗尽或非 429 返回 false 跳过该条 + for (int attempt = 0; attempt <= migrateImportMaxRetry; attempt++) { + try { + String resp = callApi("group_open_http_svc/import_group_msg", body); + return isOk(resp); + } catch (BizException e) { + if (e.getCode() != 429 || attempt == migrateImportMaxRetry) { + log.warn("导入群消息失败 group={} from={} ts={} : {}", groupId, fromAccount, msgTimeStamp, e.getMessage()); + return false; + } + sleepRetry(); + } + } + return false; + } + + /** + * 消息导入限速重试间隔（T14 V3） + *

+ * 429 配额超限时 sleep 后重试；线程被中断则提前返回（中断状态已恢复，由上层决定是否继续）。 + */ + private void sleepRetry() { + try { + Thread.sleep(migrateImportRetryMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } } // ==================== 关系链迁移相关（T14 S5） ==================== diff --git a/src/main/resources/app.yml b/src/main/resources/app.yml index c1030ab..3ec6679 100644 --- a/src/main/resources/app.yml +++ b/src/main/resources/app.yml @@ -90,6 +90,12 @@ imutil.pull: maxMsgPerConv: 20 # 每会话每轮拉取条数上限 lookbackMinutes: 30 # C2C 拉取时间窗（最近N分钟） +# 数据迁移配置（T14 消息迁移） +imutil.migrate: + msgLookbackDays: 7 # 消息迁移回溯天数，0=全量（受套餐云端保存时长限制） + importMaxRetry: 3 # 消息导入 429 限流重试次数 + importRetryMs: 1000 # 消息导入重试间隔（毫秒） + # 限流配置（按租户令牌桶） imutil.ratelimit: defaultImQps: 50 # 默认每租户 IM API QPS diff --git a/src/main/resources/templates/migrate.ftl b/src/main/resources/templates/migrate.ftl index 9a8496d..662ba74 100644 --- a/src/main/resources/templates/migrate.ftl +++ b/src/main/resources/templates/migrate.ftl @@ -4,6 +4,7 @@

<#if msg=='done'>✅ 迁移完成（成功导入 ${imported!'0'} 用户，详见任务记录） <#elseif msg=='gdone'>✅ 群迁移完成（迁移 ${migrated!'0'} 群，详见任务记录） + <#elseif msg=='gmdone'>✅ 群消息迁移完成（导入 ${imported!'0'} 条，详见任务记录） <#elseif msg=='mdone'>✅ 消息迁移完成（导入 ${imported!'0'} 条，详见任务记录） <#elseif msg=='rdone'>✅ 关系链迁移完成（导入 ${imported!'0'} 用户好友，详见任务记录） <#elseif msg=='empty'>❌ 用户清单为空（每行一个 UserID，# 开头为注释） @@ -53,6 +54,27 @@
+
+
群消息迁移
+

上传/粘贴老应用群 ID 清单（每行一个），选老应用触发。逐群 getGroupMsg（老密钥）分页拉历史 → 加前缀 from → importGroupMsg（主密钥）。群 ID 需已迁移（先做群迁移）。

+
+
+
+ + +
+
+
+
+ +
+
+
单聊消息迁移

上传会话清单（每行 from,to 逗号或空格分隔，# 注释）或粘贴，选老应用后触发。getRoamMsg（老密钥）拉 → 加前缀 → importMsg（主密钥，不计未读）。v1 每会话拉最近 100 条，断点续传按会话索引。用户需已迁移。