From 7ca5bd429030099e6e53c16a36deddca3e88a5c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=BF=9E=E9=BE=99=E5=88=9A?= Date: Fri, 10 Jul 2026 11:21:49 +0800 Subject: [PATCH] =?UTF-8?q?feat(t17):=20=E5=8D=95=E8=81=8A=E5=90=8C?= =?UTF-8?q?=E6=AD=A5+=E8=A1=A5=E6=8B=89=E5=AF=B9=E8=B4=A6+=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E6=B6=88=E6=81=AF=E8=BF=9B=E5=88=86=E5=8F=91=EF=BC=9B?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20get=5Froam=5Fmsg=2060008?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增单聊(C2C)同步、补拉对账定时任务，同步消息纳入分发队列推送业务系统。 1.单聊同步④:user遍历→recentcontact/get_list收集C2C会话→会话对去重→ get_roam_msg时间窗向前滚动全量→落库(conv_type=1/SYNC/进分发) 2.补拉对账⑤:SyncCheckTask定时(1h)+手动按钮，重跑群+单聊幂等补缺失 3.同步消息进分发:群/单聊distStatus=0入DistQueue；限速分批入队 (next_retry_at按序号错开delay=seq/dispatchQps) 4.重构:enqueue抽到DispatchService公共方法，Pull/Sync共用 修复(get_roam_msg一直60008静默失败,C2C补拉从未工作): -命令字 openim_admin/get_roam_msg → openim/admin_getroammsg -字段 From/To/MaxTimeInterval → Operator_Account/Peer_Account/MaxTime -新增 getRoamMsgRangeAs(MinTime/MaxTime+LastMsgKey续拉)+getRecentContactListAs 联调(sa 16用户):单聊链路跑通(get_list32+admin_getroammsg OK/Complete=1), 幂等正常,补拉定时工作,无死循环。enqueue实证待配callbackUrl+新消息。 Co-Authored-By: Claude --- .../imutil/controller/AdminController.java | 26 ++ .../com/imutil/service/DispatchService.java | 22 ++ .../java/com/imutil/service/SyncService.java | 25 +- .../service/impl/DispatchServiceImpl.java | 54 ++++ .../imutil/service/impl/PullServiceImpl.java | 44 +-- .../imutil/service/impl/SyncServiceImpl.java | 257 +++++++++++++++++- .../java/com/imutil/task/SyncCheckTask.java | 59 ++++ .../com/imutil/tencent/TencentImClient.java | 71 ++++- src/main/resources/templates/sync.ftl | 34 ++- 9 files changed, 536 insertions(+), 56 deletions(-) create mode 100644 src/main/java/com/imutil/task/SyncCheckTask.java diff --git a/src/main/java/com/imutil/controller/AdminController.java b/src/main/java/com/imutil/controller/AdminController.java index d951b03..f08281c 100644 --- a/src/main/java/com/imutil/controller/AdminController.java +++ b/src/main/java/com/imutil/controller/AdminController.java @@ -634,6 +634,32 @@ public class AdminController { ctx.redirect(basePath + "/admin/sync?msg=smdone"); } + /** 同步单聊消息：需先同步用户（群成员） */ + @Post + @Mapping("/sync/c2cmsg") + public void syncC2CMsg(@Param(required = false) Long sourceAppId, + @Param(defaultValue = "") String tenantId, + Context ctx) throws Throwable { + if (sourceAppId == null && (tenantId == null || tenantId.isEmpty())) { + ctx.redirect(basePath + "/admin/sync?msg=syncopt"); + return; + } + syncService.syncC2CMessages(sourceAppId, tenantId); + ctx.redirect(basePath + "/admin/sync?msg=scdone"); + } + + /** 补拉对账：重跑群+单聊（幂等补缺失），仅主应用 + 指定租户 */ + @Post + @Mapping("/sync/check") + public void syncCheck(@Param String tenantId, Context ctx) throws Throwable { + if (tenantId == null || tenantId.isEmpty()) { + ctx.redirect(basePath + "/admin/sync?msg=syncopt"); + return; + } + syncService.checkAndPull(tenantId); + ctx.redirect(basePath + "/admin/sync?msg=chkdone"); + } + // ==================== 公共：构造页面模型 ==================== /** diff --git a/src/main/java/com/imutil/service/DispatchService.java b/src/main/java/com/imutil/service/DispatchService.java index cd3b96e..a843ce6 100644 --- a/src/main/java/com/imutil/service/DispatchService.java +++ b/src/main/java/com/imutil/service/DispatchService.java @@ -1,7 +1,9 @@ package com.imutil.service; import com.imutil.entity.DistQueue; +import com.imutil.entity.ImMessage; +import java.time.OffsetDateTime; import java.util.List; /** @@ -32,4 +34,24 @@ public interface DispatchService { * @return 重置条数 */ int recoverStuck(); + + /** + * 写入分发队列（默认 nextRetryAt=now，立即可被消费分发） + *

