feat(t14-s1s2): 数据迁移基础设施 + 用户迁移

T14 数据迁移工具 S1/S2 阶段:

S1 基础设施:
- source_app/migrate_task 建表 + 实体 + Mapper
- TencentImClient 双密钥 callApiAs(老应用拉取,不走主应用限流,审计记 migrate)
- 后台「老应用配置」页(CRUD + 测试连通)

S2 用户迁移:
- batchImport(multiaccount_import,每批≤100)
- MigrateService.migrateUsers(前缀转换+导入+映射+校验,幂等)
- 后台「数据迁移」页(上传清单文件/粘贴列表 + 任务记录)

联调通过:3用户清单→加sa前缀→腾讯导入OK→user_mapping→校验3/3/3。

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
yixiong
2026-07-09 11:12:25 +08:00
co-authored by Claude
parent b4c6652f66
commit 7b1385e160
14 changed files with 754 additions and 11 deletions
@@ -7,11 +7,20 @@ import com.imutil.entity.CrossTenantGrant;
import com.imutil.entity.DistQueue;
import com.imutil.entity.Tenant;
import com.imutil.entity.UsageStat;
import com.imutil.entity.SourceApp;
import com.imutil.mapper.CrossTenantGrantMapper;
import com.imutil.mapper.DistQueueMapper;
import com.imutil.mapper.TenantMapper;
import com.imutil.mapper.UsageStatMapper;
import com.imutil.service.AdminUserService;
import com.imutil.service.SourceAppService;
import com.imutil.tencent.TencentImClient;
import com.imutil.entity.MigrateTask;
import com.imutil.service.MigrateService;
import org.noear.solon.core.handle.UploadedFile;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.noear.solon.annotation.Controller;
import org.noear.solon.annotation.Get;
@@ -53,6 +62,15 @@ public class AdminController {
@Inject
private UsageStatMapper usageStatMapper;
@Inject
private SourceAppService sourceAppService;
@Inject
private TencentImClient tencentImClient;
@Inject
private MigrateService migrateService;
// ==================== 登录 / 登出 ====================
@Get
@@ -234,6 +252,121 @@ public class AdminController {
return mv;
}
// ==================== 老应用配置(数据迁移源) ====================
@Get
@Mapping("/sourceapp")
public Object sourceAppPage(@Param(defaultValue = "") String msg) {
ModelAndView mv = view("sourceapp.ftl", "老应用配置", "sourceapp");
mv.put("apps", sourceAppService.list());
mv.put("tenants", tenantMapper.selectList(null));
mv.put("msg", msg);
return mv;
}
@Post
@Mapping("/sourceapp/save")
public void sourceAppSave(SourceApp app, Context ctx) throws Throwable {
if (app.getSdkAppId() == null || app.getSdkAppId().isEmpty()) {
ctx.redirect(basePath + "/admin/sourceapp?msg=sdkappid_required");
return;
}
if (app.getTenantId() == null || app.getTenantId().isEmpty()) {
ctx.redirect(basePath + "/admin/sourceapp?msg=tenant_required");
return;
}
sourceAppService.save(app);
ctx.redirect(basePath + "/admin/sourceapp?msg=saved");
}
@Post
@Mapping("/sourceapp/toggle")
public void sourceAppToggle(@Param Long id, @Param Integer status, Context ctx) throws Throwable {
sourceAppService.toggle(id, status);
ctx.redirect(basePath + "/admin/sourceapp?msg=toggled");
}
/**
* 测试老应用连通性:用老应用密钥调 account_import(幂等,导入 administrator,无实质副作用),
* 验证 SDKAppId/密钥/sig 正确。腾讯返回业务级响应(非 70003)即说明 sig 鉴权通过、连通成功。
*/
@Post
@Mapping("/sourceapp/test")
public void sourceAppTest(@Param Long id, Context ctx) throws Throwable {
SourceApp app = sourceAppService.getById(id);
if (app == null) {
ctx.redirect(basePath + "/admin/sourceapp?msg=notfound");
return;
}
// account_import:幂等导入 administrator(主应用/老应用本就需要该管理员账号),必返 ActionStatus=OK
String body = "{\"Identifier\":\"administrator\"}";
String resp = tencentImClient.callApiAs("im_open_login_svc/account_import", body,
Long.parseLong(app.getSdkAppId()), app.getSecretKey());
// 连通判定:网络异常(ErrorCode=-1doCallApi 兜底)→失败;70003=UserSig 错误(密钥不匹配);
// 其余任何腾讯业务级响应都说明请求已达腾讯且 sig 鉴权通过 → 连通成功
String msg;
if (resp.contains("\"ErrorCode\":-1")) {
msg = "test_fail";
} else if (resp.contains("70003")) {
msg = "test_sigfail";
} else {
msg = "test_ok";
}
log.info("老应用连通测试 id={} sdk={} resp={}", id, app.getSdkAppId(), resp);
ctx.redirect(basePath + "/admin/sourceapp?msg=" + msg);
}
// ==================== 数据迁移(T14 ====================
@Get
@Mapping("/migrate")
public Object migratePage(@Param(defaultValue = "") String msg,
@Param(defaultValue = "") String imported) {
ModelAndView mv = view("migrate.ftl", "数据迁移", "migrate");
mv.put("tasks", migrateService.listTasks());
mv.put("apps", sourceAppService.list());
mv.put("msg", msg);
mv.put("imported", imported);
return mv;
}
/**
* 用户迁移:上传用户清单文件(每行一个老 UserID)或粘贴列表 → 加前缀 → 批量导入主应用 → 写映射 → 校验
*/
@Post
@Mapping("/migrate/users")
public void migrateUsers(@Param Long sourceAppId,
@Param(defaultValue = "") String userText,
Context ctx) throws Throwable {
List<String> userIds = new ArrayList<>();
UploadedFile file = ctx.file("file");
if (file != null) {
parseUserIds(new String(file.getContentAsBytes(), StandardCharsets.UTF_8), userIds);
} else if (!userText.isEmpty()) {
parseUserIds(userText, userIds);
}
if (sourceAppId == null) {
ctx.redirect(basePath + "/admin/migrate?msg=noapp");
return;
}
if (userIds.isEmpty()) {
ctx.redirect(basePath + "/admin/migrate?msg=empty");
return;
}
MigrateTask task = migrateService.migrateUsers(sourceAppId, userIds);
ctx.redirect(basePath + "/admin/migrate?msg=done&imported=" + task.getProcessedCount());
}
/** 解析用户清单:每行一个 UserID,空行与 # 注释跳过 */
private void parseUserIds(String content, List<String> out) {
for (String line : content.split("\\r?\\n")) {
line = line.trim();
if (!line.isEmpty() && !line.startsWith("#")) {
out.add(line);
}
}
}
// ==================== 公共:构造页面模型 ====================
/**
@@ -0,0 +1,54 @@
package com.imutil.entity;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import java.time.OffsetDateTime;
/**
* 迁移任务记录实体(支持断点续传与校验)
* <p>
* 一次迁移对应一条记录:按对象类型(USER/GROUP/MSG/ROSTER)从老应用迁到主应用。
* pos_cursor 记录断点游标,任务中断后可从游标续传;verify_result 存老 vs 新条数对比。
*
* @author imutil
*/
@Data
@TableName("migrate_task")
public class MigrateTask {
@TableId(type = IdType.AUTO)
private Long id;
/** 关联 source_app.id */
private Long sourceAppId;
/** 对象类型:USER/GROUP/MSG/ROSTER */
private String objectType;
/** 0=pending 1=running 2=paused 3=done 4=failed */
private Integer status;
/** 应处理总数 */
private Long totalCount;
/** 已处理数 */
private Long processedCount;
/** 断点游标(时间戳/会话+seq,按对象类型语义不同) */
private String posCursor;
/** 校验结果 JSON(老 vs 新 条数对比) */
private String verifyResult;
/** 失败原因 */
private String errorMsg;
private OffsetDateTime startedAt;
private OffsetDateTime finishedAt;
private OffsetDateTime createdAt;
}
@@ -0,0 +1,45 @@
package com.imutil.entity;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import java.time.OffsetDateTime;
/**
* 老应用配置实体(数据迁移源)
* <p>
* 每个老应用对应主应用一个租户(决定 UserID 前缀)。迁移时用本应用的 sdkAppId/secretKey
* 调腾讯 API 拉取存量数据,加前缀后导入主应用。
* <p>
* 密钥仅本工具持有,禁止下发到业务系统。
*
* @author imutil
*/
@Data
@TableName("source_app")
public class SourceApp {
@TableId(type = IdType.AUTO)
private Long id;
/** 老应用名称 */
private String appName;
/** 老应用 SDKAppID */
private String sdkAppId;
/** 老应用密钥(迁移拉取用) */
private String secretKey;
/** 对应主应用租户(决定 UserID 前缀),关联 tenant.tenant_id */
private String tenantId;
/** 1=启用 0=停用 */
private Integer status;
private String remark;
private OffsetDateTime createdAt;
}
@@ -0,0 +1,12 @@
package com.imutil.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.imutil.entity.MigrateTask;
/**
* 迁移任务记录 Mapper
*
* @author imutil
*/
public interface MigrateTaskMapper extends BaseMapper<MigrateTask> {
}
@@ -0,0 +1,12 @@
package com.imutil.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.imutil.entity.SourceApp;
/**
* 老应用配置 Mapper
*
* @author imutil
*/
public interface SourceAppMapper extends BaseMapper<SourceApp> {
}
@@ -0,0 +1,33 @@
package com.imutil.service;
import com.imutil.entity.MigrateTask;
import java.util.List;
/**
* 数据迁移服务(T14
* <p>
* 把老应用存量数据迁到主应用:拉取/读取 → 加租户前缀 → 导入主应用 → 写映射 → 校验。
* 各对象类型(USER/GROUP/MSG/ROSTER)分方法实现,均记 migrate_task 支持进度与校验。
*
* @author imutil
*/
public interface MigrateService {
/**
* 用户迁移:老应用用户 ID 清单 → 加前缀 → 批量导入主应用 → 写 user_mapping → 校验。
* <p>
* 老应用用户清单由调用方提供(后台上传文件解析),因腾讯 IM 无「列出所有用户」API。
* 幂等:已导入账号与已存在映射跳过,可重复执行。
*
* @param sourceAppId 老应用配置 id
* @param oldUserIds 老应用用户 ID 清单
* @return 迁移任务记录(含进度与校验结果)
*/
MigrateTask migrateUsers(Long sourceAppId, List<String> oldUserIds);
/**
* 迁移任务列表(最近 100 条,按 id 倒序)
*/
List<MigrateTask> listTasks();
}
@@ -0,0 +1,28 @@
package com.imutil.service;
import com.imutil.entity.SourceApp;
import java.util.List;
/**
* 老应用配置服务(数据迁移源)
* <p>
* 供管理后台 CRUD 与迁移逻辑查询老应用配置(sdkAppId/密钥/对应租户前缀)。
* 老应用数量少,不走缓存。
*
* @author imutil
*/
public interface SourceAppService {
/** 全部老应用 */
List<SourceApp> list();
/** 按 id 查 */
SourceApp getById(Long id);
/** 新增或更新(id 为空则新增) */
void save(SourceApp app);
/** 启用/停用 */
void toggle(Long id, Integer status);
}
@@ -0,0 +1,137 @@
package com.imutil.service.impl;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.imutil.common.BizException;
import com.imutil.entity.MigrateTask;
import com.imutil.entity.SourceApp;
import com.imutil.entity.Tenant;
import com.imutil.entity.UserMapping;
import com.imutil.mapper.MigrateTaskMapper;
import com.imutil.mapper.SourceAppMapper;
import com.imutil.mapper.TenantMapper;
import com.imutil.mapper.UserMappingMapper;
import com.imutil.service.MigrateService;
import com.imutil.tencent.TencentImClient;
import lombok.extern.slf4j.Slf4j;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* 数据迁移服务实现(T14
*
* @author imutil
*/
@Slf4j
@Component
public class MigrateServiceImpl implements MigrateService {
@Inject
private SourceAppMapper sourceAppMapper;
@Inject
private TenantMapper tenantMapper;
@Inject
private UserMappingMapper userMappingMapper;
@Inject
private MigrateTaskMapper migrateTaskMapper;
@Inject
private TencentImClient tencentImClient;
@Override
public MigrateTask migrateUsers(Long sourceAppId, List<String> oldUserIds) {
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();
// 建迁移任务记录(running
MigrateTask task = new MigrateTask();
task.setSourceAppId(sourceAppId);
task.setObjectType("USER");
task.setStatus(1);
task.setTotalCount(oldUserIds == null ? 0L : (long) oldUserIds.size());
task.setStartedAt(OffsetDateTime.now());
migrateTaskMapper.insert(task);
try {
// 老用户 ID 加租户前缀(对齐 user_mapping: im_user_id = prefix + '_' + biz_user_id
List<String> newImUserIds = new ArrayList<>();
Map<String, String> oldToNew = new HashMap<>();
if (oldUserIds != null) {
for (String oldId : oldUserIds) {
if (oldId == null || oldId.trim().isEmpty()) {
continue;
}
String oldTrim = oldId.trim();
String newId = prefix + "_" + oldTrim;
newImUserIds.add(newId);
oldToNew.put(oldTrim, newId);
}
}
// 批量导入主应用(走主应用密钥+限流;幂等,已存在账号亦成功)
int imported = tencentImClient.batchImport(newImUserIds);
// 写 user_mapping(幂等:同 tenant+bizUserId 已存在则跳过)
int mapped = 0;
for (Map.Entry<String, String> e : oldToNew.entrySet()) {
UserMapping exist = userMappingMapper.selectOne(Wrappers.<UserMapping>lambdaQuery()
.eq(UserMapping::getTenantId, app.getTenantId())
.eq(UserMapping::getBizUserId, e.getKey())
.last("LIMIT 1"));
if (exist != null) {
continue;
}
UserMapping um = new UserMapping();
um.setTenantId(app.getTenantId());
um.setBizUserId(e.getKey());
um.setImUserId(e.getValue());
um.setIsDefault(false);
um.setIsGlobal(false);
um.setStatus(1);
userMappingMapper.insert(um);
mapped++;
}
// 校验:老清单数 vs 导入成功数 vs 新建映射数
task.setProcessedCount((long) imported);
task.setStatus(3); // done
task.setFinishedAt(OffsetDateTime.now());
task.setVerifyResult(String.format(
"{\"oldCount\":%d,\"imported\":%d,\"mapped\":%d}",
oldToNew.size(), imported, mapped));
migrateTaskMapper.updateById(task);
log.info("用户迁移完成 sourceApp={} old={} imported={} mapped={}",
sourceAppId, oldToNew.size(), imported, mapped);
return task;
} catch (Exception e) {
task.setStatus(4); // failed
task.setErrorMsg(e.getClass().getSimpleName() + ": " + e.getMessage());
task.setFinishedAt(OffsetDateTime.now());
migrateTaskMapper.updateById(task);
log.error("用户迁移失败 sourceApp={}", sourceAppId, e);
throw e;
}
}
@Override
public List<MigrateTask> listTasks() {
return migrateTaskMapper.selectList(Wrappers.<MigrateTask>lambdaQuery()
.orderByDesc(MigrateTask::getId)
.last("LIMIT 100"));
}
}
@@ -0,0 +1,54 @@
package com.imutil.service.impl;
import com.imutil.entity.SourceApp;
import com.imutil.mapper.SourceAppMapper;
import com.imutil.service.SourceAppService;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import java.util.List;
/**
* 老应用配置服务实现
*
* @author imutil
*/
@Component
public class SourceAppServiceImpl implements SourceAppService {
@Inject
private SourceAppMapper sourceAppMapper;
@Override
public List<SourceApp> list() {
return sourceAppMapper.selectList(null);
}
@Override
public SourceApp getById(Long id) {
if (id == null) {
return null;
}
return sourceAppMapper.selectById(id);
}
@Override
public void save(SourceApp app) {
if (app.getId() == null) {
if (app.getStatus() == null) {
app.setStatus(1);
}
sourceAppMapper.insert(app);
} else {
sourceAppMapper.updateById(app);
}
}
@Override
public void toggle(Long id, Integer status) {
SourceApp app = new SourceApp();
app.setId(id);
app.setStatus(status == null ? 0 : status);
sourceAppMapper.updateById(app);
}
}
@@ -66,14 +66,7 @@ public class TencentImClient {
private LocalCache localCache;
/**
* 生成管理员 UserSig(长效,用于调后台 API
*/
private String genAdminSig() {
return UserSigUtil.genSig(sdkAppId, secretKey, adminUserId, 30L * 86400);
}
/**
* 调用 IM REST API
* 调用主应用 IM REST API(用主应用密钥,走出站限流
*
* @param command 命令路径,如 im_open_login_svc/account_import
* @param bodyJson 请求体 JSON
@@ -83,16 +76,38 @@ public class TencentImClient {
// 租户来源:当前请求上下文,无则记 system(admin 后台调用等无租户上下文场景)
String tenantId = TenantContext.get();
String tid = (tenantId == null || tenantId.isEmpty()) ? "system" : tenantId;
// 出站腾讯 API 限流:按租户配额 + 全局共享配额(对齐设计 3.2),超限抛 429
// system(无租户上下文)用默认配额;补拉等后台任务超限由调用方 catch 跳过本轮
rateLimiter.checkApiLimit(tid, resolveTenantQps(tid));
return doCallApi(command, bodyJson, sdkAppId, secretKey, tid);
}
/**
* 调用老应用 IM REST API(数据迁移拉取用,指定老应用密钥)
* <p>
* 老应用拉取消耗的是老应用自身套餐配额(非主应用),故不走主应用限流;
* 审计 tid 记 "migrate" 便于与业务调用区分。管理员 identifier 默认 administrator
* (腾讯各应用默认管理员账号),如某老应用管理员不同,后续扩展 source_app 字段。
*
* @param srcSdkAppId 老应用 SDKAppId
* @param srcSecretKey 老应用密钥
* @return 响应 JSON 字符串
*/
public String callApiAs(String command, String bodyJson, long srcSdkAppId, String srcSecretKey) {
return doCallApi(command, bodyJson, srcSdkAppId, srcSecretKey, "migrate");
}
/**
* 调用腾讯 API 的公共实现:生成管理员 UserSig + 拼 URL + POST + 审计落库
* <p>
* 限流由调用方决定:主应用 callApi 走限流,老应用 callApiAs 不走(消耗老应用配额)。
*/
private String doCallApi(String command, String bodyJson, long appid, String key, String tid) {
String result;
try {
String adminSig = genAdminSig();
String adminSig = UserSigUtil.genSig(appid, key, adminUserId, 30L * 86400);
String url = "https://" + apiHost + "/v4/" + command
+ "?sdkappid=" + sdkAppId
+ "?sdkappid=" + appid
+ "&identifier=" + URLEncoder.encode(adminUserId, StandardCharsets.UTF_8)
+ "&usersig=" + URLEncoder.encode(adminSig, StandardCharsets.UTF_8)
+ "&contenttype=json&platform=10&apn=1";
@@ -182,6 +197,35 @@ public class TencentImClient {
return isOk(resp);
}
/**
* 批量导入账号(multiaccount_import,每批 ≤ 100,主应用密钥)
* <p>
* 幂等:已存在账号亦成功。数据迁移用户导入用:老应用 UserID 加租户前缀后批量导入主应用。
* 分批调用,返回成功导入数(整批 ActionStatus=OK 记为全部成功;个别非法 ID 场景后续可解析 RetItemList 精确化)。
*
* @param imUserIds 已加前缀的 IM 用户 ID 列表
* @return 成功导入数
*/
public int batchImport(List<String> imUserIds) {
if (imUserIds == null || imUserIds.isEmpty()) {
return 0;
}
int success = 0;
int batchSize = 100; // 腾讯 multiaccount_import 单批上限 100
for (int i = 0; i < imUserIds.size(); i += batchSize) {
List<String> sub = imUserIds.subList(i, Math.min(i + batchSize, imUserIds.size()));
Map<String, Object> body = new HashMap<>();
body.put("Accounts", sub);
String resp = callApi("im_open_login_svc/multiaccount_import", Jsons.stringify(body));
if (isOk(resp)) {
success += sub.size();
} else {
log.warn("批量导入失败批次 {} size={} resp={}", i / batchSize, sub.size(), resp);
}
}
return success;
}
/**
* 检查账号状态(只读)
* <p>