diff --git a/.gitignore b/.gitignore index b988dc2..098cb6a 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,6 @@ logs/ # 本地配置覆盖（密钥等敏感信息） app-env.yml + +# Claude 项目记忆（含密码/密钥等敏感信息，不进 git；基准在 .claude/memory/，由 /memory-sync 同步到会话目录） +.claude/memory/ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..1b8d0bb --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,82 @@ + +# tencent-im-util + +腾讯 IM 回调分发工具：接收腾讯 IM 的消息回调，落库后按租户/源应用分发。配套管理后台、用量统计、关系链/消息迁移等。 + +> 本文件为**项目级技术上下文**（行为准则见全局 `~/.claude/CLAUDE.md`）。密钥一律不写在此处，配置项以 `app.yml` 为准。 + +## 技术栈 + +- **框架**：Solon（`solon-parent`，非 Spring），Java 21，`solon-web` +- **鉴权**：Sa-Token（`sa-token-solon-plugin`） +- **持久层**：MyBatis-Plus（`mybatis-plus-extension-solon-plugin`）+ PostgreSQL + HikariCP +- **缓存/队列**：Redis（Jedis）+ Caffeine 本地缓存 +- **视图**：FreeMarker（`solon-view-freemarker`） +- **日志**：logback（`solon-logging-logback-jakarta`） +- **腾讯 IM**：`tls-sig-api-v2`（UserSig 签名）+ 自封装 `TencentImClient`（REST API） +- **定时**：`solon-scheduling-simple`；**工作线程**：自实现 `worker/DispatchWorker` + +## 构建与运行 + +```bash +# 打包（离线模式，跳过测试） +mvn -o package -DskipTests +# 运行 +java -jar target/tencent-im-util.jar +``` + +> ⚠️ **不要用 `mvn solon:solon`**（本项目下不可用，会失败）。详见 `.claude/memory/run-app-jar.md`。 +> 端口 `8092`，contextPath `/imutil`（即所有路由前缀 `/imutil`）。控制台中文需 UTF-8 终端（已在 `logback.xml` 修正 charset）。 + +## 目录结构（`src/main/java/com/imutil`） + +| 包 | 职责 | +|---|---| +| `controller` | `AdminController`(`/admin/*` 后台)、`CallbackController`(`/callback/im` 回调入口)、`HealthController`、`SigController`(UserSig 下发) | +| `service` + `service/impl` | 业务逻辑（接口 + 实现） | +| `entity` | MyBatis-Plus 实体：`Tenant`/`SourceApp`/`ImMessage`/`DistQueue`/`UsageStat`/`MigrateTask`/`UserMapping`/`GroupMapping`/`Recording`/`TrtcRoom`/`CrossTenantGrant`/`CrossTenantAudit`/`PullWatermark`/`ApiCallLog`/`AdminUser` | +| `mapper` | MyBatis 映射，含 `PartitionMapper`（PG 分区管理） | +| `filter` | `AdminAuthFilter`（后台鉴权）、`TenantAuthFilter`（租户鉴权）、`GlobalExceptionFilter`（全局异常） | +| `task` | 定时任务：`DispatchRecoverTask`（死信重投）、`PartitionCreateTask`（分区创建）、`PullCheckTask`、`UsageStatTask` | +| `tencent` | `TencentImClient`（REST API + 429 限速重试）、`TencentCallbackSign`（回调签名校验）、`UserSigUtil` | +| `worker` | `DispatchWorker`（消费分发队列） | +| `common` | 工具类：`PasswordUtil`(PBKDF2)、`RedisService`、`LocalCache`、`RateLimiter`、`Jsons`、`Httpx`、`Ids`、`MsgKeys`、`TenantContext`、`HealthService`、`BizException` | +| `model` | `Result`（统一响应体） | + +## 核心调用链 + +1. **回调分发（主链路）**：腾讯服务器 → `CallbackController#callback`(`/imutil/callback/im`) → `CallbackServiceImpl`（验签 + 落 `ImMessage` + 入 `DistQueue`）→ `DispatchWorker` 消费 → 分发到目标租户/源应用；失败由 `DispatchRecoverTask` 重投。 +2. **管理后台**：`AdminController`（Sa-Token 会话）+ FreeMarker 页面（`resources/templates/*.ftl`）：login / home / tenant / sourceapp / grant / queue / usage / password / messages / migrate。 +3. **数据迁移**：`MigrateService`（C2C 全量分页 + 群消息，`msgLookbackDays` 回溯天数，429 自动重试）。 +4. **跨租户授权**：`CrossTenantService`（`CrossTenantGrant` 授权 + `CrossTenantAudit` 审计）。 +5. **分区与统计**：`PartitionService`（按 `msg_time` 分区裁剪）+ `PartitionCreateTask`；`UsageStatService` + `UsageStatTask`。 + +## 项目约定（务必遵守） + +- **Long ID → String**：雪花 ID 等 `Long` 序列化给前端**必须** `String.valueOf()`，否则 JS 精度丢失（全局规则 10.1）。 +- **FreeMarker + java.time 坑**：ftl 里**禁止**对 `LocalDateTime`/`OffsetDateTime` 用 `?string(pattern)`（抛 `NonMethodException`）。改为 Controller 端 `DateTimeFormatter.format` 预格式化成 `xxxStr` 字段，ftl 用 `${obj.xxxStr!}`。详见 `.claude/memory/freemarker-java-time.md`。 +- **LocalDate / DateTimeFormatter**：pattern 含 `H/m/s` 时只能用 `LocalDateTime`，禁用 `LocalDate.now()`（全局规则 10.2）。 +- **中文注释**：所有注释用简体中文，标识符保持英文（全局规则 9）。 +- **密码**：`PasswordUtil` PBKDF2 哈希存 `admin_user.password_hash`（`iterations:salt:hash`）；改密走 `/admin/password`。 +- **配置不提交**：`app.yml` 含密钥（Redis 密码 / `tencent.secretKey` / admin 默认密码），长期处于未提交状态，**勿提交**；如需本地覆盖敏感值用 `app-env.yml`（已 gitignore）。 + +## 关键配置（`src/main/resources/app.yml`，值为准） + +- `server.port` / `server.contextPath` +- `solon.dataSources.db1`（PostgreSQL 连接） +- `imutil.redis`（host/port/password/database） +- `imutil.tencent`（`sdkAppId`/`secretKey`/`adminUserId`/`apiHost`/`usersigExpireDays`/`callbackToken`——`callbackToken` 留空则跳过回调签名校验，仅联调用） +- `imutil.admin`（`defaultUsername`/`defaultPassword`——仅首次初始化 `admin_user`） +- `imutil.migrate`（`msgLookbackDays` 等） +- `sa-token.*` + +## 文档与外部资源 + +- **实施记录与测试用例**：Obsidian 库 `8 腾讯IM&音视频分发\实施记录与测试用例.md`（项目外，按 T1–T16 任务编号组织；**不进 git**）。 +- **项目记忆**：`.claude/memory/`（已 gitignore，含账号/密码/密钥提醒等；索引见其 `MEMORY.md`）。召回失效时跑 `/memory-sync` 重新同步到会话目录。 +- **数据库操作**：本机无 `psql`，用 DBX MCP（`192.168.10.118-pg`，database `tencen_im`）查改数据。详见 `.claude/memory/pg-via-dbx.md`。 +- **回调测试**：端点 `/imutil/callback/im`，本地可用 curl 模拟（见 `.claude/memory/tencent-callback-test.md`）。 + +## 当前进度 + +核心功能 T1–T15 已完成（T12/T13 TRTC 音视频按需暂缓），T14 消息迁移 V1–V4、T16 消息记录查看已完成。详见实施记录文档与 `.claude/memory/git-author-and-progress.md`。 diff --git a/src/main/java/com/imutil/controller/AdminController.java b/src/main/java/com/imutil/controller/AdminController.java index 62e5c9b..d951b03 100644 --- a/src/main/java/com/imutil/controller/AdminController.java +++ b/src/main/java/com/imutil/controller/AdminController.java @@ -2,6 +2,7 @@ package com.imutil.controller; import cn.dev33.satoken.stp.StpUtil; import com.baomidou.mybatisplus.core.toolkit.Wrappers; +import com.imutil.common.Ids; import com.imutil.entity.AdminUser; import com.imutil.entity.CrossTenantGrant; import com.imutil.entity.DistQueue; @@ -14,10 +15,12 @@ import com.imutil.mapper.TenantMapper; import com.imutil.mapper.UsageStatMapper; import com.imutil.service.AdminUserService; import com.imutil.service.SourceAppService; +import com.imutil.service.TenantService; import com.imutil.tencent.TencentImClient; import com.imutil.entity.ImMessage; import com.imutil.entity.MigrateTask; import com.imutil.service.MigrateService; +import com.imutil.service.SyncService; import com.imutil.mapper.ImMessageMapper; import org.noear.solon.core.handle.UploadedFile; import java.nio.charset.StandardCharsets; @@ -61,6 +64,9 @@ public class AdminController { @Inject private TenantMapper tenantMapper; + @Inject + private TenantService tenantService; + @Inject private CrossTenantGrantMapper grantMapper; @@ -82,6 +88,9 @@ public class AdminController { @Inject private MigrateService migrateService; + @Inject + private SyncService syncService; + // ==================== 登录 / 登出 ==================== @Get @@ -190,21 +199,30 @@ public class AdminController { @Post @Mapping("/tenant/save") public void tenantSave(Tenant t, Context ctx) throws Throwable { - if (t.getTenantId() == null || t.getTenantId().isEmpty()) { - ctx.redirect(basePath + "/admin/tenant?msg=tenant_id_required"); - return; - } - if (t.getPrefixCode() == null || t.getPrefixCode().isEmpty()) { - t.setPrefixCode(t.getTenantId()); - } - if (t.getStatus() == null) { - t.setStatus(1); - } - if (tenantMapper.selectById(t.getTenantId()) == null) { + // tenantId 页面不可见：新增时雪花生成，编辑时由 hidden 携带（主键不可改） + boolean isNew = (t.getTenantId() == null || t.getTenantId().isEmpty()); + if (isNew) { + // 前缀码（IM 账号前缀）必填且全局唯一 + if (t.getPrefixCode() == null || t.getPrefixCode().isEmpty()) { + ctx.redirect(basePath + "/admin/tenant?msg=prefix_code_required"); + return; + } + Tenant exist = tenantMapper.selectOne(Wrappers.lambdaQuery() + .eq(Tenant::getPrefixCode, t.getPrefixCode()).last("LIMIT 1")); + if (exist != null) { + ctx.redirect(basePath + "/admin/tenant?msg=prefix_code_duplicate"); + return; + } + t.setTenantId(Ids.nextIdStr()); + if (t.getStatus() == null) { + t.setStatus(1); + } tenantMapper.insert(t); } else { tenantMapper.updateById(t); } + // 失效缓存，使后续按 id/appKey/prefix 查询拿到最新值 + tenantService.evictCache(t.getTenantId()); ctx.redirect(basePath + "/admin/tenant?msg=saved"); } @@ -215,6 +233,7 @@ public class AdminController { t.setTenantId(tenantId); t.setStatus(status == null ? 0 : status); tenantMapper.updateById(t); + tenantService.evictCache(tenantId); ctx.redirect(basePath + "/admin/tenant?msg=toggled"); } @@ -262,11 +281,26 @@ public class AdminController { @Get @Mapping("/queue") public Object queuePage(@Param(defaultValue = "") String msg) { - ModelAndView mv = view("queue.ftl", "队列监控", "queue"); - mv.put("pendings", distQueueMapper.selectList(Wrappers.lambdaQuery() + // nextRetryAt 为 OffsetDateTime，FreeMarker ?string 对 java.time 不支持，Controller 端预格式化 + DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + List rawPendings = distQueueMapper.selectList(Wrappers.lambdaQuery() .eq(DistQueue::getStatus, 0) .orderByAsc(DistQueue::getId) - .last("LIMIT 50"))); + .last("LIMIT 50")); + List> pendings = new ArrayList<>(); + if (rawPendings != null) { + for (DistQueue q : rawPendings) { + Map v = new LinkedHashMap<>(); + v.put("id", q.getId()); + v.put("tenantId", q.getTenantId()); + v.put("convId", q.getConvId()); + v.put("retryCount", q.getRetryCount()); + v.put("nextRetryAtStr", q.getNextRetryAt() == null ? "" : q.getNextRetryAt().format(fmt)); + pendings.add(v); + } + } + ModelAndView mv = view("queue.ftl", "队列监控", "queue"); + mv.put("pendings", pendings); mv.put("deads", distQueueMapper.selectList(Wrappers.lambdaQuery() .eq(DistQueue::getStatus, 3) .orderByDesc(DistQueue::getId) @@ -289,10 +323,27 @@ public class AdminController { @Get @Mapping("/usage") public Object usagePage() { - ModelAndView mv = view("usage.ftl", "用量报表", "usage"); - mv.put("stats", usageStatMapper.selectList(Wrappers.lambdaQuery() + // statTime 为 OffsetDateTime，FreeMarker ?string 对 java.time 不支持，Controller 端预格式化 + List raw = usageStatMapper.selectList(Wrappers.lambdaQuery() .orderByDesc(UsageStat::getStatTime) - .last("LIMIT 100"))); + .last("LIMIT 100")); + DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm"); + List> stats = new ArrayList<>(); + if (raw != null) { + for (UsageStat s : raw) { + Map v = new LinkedHashMap<>(); + v.put("tenantId", s.getTenantId()); + v.put("statTimeStr", s.getStatTime() == null ? "" : s.getStatTime().format(fmt)); + v.put("statLevel", s.getStatLevel()); + v.put("imMsgCount", s.getImMsgCount()); + v.put("imDau", s.getImDau()); + v.put("apiCallCount", s.getApiCallCount()); + v.put("trtcDurationSec", s.getTrtcDurationSec()); + stats.add(v); + } + } + ModelAndView mv = view("usage.ftl", "用量报表", "usage"); + mv.put("stats", stats); return mv; } @@ -522,6 +573,67 @@ public class AdminController { } } + // ==================== 数据同步（T17） ==================== + + /** + * 数据同步页：只读拉取腾讯 IM 群组/群成员/群消息到本地表，供按租户查看。 + * 与 migrate（带建群/建号写回）不同，本功能在腾讯侧无任何写操作。 + */ + @Get + @Mapping("/sync") + public Object syncPage(@Param(defaultValue = "") String msg) { + ModelAndView mv = view("sync.ftl", "数据同步", "sync"); + mv.put("tasks", migrateService.listTasks()); + mv.put("apps", sourceAppService.list()); + mv.put("tenants", tenantMapper.selectList(null)); + mv.put("msg", msg); + return mv; + } + + /** + * 同步群组：sourceAppId 空=主应用（需选 tenantId）；非空=老应用（密钥+租户取其配置） + */ + @Post + @Mapping("/sync/groups") + public void syncGroups(@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.syncGroups(sourceAppId, tenantId); + ctx.redirect(basePath + "/admin/sync?msg=sgdone"); + } + + /** 同步群成员（→用户）：需先同步群组 */ + @Post + @Mapping("/sync/users") + public void syncUsers(@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.syncGroupMembers(sourceAppId, tenantId); + ctx.redirect(basePath + "/admin/sync?msg=sudone"); + } + + /** 同步群消息：需先同步群组 */ + @Post + @Mapping("/sync/groupmsg") + public void syncGroupMsg(@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.syncGroupMessages(sourceAppId, tenantId); + ctx.redirect(basePath + "/admin/sync?msg=smdone"); + } + // ==================== 公共：构造页面模型 ==================== /** @@ -542,7 +654,7 @@ public class AdminController { @Mapping("/messages") public Object messagesPage(@Param(defaultValue = "") String tenantId, @Param(required = false) Integer convType, - @Param(defaultValue = "1") Integer distStatus, + @Param(required = false) String distStatus, @Param(defaultValue = "") String dateFrom, @Param(defaultValue = "") String dateTo, @Param(defaultValue = "") String keyword, @@ -556,10 +668,22 @@ public class AdminController { int safeSize = size <= 0 ? 20 : Math.min(size, 200); int offset = (safePage - 1) * safeSize; + // distStatus：null=首次进入(默认已分发) / ""=用户选"全部" / "0,1,2"=对应状态 + // 前端"全部"提交空串，Solon 对带空值的 Integer 参数会判为缺失(defaultValue 不兜底)， + // 故改用 String 接收后手动解析，Mapper 以 null 表示不限分发状态。 + Integer distStatusInt; + if (distStatus == null) { + distStatusInt = 1; + } else if (distStatus.isEmpty()) { + distStatusInt = null; + } else { + distStatusInt = Integer.valueOf(distStatus); + } + String tid = (tenantId == null || tenantId.isEmpty()) ? null : tenantId; String kw = (keyword == null || keyword.isEmpty()) ? null : keyword; - List rawRows = imMessageMapper.selectMessagePage(tid, convType, distStatus, from, to, kw, offset, safeSize); - long total = imMessageMapper.countMessages(tid, convType, distStatus, from, to, kw); + List rawRows = imMessageMapper.selectMessagePage(tid, convType, distStatusInt, from, to, kw, offset, safeSize); + long total = imMessageMapper.countMessages(tid, convType, distStatusInt, from, to, kw); int totalPages = (int) ((total + safeSize - 1) / Math.max(1, safeSize)); // 转成 Map 并预格式化时间（FreeMarker 对 java.time OffsetDateTime 格式化支持不稳，Controller 端处理） @@ -595,7 +719,7 @@ public class AdminController { mv.put("totalPages", totalPages); mv.put("tenantId", tenantId == null ? "" : tenantId); mv.put("convType", convType); - mv.put("distStatus", distStatus); + mv.put("distStatus", distStatusInt); mv.put("dateFrom", dateFrom == null ? "" : dateFrom); mv.put("dateTo", dateTo == null ? "" : dateTo); mv.put("keyword", keyword == null ? "" : keyword); diff --git a/src/main/java/com/imutil/controller/CallbackController.java b/src/main/java/com/imutil/controller/CallbackController.java index 4f04479..598fcb0 100644 --- a/src/main/java/com/imutil/controller/CallbackController.java +++ b/src/main/java/com/imutil/controller/CallbackController.java @@ -42,6 +42,7 @@ public class CallbackController { @Param(value = "RequestTime", required = false) String requestTime, @Param(value = "SdkAppid", required = false) String sdkAppid) throws Throwable { String body = ctx.body(); + log.info("收到腾讯回调 command={} sdkAppid={} requestTime={} body={}", command, sdkAppid, requestTime, body); // 1. 签名校验（未配置 Token 时跳过，便于联调，生产必须配置） if (callbackToken != null && !callbackToken.isEmpty()) { @@ -62,6 +63,7 @@ public class CallbackController { // 落库失败的消息由补拉服务（T10）兜底 result = "{\"ActionStatus\":\"OK\",\"ErrorCode\":0,\"ErrorInfo\":\"\"}"; } + log.info("回调处理完成 command={} result={}", command, result); ctx.output(result); } } diff --git a/src/main/java/com/imutil/entity/GroupMapping.java b/src/main/java/com/imutil/entity/GroupMapping.java index c41ef9b..5c9a74f 100644 --- a/src/main/java/com/imutil/entity/GroupMapping.java +++ b/src/main/java/com/imutil/entity/GroupMapping.java @@ -30,5 +30,17 @@ public class GroupMapping { /** 群类型：Public/Private/ChatRoom/AVChatRoom */ private String groupType; + /** 群名称（同步拉取自腾讯 group_info.Name） */ + private String name; + + /** 群主账号（同步拉取自腾讯 group_info.Owner_Account） */ + private String ownerAccount; + + /** 群成员数（同步拉取自腾讯 group_info.MemberNum / 成员列表长度） */ + private Integer memberCount; + + /** 最近一次同步时间（数据同步功能刷新） */ + private OffsetDateTime lastSyncedAt; + private OffsetDateTime createdAt; } diff --git a/src/main/java/com/imutil/entity/UserMapping.java b/src/main/java/com/imutil/entity/UserMapping.java index 7b76776..483e985 100644 --- a/src/main/java/com/imutil/entity/UserMapping.java +++ b/src/main/java/com/imutil/entity/UserMapping.java @@ -40,5 +40,11 @@ public class UserMapping { /** 状态：1=正常 0=封禁 */ private Integer status; + /** 用户昵称（同步拉取自腾讯群成员资料） */ + private String nick; + + /** 最近一次同步时间（数据同步功能刷新） */ + private OffsetDateTime lastSyncedAt; + private OffsetDateTime createdAt; } diff --git a/src/main/java/com/imutil/service/SyncService.java b/src/main/java/com/imutil/service/SyncService.java new file mode 100644 index 0000000..ee7fb64 --- /dev/null +++ b/src/main/java/com/imutil/service/SyncService.java @@ -0,0 +1,41 @@ +package com.imutil.service; + +import com.imutil.entity.MigrateTask; + +/** + * 数据同步服务（T17） + *