+ * 取租户 callbackUrl 作为目标；租户未配置 callbackUrl 则跳过（不报错，本地留底仍有效）。 + * + * @param tenantId 归属租户 + * @param msg 已落库消息（payload 取其 msgKey/convType/convId/from/to/msgTime/msgType/msgBody/source） + */ + void enqueue(String tenantId, ImMessage msg); + + /** + * 写入分发队列并指定下次分发时间（限速平滑分发用） + *

+ * 同步批量入队时按序号错开 nextRetryAt（delay = seq / dispatchQps 秒）， + * DispatchWorker 按时间消费，避免历史消息瞬时打爆业务系统 callbackUrl。 + * + * @param nextRetryAt 下次可分发时间 + */ + void enqueue(String tenantId, ImMessage msg, OffsetDateTime nextRetryAt); } diff --git a/src/main/java/com/imutil/service/SyncService.java b/src/main/java/com/imutil/service/SyncService.java index ee7fb64..61e5873 100644 --- a/src/main/java/com/imutil/service/SyncService.java +++ b/src/main/java/com/imutil/service/SyncService.java @@ -9,8 +9,9 @@ import com.imutil.entity.MigrateTask; * 本服务定位为只读拉取：把指定腾讯 IM 应用里的群组/群成员/群消息拉到本地表， * 供管理后台按租户查看，不在腾讯侧产生任何写操作。 *

