feat:昵称重复、显示分销人分销tree
This commit is contained in:
1 parent
4518856ccd
commit
074307b07b
50 files changed
+3386
-116
No files matched your search
@@ -0,0 +1,3 @@
|
||||
node_modules/
|
||||
runs/
|
||||
config.local.json
|
||||
@@ -0,0 +1,121 @@
|
||||
# 聊天历史用户修复工具
|
||||
|
||||
按会员 ID 修正 IM 身份,保留同昵称会员的各自账号与历史关系。支持 MySQL 5.7、8.0 和 Node.js 20.19 及以上版本。工具可以在能连接两库的运维电脑运行。
|
||||
|
||||
## 准备连接
|
||||
|
||||
```bash
|
||||
cd repair-im-users
|
||||
npm ci --omit=dev
|
||||
cp config.example.json config.local.json
|
||||
chmod 600 config.local.json
|
||||
```
|
||||
|
||||
在 `config.local.json` 分别填写业务库和 IM 库的地址、端口、库名、账号。`passwordEnv` 指向密码环境变量。按现有数据库要求设置可选的 `ssl` 参数。
|
||||
|
||||
下面的密码输入示例使用 Bash。
|
||||
|
||||
```bash
|
||||
read -r -s -p '业务数据库密码 ' CHAT_REPAIR_BUSINESS_PASSWORD
|
||||
printf '\n'
|
||||
read -r -s -p 'IM 数据库密码 ' CHAT_REPAIR_IM_PASSWORD
|
||||
printf '\n'
|
||||
export CHAT_REPAIR_BUSINESS_PASSWORD CHAT_REPAIR_IM_PASSWORD
|
||||
umask 077
|
||||
```
|
||||
|
||||
`plan`、`verify` 使用只读事务。`apply`、`rollback` 需要相应表的查询、更新、插入权限,`apply` 还需要业务会员表的 ALTER 权限。结构迁移通过独立 SQL 文件执行。
|
||||
|
||||
## 四个命令
|
||||
|
||||
```bash
|
||||
# 提前只读检查,允许新结构尚未安装
|
||||
node repair-im-users.mjs plan --config config.local.json --out runs/preflight.plan.json
|
||||
|
||||
# 维护窗口内,执行结构迁移后,重新生成正式清单
|
||||
node repair-im-users.mjs plan --config config.local.json --out runs/maintenance.plan.json
|
||||
|
||||
# 保持两库写入、旧服务和同步任务暂停
|
||||
node repair-im-users.mjs apply --config config.local.json --plan runs/maintenance.plan.json --maintenance
|
||||
|
||||
# 服务启动前核验资料、关系、消息数量和编号水位
|
||||
node repair-im-users.mjs verify --config config.local.json --plan runs/maintenance.plan.json
|
||||
|
||||
# 启动新版服务并完成后台补偿后,开放流量前再次核验
|
||||
node repair-im-users.mjs verify --config config.local.json --plan runs/maintenance.plan.json --ready
|
||||
|
||||
# 需要回滚时,重新进入维护窗口并暂停服务
|
||||
node repair-im-users.mjs rollback --config config.local.json --plan runs/maintenance.plan.json --maintenance
|
||||
```
|
||||
|
||||
每条命令成功退出码为 0,异常退出码为 1。`--maintenance` 表示操作者已经完成停写,工具依据数据库快照再次检查数据。
|
||||
|
||||
## 清单和执行记录
|
||||
|
||||
| 文件 | 用途 |
|
||||
| --- | --- |
|
||||
| `*.plan.json` | 原字段值、目标字段值、数据库身份、批次号和 SHA256 |
|
||||
| `*.plan.json.summary.json` | 状态数量、缺失用户、同昵称清单、结构预检查 |
|
||||
| `*.plan.json.journal.json` | IM 提交、业务任务提交、自增起点调整和回滚断点 |
|
||||
| `*.plan.json.verification.json` | 最近一次成功核验结果 |
|
||||
|
||||
清单包含会员手机号、头像地址等资料,以及用于核验密码未变化的哈希指纹。请与两库完整备份一起保存至受控目录。生成文件采用私有权限和原子替换。清单、记录、配置文件须保持成套保存。
|
||||
|
||||
## 处理规则
|
||||
|
||||
| 数据分类 | 用户名 | 会员状态 |
|
||||
| --- | --- | --- |
|
||||
| 正常业务会员 | `u<ID>` | 0 |
|
||||
| 禁用业务会员 | `u<ID>` | 1 |
|
||||
| 逻辑删除会员 | `u<ID>` | 2 |
|
||||
| IM 独有正数用户 | `u<ID>` | 3 |
|
||||
| 业务会员缺少 IM 用户 | 创建同步任务,启动新版接口后补建 | 按业务当前状态 |
|
||||
| 系统账号 -100、-200 | 保留专用初始化规则 | 保留并核验 |
|
||||
|
||||
工具使用精确十进制字符串和 BigInt 处理 ID、同步版本和撤销版本。头像、昵称同步至用户、好友和群成员的资料副本。群备注沿用原值。
|
||||
|
||||
正式执行前检查唯一索引、字段容量、InnoDB 引擎、异常 ID、保留账号占用,并通过数据库实际排序规则检查用户名冲突。两库身份包含库名、服务器 UUID、服务器名称和版本。
|
||||
|
||||
## 提交和重跑
|
||||
|
||||
1. 对原始业务资料和 IM 快照重新核对。
|
||||
2. 在一个 IM 事务内,将所有待迁移用户改为批次临时用户名,再统一改为 `u<ID>`,更新资料副本和账号状态。
|
||||
3. 核对用户、关系和消息快照后提交 IM 事务。
|
||||
4. 在业务事务内写入同步任务和持久化编号水位。
|
||||
5. 单独校正会员表 AUTO_INCREMENT,保存完成断点。
|
||||
|
||||
编号水位覆盖业务用户、IM 用户,以及好友、群和历史消息中的用户编号。新版业务注册使用持久化水位分配 ID,可覆盖 MySQL 5.7 重启后重建自增值的情况。
|
||||
|
||||
进程中断或提交响应丢失后,保持停写,使用同一份清单重新运行 `apply`。工具识别已提交快照并补齐后续阶段。完成后的重复执行保持幂等。`imApplying`、`imCommitted`、`businessCommitted` 是执行过程中的正常断点名称。
|
||||
|
||||
## 核验边界
|
||||
|
||||
`verify` 用于服务启动前,要求原有消息数量与快照完全一致。`verify --ready` 允许补建账号、默认好友和欢迎消息,要求原有关系仍存在、消息总量未减少、任务已确认且正常会员已完成初始化。
|
||||
|
||||
核验期间继续暂停用户业务写入。若已经开放流量,会员改资料、新消息或群变动会影响快照核验,应依据新清单及业务记录检查。
|
||||
|
||||
`--ready` 同时要求系统账号存在且可用。该步骤在业务服务运行专用系统账号初始化后执行。
|
||||
|
||||
## 回滚规则
|
||||
|
||||
回滚前再次备份当前两库,暂停同步、写入和所有 IM 服务。
|
||||
|
||||
工具先检查迁移后字段、好友、群成员和消息是否发生后续变化。发现差异会打印字段路径及原值、现值并终止覆盖。回滚采用两阶段改名,恢复原用户名、昵称和头像副本。
|
||||
|
||||
会员停用状态、人工封禁、登录撤销版本和持久化编号水位继续保持当前有效约束。同步版本和登录撤销版本递增,业务任务重新排入补偿队列。数据回滚完成后应保持维护状态,再生成正式清单执行修复,或依据差异进行人工处理。新版同步任务随后执行时会再次采用 `u<ID>`。
|
||||
|
||||
服务启动产生的新好友、欢迎消息,以及恢复流量后的资料和消息变化,都可能触发回滚保护。此时依据完整数据库备份和差异记录制定定向恢复步骤。
|
||||
|
||||
回滚命令只负责本批次资料字段及同步任务,表结构与历史账号继续保留。恢复后启动新版 IM 接口重建 Redis 状态,再启动连接节点和业务服务,按当前账号状态重新领取凭证。
|
||||
|
||||
## 本地测试
|
||||
|
||||
```bash
|
||||
# 算法和清单测试
|
||||
npm test
|
||||
|
||||
# 指向本机隔离 MySQL 测试端口,每个测试创建并清理独立库
|
||||
CHAT_REPAIR_TEST_PORTS=33907,33909 npm test
|
||||
```
|
||||
|
||||
MySQL 测试夹具使用本地专用凭据,限制为显式指定的 `127.0.0.1` 端口。HTTP、WebSocket 测试依赖完整源码树和独立本地服务,运行方式见发布核验报告。
|
||||
@@ -0,0 +1,16 @@
|
||||
{
|
||||
"business": {
|
||||
"host": "127.0.0.1",
|
||||
"port": 3306,
|
||||
"user": "chat_v1",
|
||||
"database": "chat_v1",
|
||||
"passwordEnv": "CHAT_REPAIR_BUSINESS_PASSWORD"
|
||||
},
|
||||
"im": {
|
||||
"host": "127.0.0.1",
|
||||
"port": 3306,
|
||||
"user": "im_platform",
|
||||
"database": "im_platform",
|
||||
"passwordEnv": "CHAT_REPAIR_IM_PASSWORD"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
{
|
||||
"business": {
|
||||
"host": "139.224.25.207",
|
||||
"port": 3306,
|
||||
"user": "chat_v1",
|
||||
"database": "chat_v1",
|
||||
"passwordEnv": "CHAT_REPAIR_BUSINESS_PASSWORD"
|
||||
},
|
||||
"im": {
|
||||
"host": "139.224.25.207",
|
||||
"port": 3306,
|
||||
"user": "im_platform",
|
||||
"database": "im_platform",
|
||||
"passwordEnv": "CHAT_REPAIR_IM_PASSWORD"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,247 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
|
||||
export const MAX_MEMBER_ID = 9223372036854775807n;
|
||||
export const USER_COLUMNS = ['user_name', 'nick_name', 'head_image', 'head_image_thumb',
|
||||
'member_state', 'sync_version', 'auth_version', 'access_disabled'];
|
||||
export const FRIEND_COLUMNS = ['friend_nick_name', 'friend_head_image'];
|
||||
export const GROUP_MEMBER_COLUMNS = ['user_nick_name', 'head_image'];
|
||||
|
||||
export function stable(value) {
|
||||
if (Array.isArray(value)) return value.map(stable);
|
||||
if (value !== null && typeof value === 'object') {
|
||||
return Object.fromEntries(Object.keys(value).sort().map(key => [key, stable(value[key])]));
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
export function fingerprint(value) {
|
||||
return createHash('sha256').update(JSON.stringify(stable(value)) ?? 'undefined').digest('hex');
|
||||
}
|
||||
|
||||
export function assertSame(actual, expected, label) {
|
||||
if (fingerprint(actual) !== fingerprint(expected)) {
|
||||
const differences = [];
|
||||
const inspect = (left, right, path) => {
|
||||
if (differences.length >= 10 || fingerprint(left) === fingerprint(right)) return;
|
||||
if (left != null && right != null && typeof left === 'object' && typeof right === 'object') {
|
||||
for (const key of new Set([...Object.keys(left), ...Object.keys(right)])) inspect(left[key], right[key], `${path}.${key}`);
|
||||
} else {
|
||||
differences.push({ path, expected: right ?? null, actual: left ?? null });
|
||||
}
|
||||
};
|
||||
inspect(actual, expected, label);
|
||||
throw new Error(`${label}与清单不一致,请保持维护状态并检查差异 ${JSON.stringify(differences)}`);
|
||||
}
|
||||
}
|
||||
|
||||
export function memberId(value) {
|
||||
const text = String(value);
|
||||
if (!/^-?[0-9]+$/.test(text)) throw new Error('用户编号须为精确的十进制整数');
|
||||
const id = BigInt(text);
|
||||
if (id > MAX_MEMBER_ID || id < -MAX_MEMBER_ID) throw new Error('用户编号超出 Long 范围');
|
||||
if (id <= 0n && id !== -100n && id !== -200n) throw new Error(`异常用户编号 ${text}`);
|
||||
return id.toString();
|
||||
}
|
||||
|
||||
export function profile(member) {
|
||||
const id = memberId(member.id);
|
||||
if (BigInt(id) <= 0n) throw new Error(`业务会员编号须为正数 ${id}`);
|
||||
const nickname = member.nickname?.trim() ? member.nickname : `u${id}`;
|
||||
if (nickname.length > 30) throw new Error(`会员 ${id} 的昵称超过接口支持的30字符`);
|
||||
return {
|
||||
userId: id,
|
||||
userName: `u${id}`,
|
||||
nickName: nickname,
|
||||
avatar: member.avatar ?? '',
|
||||
phone: member.mobile ?? null,
|
||||
memberState: member.deleted ? 2 : Number(member.status) === 0 ? 0 : 1,
|
||||
};
|
||||
}
|
||||
|
||||
export function temporaryName(id, runId) {
|
||||
return `~${runId.slice(0, 4)}_${BigInt(id).toString(36)}`;
|
||||
}
|
||||
|
||||
export function createPlan(before, identities, runId) {
|
||||
const members = new Map(before.members.map(row => [memberId(row.id), row]));
|
||||
if (members.size !== before.members.length) throw new Error('业务会员编号重复');
|
||||
const users = new Map(before.users.map(row => [memberId(row.id), row]));
|
||||
if (users.size !== before.users.length) throw new Error('IM 用户编号重复');
|
||||
const tasks = new Map(before.tasks.map(row => [memberId(row.member_id), row]));
|
||||
const profiles = new Map([...members].map(([id, member]) => [id, profile(member)]));
|
||||
for (const [id, task] of tasks) {
|
||||
if (!profiles.has(id)) {
|
||||
memberId(id);
|
||||
const previous = JSON.parse(task.profile_json);
|
||||
const user = users.get(id);
|
||||
profiles.set(id, { ...previous, userId: id, userName: `u${id}`,
|
||||
nickName: user?.nick_name ?? previous.nickName,
|
||||
avatar: user?.head_image ?? previous.avatar, memberState: 3 });
|
||||
}
|
||||
}
|
||||
const targetVersions = new Map();
|
||||
const taskChanges = [];
|
||||
let maximum = 0n;
|
||||
const referenced = [
|
||||
...before.friends.flatMap(row => [row.user_id, row.friend_id]),
|
||||
...before.groupMembers.map(row => row.user_id),
|
||||
...before.groups.map(row => row.owner_id),
|
||||
...Object.values(before.messages).flatMap(row => [row.maximumSender, row.maximumReceiver]),
|
||||
].filter(value => value != null);
|
||||
for (const id of [...members.keys(), ...users.keys(), ...tasks.keys(), ...referenced.map(memberId)]) {
|
||||
if (BigInt(id) > maximum) maximum = BigInt(id);
|
||||
}
|
||||
if (maximum === MAX_MEMBER_ID) throw new Error('会员编号已经耗尽,无法安全提高自增起点');
|
||||
for (const [id, desired] of profiles) {
|
||||
const userVersion = BigInt(users.get(id)?.sync_version ?? 0);
|
||||
const taskVersion = BigInt(tasks.get(id)?.version ?? 0);
|
||||
const version = (userVersion > taskVersion ? userVersion : taskVersion) + 1n;
|
||||
if (version > MAX_MEMBER_ID) throw new Error(`会员 ${id} 的同步版本已耗尽`);
|
||||
targetVersions.set(id, version.toString());
|
||||
taskChanges.push({
|
||||
member_id: id,
|
||||
tenant_id: String(members.get(id)?.tenant_id ?? tasks.get(id).tenant_id),
|
||||
profile_json: JSON.stringify(desired),
|
||||
member_state: desired.memberState,
|
||||
version: version.toString(),
|
||||
synced_version: '0',
|
||||
initialized: Boolean(tasks.get(id)?.initialized),
|
||||
retry_count: 0,
|
||||
last_error: '',
|
||||
});
|
||||
}
|
||||
const userChanges = [];
|
||||
for (const user of before.users) {
|
||||
const id = memberId(user.id);
|
||||
if (BigInt(id) < 0n) continue;
|
||||
const desired = profiles.get(id);
|
||||
const state = desired?.memberState ?? 3;
|
||||
const disabled = state !== 0 || Boolean(user.is_banned);
|
||||
let authVersion = BigInt(user.auth_version ?? 0);
|
||||
if (Number(user.member_state ?? 0) !== state || Boolean(user.access_disabled) !== disabled) authVersion++;
|
||||
if (disabled && authVersion === 0n) authVersion = 1n;
|
||||
const version = targetVersions.get(id) ?? (BigInt(user.sync_version ?? 0) + 1n).toString();
|
||||
userChanges.push({
|
||||
id,
|
||||
user_name: `u${id}`,
|
||||
nick_name: desired?.nickName ?? user.nick_name,
|
||||
head_image: desired?.avatar ?? user.head_image,
|
||||
head_image_thumb: desired?.avatar ?? user.head_image_thumb,
|
||||
member_state: state,
|
||||
sync_version: version,
|
||||
auth_version: authVersion.toString(),
|
||||
access_disabled: disabled,
|
||||
});
|
||||
}
|
||||
const changes = new Map(userChanges.map(row => [row.id, row]));
|
||||
const friendChanges = before.friends.filter(row => changes.has(String(row.friend_id))).map(row => {
|
||||
const user = changes.get(String(row.friend_id));
|
||||
return { id: row.id, friend_nick_name: user.nick_name, friend_head_image: user.head_image_thumb };
|
||||
});
|
||||
const groupMemberChanges = before.groupMembers.filter(row => changes.has(String(row.user_id))).map(row => {
|
||||
const user = changes.get(String(row.user_id));
|
||||
return { id: row.id, user_nick_name: user.nick_name, head_image: user.head_image_thumb };
|
||||
});
|
||||
const reserved = new Set(before.users.filter(row => BigInt(row.id) < 0n).map(row => row.user_name.toLowerCase()));
|
||||
for (const task of taskChanges) {
|
||||
if (reserved.has(`u${task.member_id}`)) throw new Error(`会员 ${task.member_id} 的目标用户名与保留账号冲突`);
|
||||
}
|
||||
const temporary = new Set();
|
||||
for (const user of userChanges) {
|
||||
const name = temporaryName(user.id, runId);
|
||||
if (reserved.has(user.user_name.toLowerCase()) || reserved.has(name.toLowerCase()) || temporary.has(name)) {
|
||||
throw new Error(`会员 ${user.id} 的目标用户名与保留账号冲突`);
|
||||
}
|
||||
temporary.add(name);
|
||||
}
|
||||
const plan = {
|
||||
formatVersion: 1,
|
||||
runId,
|
||||
identities,
|
||||
createdAt: new Date().toISOString(),
|
||||
before,
|
||||
userChanges,
|
||||
friendChanges,
|
||||
groupMemberChanges,
|
||||
taskChanges,
|
||||
nextMemberId: (maximum + 1n).toString(),
|
||||
};
|
||||
plan.hash = fingerprint(plan);
|
||||
return plan;
|
||||
}
|
||||
|
||||
export function validatePlan(plan) {
|
||||
const { hash, ...content } = plan;
|
||||
if (plan.formatVersion !== 1 || fingerprint(content) !== hash) throw new Error('修复清单格式或校验值错误');
|
||||
memberId(plan.nextMemberId);
|
||||
}
|
||||
|
||||
function applyChanges(rows, changes) {
|
||||
const byId = new Map(changes.map(row => [String(row.id), row]));
|
||||
return rows.map(row => ({ ...row, ...byId.get(String(row.id)) }));
|
||||
}
|
||||
|
||||
export function expectedIm(plan) {
|
||||
return {
|
||||
users: applyChanges(plan.before.users, plan.userChanges),
|
||||
friends: applyChanges(plan.before.friends, plan.friendChanges),
|
||||
groupMembers: applyChanges(plan.before.groupMembers, plan.groupMemberChanges),
|
||||
groups: plan.before.groups,
|
||||
messages: plan.before.messages,
|
||||
};
|
||||
}
|
||||
|
||||
export function imPart(snapshot) {
|
||||
const { users, friends, groupMembers, groups, messages } = snapshot;
|
||||
return { users, friends, groupMembers, groups, messages };
|
||||
}
|
||||
|
||||
export function rollbackChanges(plan, current) {
|
||||
const expected = expectedIm(plan);
|
||||
const liveUsers = new Map(current.users.map(row => [String(row.id), row]));
|
||||
const expectedUsers = new Map(expected.users.map(row => [String(row.id), row]));
|
||||
const users = [];
|
||||
for (const before of plan.before.users) {
|
||||
if (BigInt(before.id) < 0n) continue;
|
||||
const live = liveUsers.get(String(before.id));
|
||||
if (!live) throw new Error(`会员 ${before.id} 已不存在,停止回滚`);
|
||||
const after = expectedUsers.get(String(before.id));
|
||||
const comparable = { ...live, auth_version: after.auth_version };
|
||||
assertSame(comparable, after, `会员 ${before.id}`);
|
||||
const authVersion = BigInt(live.auth_version) + 1n;
|
||||
const syncVersion = BigInt(live.sync_version) + 1n;
|
||||
users.push({ ...live, user_name: before.user_name, nick_name: before.nick_name,
|
||||
head_image: before.head_image, head_image_thumb: before.head_image_thumb,
|
||||
auth_version: authVersion.toString(), sync_version: syncVersion.toString() });
|
||||
}
|
||||
assertSame(current.friends, expected.friends, '好友资料');
|
||||
assertSame(current.groupMembers, expected.groupMembers, '群成员资料');
|
||||
assertSame(current.groups, expected.groups, '群关系');
|
||||
assertSame(current.messages, expected.messages, '消息数量');
|
||||
return { users, friends: plan.before.friends, groupMembers: plan.before.groupMembers };
|
||||
}
|
||||
|
||||
export function summary(plan) {
|
||||
const states = { active: 0, disabled: 0, deleted: 0, orphan: 0 };
|
||||
const names = ['active', 'disabled', 'deleted', 'orphan'];
|
||||
for (const row of plan.userChanges) states[names[row.member_state]]++;
|
||||
const nicknames = new Map();
|
||||
for (const member of plan.before.members) {
|
||||
const same = nicknames.get(member.nickname) ?? [];
|
||||
same.push(member.id);
|
||||
nicknames.set(member.nickname, same);
|
||||
}
|
||||
const duplicateNicknames = [...nicknames].filter(([, ids]) => ids.length > 1)
|
||||
.map(([nickname, memberIds]) => ({ nickname, memberIds }));
|
||||
return {
|
||||
runId: plan.runId,
|
||||
businessMembers: plan.before.members.length,
|
||||
imUsers: plan.before.users.length,
|
||||
...states,
|
||||
missingImUsers: plan.taskChanges.filter(row => !plan.before.users.some(user => user.id === row.member_id)).length,
|
||||
nextMemberId: plan.nextMemberId,
|
||||
duplicateNicknameGroups: duplicateNicknames.length,
|
||||
duplicateNicknameMembers: duplicateNicknames.reduce((count, group) => count + group.memberIds.length, 0),
|
||||
duplicateNicknames,
|
||||
};
|
||||
}
|
||||
Generated
+138
@@ -0,0 +1,138 @@
|
||||
{
|
||||
"name": "friendship-repair-im-users",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "friendship-repair-im-users",
|
||||
"dependencies": {
|
||||
"mysql2": "3.15.3"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=20.19"
|
||||
}
|
||||
},
|
||||
"node_modules/aws-ssl-profiles": {
|
||||
"version": "1.1.2",
|
||||
"resolved": "https://registry.npmmirror.com/aws-ssl-profiles/-/aws-ssl-profiles-1.1.2.tgz",
|
||||
"integrity": "sha512-NZKeq9AfyQvEeNlN0zSYAaWrmBffJh3IELMZfRpJVWgrpEbtEpnjvzqBPf+mxoI287JohRDoa+/nsfqqiZmF6g==",
|
||||
"license": "MIT",
|
||||
"engines": {
|
||||
"node": ">= 6.0.0"
|
||||
}
|
||||
},
|
||||
"node_modules/denque": {
|
||||
"version": "2.1.0",
|
||||
"resolved": "https://registry.npmmirror.com/denque/-/denque-2.1.0.tgz",
|
||||
"integrity": "sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==",
|
||||
"license": "Apache-2.0",
|
||||
"engines": {
|
||||
"node": ">=0.10"
|
||||
}
|
||||
},
|
||||
"node_modules/generate-function": {
|
||||
"version": "2.3.1",
|
||||
"resolved": "https://registry.npmmirror.com/generate-function/-/generate-function-2.3.1.tgz",
|
||||
"integrity": "sha512-eeB5GfMNeevm/GRYq20ShmsaGcmI81kIX2K9XQx5miC8KdHaC6Jm0qQ8ZNeGOi7wYB8OsdxKs+Y2oVuTFuVwKQ==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"is-property": "^1.0.2"
|
||||
}
|
||||
},
|
||||
"node_modules/iconv-lite": {
|
||||
"version": "0.7.3",
|
||||
"resolved": "https://registry.npmmirror.com/iconv-lite/-/iconv-lite-0.7.3.tgz",
|
||||
"integrity": "sha512-IKXpvIzjnC9XTAUbVBcMfGS0EPaIXtW6v+zr+RRp+hqULEpo0owZax6wyRwPOJbWbzjYspQwusTsfVr0ifh4uQ==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"safer-buffer": ">= 2.1.2 < 3.0.0"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=0.10.0"
|
||||
},
|
||||
"funding": {
|
||||
"type": "opencollective",
|
||||
"url": "https://opencollective.com/express"
|
||||
}
|
||||
},
|
||||
"node_modules/is-property": {
|
||||
"version": "1.0.2",
|
||||
"resolved": "https://registry.npmmirror.com/is-property/-/is-property-1.0.2.tgz",
|
||||
"integrity": "sha512-Ks/IoX00TtClbGQr4TWXemAnktAQvYB7HzcCxDGqEZU6oCmb2INHuOoKxbtR+HFkmYWBKv/dOZtGRiAjDhj92g==",
|
||||
"license": "MIT"
|
||||
},
|
||||
"node_modules/long": {
|
||||
"version": "5.3.2",
|
||||
"resolved": "https://registry.npmmirror.com/long/-/long-5.3.2.tgz",
|
||||
"integrity": "sha512-mNAgZ1GmyNhD7AuqnTG3/VQ26o760+ZYBPKjPvugO8+nLbYfX6TVpJPseBvopbdY+qpZ/lKUnmEc1LeZYS3QAA==",
|
||||
"license": "Apache-2.0"
|
||||
},
|
||||
"node_modules/lru.min": {
|
||||
"version": "1.1.5",
|
||||
"resolved": "https://registry.npmmirror.com/lru.min/-/lru.min-1.1.5.tgz",
|
||||
"integrity": "sha512-5J9ysMYUpYIg9RF2vJpy9SinEmSviFSe0GyPpCQ4L5QSkLAgeLXlTAOu2ZwWUU5m+0SBl6gUU1R1ZQB3aKypfA==",
|
||||
"license": "MIT",
|
||||
"engines": {
|
||||
"bun": ">=1.0.0",
|
||||
"deno": ">=1.30.0",
|
||||
"node": ">=8.0.0"
|
||||
},
|
||||
"funding": {
|
||||
"type": "github",
|
||||
"url": "https://github.com/sponsors/wellwelwel"
|
||||
}
|
||||
},
|
||||
"node_modules/mysql2": {
|
||||
"version": "3.15.3",
|
||||
"resolved": "https://registry.npmmirror.com/mysql2/-/mysql2-3.15.3.tgz",
|
||||
"integrity": "sha512-FBrGau0IXmuqg4haEZRBfHNWB5mUARw6hNwPDXXGg0XzVJ50mr/9hb267lvpVMnhZ1FON3qNd4Xfcez1rbFwSg==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"aws-ssl-profiles": "^1.1.1",
|
||||
"denque": "^2.1.0",
|
||||
"generate-function": "^2.3.1",
|
||||
"iconv-lite": "^0.7.0",
|
||||
"long": "^5.2.1",
|
||||
"lru.min": "^1.0.0",
|
||||
"named-placeholders": "^1.1.3",
|
||||
"seq-queue": "^0.0.5",
|
||||
"sqlstring": "^2.3.2"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">= 8.0"
|
||||
}
|
||||
},
|
||||
"node_modules/named-placeholders": {
|
||||
"version": "1.1.6",
|
||||
"resolved": "https://registry.npmmirror.com/named-placeholders/-/named-placeholders-1.1.6.tgz",
|
||||
"integrity": "sha512-Tz09sEL2EEuv5fFowm419c1+a/jSMiBjI9gHxVLrVdbUkkNUUfjsVYs9pVZu5oCon/kmRh9TfLEObFtkVxmY0w==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"lru.min": "^1.1.0"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=8.0.0"
|
||||
}
|
||||
},
|
||||
"node_modules/safer-buffer": {
|
||||
"version": "2.1.2",
|
||||
"resolved": "https://registry.npmmirror.com/safer-buffer/-/safer-buffer-2.1.2.tgz",
|
||||
"integrity": "sha512-YZo3K82SD7Riyi0E1EQPojLz7kpepnSQI9IyPbHHg1XXXevb5dJI7tpyN2ADxGcQbHG7vcyRHk0cbwqcQriUtg==",
|
||||
"license": "MIT"
|
||||
},
|
||||
"node_modules/seq-queue": {
|
||||
"version": "0.0.5",
|
||||
"resolved": "https://registry.npmmirror.com/seq-queue/-/seq-queue-0.0.5.tgz",
|
||||
"integrity": "sha512-hr3Wtp/GZIc/6DAGPDcV4/9WoZhjrkXsi5B/07QgX8tsdc6ilr7BFM6PM6rbdAX1kFSDYeZGLipIZZKyQP0O5Q=="
|
||||
},
|
||||
"node_modules/sqlstring": {
|
||||
"version": "2.3.3",
|
||||
"resolved": "https://registry.npmmirror.com/sqlstring/-/sqlstring-2.3.3.tgz",
|
||||
"integrity": "sha512-qC9iz2FlN7DQl3+wjwn3802RTyjCx7sDvfQEXchwa6CWOx07/WVfh91gBmQ9fahw8snwGEWU3xGzOt4tFyHLxg==",
|
||||
"license": "MIT",
|
||||
"engines": {
|
||||
"node": ">= 0.6"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
{
|
||||
"name": "friendship-repair-im-users",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"engines": { "node": ">=20.19" },
|
||||
"scripts": { "test": "node --test test/*.test.mjs" },
|
||||
"dependencies": { "mysql2": "3.15.3" }
|
||||
}
|
||||
@@ -0,0 +1,457 @@
|
||||
import mysql from 'mysql2/promise';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { mkdir, open, readFile, rename } from 'node:fs/promises';
|
||||
import { dirname, resolve } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import {
|
||||
USER_COLUMNS, FRIEND_COLUMNS, GROUP_MEMBER_COLUMNS, assertSame, createPlan,
|
||||
expectedIm, fingerprint, imPart, rollbackChanges, summary, temporaryName, validatePlan,
|
||||
} from './core.mjs';
|
||||
|
||||
const TASK_COLUMNS = ['tenant_id', 'profile_json', 'member_state', 'version', 'synced_version',
|
||||
'initialized', 'retry_count', 'last_error'];
|
||||
const NEW_COLUMNS = ['member_state', 'sync_version', 'auth_version', 'access_disabled'];
|
||||
|
||||
function argumentsFor(argv) {
|
||||
const [command, ...args] = argv;
|
||||
const options = {};
|
||||
for (let index = 0; index < args.length; index++) {
|
||||
const key = args[index];
|
||||
if (key === '--maintenance' || key === '--ready') {
|
||||
options[key.slice(2)] = true;
|
||||
} else if (['--config', '--plan', '--out'].includes(key) && args[index + 1]) {
|
||||
options[key.slice(2)] = args[++index];
|
||||
} else {
|
||||
throw new Error(`无法识别的参数 ${key}`);
|
||||
}
|
||||
}
|
||||
if (!['plan', 'apply', 'verify', 'rollback'].includes(command) || !options.config) {
|
||||
throw new Error('用法为 node repair-im-users.mjs plan|apply|verify|rollback --config <配置文件> [--plan <清单>] [--out <清单>] [--maintenance] [--ready]');
|
||||
}
|
||||
if (command !== 'plan' && !options.plan) throw new Error('请使用 --plan 指定已生成的修复清单');
|
||||
if (['apply', 'rollback'].includes(command) && !options.maintenance) {
|
||||
throw new Error('执行前须暂停业务写入、所有同步任务和 IM 服务,再指定 --maintenance');
|
||||
}
|
||||
return { command, options };
|
||||
}
|
||||
|
||||
async function atomicJson(path, value) {
|
||||
await mkdir(dirname(path), { recursive: true, mode: 0o700 });
|
||||
const temporary = `${path}.tmp-${process.pid}`;
|
||||
const handle = await open(temporary, 'w', 0o600);
|
||||
try {
|
||||
await handle.writeFile(JSON.stringify(value, null, 2) + '\n');
|
||||
await handle.sync();
|
||||
} finally {
|
||||
await handle.close();
|
||||
}
|
||||
await rename(temporary, path);
|
||||
}
|
||||
|
||||
async function loadJson(path, fallback) {
|
||||
try {
|
||||
return JSON.parse(await readFile(path, 'utf8'));
|
||||
} catch (error) {
|
||||
if (error.code === 'ENOENT' && fallback !== undefined) return fallback;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function connect(config) {
|
||||
if (!config?.host || !config.user || !config.database || !config.passwordEnv) {
|
||||
throw new Error('每个数据库连接均须明确填写 host、user、database、passwordEnv');
|
||||
}
|
||||
const password = process.env[config.passwordEnv];
|
||||
if (password === undefined) throw new Error(`缺少密码环境变量 ${config.passwordEnv}`);
|
||||
return mysql.createConnection({
|
||||
host: config.host, port: config.port ?? 3306, user: config.user,
|
||||
database: config.database, password, ssl: config.ssl,
|
||||
connectTimeout: 10000, supportBigNumbers: true, bigNumberStrings: true,
|
||||
dateStrings: true, charset: 'utf8mb4', multipleStatements: false,
|
||||
});
|
||||
}
|
||||
|
||||
function normalize(row) {
|
||||
const output = { ...row };
|
||||
for (const key of ['id', 'member_id', 'tenant_id', 'friend_id', 'user_id', 'group_id', 'owner_id',
|
||||
'version', 'synced_version', 'sync_version', 'auth_version']) {
|
||||
if (output[key] != null) output[key] = String(output[key]);
|
||||
}
|
||||
for (const key of ['deleted', 'is_banned', 'access_disabled', 'initialized']) {
|
||||
if (key in output) {
|
||||
const value = output[key];
|
||||
output[key] = Buffer.isBuffer(value) ? value[0] !== 0 : Boolean(Number(value));
|
||||
}
|
||||
}
|
||||
return output;
|
||||
}
|
||||
|
||||
async function rows(connection, sql, parameters = []) {
|
||||
const [result] = await connection.query(sql, parameters);
|
||||
return result.map(normalize);
|
||||
}
|
||||
|
||||
async function identity(connection) {
|
||||
const [row] = await rows(connection, 'SELECT DATABASE() AS db, @@server_uuid AS serverUuid, @@hostname AS serverName, VERSION() AS version');
|
||||
return row;
|
||||
}
|
||||
|
||||
async function inspectSchema(business, im) {
|
||||
const columns = await rows(im, 'SELECT COLUMN_NAME AS name, CHARACTER_MAXIMUM_LENGTH AS maximum, COLLATION_NAME AS collation FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?', ['im_user']);
|
||||
const names = new Set(columns.map(row => row.name));
|
||||
const [indexes] = await im.query('SHOW INDEX FROM im_user');
|
||||
const usernameIndexes = indexes.filter(row => Number(row.Non_unique) === 0 && row.Column_name === 'user_name' && row.Sub_part == null);
|
||||
const hasUniqueUsername = usernameIndexes.some(index => indexes.filter(row => row.Key_name === index.Key_name).length === 1);
|
||||
if (!hasUniqueUsername) throw new Error('IM 用户表须具备完整 user_name 唯一索引');
|
||||
const username = columns.find(row => row.name === 'user_name');
|
||||
if (Number(username?.maximum) < 20) throw new Error('IM 用户名字段长度须至少为20字符');
|
||||
const [tables] = await business.query('SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?', ['member_chat_sync']);
|
||||
const [sequences] = await business.query('SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?', ['member_chat_sync_sequence']);
|
||||
for (const [connection, tables] of [[business, ['member_user', 'member_chat_sync', 'member_chat_sync_sequence']],
|
||||
[im, ['im_user', 'im_friend', 'im_group_member']]]) {
|
||||
const definitions = await rows(connection, 'SELECT TABLE_NAME AS name,ENGINE AS engine FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME IN (?)', [tables]);
|
||||
for (const table of definitions) {
|
||||
if (table.engine !== 'InnoDB') throw new Error(`表 ${table.name} 必须使用 InnoDB 才能保证修复事务完整性`);
|
||||
}
|
||||
}
|
||||
const capacities = await rows(im, 'SELECT TABLE_NAME AS tableName,COLUMN_NAME AS columnName,CHARACTER_MAXIMUM_LENGTH AS maximum FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME IN (?)',
|
||||
[['im_user', 'im_friend', 'im_group_member']]);
|
||||
return { imColumns: [...names], usernameCollation: username.collation,
|
||||
missingImColumns: NEW_COLUMNS.filter(name => !names.has(name)), taskTableExists: tables.length === 1,
|
||||
sequenceTableExists: sequences.length === 1, capacities };
|
||||
}
|
||||
|
||||
async function validateTargets(im, schema, plan) {
|
||||
const tables = { im_user: plan.userChanges, im_friend: plan.friendChanges, im_group_member: plan.groupMemberChanges };
|
||||
for (const definition of schema.capacities) {
|
||||
if (definition.maximum == null) continue;
|
||||
for (const row of tables[definition.tableName]) {
|
||||
const value = row[definition.columnName];
|
||||
if (value != null && Array.from(String(value)).length > Number(definition.maximum)) {
|
||||
throw new Error(`表 ${definition.tableName} 编号 ${row.id} 的 ${definition.columnName} 超出字段长度 ${definition.maximum}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
const temporary = plan.userChanges.map(user => temporaryName(user.id, plan.runId));
|
||||
const finalNames = [...new Set([...plan.userChanges.map(user => user.user_name), ...plan.taskChanges.map(task => `u${task.member_id}`)])];
|
||||
for (const [names, condition] of [[temporary, ''], [finalNames, 'id < 0 AND ']]) {
|
||||
for (let offset = 0; offset < names.length; offset += 500) {
|
||||
const conflicts = await rows(im, `SELECT id,user_name FROM im_user WHERE ${condition}user_name IN (?)`, [names.slice(offset, offset + 500)]);
|
||||
if (conflicts.length) throw new Error(`数据库排序规则下存在保留用户名或临时用户名占用 ${JSON.stringify(conflicts)}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function readMembers(connection, lock = false) {
|
||||
return rows(connection, 'SELECT id,tenant_id,nickname,avatar,mobile,status,deleted FROM member_user ORDER BY id' + (lock ? ' FOR UPDATE' : ''));
|
||||
}
|
||||
|
||||
async function readTasks(connection, exists, lock = false) {
|
||||
if (!exists) return [];
|
||||
return rows(connection, `SELECT member_id,${TASK_COLUMNS.join(',')},deleted FROM member_chat_sync ORDER BY member_id` + (lock ? ' FOR UPDATE' : ''));
|
||||
}
|
||||
|
||||
async function readIm(connection, schema, lock = false) {
|
||||
const defaults = NEW_COLUMNS.map(name => schema.imColumns.includes(name) ? name : `0 AS ${name}`);
|
||||
const suffix = lock ? ' FOR UPDATE' : '';
|
||||
const users = await rows(connection, 'SELECT id,user_name,nick_name,head_image,head_image_thumb,is_banned,'
|
||||
+ defaults.join(',') + ',SHA2(password,256) AS password_hash FROM im_user ORDER BY id' + suffix);
|
||||
const friends = await rows(connection, 'SELECT id,user_id,friend_id,friend_nick_name,friend_head_image FROM im_friend ORDER BY id' + suffix);
|
||||
const groupMembers = await rows(connection, 'SELECT id,group_id,user_id,user_nick_name,head_image,remark_nick_name,remark_group_name,quit FROM im_group_member ORDER BY id' + suffix);
|
||||
const groups = await rows(connection, 'SELECT id,owner_id,dissolve FROM im_group ORDER BY id' + suffix);
|
||||
const messages = {};
|
||||
for (const table of ['im_private_message', 'im_group_message']) {
|
||||
const receiver = table === 'im_private_message' ? ',MAX(recv_id) AS maximumReceiver' : '';
|
||||
const [count] = await rows(connection, `SELECT COUNT(*) AS count,MAX(id) AS maximumId,MAX(send_id) AS maximumSender${receiver} FROM ${table}`);
|
||||
messages[table] = { count: String(count.count), maximumId: count.maximumId == null ? null : String(count.maximumId),
|
||||
maximumSender: count.maximumSender == null ? null : String(count.maximumSender),
|
||||
maximumReceiver: count.maximumReceiver == null ? null : String(count.maximumReceiver) };
|
||||
}
|
||||
return { users, friends, groupMembers, groups, messages };
|
||||
}
|
||||
|
||||
async function readSnapshot(business, im, schema) {
|
||||
return { members: await readMembers(business), tasks: await readTasks(business, schema.taskTableExists),
|
||||
...await readIm(im, schema) };
|
||||
}
|
||||
|
||||
async function updateRows(connection, table, columns, changes) {
|
||||
for (let offset = 0; offset < changes.length; offset += 100) {
|
||||
const batch = changes.slice(offset, offset + 100);
|
||||
const parameters = [];
|
||||
const assignments = columns.map(column => {
|
||||
const cases = batch.map(row => {
|
||||
parameters.push(row.id, row[column]);
|
||||
return 'WHEN ? THEN ?';
|
||||
});
|
||||
return `\`${column}\` = CASE id ${cases.join(' ')} ELSE \`${column}\` END`;
|
||||
});
|
||||
parameters.push(...batch.map(row => row.id));
|
||||
const placeholders = batch.map(() => '?').join(',');
|
||||
await connection.query(`UPDATE ${table} SET ${assignments.join(',')} WHERE id IN (${placeholders})`, parameters);
|
||||
}
|
||||
}
|
||||
|
||||
async function renameUsers(connection, users, runId) {
|
||||
await updateRows(connection, 'im_user', ['user_name'], users.map(user => ({
|
||||
id: user.id, user_name: temporaryName(user.id, runId),
|
||||
})));
|
||||
await updateRows(connection, 'im_user', USER_COLUMNS, users);
|
||||
}
|
||||
|
||||
async function seedTasks(connection, changes) {
|
||||
for (const task of changes) {
|
||||
const columns = ['member_id', ...TASK_COLUMNS];
|
||||
const values = columns.map(column => task[column]);
|
||||
const assignments = TASK_COLUMNS.map(column => `${column} = VALUES(${column})`);
|
||||
await connection.query(`INSERT INTO member_chat_sync (${columns.join(',')},next_retry_time,deleted)
|
||||
VALUES (${columns.map(() => '?').join(',')},CURRENT_TIMESTAMP,b'0')
|
||||
ON DUPLICATE KEY UPDATE ${assignments.join(',')},next_retry_time=CURRENT_TIMESTAMP,deleted=b'0'`, values);
|
||||
}
|
||||
}
|
||||
|
||||
async function requireSchemas(schema) {
|
||||
if (schema.missingImColumns.length || !schema.taskTableExists || !schema.sequenceTableExists) {
|
||||
throw new Error('请先在维护窗口执行两侧结构迁移,再重新生成清单');
|
||||
}
|
||||
}
|
||||
|
||||
async function executeApply(context, plan, journal, saveJournal) {
|
||||
const { business, im, schema } = context;
|
||||
await requireSchemas(schema);
|
||||
if (journal.phase === 'rolledBack') throw new Error('该批次已回滚,请重新生成修复清单');
|
||||
if (journal.phase === 'complete') return;
|
||||
assertSame(await readMembers(business), plan.before.members, '业务会员资料');
|
||||
const current = await readIm(im, schema);
|
||||
const after = expectedIm(plan);
|
||||
if (fingerprint(current) !== fingerprint(after)) {
|
||||
assertSame(current, imPart(plan.before), 'IM 原始资料');
|
||||
journal.phase = 'imApplying';
|
||||
await saveJournal();
|
||||
await im.beginTransaction();
|
||||
try {
|
||||
assertSame(await readIm(im, schema, true), imPart(plan.before), 'IM 锁定资料');
|
||||
await renameUsers(im, plan.userChanges, plan.runId);
|
||||
await updateRows(im, 'im_friend', FRIEND_COLUMNS, plan.friendChanges);
|
||||
await updateRows(im, 'im_group_member', GROUP_MEMBER_COLUMNS, plan.groupMemberChanges);
|
||||
assertSame(await readIm(im, schema), after, 'IM 修复结果');
|
||||
await im.commit();
|
||||
} catch (error) {
|
||||
await im.rollback();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
journal.phase = 'imCommitted';
|
||||
await saveJournal();
|
||||
await business.beginTransaction();
|
||||
try {
|
||||
assertSame(await readMembers(business, true), plan.before.members, '业务锁定资料');
|
||||
const existing = await readTasks(business, true, true);
|
||||
const desired = plan.taskChanges.map(task => ({ ...task, deleted: false }));
|
||||
const expectedTasks = new Map(plan.before.tasks.map(task => [task.member_id, task]));
|
||||
for (const task of desired) expectedTasks.set(task.member_id, task);
|
||||
const expected = [...expectedTasks.values()].sort((left, right) => BigInt(left.member_id) < BigInt(right.member_id) ? -1 : 1);
|
||||
if (fingerprint(existing) !== fingerprint(expected)) {
|
||||
assertSame(existing, plan.before.tasks, '业务同步任务');
|
||||
await seedTasks(business, plan.taskChanges);
|
||||
}
|
||||
await business.query('UPDATE member_chat_sync_sequence SET last_id = GREATEST(last_id, ?) WHERE id = 1',
|
||||
[(BigInt(plan.nextMemberId) - 1n).toString()]);
|
||||
await business.commit();
|
||||
} catch (error) {
|
||||
await business.rollback();
|
||||
throw error;
|
||||
}
|
||||
journal.phase = 'businessCommitted';
|
||||
await saveJournal();
|
||||
// AUTO_INCREMENT is DDL and is deliberately a separately resumable phase.
|
||||
const [definition] = await rows(business, 'SHOW CREATE TABLE member_user');
|
||||
const nextId = definition['Create Table'].match(/\bAUTO_INCREMENT=(\d+)/)?.[1] ?? '1';
|
||||
if (BigInt(nextId) < BigInt(plan.nextMemberId)) {
|
||||
await business.query(`ALTER TABLE member_user AUTO_INCREMENT = ${plan.nextMemberId}`);
|
||||
}
|
||||
journal.phase = 'complete';
|
||||
await saveJournal();
|
||||
}
|
||||
|
||||
async function executeRollback(context, plan, journal, saveJournal) {
|
||||
const { business, im, schema } = context;
|
||||
await requireSchemas(schema);
|
||||
if (journal.phase === 'rolledBack') return;
|
||||
if (!['imApplying', 'imCommitted', 'businessCommitted', 'complete', 'rollingBack', 'imRolledBack'].includes(journal.phase)) {
|
||||
throw new Error('该批次尚未开始执行');
|
||||
}
|
||||
assertSame(await readMembers(business), plan.before.members, '业务会员资料');
|
||||
if (journal.phase !== 'imRolledBack') {
|
||||
await im.beginTransaction();
|
||||
try {
|
||||
const current = await readIm(im, schema, true);
|
||||
if (journal.rollbackUsers && fingerprint(current.users) === fingerprint(journal.rollbackUsers)) {
|
||||
// A preceding commit may have succeeded before its acknowledgement was written.
|
||||
} else {
|
||||
const changes = rollbackChanges(plan, current);
|
||||
journal.phase = 'rollingBack';
|
||||
const byId = new Map(changes.users.map(user => [String(user.id), user]));
|
||||
journal.rollbackUsers = current.users.map(user => byId.get(String(user.id)) ?? user);
|
||||
await saveJournal();
|
||||
await renameUsers(im, changes.users, randomUUID().replaceAll('-', ''));
|
||||
await updateRows(im, 'im_friend', FRIEND_COLUMNS, changes.friends);
|
||||
await updateRows(im, 'im_group_member', GROUP_MEMBER_COLUMNS, changes.groupMembers);
|
||||
}
|
||||
await im.commit();
|
||||
} catch (error) {
|
||||
await im.rollback();
|
||||
throw error;
|
||||
}
|
||||
journal.phase = 'imRolledBack';
|
||||
await saveJournal();
|
||||
}
|
||||
await business.beginTransaction();
|
||||
try {
|
||||
// Keep versions monotonic while restoring the pre-repair desired profiles.
|
||||
const currentTasks = new Map((await readTasks(business, true, true)).map(task => [task.member_id, task]));
|
||||
const restored = [];
|
||||
const previousTasks = new Map(plan.before.tasks.map(task => [task.member_id, task]));
|
||||
const users = new Map(journal.rollbackUsers.map(user => [String(user.id), user]));
|
||||
for (const task of plan.taskChanges) {
|
||||
const current = currentTasks.get(task.member_id);
|
||||
if (current && current.version !== task.version && current.version !== previousTasks.get(task.member_id)?.version
|
||||
&& current.version !== journal.restoredVersions?.[task.member_id]) {
|
||||
throw new Error(`会员 ${task.member_id} 的同步任务已变化,停止回滚`);
|
||||
}
|
||||
const base = BigInt(users.get(task.member_id)?.sync_version ?? task.version);
|
||||
const version = journal.restoredVersions?.[task.member_id] ?? (base + 1n).toString();
|
||||
restored.push({ ...task, member_id: task.member_id, version, synced_version: '0', retry_count: 0, last_error: '' });
|
||||
}
|
||||
journal.restoredVersions = Object.fromEntries(restored.map(task => [task.member_id, task.version]));
|
||||
await saveJournal();
|
||||
await seedTasks(business, restored);
|
||||
await business.commit();
|
||||
} catch (error) {
|
||||
await business.rollback();
|
||||
throw error;
|
||||
}
|
||||
journal.phase = 'rolledBack';
|
||||
await saveJournal();
|
||||
}
|
||||
|
||||
async function verify(context, plan, ready) {
|
||||
const { business, im, schema } = context;
|
||||
await requireSchemas(schema);
|
||||
const live = await readIm(im, schema);
|
||||
const users = new Map(live.users.map(user => [user.id, user]));
|
||||
const tasks = new Map((await readTasks(business, true)).map(task => [task.member_id, task]));
|
||||
assertSame(await readMembers(business), plan.before.members, '业务会员资料');
|
||||
for (const expected of plan.userChanges) {
|
||||
const actual = users.get(expected.id);
|
||||
if (!actual) throw new Error(`IM 会员 ${expected.id} 缺失`);
|
||||
const original = plan.before.users.find(user => user.id === expected.id);
|
||||
assertSame(actual.password_hash, original.password_hash, `会员 ${expected.id} 的原密码校验值`);
|
||||
for (const column of USER_COLUMNS) {
|
||||
if (column === 'auth_version') {
|
||||
if (BigInt(actual[column]) < BigInt(expected[column])) throw new Error('登录撤销版本发生倒退');
|
||||
} else {
|
||||
assertSame(actual[column], expected[column], `会员 ${expected.id} 的 ${column}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
let pending = 0;
|
||||
let pendingInitialization = 0;
|
||||
for (const expected of plan.taskChanges) {
|
||||
const task = tasks.get(expected.member_id);
|
||||
if (!task || task.version !== expected.version) throw new Error(`会员 ${expected.member_id} 的任务版本不匹配`);
|
||||
if (BigInt(task.synced_version) < BigInt(task.version)) pending++;
|
||||
if (task.member_state === 0 && !task.initialized) pendingInitialization++;
|
||||
if (ready && !users.has(expected.member_id)) throw new Error(`会员 ${expected.member_id} 尚未完成 IM 补建`);
|
||||
if (ready) {
|
||||
const desired = JSON.parse(expected.profile_json);
|
||||
const user = users.get(expected.member_id);
|
||||
assertSame(user.user_name, desired.userName, `会员 ${expected.member_id} 的用户名`);
|
||||
assertSame(user.member_state, desired.memberState, `会员 ${expected.member_id} 的可用状态`);
|
||||
assertSame(user.sync_version, expected.version, `会员 ${expected.member_id} 的应用版本`);
|
||||
}
|
||||
}
|
||||
const expected = expectedIm(plan);
|
||||
const liveFriends = new Map(live.friends.map(row => [row.id, row]));
|
||||
const liveGroupMembers = new Map(live.groupMembers.map(row => [row.id, row]));
|
||||
for (const friend of expected.friends) assertSame(liveFriends.get(friend.id), friend, `好友关系 ${friend.id}`);
|
||||
for (const member of expected.groupMembers) assertSame(liveGroupMembers.get(member.id), member, `群成员关系 ${member.id}`);
|
||||
assertSame(live.groups, plan.before.groups, '群关系');
|
||||
if (!ready) assertSame(live.messages, plan.before.messages, '消息数量');
|
||||
const [counter] = await rows(business, 'SELECT last_id FROM member_chat_sync_sequence WHERE id = 1');
|
||||
if (BigInt(counter?.last_id ?? 0) < BigInt(plan.nextMemberId) - 1n) throw new Error('业务持久化编号水位未提升至安全起点');
|
||||
if (ready && (pending || pendingInitialization)) throw new Error(`仍有 ${pending} 个待同步账号、${pendingInitialization} 个待初始化账号`);
|
||||
if (ready) {
|
||||
for (const [table, before] of Object.entries(plan.before.messages)) {
|
||||
if (BigInt(live.messages[table].count) < BigInt(before.count)) throw new Error(`表 ${table} 的消息数量减少`);
|
||||
}
|
||||
for (const id of ['-100', '-200']) {
|
||||
const user = users.get(id);
|
||||
if (!user || user.member_state !== 0 || user.is_banned) throw new Error(`系统账号 ${id} 尚未正常初始化`);
|
||||
}
|
||||
}
|
||||
return { ...summary(plan), verified: true, pending, pendingInitialization, ready: Boolean(ready) };
|
||||
}
|
||||
|
||||
export async function main(argv = process.argv.slice(2)) {
|
||||
const { command, options } = argumentsFor(argv);
|
||||
const config = await loadJson(resolve(options.config));
|
||||
const business = await connect(config.business);
|
||||
let im;
|
||||
try {
|
||||
im = await connect(config.im);
|
||||
const identities = { business: await identity(business), im: await identity(im) };
|
||||
const schema = await inspectSchema(business, im);
|
||||
const context = { business, im, schema };
|
||||
if (command === 'plan' || command === 'verify') {
|
||||
await business.query('SET SESSION TRANSACTION READ ONLY');
|
||||
await im.query('SET SESSION TRANSACTION READ ONLY');
|
||||
await business.query('START TRANSACTION READ ONLY');
|
||||
await im.query('START TRANSACTION READ ONLY');
|
||||
}
|
||||
if (command === 'plan') {
|
||||
const before = await readSnapshot(business, im, schema);
|
||||
const plan = createPlan(before, identities, randomUUID().replaceAll('-', ''));
|
||||
await validateTargets(im, schema, plan);
|
||||
const path = resolve(options.out ?? `runs/${plan.runId}.plan.json`);
|
||||
await atomicJson(path, plan);
|
||||
const report = { ...summary(plan), createdAt: plan.createdAt,
|
||||
schemaReady: !schema.missingImColumns.length && schema.taskTableExists && schema.sequenceTableExists,
|
||||
missingBusinessTables: [!schema.taskTableExists ? 'member_chat_sync' : null,
|
||||
!schema.sequenceTableExists ? 'member_chat_sync_sequence' : null].filter(Boolean),
|
||||
missingImColumns: schema.missingImColumns, usernameCollation: schema.usernameCollation, plan: path };
|
||||
await atomicJson(`${path}.summary.json`, report);
|
||||
console.log(JSON.stringify(report, null, 2));
|
||||
return;
|
||||
}
|
||||
const path = resolve(options.plan);
|
||||
const plan = await loadJson(path);
|
||||
validatePlan(plan);
|
||||
assertSame(identities, plan.identities, '数据库身份');
|
||||
if (command === 'verify') {
|
||||
const report = await verify(context, plan, options.ready);
|
||||
await atomicJson(`${path}.verification.json`, report);
|
||||
console.log(JSON.stringify(report, null, 2));
|
||||
return;
|
||||
}
|
||||
const journalPath = `${path}.journal.json`;
|
||||
const journal = await loadJson(journalPath, { runId: plan.runId, planHash: plan.hash, phase: 'created' });
|
||||
if (journal.planHash !== plan.hash) throw new Error('执行记录与清单不匹配');
|
||||
const saveJournal = () => atomicJson(journalPath, { ...journal, updatedAt: new Date().toISOString() });
|
||||
if (command === 'apply') await executeApply(context, plan, journal, saveJournal);
|
||||
if (command === 'rollback') await executeRollback(context, plan, journal, saveJournal);
|
||||
console.log(JSON.stringify({ ...summary(plan), phase: journal.phase, journal: journalPath }, null, 2));
|
||||
} finally {
|
||||
await business.rollback().catch(() => {});
|
||||
await im?.rollback().catch(() => {});
|
||||
await business.end();
|
||||
await im?.end();
|
||||
}
|
||||
}
|
||||
|
||||
if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
|
||||
main().catch(error => {
|
||||
console.error(error.message);
|
||||
process.exitCode = 1;
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { assertSame, createPlan, expectedIm, memberId, profile, rollbackChanges, temporaryName, validatePlan } from '../core.mjs';
|
||||
|
||||
function user(id, name) {
|
||||
return { id: String(id), user_name: name, nick_name: name, head_image: '', head_image_thumb: '',
|
||||
member_state: 0, sync_version: '0', auth_version: '0', access_disabled: false, is_banned: false, password_hash: 'preserved' };
|
||||
}
|
||||
|
||||
function fixture() {
|
||||
return {
|
||||
members: [
|
||||
{ id: '100', tenant_id: '1', nickname: '老芋头', avatar: '', mobile: null, status: 1, deleted: true },
|
||||
{ id: '131', tenant_id: '1', nickname: '老芋头', avatar: 'new-avatar', mobile: null, status: 0, deleted: false },
|
||||
],
|
||||
tasks: [],
|
||||
users: [user('-200', '客服'), user('-100', '消息'), user('100', 'u131'), user('131', 'u100'), user('900', '孤立账号')],
|
||||
friends: [{ id: '1', user_id: '1', friend_id: '131', friend_nick_name: '旧昵称', friend_head_image: '' }],
|
||||
groupMembers: [{ id: '1', group_id: '1', user_id: '131', user_nick_name: '旧昵称', head_image: '', remark_nick_name: '我的备注' }],
|
||||
groups: [{ id: '1', owner_id: '100', dissolve: 0 }],
|
||||
messages: { private: { count: '10', maximumId: '10' } },
|
||||
};
|
||||
}
|
||||
|
||||
test('same nicknames remain distinct and deleted/orphan accounts are archived', () => {
|
||||
const plan = createPlan(fixture(), {}, 'abcdef');
|
||||
assert.deepEqual(plan.userChanges.map(row => [row.id, row.user_name, row.member_state]),
|
||||
[['100', 'u100', 2], ['131', 'u131', 0], ['900', 'u900', 3]]);
|
||||
assert.equal(plan.nextMemberId, '901');
|
||||
assert.equal(plan.userChanges[0].auth_version, '1');
|
||||
assert.equal(plan.userChanges[1].auth_version, '0');
|
||||
assert.equal(expectedIm(plan).groupMembers[0].remark_nick_name, '我的备注');
|
||||
assert.equal(expectedIm(plan).users[3].password_hash, 'preserved');
|
||||
});
|
||||
|
||||
test('IDs and versions preserve precision beyond the JavaScript safe-integer range', () => {
|
||||
const before = fixture();
|
||||
before.members[1].id = '9007199254740993';
|
||||
before.users[3].id = '9007199254740993';
|
||||
before.users[3].sync_version = '9007199254740994';
|
||||
const plan = createPlan(before, {}, 'abcdef');
|
||||
assert.equal(plan.nextMemberId, '9007199254740994');
|
||||
assert.equal(plan.userChanges[1].user_name, 'u9007199254740993');
|
||||
assert.equal(plan.userChanges[1].sync_version, '9007199254740995');
|
||||
});
|
||||
|
||||
test('historical message references reserve IDs even when the old IM user row is absent', () => {
|
||||
const before = fixture();
|
||||
before.messages.private.maximumSender = '9007199254740993';
|
||||
assert.equal(createPlan(before, {}, 'abcdef').nextMemberId, '9007199254740994');
|
||||
});
|
||||
|
||||
test('precheck rejects invalid historical identities and differences identify missing fields', () => {
|
||||
const before = fixture();
|
||||
before.messages.private.maximumSender = '9223372036854775808';
|
||||
assert.throws(() => createPlan(before, {}, 'abcdef'), /Long/);
|
||||
assert.throws(() => assertSame({ id: '131' }, { id: '131', nickname: '原昵称' }, '会员'), /会员.nickname/);
|
||||
});
|
||||
|
||||
test('maximum Long fits protocol and temporary usernames', () => {
|
||||
const member = fixture().members[1];
|
||||
member.id = '9223372036854775807';
|
||||
assert.equal(profile(member).userName.length, 20);
|
||||
assert.ok(temporaryName(member.id, 'abcdef').length <= 20);
|
||||
assert.throws(() => memberId('9223372036854775808'));
|
||||
assert.throws(() => memberId(0));
|
||||
assert.throws(() => memberId(-1));
|
||||
});
|
||||
|
||||
test('unknown accounts in the durable queue receive orphan snapshots', () => {
|
||||
const before = fixture();
|
||||
before.tasks.push({ member_id: '900', tenant_id: '1', version: '9',
|
||||
profile_json: JSON.stringify({ userId: '900', nickName: 'old' }) });
|
||||
const plan = createPlan(before, {}, 'abcdef');
|
||||
assert.equal(plan.taskChanges[2].member_state, 3);
|
||||
assert.equal(plan.taskChanges[2].version, '10');
|
||||
assert.equal(plan.userChanges[2].sync_version, '10');
|
||||
});
|
||||
|
||||
test('precheck catches system namespace conflicts and nickname limits', () => {
|
||||
const before = fixture();
|
||||
before.users[0].user_name = 'U131';
|
||||
assert.throws(() => createPlan(before, {}, 'abcdef'), /冲突/);
|
||||
before.users[0].user_name = '客服';
|
||||
before.members[1].nickname = '字'.repeat(31);
|
||||
assert.throws(() => createPlan(before, {}, 'abcdef'), /30字符/);
|
||||
});
|
||||
|
||||
test('plan checksum detects edits', () => {
|
||||
const plan = createPlan(fixture(), {}, 'abcdef');
|
||||
validatePlan(plan);
|
||||
plan.userChanges[0].user_name = 'other';
|
||||
assert.throws(() => validatePlan(plan), /校验值/);
|
||||
});
|
||||
|
||||
test('rollback restores profiles and advances both versions', () => {
|
||||
const plan = createPlan(fixture(), {}, 'abcdef');
|
||||
const current = expectedIm(plan);
|
||||
const restored = rollbackChanges(plan, current);
|
||||
assert.equal(restored.users[0].user_name, 'u131');
|
||||
assert.equal(restored.users[0].auth_version, '2');
|
||||
assert.equal(restored.users[0].sync_version, '2');
|
||||
assert.equal(restored.users[0].member_state, 2);
|
||||
assert.equal(restored.users[0].access_disabled, true);
|
||||
current.friends[0].friend_nick_name = 'later edit';
|
||||
assert.throws(() => rollbackChanges(plan, current), /清单不一致/);
|
||||
});
|
||||
|
||||
test('rollback rejects later profile, relation and message changes', () => {
|
||||
const plan = createPlan(fixture(), {}, 'abcdef');
|
||||
const current = expectedIm(plan);
|
||||
current.users[3].nick_name = 'later edit';
|
||||
assert.throws(() => rollbackChanges(plan, current), /清单不一致/);
|
||||
});
|
||||
@@ -0,0 +1,177 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import mysql from 'mysql2/promise';
|
||||
import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { resolve } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { main } from '../repair-im-users.mjs';
|
||||
|
||||
const ports = (process.env.CHAT_REPAIR_TEST_PORTS ?? '').split(',').filter(Boolean).map(Number);
|
||||
const apiRoot = fileURLToPath(new URL('../../../', import.meta.url));
|
||||
const imRoot = resolve(apiRoot, '../Friendship-and-chat-im/box-im-master');
|
||||
process.env.CHAT_REPAIR_FIXTURE_PASSWORD = 'chat_sync_test';
|
||||
|
||||
async function fixture(port) {
|
||||
const token = randomUUID().replaceAll('-', '');
|
||||
const businessName = `codex_chat_business_${token}`;
|
||||
const imName = `codex_chat_im_${token}`;
|
||||
const base = { host: '127.0.0.1', port, user: 'root', password: 'chat_sync_test',
|
||||
supportBigNumbers: true, bigNumberStrings: true, multipleStatements: true };
|
||||
const admin = await mysql.createConnection(base);
|
||||
const directory = await mkdtemp(resolve(tmpdir(), 'chat-repair-test-'));
|
||||
const config = resolve(directory, 'config.json');
|
||||
const plan = resolve(directory, 'plan.json');
|
||||
await admin.query(`CREATE DATABASE ${businessName} CHARACTER SET utf8mb4; CREATE DATABASE ${imName} CHARACTER SET utf8mb4`);
|
||||
const business = await mysql.createConnection({ ...base, database: businessName });
|
||||
const im = await mysql.createConnection({ ...base, database: imName });
|
||||
try {
|
||||
await business.query(`CREATE TABLE member_user (
|
||||
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, tenant_id BIGINT NOT NULL DEFAULT 1,
|
||||
nickname VARCHAR(30) NOT NULL DEFAULT '', avatar VARCHAR(255), mobile VARCHAR(11),
|
||||
status TINYINT NOT NULL DEFAULT 0, deleted BIT(1) NOT NULL DEFAULT b'0') ENGINE=InnoDB;
|
||||
CREATE TABLE infra_job (id BIGINT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(32), status TINYINT,
|
||||
handler_name VARCHAR(64),handler_param VARCHAR(255),cron_expression VARCHAR(32),retry_count INT,
|
||||
retry_interval INT,monitor_timeout INT,deleted BIT(1) DEFAULT b'0');`);
|
||||
await business.query(await readFile(resolve(apiRoot, 'sql/mysql/member/member_chat_sync.sql'), 'utf8'));
|
||||
await im.query(await readFile(resolve(imRoot, 'db/im-platform.sql'), 'utf8'));
|
||||
await business.query(`INSERT INTO member_user (id,nickname,status,deleted) VALUES
|
||||
(100,'老芋头',1,b'1'),(131,'老芋头',0,b'0'),(501,'新会员',0,b'0')`);
|
||||
await im.query(`INSERT INTO im_user (id,user_name,nick_name,password) VALUES
|
||||
(-100,'消息通知','消息通知','retained'),(-200,'客服','客服','retained'),
|
||||
(100,'u131','旧100','retained'),(131,'u100','旧131','retained'),(900,'孤立账号','孤立账号','retained');
|
||||
INSERT INTO im_friend (id,user_id,friend_id,friend_nick_name,friend_head_image) VALUES (1,100,131,'旧131','');
|
||||
INSERT INTO im_group (id,owner_id,name) VALUES (1,100,'原群');
|
||||
INSERT INTO im_group_member (id,group_id,user_id,user_nick_name,remark_nick_name,head_image)
|
||||
VALUES (1,1,131,'旧131','原备注','');
|
||||
INSERT INTO im_private_message (send_id,recv_id,content,type,status) VALUES (100,131,'原消息',0,3);`);
|
||||
const connection = database => ({ host: '127.0.0.1', port, user: 'root', database,
|
||||
passwordEnv: 'CHAT_REPAIR_FIXTURE_PASSWORD' });
|
||||
await writeFile(config, JSON.stringify({ business: connection(businessName), im: connection(imName) }));
|
||||
return {
|
||||
business, im, config, plan,
|
||||
run: (command, ...extra) => main([command, '--config', config, '--plan', plan, ...extra]),
|
||||
async close() {
|
||||
await business.end();
|
||||
await im.end();
|
||||
await admin.query(`DROP DATABASE ${businessName}; DROP DATABASE ${imName}`);
|
||||
await admin.end();
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
await business.end();
|
||||
await im.end();
|
||||
await admin.query(`DROP DATABASE IF EXISTS ${businessName}; DROP DATABASE IF EXISTS ${imName}`);
|
||||
await admin.end();
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
test('real MySQL repair integration requires explicit isolated localhost ports', { skip: ports.length !== 0 }, () => {});
|
||||
|
||||
for (const port of ports) {
|
||||
test(`MySQL ${port} prechecks field capacity, actual collation and transactional storage`, async () => {
|
||||
const context = await fixture(port);
|
||||
try {
|
||||
await context.im.query('ALTER TABLE im_user MODIFY head_image VARCHAR(4)');
|
||||
await context.business.query("UPDATE member_user SET avatar='too-long' WHERE id=131");
|
||||
await assert.rejects(context.run('plan', '--out', context.plan), /head_image.*长度/);
|
||||
await context.business.query("UPDATE member_user SET avatar='' WHERE id=131");
|
||||
await context.im.query('ALTER TABLE im_user MODIFY user_name VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL');
|
||||
await context.im.query("UPDATE im_user SET user_name='u501' WHERE id=-100");
|
||||
await assert.rejects(context.run('plan', '--out', context.plan), /排序规则/);
|
||||
await context.im.query("UPDATE im_user SET user_name='消息通知' WHERE id=-100");
|
||||
await context.im.query('ALTER TABLE im_friend ENGINE=MyISAM');
|
||||
await assert.rejects(context.run('plan', '--out', context.plan), /InnoDB/);
|
||||
} finally {
|
||||
await context.close();
|
||||
}
|
||||
});
|
||||
|
||||
test(`MySQL ${port} repairs cyclic names, repeats safely and rolls back`, async () => {
|
||||
const context = await fixture(port);
|
||||
try {
|
||||
await context.run('plan', '--out', context.plan);
|
||||
await context.run('apply', '--maintenance');
|
||||
await context.run('apply', '--maintenance');
|
||||
await context.run('verify');
|
||||
const [users] = await context.im.query('SELECT id,user_name,member_state,auth_version FROM im_user WHERE id > 0 ORDER BY id');
|
||||
assert.deepEqual(users.map(user => [user.id, user.user_name, user.member_state]),
|
||||
[['100', 'u100', 2], ['131', 'u131', 0], ['900', 'u900', 3]]);
|
||||
const [[counter]] = await context.business.query('SELECT last_id FROM member_chat_sync_sequence WHERE id=1');
|
||||
assert.equal(String(counter.last_id), '900');
|
||||
await assert.rejects(context.run('verify', '--ready'), /尚未完成|待同步/);
|
||||
await context.im.query("UPDATE im_user SET nick_name='后续修改' WHERE id=131");
|
||||
await assert.rejects(context.run('rollback', '--maintenance'), /清单不一致/);
|
||||
await context.im.query("UPDATE im_user SET nick_name='老芋头' WHERE id=131");
|
||||
await context.run('rollback', '--maintenance');
|
||||
await context.run('rollback', '--maintenance');
|
||||
const [[restored]] = await context.im.query('SELECT user_name,auth_version FROM im_user WHERE id=100');
|
||||
assert.equal(restored.user_name, 'u131');
|
||||
assert.ok(BigInt(restored.auth_version) > BigInt(users[0].auth_version));
|
||||
const [[message]] = await context.im.query('SELECT COUNT(*) AS n FROM im_private_message');
|
||||
assert.equal(Number(message.n), 1);
|
||||
} finally {
|
||||
await context.close();
|
||||
}
|
||||
});
|
||||
|
||||
test(`MySQL ${port} resumes after lost commit acknowledgement`, async () => {
|
||||
const context = await fixture(port);
|
||||
try {
|
||||
await context.run('plan', '--out', context.plan);
|
||||
await context.run('apply', '--maintenance');
|
||||
const journalPath = `${context.plan}.journal.json`;
|
||||
const journal = JSON.parse(await readFile(journalPath, 'utf8'));
|
||||
journal.phase = 'imApplying';
|
||||
await writeFile(journalPath, JSON.stringify(journal));
|
||||
await context.run('apply', '--maintenance');
|
||||
await context.run('verify');
|
||||
} finally {
|
||||
await context.close();
|
||||
}
|
||||
});
|
||||
|
||||
test(`MySQL ${port} verifies readiness after missing users and initialization complete`, async () => {
|
||||
const context = await fixture(port);
|
||||
try {
|
||||
await context.run('plan', '--out', context.plan);
|
||||
await context.run('apply', '--maintenance');
|
||||
const plan = JSON.parse(await readFile(context.plan, 'utf8'));
|
||||
const missing = plan.taskChanges.find(task => task.member_id === '501');
|
||||
const desired = JSON.parse(missing.profile_json);
|
||||
await context.im.query('INSERT INTO im_user (id,user_name,nick_name,password,head_image,head_image_thumb,member_state,sync_version) VALUES (?,?,?,?,?,?,?,?)',
|
||||
['501', desired.userName, desired.nickName, 'new-random-password-hash', desired.avatar, desired.avatar, desired.memberState, missing.version]);
|
||||
await context.business.query("UPDATE member_chat_sync SET synced_version=version,initialized=b'1'");
|
||||
await context.im.query("INSERT INTO im_friend (id,user_id,friend_id,friend_nick_name) VALUES (2,501,-200,'客服')");
|
||||
await context.im.query("INSERT INTO im_private_message (send_id,recv_id,content,type,status) VALUES (-200,501,'欢迎',0,0)");
|
||||
await context.run('verify', '--ready');
|
||||
const report = JSON.parse(await readFile(`${context.plan}.verification.json`, 'utf8'));
|
||||
assert.equal(report.verified, true);
|
||||
assert.equal(report.pending, 0);
|
||||
assert.equal(report.pendingInitialization, 0);
|
||||
} finally {
|
||||
await context.close();
|
||||
}
|
||||
});
|
||||
|
||||
test(`MySQL ${port} rejects database identity and concurrent source changes`, async () => {
|
||||
const context = await fixture(port);
|
||||
try {
|
||||
await context.run('plan', '--out', context.plan);
|
||||
await context.business.query("UPDATE member_user SET nickname='后续修改' WHERE id=131");
|
||||
await assert.rejects(context.run('apply', '--maintenance'), /清单不一致/);
|
||||
const [[user]] = await context.im.query('SELECT user_name FROM im_user WHERE id=100');
|
||||
assert.equal(user.user_name, 'u131');
|
||||
const config = JSON.parse(await readFile(context.config, 'utf8'));
|
||||
[config.business.database, config.im.database] = [config.im.database, config.business.database];
|
||||
await writeFile(context.config, JSON.stringify(config));
|
||||
await assert.rejects(context.run('apply', '--maintenance'));
|
||||
} finally {
|
||||
await context.close();
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createHash } from 'node:crypto';
|
||||
import { readFile } from 'node:fs/promises';
|
||||
|
||||
const endpoint = process.env.CHAT_SYNC_RUNTIME_HTTP;
|
||||
const socketEndpoint = process.env.CHAT_SYNC_RUNTIME_SOCKET;
|
||||
const secondSocketEndpoint = process.env.CHAT_SYNC_RUNTIME_SOCKET_SECOND;
|
||||
|
||||
test('local HTTP and WebSocket enforce disable, restore and credential revocation', {
|
||||
skip: !endpoint || !socketEndpoint,
|
||||
timeout: 30000,
|
||||
}, async () => {
|
||||
assert.match(endpoint, /^http:\/\/(127\.0\.0\.1|localhost):[0-9]+$/);
|
||||
assert.match(socketEndpoint, /^ws:\/\/(127\.0\.0\.1|localhost):[0-9]+$/);
|
||||
assert.equal(typeof WebSocket, 'function', 'Node 20 requires --experimental-websocket');
|
||||
const source = await readFile(new URL('../../../../Friendship-and-chat-im/box-im-master/im-platform/src/main/java/com/bx/implatform/enums/third/ThirdKeyTypeEnum.java', import.meta.url), 'utf8');
|
||||
const match = source.match(/\(\s*"10"\s*,\s*"[^"]*"\s*,\s*"([^"]+)"\s*,\s*"([^"]+)"\s*\)/);
|
||||
assert.ok(match, 'Fixture signing configuration must be available locally');
|
||||
const [, key, secret] = match;
|
||||
|
||||
async function third(path, data) {
|
||||
const body = JSON.stringify(data);
|
||||
const timestamp = String(Math.floor(Date.now() / 1000));
|
||||
const sign = createHash('md5').update(key + secret + '10' + timestamp + body).digest('hex');
|
||||
const response = await fetch(endpoint + path, { method: 'POST', body,
|
||||
headers: { 'Content-Type': 'application/json', 'APP-KEY': key, 'BIZ-TYPE': '10', TIMESTAMP: timestamp, SIGN: sign } });
|
||||
return response.json();
|
||||
}
|
||||
|
||||
async function connect(token, target = socketEndpoint) {
|
||||
assert.match(target, /^ws:\/\/(127\.0\.0\.1|localhost):[0-9]+$/);
|
||||
const socket = new WebSocket(new URL('/im', target));
|
||||
const packets = [];
|
||||
let notify;
|
||||
const first = new Promise(resolve => { notify = resolve; });
|
||||
socket.addEventListener('message', message => {
|
||||
const packet = JSON.parse(message.data);
|
||||
packets.push(packet);
|
||||
notify(packet);
|
||||
});
|
||||
const closed = new Promise(resolve => socket.addEventListener('close', resolve));
|
||||
await new Promise((resolve, reject) => {
|
||||
socket.addEventListener('open', resolve, { once: true });
|
||||
socket.addEventListener('error', reject, { once: true });
|
||||
});
|
||||
socket.send(JSON.stringify({ cmd: 0, data: { accessToken: token } }));
|
||||
return { socket, packets, first, closed };
|
||||
}
|
||||
|
||||
const current = await third('/third/user/getUserByUserId', { userId: 131 });
|
||||
const version = Number(current.data?.syncVersion ?? 0) + 1;
|
||||
const profile = { userId: 131, userName: '老芋头', nickName: '老芋头', avatar: '', memberState: 0, syncVersion: version };
|
||||
const synchronized = await third('/third/user/createOrUpdate', profile);
|
||||
assert.equal(synchronized.code, 200, synchronized.message);
|
||||
assert.equal(synchronized.data.appliedVersion, version);
|
||||
const initial = await third('/third/user/login', { userId: 131, terminal: 1 });
|
||||
assert.equal(initial.code, 200);
|
||||
const old = await connect(initial.data.accessToken);
|
||||
const otherNode = secondSocketEndpoint ? await connect(initial.data.accessToken, secondSocketEndpoint) : null;
|
||||
try {
|
||||
assert.equal((await old.first).cmd, 0);
|
||||
if (otherNode) assert.equal((await otherNode.first).cmd, 0);
|
||||
assert.equal((await third('/third/user/createOrUpdate', { ...profile, memberState: 1, syncVersion: version + 1 })).code, 200);
|
||||
await old.closed;
|
||||
if (otherNode) await otherNode.closed;
|
||||
assert.ok(old.packets.some(packet => packet.cmd === 2 && packet.data.includes('禁用')));
|
||||
const blocked = await fetch(endpoint + '/user/self', { headers: { accessToken: initial.data.accessToken } }).then(response => response.json());
|
||||
assert.equal(blocked.code, 10005);
|
||||
const refresh = await fetch(endpoint + '/refreshToken', { method: 'PUT', headers: { refreshToken: initial.data.refreshToken } }).then(response => response.json());
|
||||
assert.equal(refresh.code, 10005);
|
||||
assert.equal((await third('/third/user/createOrUpdate', { ...profile, syncVersion: version + 2 })).code, 200);
|
||||
const restored = await third('/third/user/login', { userId: 131, terminal: 1 });
|
||||
assert.equal(restored.code, 200);
|
||||
assert.notEqual(restored.data.accessToken, initial.data.accessToken);
|
||||
const stale = await connect(initial.data.accessToken);
|
||||
assert.equal((await stale.first).cmd, 2);
|
||||
await stale.closed;
|
||||
const fresh = await connect(restored.data.accessToken);
|
||||
try {
|
||||
assert.equal((await fresh.first).cmd, 0);
|
||||
const revoked = await fetch(endpoint + '/user/self', { headers: { accessToken: initial.data.accessToken } }).then(response => response.json());
|
||||
assert.equal(revoked.code, 401);
|
||||
const active = await fetch(endpoint + '/user/self', { headers: { accessToken: restored.data.accessToken } }).then(response => response.json());
|
||||
assert.equal(active.code, 200);
|
||||
assert.equal(active.data.userName, 'u131');
|
||||
const independent = await fetch(endpoint + '/login', { method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ userName: 'testuser', password: 'Password123', terminal: 1 }) }).then(response => response.json());
|
||||
assert.equal(independent.code, 10007);
|
||||
const missedTransition = await third('/third/user/createOrUpdate', {
|
||||
...profile, syncVersion: version + 4, accessChangeVersion: version + 4,
|
||||
});
|
||||
assert.equal(missedTransition.code, 200);
|
||||
const revokedAfterMissedDisable = await fetch(endpoint + '/user/self', {
|
||||
headers: { accessToken: restored.data.accessToken },
|
||||
}).then(response => response.json());
|
||||
assert.equal(revokedAfterMissedDisable.code, 401);
|
||||
fresh.socket.send(JSON.stringify({ cmd: 1, data: {} }));
|
||||
await fresh.closed;
|
||||
assert.ok(fresh.packets.some(packet => packet.cmd === 2));
|
||||
} finally {
|
||||
fresh.socket.close();
|
||||
}
|
||||
} finally {
|
||||
old.socket.close();
|
||||
otherNode?.socket.close();
|
||||
}
|
||||
});
|
||||
Reference in new issue
Block a user