+ * 与 {@link MigrateService}（老应用→主应用的"迁移"，带建群/建号写回副作用）不同， + * 本服务定位为只读拉取：把指定腾讯 IM 应用里的群组/群成员/群消息拉到本地表， + * 供管理后台按租户查看，不在腾讯侧产生任何写操作。 + *

+ * 腾讯 IM 现实约束：无"全量用户列表"接口（用户靠群成员反推）；无"全量 C2C 会话列表"接口 + * （单聊历史消息本轮不做全量，回调增量已由 {@link com.imutil.service.PullService} 覆盖）。 + * + * @author imutil + */ +public interface SyncService { + + /** + * 同步群组：get_appid_group_list 全量 Next 分页 + 逐群 get_group_info 拉资料 → upsert group_mapping + * + * @param sourceAppId 数据源：null=主应用（全局密钥）；非空=该老应用密钥 + * @param tenantId 数据归属租户（主应用时必填；源应用时以其 tenant_id 为准） + * @return 同步任务记录（含 total/processed/校验结果） + */ + MigrateTask syncGroups(Long sourceAppId, String tenantId); + + /** + * 同步群成员（→用户）：遍历该租户已同步的群，逐群 get_group_info 取成员 → upsert user_mapping + * + * @see #syncGroups(Long, String) 先同步群组，群成员才有来源 + */ + MigrateTask syncGroupMembers(Long sourceAppId, String tenantId); + + /** + * 同步群消息：遍历该租户已同步的群，逐群 getGroupMsg + IsFinished 滚动全量 → 写 im_message(source=SYNC) + * + * @see #syncGroups(Long, String) 先同步群组，群消息才有目标群清单 + */ + MigrateTask syncGroupMessages(Long sourceAppId, String tenantId); +} diff --git a/src/main/java/com/imutil/service/TenantService.java b/src/main/java/com/imutil/service/TenantService.java index db7cce7..bcef641 100644 --- a/src/main/java/com/imutil/service/TenantService.java +++ b/src/main/java/com/imutil/service/TenantService.java @@ -22,6 +22,16 @@ public interface TenantService { */ Tenant getById(String tenantId); + /** + * 根据 prefix_code（IM 账号前缀）查询租户（带缓存） + *

+ * 回调识别租户时，账号前缀经此反查得 tenant_id。 + * + * @param prefixCode IM 前缀码 + * @return 租户实体，不存在返回 null + */ + Tenant getByPrefixCode(String prefixCode); + /** * 失效租户缓存（租户变更时调用） */ diff --git a/src/main/java/com/imutil/service/impl/CallbackServiceImpl.java b/src/main/java/com/imutil/service/impl/CallbackServiceImpl.java index 8c68427..b1d0382 100644 --- a/src/main/java/com/imutil/service/impl/CallbackServiceImpl.java +++ b/src/main/java/com/imutil/service/impl/CallbackServiceImpl.java @@ -83,7 +83,7 @@ public class CallbackServiceImpl implements CallbackService { String tenantId = identifyTenant(callbackCommand, node); if (tenantId == null) { // 无法识别租户（如腾讯系统消息 administrator），不落库不分发 - log.debug("回调无法识别租户，跳过 command={} from={}", callbackCommand, node.get("FromAccount").getString()); + log.debug("回调无法识别租户，跳过 command={} from={}", callbackCommand, node.get("From_Account").getString()); return ok(); } @@ -99,7 +99,8 @@ public class CallbackServiceImpl implements CallbackService { // 跨租户标记：复查授权确认（before 已拦截未授权）。 // 若 before 未开启/失效，未授权消息会漏到 after，此处查不到授权即告警（兜底发现隔离异常）。 if (msg.getConvType() != null && msg.getConvType() == 1) { - String toTenant = parsePrefix(msg.getConvId()); + Tenant toT = tenantService.getByPrefixCode(parsePrefix(msg.getConvId())); + String toTenant = toT == null ? null : toT.getTenantId(); if (toTenant != null && !toTenant.equals(tenantId)) { if (crossTenantService.checkSendMsgGrant( tenantId, msg.getFromAccount(), toTenant, msg.getConvId()) != null) { @@ -138,13 +139,16 @@ public class CallbackServiceImpl implements CallbackService { * @return 放行 ok() 或拦截 fail() */ private String handleBeforeSendMsg(ONode node) { - String from = node.get("FromAccount").getString(); - String to = node.get("ToAccount").getString(); + String from = node.get("From_Account").getString(); + String to = node.get("To_Account").getString(); if (to == null || to.isEmpty()) { to = node.get("ToPlayerId").getString(); } - String fromTenant = parsePrefix(from); - String toTenant = parsePrefix(to); + // 前缀(prefixCode) → 反查 tenantId（主键雪花化后，授权记录按 tenant_id 存储） + Tenant fromT = tenantService.getByPrefixCode(parsePrefix(from)); + Tenant toT = tenantService.getByPrefixCode(parsePrefix(to)); + String fromTenant = fromT == null ? null : fromT.getTenantId(); + String toTenant = toT == null ? null : toT.getTenantId(); // 同租户或任一租户无法识别（如系统消息）：不拦截，放行正常通讯 if (fromTenant == null || toTenant == null || fromTenant.equals(toTenant)) { return ok(); @@ -160,7 +164,7 @@ public class CallbackServiceImpl implements CallbackService { String msgKey = null; try { long msgSeq = node.get("MsgSeq").getLong(); - long msgRandom = node.get("MsgRandom").getLong(); + long msgRandom = pickMsgRandom(node); msgKey = MsgKeys.build(from, to, msgSeq, msgRandom); } catch (Exception ignore) { // 回调体缺字段无法算 msgKey，审计仍写（msgKey=null） @@ -215,14 +219,18 @@ public class CallbackServiceImpl implements CallbackService { * @return 租户ID，识别失败返回 null */ private String identifyTenant(String command, ONode node) { - String from = node.get("FromAccount").getString(); + // 账号前缀即 prefixCode（短码），按 prefix_code 反查得 tenantId（主键已雪花化，与前缀解耦） + String from = node.get("From_Account").getString(); if (from != null && !from.isEmpty()) { - String tenantId = parsePrefix(from); - if (tenantId != null && tenantService.getById(tenantId) != null) { - return tenantId; + String prefix = parsePrefix(from); + if (prefix != null) { + Tenant t = tenantService.getByPrefixCode(prefix); + if (t != null) { + return t.getTenantId(); + } } } - // 群回调：GroupId 查映射 + // 群回调：GroupId 查映射（group_mapping.tenant_id 即租户主键，雪花/短码皆正确） String groupId = node.get("GroupId").getString(); if (groupId != null && !groupId.isEmpty()) { GroupMapping gm = groupMappingMapper.selectOne(Wrappers.lambdaQuery() @@ -230,11 +238,6 @@ public class CallbackServiceImpl implements CallbackService { if (gm != null) { return gm.getTenantId(); } - // 群ID也带前缀 - String tid = parsePrefix(groupId); - if (tid != null && tenantService.getById(tid) != null) { - return tid; - } } return null; } @@ -269,15 +272,15 @@ public class CallbackServiceImpl implements CallbackService { * 解析消息体，构造 ImMessage（含幂等键 msg_key） */ private ImMessage parseMessage(String command, ONode node, String tenantId) { - String from = node.get("FromAccount").getString(); - String to = node.get("ToAccount").getString(); + String from = node.get("From_Account").getString(); + String to = node.get("To_Account").getString(); if (to == null || to.isEmpty()) { to = node.get("ToPlayerId").getString(); } String groupId = node.get("GroupId").getString(); long msgSeq = node.get("MsgSeq").getLong(); - long msgRandom = node.get("MsgRandom").getLong(); - long msgTs = node.get("MsgTimeStamp").getLong(); + long msgRandom = pickMsgRandom(node); + long msgTs = pickMsgTime(node); if (msgTs == 0) { msgTs = System.currentTimeMillis() / 1000L; @@ -297,7 +300,7 @@ public class CallbackServiceImpl implements CallbackService { msg.setFromAccount(from); msg.setToAccount(groupId != null && !groupId.isEmpty() ? null : to); msg.setGroupId(groupId); - msg.setMsgType(node.get("MsgType").getString()); + msg.setMsgType(pickMsgType(node)); msg.setMsgBody(node.get("MsgBody").toString()); msg.setSource("CALLBACK"); msg.setIsCrossTenant(false); @@ -313,13 +316,43 @@ public class CallbackServiceImpl implements CallbackService { if (groupId != null && !groupId.isEmpty()) { return groupId; } - String to = node.get("ToAccount").getString(); + String to = node.get("To_Account").getString(); if (to == null || to.isEmpty()) { to = node.get("ToPlayerId").getString(); } return to; } + /** + * 消息随机数：C2C=MsgRandom，群=Random，取到非 0 者为准（避免字段差异导致 msg_key 退化为 :0） + */ + private long pickMsgRandom(ONode node) { + long v = node.get("MsgRandom").getLong(); + return v != 0 ? v : node.get("Random").getLong(); + } + + /** + * 消息时间（秒）：优先 MsgTime（腾讯标准），回退 MsgTimeStamp（兼容历史/异常回调体） + */ + private long pickMsgTime(ONode node) { + long v = node.get("MsgTime").getLong(); + return v != 0 ? v : node.get("MsgTimeStamp").getLong(); + } + + /** + * 消息类型：优先 MsgBody[0].MsgType（群/C2C 均在数组元素内），回退顶层 MsgType + */ + private String pickMsgType(ONode node) { + ONode body = node.get("MsgBody"); + if (body.isArray() && body.size() > 0) { + String t = body.get(0).get("MsgType").getString(); + if (t != null && !t.isEmpty()) { + return t; + } + } + return node.get("MsgType").getString(); + } + /** * 返回腾讯回调成功响应 */ diff --git a/src/main/java/com/imutil/service/impl/SyncServiceImpl.java b/src/main/java/com/imutil/service/impl/SyncServiceImpl.java new file mode 100644 index 0000000..82d2f11 --- /dev/null +++ b/src/main/java/com/imutil/service/impl/SyncServiceImpl.java @@ -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） + *

+ * 只读拉取腾讯 IM 数据到本地表，供后台按租户查看。与 MigrateService（带写回副作用）互补。 + *

    + *
  • 群组：get_appid_group_list 全量 Next 分页 + 逐群 get_group_info → upsert group_mapping
    • + *
  • 群成员(→用户)：遍历该租户已同步群，取 MemberList → upsert user_mapping（腾讯无全量用户接口，靠群成员反推）
    • + *
  • 群消息：遍历该租户已同步群，getGroupMsg + IsFinished 滚动全量 → 写 im_message(source=SYNC)
    • + *
+ * 数据源密钥：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.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 groups = groupMappingMapper.selectList(Wrappers.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.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 groups = groupMappingMapper.selectList(Wrappers.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.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()); + } +} diff --git a/src/main/java/com/imutil/service/impl/TenantServiceImpl.java b/src/main/java/com/imutil/service/impl/TenantServiceImpl.java index d97c164..db7b161 100644 --- a/src/main/java/com/imutil/service/impl/TenantServiceImpl.java +++ b/src/main/java/com/imutil/service/impl/TenantServiceImpl.java @@ -24,6 +24,7 @@ public class TenantServiceImpl implements TenantService { private static final String CACHE_KEY_BY_KEY = "imutil:tenant:bykey:"; private static final String CACHE_KEY_BY_ID = "imutil:tenant:byid:"; + private static final String CACHE_KEY_BY_PREFIX = "imutil:tenant:byprefix:"; private static final long CACHE_TTL_SEC = 300; @Inject @@ -84,6 +85,33 @@ public class TenantServiceImpl implements TenantService { return t; } + @Override + public Tenant getByPrefixCode(String prefixCode) { + if (prefixCode == null || prefixCode.isEmpty()) { + return null; + } + String key = CACHE_KEY_BY_PREFIX + prefixCode; + // 1. 本地缓存 + Tenant t = localCache.get(key, Tenant.class); + if (t != null) { + return t; + } + // 2. Redis + t = redisService.getJson(key, Tenant.class); + if (t == null) { + // 3. DB + t = tenantMapper.selectOne(Wrappers.lambdaQuery() + .eq(Tenant::getPrefixCode, prefixCode).last("LIMIT 1")); + if (t != null) { + redisService.setJson(key, t, CACHE_TTL_SEC, TimeUnit.SECONDS); + } + } + if (t != null) { + localCache.put(key, t); + } + return t; + } + @Override public void evictCache(String tenantId) { Tenant t = tenantMapper.selectById(tenantId); @@ -94,6 +122,9 @@ public class TenantServiceImpl implements TenantService { String k2 = CACHE_KEY_BY_KEY + t.getAppKey(); redisService.del(k2); localCache.invalidate(k2); + String k3 = CACHE_KEY_BY_PREFIX + t.getPrefixCode(); + redisService.del(k3); + localCache.invalidate(k3); } } } diff --git a/src/main/java/com/imutil/tencent/TencentImClient.java b/src/main/java/com/imutil/tencent/TencentImClient.java index 0299fc0..d389bcd 100644 --- a/src/main/java/com/imutil/tencent/TencentImClient.java +++ b/src/main/java/com/imutil/tencent/TencentImClient.java @@ -341,6 +341,36 @@ public class TencentImClient { return callApiAs("group_open_http_svc/get_appid_groups", Jsons.stringify(body), srcSdkAppId, srcSecretKey); } + /** + * 全量分页列出 SDKAppId 下所有群（T17 数据同步，主应用密钥） + *

+ * 命令字 group_open_http_svc/get_appid_group_list，支持 Next 游标分页： + * 首次 Next 传空串，循环把上次返回的 Next 再传入，直到返回 Next 为空。 + * 与 {@link #getAppidGroups} 的区别：标准接口、支持分页、GroupIdList 元素为纯字符串。 + * + * @param limit 每页数量（建议 100） + * @param next 分页游标，首次传空串 + * @return 腾讯响应原始 JSON（含 Next/Total/GroupIdList），由调用方解析 + */ + public String getAppidGroupList(int limit, String next) { + Map body = new HashMap<>(); + body.put("Limit", limit); + body.put("Next", next == null ? "" : next); + return callApi("group_open_http_svc/get_appid_group_list", Jsons.stringify(body)); + } + + /** + * 全量分页列出 SDKAppId 下所有群（T17 数据同步，指定应用密钥，用于老应用） + * + * @see #getAppidGroupList(int, String) + */ + public String getAppidGroupList(int limit, String next, long sdkAppId, String secretKey) { + Map body = new HashMap<>(); + body.put("Limit", limit); + body.put("Next", next == null ? "" : next); + return callApiAs("group_open_http_svc/get_appid_group_list", Jsons.stringify(body), sdkAppId, secretKey); + } + /** * 查群资料 + 成员列表（老应用拉群详情，用老密钥） *

diff --git a/src/main/resources/db/init.sql b/src/main/resources/db/init.sql index cc22944..cbc7b6c 100644 --- a/src/main/resources/db/init.sql +++ b/src/main/resources/db/init.sql @@ -19,6 +19,7 @@ CREATE TABLE IF NOT EXISTS tenant ( CONSTRAINT pk_tenant PRIMARY KEY (tenant_id) ); CREATE UNIQUE INDEX IF NOT EXISTS uk_tenant_app_key ON tenant(app_key); +CREATE UNIQUE INDEX IF NOT EXISTS uk_tenant_prefix ON tenant(prefix_code); -- ===== 2. 用户映射（业务用户 ↔ IM 用户） ===== CREATE TABLE IF NOT EXISTS user_mapping ( @@ -29,6 +30,8 @@ CREATE TABLE IF NOT EXISTS user_mapping ( is_default boolean NOT NULL DEFAULT false, is_global boolean NOT NULL DEFAULT false, status smallint NOT NULL DEFAULT 1, -- 1=正常 0=封禁 + nick varchar(128), -- 用户昵称（T17 数据同步拉取自群成员资料） + last_synced_at timestamptz, -- 最近一次同步时间（T17） created_at timestamptz NOT NULL DEFAULT now(), CONSTRAINT pk_user_mapping PRIMARY KEY (id) ); @@ -42,6 +45,10 @@ CREATE TABLE IF NOT EXISTS group_mapping ( biz_group_id varchar(128) NOT NULL, im_group_id varchar(128) NOT NULL, group_type varchar(16) NOT NULL DEFAULT 'Public', + name varchar(256), -- 群名称（T17 数据同步拉取） + owner_account varchar(128), -- 群主账号（T17 数据同步拉取） + member_count int, -- 群成员数（T17 数据同步拉取） + last_synced_at timestamptz, -- 最近一次同步时间（T17） created_at timestamptz NOT NULL DEFAULT now(), CONSTRAINT pk_group_mapping PRIMARY KEY (id) ); @@ -243,3 +250,11 @@ CREATE TABLE IF NOT EXISTS migrate_task ( CONSTRAINT pk_migrate_task PRIMARY KEY (id) ); CREATE INDEX IF NOT EXISTS idx_migrate_task_app ON migrate_task(source_app_id, object_type, status); + +-- ===== T17 数据同步：老库升级补丁（ADD COLUMN IF NOT EXISTS，幂等可重跑） ===== +ALTER TABLE group_mapping ADD COLUMN IF NOT EXISTS name varchar(256); +ALTER TABLE group_mapping ADD COLUMN IF NOT EXISTS owner_account varchar(128); +ALTER TABLE group_mapping ADD COLUMN IF NOT EXISTS member_count int; +ALTER TABLE group_mapping ADD COLUMN IF NOT EXISTS last_synced_at timestamptz; +ALTER TABLE user_mapping ADD COLUMN IF NOT EXISTS nick varchar(128); +ALTER TABLE user_mapping ADD COLUMN IF NOT EXISTS last_synced_at timestamptz; diff --git a/src/main/resources/templates/_macros.ftl b/src/main/resources/templates/_macros.ftl index c432047..992ebf0 100644 --- a/src/main/resources/templates/_macros.ftl +++ b/src/main/resources/templates/_macros.ftl @@ -69,6 +69,7 @@ 消息记录 老应用配置 数据迁移 + 数据同步

修改密码 diff --git a/src/main/resources/templates/queue.ftl b/src/main/resources/templates/queue.ftl index dbf01c6..0883c5e 100644 --- a/src/main/resources/templates/queue.ftl +++ b/src/main/resources/templates/queue.ftl @@ -15,7 +15,7 @@ ${q.tenantId!} ${q.convId!} ${q.retryCount!0} - <#if q.nextRetryAt??>${q.nextRetryAt?string('yyyy-MM-dd HH:mm:ss')}<#else>- + ${(q.nextRetryAtStr)!} diff --git a/src/main/resources/templates/sourceapp.ftl b/src/main/resources/templates/sourceapp.ftl index 30b239a..e17298a 100644 --- a/src/main/resources/templates/sourceapp.ftl +++ b/src/main/resources/templates/sourceapp.ftl @@ -41,7 +41,16 @@ 停用 - + +
<#if a.status?? && a.status == 1> @@ -78,7 +87,7 @@
@@ -91,4 +100,52 @@ + + <#-- 编辑弹窗：主键 id 自增 hidden 携带；secretKey 回填明文（与租户 appSecret 一致） --> + + diff --git a/src/main/resources/templates/sync.ftl b/src/main/resources/templates/sync.ftl new file mode 100644 index 0000000..3069f2b --- /dev/null +++ b/src/main/resources/templates/sync.ftl @@ -0,0 +1,99 @@ +<#import "_macros.ftl" as m> +<@m.layout active="sync" title="数据同步"> + <#if msg?has_content> +
+ <#if msg=='sgdone'>✅ 群组同步完成（详见任务记录） + <#elseif msg=='sudone'>✅ 群成员/用户同步完成（详见任务记录） + <#elseif msg=='smdone'>✅ 群消息同步完成（详见任务记录） + <#elseif msg=='syncopt'>❌ 请选择数据源（主应用时须同时选归属租户） + <#else>${msg} + +
+ + +
+ 说明 + ：本页为只读拉取，把腾讯 IM 数据同步到本地表用于查看，不在腾讯侧建群/建号。 + 群组=get_appid_group_list 全量分页；群成员(用户)=遍历群取成员（腾讯无全量用户接口，靠群成员反推）；群消息=逐群全量拉取。 + C2C 单聊消息无全量会话列表接口，历史单聊请用「数据迁移」页或依赖回调增量。 +
+ + <#-- 公共：数据源 + 归属租户选择（三个表单各自携带） --> + <#macro srcSelect> +
+ + +
+
+ + +
+ + +
+
① 同步群组
+

拉取数据源应用的所有群（get_appid_group_list 全量分页）+ 逐群 get_group_info（名/类型/群主/成员数）→ 写 group_mapping。幂等可重跑。

+
+
<@srcSelect/>
+ +
+
+ +
+
② 同步群成员（→ 用户）
+

遍历该租户已同步的群，逐群取成员列表 → 写 user_mapping。请先执行「同步群组」，否则无群可遍历。

+
+
<@srcSelect/>
+ +
+
+ +
+
③ 同步群消息
+

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

+
+
<@srcSelect/>
+ +
+
+ +

同步/迁移任务记录

+ <#if tasks?has_content> + + + + + + <#list tasks as t> + + + + + + + + + + + +
ID源应用对象进度(已处理/总数)状态校验结果开始时间
${t.id!}<#if t.sourceAppId?? && t.sourceAppId==0>主应用<#else>${t.sourceAppId!}${t.objectType!}${t.processedCount!'0'}/${t.totalCount!'0'} + <#if t.status?? && t.status==3>完成 + <#elseif t.status?? && t.status==1>运行 + <#elseif t.status?? && t.status==4>失败 + <#else>待 + + ${t.verifyResult!'-'}${t.startedAt!}
+ <#else> +
暂无任务
+ + diff --git a/src/main/resources/templates/tenant.ftl b/src/main/resources/templates/tenant.ftl index 75b139d..c0f8076 100644 --- a/src/main/resources/templates/tenant.ftl +++ b/src/main/resources/templates/tenant.ftl @@ -10,14 +10,13 @@ - + <#list tenants as t> - @@ -31,7 +30,18 @@ 停用 - - +
租户ID名称前缀AppKeyAppSecret名称前缀AppKeyAppSecret 回调URLIM QPS状态操作
${t.tenantId!} ${t.tenantName!} ${t.prefixCode!} ${t.appKey!} + +
<#if t.status?? && t.status == 1> @@ -56,7 +66,7 @@
新增租户
-
+
@@ -71,4 +81,48 @@
+ + <#-- 编辑弹窗：主键 tenantId 只读(readonly 仍随表单提交，供 updateById 定位)；其余字段全部放开 --> + + diff --git a/src/main/resources/templates/usage.ftl b/src/main/resources/templates/usage.ftl index d3bb27b..12ae959 100644 --- a/src/main/resources/templates/usage.ftl +++ b/src/main/resources/templates/usage.ftl @@ -9,7 +9,7 @@ <#list stats as s>
${s.tenantId!}<#if s.statTime??>${s.statTime?string('yyyy-MM-dd HH:mm')}<#else>-${(s.statTimeStr)!} <#if s.statLevel?? && s.statLevel == 1>小时<#else>天 ${s.imMsgCount!0} ${s.imDau!0}