- * 腾讯 IM 现实约束：无"全量用户列表"接口（用户靠群成员反推）；无"全量 C2C 会话列表"接口 - * （单聊历史消息本轮不做全量，回调增量已由 {@link com.imutil.service.PullService} 覆盖）。 + * 腾讯 IM 现实约束：无"全量用户列表"接口（用户靠群成员反推）；C2C 单聊会话通过 + * recentcontact/get_list 按"已同步用户"反查（{@link #syncC2CMessages}），漫游消息用 get_roam_msg 全量翻页。 + * 同步进来的消息(群/单聊)均入分发队列(dist_status=0)推送业务系统，本地同时留底。 * * @author imutil */ @@ -33,9 +34,27 @@ public interface SyncService { MigrateTask syncGroupMembers(Long sourceAppId, String tenantId); /** - * 同步群消息：遍历该租户已同步的群，逐群 getGroupMsg + IsFinished 滚动全量 → 写 im_message(source=SYNC) + * 同步群消息：遍历该租户已同步的群，逐群 getGroupMsg + IsFinished 滚动全量 → 写 im_message(source=SYNC，进分发) * * @see #syncGroups(Long, String) 先同步群组，群消息才有目标群清单 */ MigrateTask syncGroupMessages(Long sourceAppId, String tenantId); + + /** + * 同步单聊(C2C)消息：遍历该租户已同步用户 → recentcontact/get_list 收集 C2C 会话 → + * 会话对去重 → get_roam_msg 时间窗向前滚动全量 → 写 im_message(conv_type=1, source=SYNC, 进分发)。 + *

+ * 腾讯无全量 C2C 会话列表接口，靠"已同步用户"反查其最近会话；未导入用户返回 50001 跳过。 + * + * @see #syncGroupMembers(Long, String) 先同步用户(群成员)，单聊才有用户清单可遍历 + */ + MigrateTask syncC2CMessages(Long sourceAppId, String tenantId); + + /** + * 补拉对账：重跑群消息 + 单聊同步（幂等，本地已有的跳过、缺失的补入），确保本系统数据最全。 + * 定时任务 {@link com.imutil.task.SyncCheckTask} 驱动，也可手动触发。 + * + * @param tenantId 归属租户（补拉对账仅对主应用归属租户有意义） + */ + MigrateTask checkAndPull(String tenantId); } diff --git a/src/main/java/com/imutil/service/impl/DispatchServiceImpl.java b/src/main/java/com/imutil/service/impl/DispatchServiceImpl.java index 08900ce..4d36178 100644 --- a/src/main/java/com/imutil/service/impl/DispatchServiceImpl.java +++ b/src/main/java/com/imutil/service/impl/DispatchServiceImpl.java @@ -2,9 +2,13 @@ package com.imutil.service.impl; import com.imutil.common.Httpx; import com.imutil.entity.DistQueue; +import com.imutil.entity.ImMessage; +import com.imutil.entity.Tenant; import com.imutil.mapper.DistQueueMapper; import com.imutil.service.DispatchService; +import com.imutil.service.TenantService; import lombok.extern.slf4j.Slf4j; +import org.noear.snack4.ONode; import org.noear.solon.annotation.Component; import org.noear.solon.annotation.Inject; import org.noear.solon.data.annotation.Tran; @@ -29,6 +33,9 @@ public class DispatchServiceImpl implements DispatchService { @Inject private DistQueueMapper distQueueMapper; + @Inject + private TenantService tenantService; + @Inject("${imutil.dispatch.fetchBatch:50}") private int fetchBatch; @@ -91,4 +98,51 @@ public class DispatchServiceImpl implements DispatchService { OffsetDateTime threshold = now.minusMinutes(lockTimeoutMin); return distQueueMapper.recoverStuck(threshold, now); } + + @Override + public void enqueue(String tenantId, ImMessage msg) { + enqueue(tenantId, msg, OffsetDateTime.now()); + } + + /** + * 写入分发队列（复用回调分发链路，payload 为结构化 JSON） + *

+ * 取租户 callbackUrl 作为目标；未配置则跳过（不报错，本地留底仍有效）。 + * nextRetryAt 由调用方控制：同步批量入队时按序号错开以限速平滑分发。 + */ + @Override + public void enqueue(String tenantId, ImMessage msg, OffsetDateTime nextRetryAt) { + Tenant t = tenantService.getById(tenantId); + if (t == null || t.getCallbackUrl() == null || t.getCallbackUrl().isEmpty()) { + return; + } + ONode payload = ONode.ofJson("{}"); + payload.set("source", msg.getSource()); + payload.set("msgKey", msg.getMsgKey()); + payload.set("tenantId", tenantId); + payload.set("convType", msg.getConvType()); + payload.set("convId", msg.getConvId()); + payload.set("fromAccount", msg.getFromAccount()); + payload.set("toAccount", msg.getToAccount()); + payload.set("groupId", msg.getGroupId()); + payload.set("msgTime", msg.getMsgTime()); + payload.set("msgType", msg.getMsgType()); + if (msg.getMsgBody() != null) { + try { + payload.set("msgBody", ONode.ofJson(msg.getMsgBody())); + } catch (Exception e) { + payload.set("msgBody", msg.getMsgBody()); + } + } + DistQueue q = new DistQueue(); + q.setMsgKey(msg.getMsgKey()); + q.setTenantId(tenantId); + q.setConvId(msg.getConvId()); + q.setTargetUrl(t.getCallbackUrl()); + q.setPayload(payload.toJson()); + q.setStatus(0); + q.setRetryCount(0); + q.setNextRetryAt(nextRetryAt); + distQueueMapper.insert(q); + } } diff --git a/src/main/java/com/imutil/service/impl/PullServiceImpl.java b/src/main/java/com/imutil/service/impl/PullServiceImpl.java index 121a64b..b961991 100644 --- a/src/main/java/com/imutil/service/impl/PullServiceImpl.java +++ b/src/main/java/com/imutil/service/impl/PullServiceImpl.java @@ -10,6 +10,7 @@ import com.imutil.entity.Tenant; import com.imutil.mapper.DistQueueMapper; import com.imutil.mapper.ImMessageMapper; import com.imutil.mapper.PullWatermarkMapper; +import com.imutil.service.DispatchService; import com.imutil.service.PullService; import com.imutil.service.TenantService; import com.imutil.tencent.TencentImClient; @@ -56,6 +57,9 @@ public class PullServiceImpl implements PullService { @Inject private TenantService tenantService; + @Inject + private DispatchService dispatchService; + @Inject("${imutil.pull.convsPerRound:20}") private int convsPerRound; @@ -168,50 +172,12 @@ public class PullServiceImpl implements PullService { imMessageMapper.insert(msg); newCount++; // 仅新插入的消息入队分发 - enqueue(wm.getTenantId(), msg); + dispatchService.enqueue(wm.getTenantId(), msg); } advanceWatermark(wm, maxSeq, maxTime); return newCount; } - /** - * 写入分发队列（复用回调分发链路，payload 用结构化 JSON） - */ - private void enqueue(String tenantId, ImMessage msg) { - Tenant t = tenantService.getById(tenantId); - if (t == null || t.getCallbackUrl() == null || t.getCallbackUrl().isEmpty()) { - return; - } - ONode payload = ONode.ofJson("{}"); - payload.set("source", msg.getSource()); - payload.set("msgKey", msg.getMsgKey()); - payload.set("tenantId", tenantId); - payload.set("convType", msg.getConvType()); - payload.set("convId", msg.getConvId()); - payload.set("fromAccount", msg.getFromAccount()); - payload.set("toAccount", msg.getToAccount()); - payload.set("groupId", msg.getGroupId()); - payload.set("msgTime", msg.getMsgTime()); - payload.set("msgType", msg.getMsgType()); - if (msg.getMsgBody() != null) { - try { - payload.set("msgBody", ONode.ofJson(msg.getMsgBody())); - } catch (Exception e) { - payload.set("msgBody", msg.getMsgBody()); - } - } - DistQueue q = new DistQueue(); - q.setMsgKey(msg.getMsgKey()); - q.setTenantId(tenantId); - q.setConvId(msg.getConvId()); - q.setTargetUrl(t.getCallbackUrl()); - q.setPayload(payload.toJson()); - q.setStatus(0); - q.setRetryCount(0); - q.setNextRetryAt(OffsetDateTime.now()); - distQueueMapper.insert(q); - } - /** * 解析群消息节点为 ImMessage */ diff --git a/src/main/java/com/imutil/service/impl/SyncServiceImpl.java b/src/main/java/com/imutil/service/impl/SyncServiceImpl.java index 1797a9a..225182d 100644 --- a/src/main/java/com/imutil/service/impl/SyncServiceImpl.java +++ b/src/main/java/com/imutil/service/impl/SyncServiceImpl.java @@ -15,6 +15,7 @@ import com.imutil.mapper.MigrateTaskMapper; import com.imutil.mapper.SourceAppMapper; import com.imutil.mapper.TenantMapper; import com.imutil.mapper.UserMappingMapper; +import com.imutil.service.DispatchService; import com.imutil.service.SyncService; import com.imutil.tencent.TencentImClient; import lombok.extern.slf4j.Slf4j; @@ -25,7 +26,10 @@ import org.noear.solon.annotation.Inject; import java.time.Instant; import java.time.OffsetDateTime; import java.time.ZoneId; +import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; /** * 数据同步服务实现（T17） @@ -89,6 +93,21 @@ public class SyncServiceImpl implements SyncService { @Inject("${imutil.sync.maxGroups:500}") private int maxGroups; + /** 同步入队分发限速（每秒入队条数，保护业务系统 callbackUrl；错开 next_retry_at 平滑分发） */ + @Inject("${imutil.sync.dispatchQps:10}") + private int syncDispatchQps; + + @Inject + private DispatchService dispatchService; + + /** 单聊同步用户上限（防海量用户 N×M 爆配额；如需全量调大） */ + @Inject("${imutil.sync.maxSyncUsers:200}") + private int maxSyncUsers; + + /** 单聊漫游回溯天数（腾讯 C2C 漫游存储期，默认 7 天，套餐更长可调大） */ + @Inject("${imutil.sync.c2cLookbackDays:7}") + private int c2cLookbackDays; + /** 数据源密钥解析结果 */ private static class SyncCtx { long sdkAppId; @@ -366,6 +385,7 @@ public class SyncServiceImpl implements SyncService { migrateTaskMapper.updateById(task); int imported = 0; + OffsetDateTime syncBase = OffsetDateTime.now(); for (int gi = 0; gi < groups.size(); gi++) { String groupId = groups.get(gi).getImGroupId(); try { @@ -399,6 +419,8 @@ public class SyncServiceImpl implements SyncService { } imMessageMapper.insert(msg); imported++; + // 限速入队分发:按序号错开 next_retry_at，DispatchWorker 按速率消费 + dispatchService.enqueue(ctx.tenantId, msg, nextRetryAt(imported, syncBase)); } boolean finished = root.get("IsFinished").getLong() == 1; // 终止：已拉完 / 空页 / 本页无有效 seq @@ -432,7 +454,7 @@ public class SyncServiceImpl implements SyncService { } /** - * 解析腾讯群消息节点为 ImMessage（source=SYNC，dist_status=1 不进分发） + * 解析腾讯群消息节点为 ImMessage（source=SYNC，dist_status=0 进分发） */ private ImMessage parseGroupMsgForSync(ONode m, String tenantId, String groupId) { // 过滤占位/系统消息（IsPlaceMsg=1：From_Account 与 MsgBody 均空，非真实消息，不入库） @@ -456,10 +478,229 @@ public class SyncServiceImpl implements SyncService { msg.setMsgBody(body.toString()); msg.setSource("SYNC"); msg.setIsCrossTenant(false); - msg.setDistStatus(1); + msg.setDistStatus(0); // 进分发 return msg; } + // ==================== 同步单聊(C2C) ==================== + + /** + * 同步单聊消息：遍历已同步用户 → recentcontact/get_list 收集 C2C 会话 → 会话对去重 → + * get_roam_msg 时间窗向前滚动全量 → 落库(conv_type=1/source=SYNC/进分发)。 + */ + @Override + public MigrateTask syncC2CMessages(Long sourceAppId, String tenantId) { + SyncCtx ctx = resolveCtx(sourceAppId, tenantId); + MigrateTask task = startTask(ctx, "C2C_SYNC"); + try { + List users = userMappingMapper.selectList(Wrappers.lambdaQuery() + .eq(UserMapping::getTenantId, ctx.tenantId)); + task.setTotalCount((long) users.size()); + migrateTaskMapper.updateById(task); + + Set seenPairs = new HashSet<>(); + int imported = 0; + OffsetDateTime syncBase = OffsetDateTime.now(); + int userCnt = 0; + for (UserMapping u : users) { + if (++userCnt > maxSyncUsers) { + log.warn("单聊同步用户达上限 {}，停止（调大 imutil.sync.maxSyncUsers 可继续）", maxSyncUsers); + break; + } + String self = u.getImUserId(); + if (self == null || self.isEmpty()) { + continue; + } + try { + List peers = collectC2CPeers(self, ctx); + for (String peer : peers) { + if (peer == null || peer.isEmpty() || peer.equals(self)) { + continue; + } + // 会话对去重：A↔B 只拉一次 + if (!seenPairs.add(pairKey(self, peer))) { + continue; + } + imported += pullRoamMsgs(self, peer, ctx, syncBase, imported); + } + } catch (Exception ue) { + // 50001(用户未导入)等按用户级跳过 + log.warn("单聊同步用户失败 {} : {}", self, ue.toString()); + } + task.setProcessedCount((long) imported); + task.setPosCursor("user:" + userCnt); + migrateTaskMapper.updateById(task); + } + task.setStatus(3); + task.setFinishedAt(OffsetDateTime.now()); + task.setVerifyResult(String.format("{\"users\":%d,\"pairs\":%d,\"imported\":%d}", + Math.min(userCnt, maxSyncUsers), seenPairs.size(), imported)); + migrateTaskMapper.updateById(task); + log.info("单聊同步完成 tenant={} users={} pairs={} imported={}", + ctx.tenantId, Math.min(userCnt, maxSyncUsers), seenPairs.size(), imported); + return task; + } catch (Exception e) { + failTask(task, e); + log.error("单聊同步失败 tenant={}", ctx.tenantId, e); + throw e; + } + } + + /** + * 分页拉取用户最近会话，收集 C2C(Type=1) 对端账号 + */ + private List collectC2CPeers(String self, SyncCtx ctx) { + List peers = new ArrayList<>(); + long timeStamp = 0; + long startIndex = 0; + int guard = 0; + while (guard++ < 1000) { + String resp = tencentImClient.getRecentContactListAs(self, timeStamp, startIndex, + ctx.sdkAppId, ctx.secretKey); + ONode root = ONode.ofJson(resp); + if (!"OK".equalsIgnoreCase(root.get("ActionStatus").getString())) { + break; + } + ONode items = root.get("SessionItem"); + int n = items.size(); + for (int i = 0; i < n; i++) { + ONode item = items.get(i); + if (item.get("Type").getLong() == 1) { // 1=C2C + String to = item.get("To_Account").getString(); + if (to != null && !to.isEmpty()) { + peers.add(to); + } + } + } + long complete = root.get("CompleteFlag").getLong(); + if (complete == 1 || n == 0) { + break; + } + long ts = root.get("TimeStamp").getLong(); + long si = root.get("StartIndex").getLong(); + if (ts == timeStamp && si == startIndex) { + break; // 游标未推进防死循环 + } + timeStamp = ts; + startIndex = si; + } + return peers; + } + + /** + * 全量拉取两人漫游消息（向旧翻页），落库 + 限速入队分发，返回新增条数 + */ + private int pullRoamMsgs(String self, String peer, SyncCtx ctx, OffsetDateTime syncBase, int seqBase) { + long minTime = Instant.now().getEpochSecond() - c2cLookbackDays * 86400L; + long maxTime = Instant.now().getEpochSecond(); + String lastMsgKey = null; + int imported = 0; + int pages = 0; + while (pages++ < groupMsgMaxPages) { + String resp = tencentImClient.getRoamMsgRangeAs(self, peer, groupMsgPerPage, minTime, maxTime, + lastMsgKey, ctx.sdkAppId, ctx.secretKey); + ONode root = ONode.ofJson(resp); + if (!"OK".equalsIgnoreCase(root.get("ActionStatus").getString())) { + break; + } + ONode list = root.get("MsgList"); + int n = list.size(); + for (int i = 0; i < n; i++) { + ONode m = list.get(i); + ImMessage msg = parseC2CMsgForSync(m, ctx.tenantId, self, peer); + if (msg == null) { + continue; + } + long cnt = imMessageMapper.selectCount(Wrappers.lambdaQuery() + .eq(ImMessage::getMsgKey, msg.getMsgKey())); + if (cnt > 0) { + continue; + } + imMessageMapper.insert(msg); + imported++; + dispatchService.enqueue(ctx.tenantId, msg, nextRetryAt(seqBase + imported, syncBase)); + } + long complete = root.get("Complete").getLong(); + long lastMsgTime = root.get("LastMsgTime").getLong(); + String nextLastMsgKey = root.get("LastMsgKey").getString(); + // 终止:已拉完 / 空页 / 无 LastMsgTime + if (complete == 1 || n == 0 || lastMsgTime <= 0) { + break; + } + // 游标未推进防死循环（LastMsgTime 未向旧推进且 LastMsgKey 未变） + if (lastMsgTime >= maxTime && (nextLastMsgKey == null || nextLastMsgKey.equals(lastMsgKey))) { + break; + } + maxTime = lastMsgTime; + lastMsgKey = nextLastMsgKey; + } + return imported; + } + + /** + * 解析腾讯单聊消息节点为 ImMessage（source=SYNC，进分发） + */ + private ImMessage parseC2CMsgForSync(ONode m, String tenantId, String from, String to) { + long msgSeq = m.get("MsgSeq").getLong(); + long msgRandom = m.get("MsgRandom").getLong(); + long msgTs = m.get("MsgTimeStamp").getLong(); + String fromAcc = m.get("From_Account").getString(); + if (fromAcc == null || fromAcc.isEmpty()) { + fromAcc = from; + } + ONode body = m.get("MsgBody"); + ImMessage msg = new ImMessage(); + msg.setMsgKey(MsgKeys.build(fromAcc, to, msgSeq, msgRandom)); + msg.setTenantId(tenantId); + msg.setMsgTime(toTime(msgTs)); + msg.setConvType(1); + msg.setConvId(to); + msg.setFromAccount(fromAcc); + msg.setToAccount(to); + msg.setMsgType(firstBodyType(body)); + msg.setMsgBody(body.toString()); + msg.setSource("SYNC"); + msg.setIsCrossTenant(false); + msg.setDistStatus(0); // 进分发 + return msg; + } + + /** 会话对去重键：min#max，使 A↔B 只拉一次 */ + private String pairKey(String a, String b) { + return (a.compareTo(b) <= 0) ? a + "#" + b : b + "#" + a; + } + + // ==================== 补拉对账 ==================== + + /** + * 补拉对账：顺序重跑群消息 + 单聊同步（幂等，本地已有的跳过、缺失的补入）。 + * 供定时任务 SyncCheckTask 与后台手动按钮调用，确保本系统数据最全。 + */ + @Override + public MigrateTask checkAndPull(String tenantId) { + SyncCtx ctx = resolveCtx(null, tenantId); + MigrateTask task = startTask(ctx, "CHECK_PULL"); + try { + task.setPosCursor("start"); + migrateTaskMapper.updateById(task); + log.info("补拉对账开始 tenant={}", tenantId); + MigrateTask g = syncGroupMessages(null, tenantId); + MigrateTask c = syncC2CMessages(null, tenantId); + long gImp = g.getProcessedCount() == null ? 0 : g.getProcessedCount(); + long cImp = c.getProcessedCount() == null ? 0 : c.getProcessedCount(); + task.setStatus(3); + task.setFinishedAt(OffsetDateTime.now()); + task.setVerifyResult(String.format("{\"groupImported\":%d,\"c2cImported\":%d}", gImp, cImp)); + migrateTaskMapper.updateById(task); + log.info("补拉对账完成 tenant={} groupImported={} c2cImported={}", tenantId, gImp, cImp); + return task; + } catch (Exception e) { + failTask(task, e); + log.error("补拉对账失败 tenant={}", tenantId, e); + throw e; + } + } + // ==================== 工具方法 ==================== private void failTask(MigrateTask task, Exception e) { @@ -491,4 +732,16 @@ public class SyncServiceImpl implements SyncService { } return OffsetDateTime.ofInstant(Instant.ofEpochSecond(epochSecond), ZoneId.systemDefault()); } + + /** + * 计算限速入队的 next_retry_at：按序号错开（delay = seq / dispatchQps 秒）， + * 使同步进来的历史消息按速率平滑分发，避免瞬时打爆业务系统 callbackUrl。 + * dispatchQps<=0 时不限速，立即 next_retry_at=now。 + */ + private OffsetDateTime nextRetryAt(long seq, OffsetDateTime base) { + if (syncDispatchQps <= 0) { + return OffsetDateTime.now(); + } + return base.plusSeconds(seq / syncDispatchQps); + } } diff --git a/src/main/java/com/imutil/task/SyncCheckTask.java b/src/main/java/com/imutil/task/SyncCheckTask.java new file mode 100644 index 0000000..899c496 --- /dev/null +++ b/src/main/java/com/imutil/task/SyncCheckTask.java @@ -0,0 +1,59 @@ +package com.imutil.task; + +import com.imutil.entity.Tenant; +import com.imutil.mapper.TenantMapper; +import com.imutil.service.SyncService; +import lombok.extern.slf4j.Slf4j; +import org.noear.solon.annotation.Component; +import org.noear.solon.annotation.Inject; +import org.noear.solon.scheduling.annotation.Scheduled; + +import java.util.List; + +/** + * 数据同步补拉对账定时任务 + *

+ * 定期重跑群消息 + 单聊同步（幂等，本地已有的跳过、缺失的补入）， + * 兜底腾讯侧有而本系统暂无的消息，确保本系统数据最全、业务系统不漏收。 + * 对应 app.yml 的 solon.scheduling.job.syncCheckJob。 + *

+ * 遍历所有租户，对每个用主应用密钥执行 checkAndPull（主应用数据为主）。 + * + * @author imutil + */ +@Slf4j +@Component +public class SyncCheckTask { + + @Inject + private SyncService syncService; + + @Inject + private TenantMapper tenantMapper; + + /** + * 由 app.yml syncCheckJob 驱动（默认 fixedDelay=1 小时） + */ + @Scheduled(name = "syncCheckJob") + public void run() { + List tenants = tenantMapper.selectList(null); + if (tenants == null || tenants.isEmpty()) { + return; + } + int ok = 0; + for (Tenant t : tenants) { + String tid = t.getTenantId(); + if (tid == null || tid.isEmpty()) { + continue; + } + try { + syncService.checkAndPull(tid); + ok++; + } catch (Throwable e) { + // 单租户异常不影响其他租户 + log.error("补拉对账定时任务异常 tenant={}", tid, e); + } + } + log.info("补拉对账定时任务完成 租户数={} 成功={}", tenants.size(), ok); + } +} diff --git a/src/main/java/com/imutil/tencent/TencentImClient.java b/src/main/java/com/imutil/tencent/TencentImClient.java index 1cd11e0..3999e67 100644 --- a/src/main/java/com/imutil/tencent/TencentImClient.java +++ b/src/main/java/com/imutil/tencent/TencentImClient.java @@ -256,7 +256,7 @@ public class TencentImClient { /** * 查询单聊历史漫游消息（C2C） *

- * 命令字 openim_admin/get_roam_msg，按时间窗返回指定两人之间最近的 MaxCnt 条消息。 + * 命令字 openim/admin_getroammsg，按时间窗返回指定两人之间最近的 MaxCnt 条消息。 * 补拉服务据此增量补全回调丢失的单聊消息。 * * @param fromAccount 发送方 IM 账号 @@ -268,12 +268,13 @@ public class TencentImClient { */ public String getRoamMsg(String fromAccount, String toAccount, int maxCnt, long minTime, long maxInterval) { Map body = new HashMap<>(); - body.put("From_Account", fromAccount); - body.put("To_Account", toAccount); + // 腾讯 get_roam_msg 请求字段为 Operator_Account/Peer_Account + MinTime/MaxTime（无 MaxTimeInterval） + body.put("Operator_Account", fromAccount); + body.put("Peer_Account", toAccount); body.put("MaxCnt", maxCnt); body.put("MinTime", minTime); - body.put("MaxTimeInterval", maxInterval); - return callApi("openim_admin/get_roam_msg", Jsons.stringify(body)); + body.put("MaxTime", minTime + maxInterval); + return callApi("openim/admin_getroammsg", Jsons.stringify(body)); } /** @@ -301,12 +302,12 @@ public class TencentImClient { 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("Operator_Account", fromAccount); + body.put("Peer_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); + body.put("MaxTime", minTime + maxInterval); + return callApiAs("openim/admin_getroammsg", Jsons.stringify(body), srcSdkAppId, srcSecretKey); } /** @@ -325,6 +326,58 @@ public class TencentImClient { return callApiAs("group_open_http_svc/group_msg_get_simple", Jsons.stringify(body), srcSdkAppId, srcSecretKey); } + /** + * 拉取用户最近会话列表（含 C2C 单聊会话）— 单聊同步用 + *

+ * 命令字 recentcontact/get_list，分页返回 From_Account 的会话 SessionItem： + * Type=1 为 C2C（带 To_Account），Type=2 为群（带 GroupId）。 + * 单聊同步据此发现用户的 C2C 会话对，再配 getRoamMsgRangeAs 全量拉取漫游消息。 + * + * @param fromAccount 本端账号 + * @param timeStamp 分页时间戳（首轮 0；后续取上轮返回的 TimeStamp） + * @param startIndex 分页起始索引（首轮 0；后续取上轮返回的 StartIndex） + * @return 腾讯响应原始 JSON（含 SessionItem[]/CompleteFlag/TimeStamp/StartIndex） + */ + public String getRecentContactListAs(String fromAccount, long timeStamp, long startIndex, + long srcSdkAppId, String srcSecretKey) { + Map body = new HashMap<>(); + body.put("From_Account", fromAccount); + body.put("TimeStamp", timeStamp); + body.put("StartIndex", startIndex); + body.put("TopTimeStamp", 0); + body.put("AssistFlags", 15); + return callApiAs("recentcontact/get_list", Jsons.stringify(body), srcSdkAppId, srcSecretKey); + } + + /** + * 查询单聊漫游消息（按时间区间，全量续拉用）— 单聊同步用 + *

+ * 腾讯 openim/admin_getroammsg 请求字段为 Operator_Account/Peer_Account + MinTime/MaxTime， + * 续拉：MaxTime 取上轮返回 LastMsgTime，并带 LastMsgKey，直到 Complete=1。 + * + * @param operatorAccount 会话一方（本端） + * @param peerAccount 会话另一方 + * @param maxCnt 单次条数上限 + * @param minTime 区间下界（秒级 epoch） + * @param maxTime 区间上界（秒级 epoch；首轮 now，续拉取上轮 LastMsgTime） + * @param lastMsgKey 上轮返回 LastMsgKey（首轮 null） + * @return 腾讯响应原始 JSON（含 MsgList/Complete/LastMsgTime/LastMsgKey） + */ + public String getRoamMsgRangeAs(String operatorAccount, String peerAccount, int maxCnt, + long minTime, long maxTime, String lastMsgKey, + long srcSdkAppId, String srcSecretKey) { + Map body = new HashMap<>(); + body.put("Operator_Account", operatorAccount); + body.put("Peer_Account", peerAccount); + body.put("MaxCnt", maxCnt); + body.put("MinTime", minTime); + body.put("MaxTime", maxTime); + if (lastMsgKey != null && !lastMsgKey.isEmpty()) { + body.put("LastMsgKey", lastMsgKey); + } + return callApiAs("openim/admin_getroammsg", Jsons.stringify(body), srcSdkAppId, srcSecretKey); + } + // ==================== 群迁移相关（T14 S3） ==================== /** diff --git a/src/main/resources/templates/sync.ftl b/src/main/resources/templates/sync.ftl index 3069f2b..b0da7db 100644 --- a/src/main/resources/templates/sync.ftl +++ b/src/main/resources/templates/sync.ftl @@ -5,6 +5,8 @@ <#if msg=='sgdone'>✅ 群组同步完成（详见任务记录） <#elseif msg=='sudone'>✅ 群成员/用户同步完成（详见任务记录） <#elseif msg=='smdone'>✅ 群消息同步完成（详见任务记录） + <#elseif msg=='scdone'>✅ 单聊消息同步完成（详见任务记录） + <#elseif msg=='chkdone'>✅ 补拉对账完成（详见任务记录） <#elseif msg=='syncopt'>❌ 请选择数据源（主应用时须同时选归属租户） <#else>${msg} @@ -14,8 +16,8 @@

说明 ：本页为只读拉取，把腾讯 IM 数据同步到本地表用于查看，不在腾讯侧建群/建号。 - 群组=get_appid_group_list 全量分页；群成员(用户)=遍历群取成员（腾讯无全量用户接口，靠群成员反推）；群消息=逐群全量拉取。 - C2C 单聊消息无全量会话列表接口，历史单聊请用「数据迁移」页或依赖回调增量。 + 群组=get_appid_group_list 全量分页；群成员(用户)=遍历群取成员（腾讯无全量用户接口，靠群成员反推）；群消息=逐群全量拉取；单聊=recentcontact/get_list 按已同步用户反查会话 + get_roam_msg 全量翻页。 + 所有同步消息均入分发队列推送业务系统（限速分批入队），本地同时留底。
<#-- 公共：数据源 + 归属租户选择（三个表单各自携带） --> @@ -60,13 +62,39 @@
③ 同步群消息
-

遍历该租户已同步的群，逐群 getGroupMsg + IsFinished 滚动全量 → 写 im_message（source=SYNC，不进分发队列）。请先执行「同步群组」。

+

遍历该租户已同步的群，逐群 getGroupMsg + IsFinished 滚动全量 → 写 im_message（source=SYNC，进分发队列，限速分批）。请先执行「同步群组」。

<@srcSelect/>
+
+
④ 同步单聊消息
+

遍历该租户已同步用户 → recentcontact/get_list 收集 C2C 会话 → 会话对去重 → get_roam_msg 全量翻页 → 写 im_message（conv_type=1，source=SYNC，进分发）。请先执行「同步群成员」。

+
+
<@srcSelect/>
+ +
+
+ +
+
⑤ 补拉对账
+

重跑群消息 + 单聊同步（幂等：本地已有的跳过、缺失的补入），确保本系统数据最全。仅主应用 + 指定租户，可定时自动或手动触发。

+
+
+ + +
+ +
+
+

同步/迁移任务记录

<#if tasks?has_content>