diff --git a/cmd/imtestaudit/main.go b/cmd/imtestaudit/main.go index 0cf08d9..34f731d 100644 --- a/cmd/imtestaudit/main.go +++ b/cmd/imtestaudit/main.go @@ -143,7 +143,6 @@ func (a *audit) run() { a.testHTTP("1", "1.8-submit", "实名提交", "POST", "/v1/im/security/realname/submissions", map[string]any{"name": "张三", "id_type": "cn_id", "id_number": "110101199001011234"}, a.tokenA, true, []int{200, 409}, nil) a.testHTTP("1", "1.11", "设置独立密码", "POST", "/v1/im/settings/independent-password", map[string]any{"password": "1234", "purpose": "youth_mode", "action_token": a.actionTok}, a.tokenA, true, []int{200, 400, 403}, nil) a.testHTTP("1", "1.12", "应用版本查询", "GET", "/v1/im/app/releases/latest?platform=ios¤t_version=1.0.0&channel=stable", nil, "", false, []int{200}, nil) - a.testHTTP("1", "1.logout", "退出登录", "POST", "/v1/im/auth/logout", map[string]any{}, a.tokenA, true, []int{200, 404, 405}, nil) a.testHTTP("2", "2.1", "Bootstrap", "GET", "/v1/im/bootstrap", nil, a.tokenA, false, []int{200}, func(r imResp) string { return fmt.Sprintf("entries=%v limits=%v", r.Data["entries"] != nil, r.Data["limits"] != nil) @@ -174,7 +173,7 @@ func (a *audit) run() { a.testHTTP("4", "4.forward", "消息逐条转发", "POST", "/v1/im/messages/forward", map[string]any{"mode": "separate", "sources": []map[string]any{{"topic": a.p2pTopic, "seq": 1}}, "target_topics": []string{"slf"}}, a.tokenA, true, []int{200, 404, 405}, nil) a.testHTTP("4", "4.bundle", "合并转发详情", "GET", "/v1/im/forward-bundles/fbundle_missing?cursor=", nil, a.tokenA, false, []int{200, 404, 405}, nil) a.testHTTP("4", "4.recall", "消息撤回不存在 seq", "POST", "/v1/im/topics/"+url.PathEscape(a.p2pTopic)+"/messages/1/recall", map[string]any{"version": 1, "reason": "audit"}, a.tokenA, true, []int{200, 400, 404, 403}, nil) - a.testHTTP("4", "4.revision", "消息编辑不存在 seq", "POST", "/v1/im/topics/"+url.PathEscape(a.p2pTopic)+"/messages/1/revisions", map[string]any{"content": map[string]any{"text": "edit"}, "client_mid": "audit-mid", "version": 1}, a.tokenA, true, []int{200, 400, 404, 403}, nil) + a.testHTTP("4", "4.revision", "消息编辑不存在 seq", "POST", "/v1/im/topics/"+url.PathEscape(a.p2pTopic)+"/messages/1/revisions", map[string]any{"content": map[string]any{"text": "edit"}, "client_mid": "audit-mid", "version": 1}, a.tokenA, true, []int{200, 400, 403, 404, 409}, nil) a.testHTTP("4", "4.receipts", "消息回执查询", "GET", "/v1/im/topics/"+url.PathEscape(a.p2pTopic)+"/messages/1/receipts?cursor=", nil, a.tokenA, false, []int{200, 400, 404, 403}, nil) a.testHTTP("5", "5.friends", "好友列表", "GET", "/v1/im/friends?query=&initial=&cursor=", nil, a.tokenA, false, []int{200}, nil) @@ -310,6 +309,7 @@ func (a *audit) run() { return "category=" + a.feedbackCat }) a.testHTTP("11", "11.feedback", "提交反馈", "POST", "/v1/im/feedback", map[string]any{"category": a.feedbackCat, "content": "自动化联调反馈", "contact": "", "attachment_file_ids": []string{}, "client_context": map[string]any{"source": "imtestaudit"}, "include_diagnostics": true}, a.tokenA, true, []int{200}, nil) + a.testHTTP("1", "1.logout", "退出登录", "POST", "/v1/im/auth/logout", map[string]any{}, a.tokenA, true, []int{200, 404, 405}, nil) } func (a *audit) register(chapter, id, username, phone, password string, primary bool) { diff --git a/cmd/imwsaudit/main.go b/cmd/imwsaudit/main.go index a841610..719c4a0 100644 --- a/cmd/imwsaudit/main.go +++ b/cmd/imwsaudit/main.go @@ -96,7 +96,7 @@ func main() { } func (a *audit) run() { - suffix := fmt.Sprintf("%08d", time.Now().UnixNano()%100000000) + suffix := fmt.Sprintf("%08d", (time.Now().UnixNano()+int64(os.Getpid())*100003)%100000000) nameSuffix := suffix[len(suffix)-6:] a.usernameA, a.phoneA, a.passA = "wsa"+nameSuffix, "157"+suffix, "AuditPass123" a.usernameB, a.phoneB, a.passB = "wsb"+nameSuffix, "158"+suffix, "AuditPass123" diff --git a/docs/IM-Redis三层数据分层改造方案.md b/docs/IM-Redis三层数据分层改造方案.md index 82e4cc3..453d2ca 100644 --- a/docs/IM-Redis三层数据分层改造方案.md +++ b/docs/IM-Redis三层数据分层改造方案.md @@ -475,6 +475,8 @@ im:{tenant_id}:... 当前代码将 Redis 配置挂在 `store_config.redis` 下；不配置则不启用 Redis 群消息发号和唯一 key，群消息仍兼容旧的本机 `lastID + 1` 发号。 +群消息物理表按自然月拆分时，落库路由、跨月查询、幂等和迁移方案见：[IM-群消息按月分表改造方案](./IM-群消息按月分表改造方案.md)。 + 配置示例： ```json diff --git a/docs/IM-群消息按月分表改造方案.md b/docs/IM-群消息按月分表改造方案.md new file mode 100644 index 0000000..af8fdeb --- /dev/null +++ b/docs/IM-群消息按月分表改造方案.md @@ -0,0 +1,555 @@ +# IM 群消息按月分表改造方案 + +## 1. 背景与目标 + +当前每个租户库使用单表 `im_group_messages` 保存全部群消息。随着群数量、消息量和保留时间增长，单表会出现索引膨胀、查询抖动、DDL 风险升高和归档困难等问题。 + +本方案将群消息正文按自然月拆分为物理表： + +```text +im_group_messages_202609 +im_group_messages_202610 +im_group_messages_202611 +``` + +目标： + +1. 每个租户库独立分表，消息表不增加 `tenant_id`。 +2. Topic 的 `seqid` 继续由 Redis 原子递增，跨月不重置。 +3. Kafka 异步落库稳定路由到首次确定的月表。 +4. 客户端继续按 `topic + seq` 拉取，不感知月份。 +5. 支持跨月分页、补拉、指定消息、搜索、撤回、编辑和删除。 +6. 旧单表平滑迁移，不停服、不丢失、不重复。 +7. Redis 保留 24 小时热消息，MySQL 月表仍是最终数据源。 + +## 2. 核心原则 + +### 2.1 分表粒度 + +- 一个租户库每月一张群消息表，不按单个群建表。 +- 表名固定为 `im_group_messages_YYYYMM`。 +- 月份统一使用 UTC，但最终写入月份由 Redis 全局 `active_month` 决定，避免多节点在月初产生交叠。 +- Redis Lua 一次性返回 `seqid + shard_month`；生产端把结果固化到 Kafka，重试、补偿不得重算。 + +### 2.2 Seq 规则 + +- `seqid` 在 Topic 内全生命周期单调递增，跨月不归零。 +- Redis `topic:last_seq` 是唯一发号源。 +- 月表只保存已分配的 Seq，不通过 `MAX(seqid)` 发号。 +- 客户端游标继续使用 `before/since/seq`。 + +### 2.3 分层职责 + +| 层级 | 职责 | +| --- | --- | +| Redis | 原子 Seq、全局活动月份、长期 Topic 月段路由、短期发送幂等、24 小时热消息、在线状态、实时推送 | +| Kafka | 异步落库、削峰、重试、死信、消费幂等 | +| MySQL 月表 | 群消息正文最终存储 | +| MySQL 路由镜像 | 月表目录、Topic 月段路由持久化镜像；Redis 丢失时用于重建，不进入正常查询热路径 | +| `im_dellog` | 用户视角删除范围，每张月表查询时继续过滤 | + +## 3. 表结构 + +### 3.1 月表模板 + +第一阶段保持当前字段兼容，降低 Store 层改造范围。 + +```sql +CREATE TABLE IF NOT EXISTS im_group_messages_202609 ( + message_id BIGINT NOT NULL COMMENT 'Snowflake 全局消息 ID', + topic VARCHAR(64) NOT NULL COMMENT '群 Topic', + group_id BIGINT NOT NULL COMMENT '业务群 ID', + seqid BIGINT NOT NULL COMMENT 'Topic 内递增，跨月不重置', + from_user BIGINT NOT NULL, + client_msg_id VARCHAR(64) NULL, + msg_type VARCHAR(32) NOT NULL DEFAULT 'text', + status VARCHAR(32) NOT NULL DEFAULT 'normal', + content JSON NULL, + head JSON NULL, + mentioned_user_ids JSON NULL, + mention_all TINYINT(1) NOT NULL DEFAULT 0, + reply_seqid BIGINT NULL, + root_seqid BIGINT NULL, + file_count INT NOT NULL DEFAULT 0, + content_size INT NOT NULL DEFAULT 0, + has_risk TINYINT(1) NOT NULL DEFAULT 0, + audit_status VARCHAR(32) NOT NULL DEFAULT 'pass', + version INT NOT NULL DEFAULT 1, + recalled_at DATETIME(3) NULL, + recalled_by BIGINT NULL, + deleted_at DATETIME(3) NULL, + expires_at DATETIME(3) NULL, + created_at DATETIME(3) NOT NULL, + updated_at DATETIME(3) NOT NULL, + PRIMARY KEY (message_id), + UNIQUE KEY uk_topic_seq (topic, seqid), + UNIQUE KEY uk_topic_sender_client_mid (topic, from_user, client_msg_id), + KEY idx_group_seq (group_id, seqid), + KEY idx_topic_created (topic, created_at), + KEY idx_topic_from_seq (topic, from_user, seqid), + KEY idx_audit_created (audit_status, created_at), + KEY idx_expires (expires_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +``` + +说明： + +1. `message_id` 使用 Snowflake 全局唯一 ID，作为月表物理主键和跨表引用 ID。 +2. 聊天协议定位、补拉和已读水位仍统一使用 `topic + seqid`，不向客户端暴露物理主键。 +3. 单表唯一索引不能覆盖跨月重试，跨月幂等由 Redis、消费日志和幂等表共同保证。 +4. 文件二进制仍在 OSS，月表仅保存消息和文件元数据。 + +### 3.2 月表目录 + +```sql +CREATE TABLE IF NOT EXISTS im_group_message_shards ( + shard_month CHAR(6) NOT NULL COMMENT 'UTC 月份 YYYYMM', + table_name VARCHAR(64) NOT NULL, + shard_start DATETIME(3) NOT NULL, + shard_end DATETIME(3) NOT NULL, + status VARCHAR(16) NOT NULL DEFAULT 'prepared' + COMMENT 'prepared/writable/sealing/sealed/archived', + row_count BIGINT NOT NULL DEFAULT 0, + min_created_at DATETIME(3) NULL, + max_created_at DATETIME(3) NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) + ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (shard_month), + UNIQUE KEY uk_table_name (table_name) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +``` + +表名只允许由服务端根据月份生成，并且必须存在于目录表，禁止使用请求参数直接拼接 SQL。 + +### 3.3 Redis Topic 月段路由 + +在线查询使用 Redis 路由，路由 Key 长期不过期： + +```text +# 当前可写月份 +im:{tenant_id}:groupmsg:active_month + STRING = 202610 + +# Topic 月份段，score 为该月首条 seq +im:{tenant_id}:groupmsg:route:{topic} + ZSET: 202609 => 1 + 202610 => 10087 + +# 月段完整范围，field 为 YYYYMM +im:{tenant_id}:groupmsg:route_meta:{topic} + HASH: 202609 => min_seq=1,max_seq=10086,count=10086 + 202610 => min_seq=10087,max_seq=12000,count=1914 +``` + +路由规则： + +1. `route` 和 `route_meta` 不设置 TTL，只在群物理删除或对应消息保留期结束后清理。 +2. 精确 Seq 定位使用 `ZREVRANGEBYSCORE route seq -inf LIMIT 0 1`，找到最大 `min_seq <= seq` 的月份，再校验 Hash 中的 `max_seq`。 +3. 倒序分页通过 `ZREVRANGE` 按月段倒序读取，正向补拉通过 `ZRANGE` 读取。 +4. 路由 Key 使用 Redis Cluster hash tag，保证 Lua 和 Pipeline 访问同一槽位。 +5. Redis 开启 AOF everysec、主从复制和持久化备份；同时保留 MySQL 镜像作为灾备重建来源。 + +活动月份和 Seq 必须在同一个 Lua 脚本中读取/分配，两个 Key 使用相同的租户 hash tag： + +```lua +local shard_month = redis.call('GET', KEYS[1]) +if not shard_month or shard_month < ARGV[1] then + redis.call('SET', KEYS[1], ARGV[1]) + shard_month = ARGV[1] +end + +local seq_id = redis.call('INCR', KEYS[2]) +return {seq_id, shard_month} +``` + +其中 `KEYS[1]` 是租户活动月份，`KEYS[2]` 是 Topic Seq，`ARGV[1]` 来自 Redis `TIME` 转换后的 UTC 月份。业务节点不得根据本机时间计算月份；首次写入可初始化活动月份，之后 Lua 只允许月份单向推进，旧节点不能回拨。 + +### 3.4 MySQL 路由镜像 + +```sql +CREATE TABLE IF NOT EXISTS im_group_message_topic_shards ( + topic VARCHAR(64) NOT NULL, + shard_month CHAR(6) NOT NULL, + min_seq BIGINT NOT NULL, + max_seq BIGINT NOT NULL, + message_count BIGINT NOT NULL DEFAULT 0, + min_created_at DATETIME(3) NOT NULL, + max_created_at DATETIME(3) NOT NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) + ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (topic, shard_month), + KEY idx_topic_max_seq (topic, max_seq), + KEY idx_month_topic (shard_month, topic) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +``` + +该表不参与正常消息查询，只用于 Kafka 事务持久化、Redis 路由重建、历史迁移、数据核对和归档。Redis 与镜像不一致时，以月表实际数据和 MySQL 镜像修复 Redis。 + +路由一致性规则： + +1. Kafka 消费事务同时提交消息和 MySQL 路由镜像，保证最终数据与镜像一致。 +2. MySQL 提交成功后再更新 Redis `route/route_meta`；Redis 更新失败不回滚消息，进入路由修复重试队列。 +3. 查询发现 Redis 路由 Key 不存在时，从 MySQL 镜像加载该 Topic 全部有效月段并回填 Redis，然后重试查询。 +4. Shard Manager 启动时可按租户增量重建路由，并定时抽样核对 Redis、MySQL 镜像和月表实际 Seq 范围。 +5. Redis 路由不设置 TTL；只有群被物理删除，或消息保留策略确认清除对应月份后，才同步删除相应月段。 + +### 3.5 跨月幂等表 + +```sql +CREATE TABLE IF NOT EXISTS im_group_message_dedup ( + topic VARCHAR(64) NOT NULL, + from_user BIGINT NOT NULL, + client_msg_id VARCHAR(64) NOT NULL, + seqid BIGINT NOT NULL, + shard_month CHAR(6) NOT NULL, + message_id BIGINT NULL, + status VARCHAR(16) NOT NULL DEFAULT 'pending', + expires_at DATETIME(3) NOT NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) + ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (topic, from_user, client_msg_id), + KEY idx_expires (expires_at), + KEY idx_topic_seq (topic, seqid) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +``` + +- Redis `SET NX` 是入口快速幂等。 +- Kafka 消费日志按 `event_id` 防止同一事件重复消费。 +- 该表防止月末发送、月初重试时同一 `client_msg_id` 落入两张表。 +- 建议保留 7 至 30 天后清理，不作为永久消息索引。 + +## 4. 建表和月切换 + +每个租户库运行统一 Shard Manager： + +1. 每天巡检上月、当月和未来两个月。 +2. 缺表时根据固定模板创建，成功后登记目录。 +3. 使用 Redis 分布式锁和 Redis `TIME` 判断月界，确认新月表可写后再原子更新 `active_month`。 +4. 当月设置为 `writable`，所有节点通过 Redis 读取同一个活动月份。 +5. 上月进入 `sealing`，建议继续接受 7 天延迟 Kafka 消息。 +6. 确认没有积压和死信后改为 `sealed`。 +7. 建表失败立即告警，不切换 `active_month`，禁止在业务写入事务中临时执行 DDL。 + +月切换示例： + +```text +2026-09-30 23:59:59 UTC seq=10086 -> im_group_messages_202609 +2026-10-01 00:00:00 UTC seq=10087 -> im_group_messages_202610 +``` + +Kafka 事件必须固定路由信息： + +```json +{ + "topic": "grp_xxx", + "seq_id": 10087, + "created_at": "2026-10-01T00:00:00.123Z", + "shard_month": "202610", + "client_msg_id": "client_mid_xxx" +} +``` + +## 5. 群消息落库 + +### 5.1 生产端 + +1. 校验群成员、禁言、群状态和内容策略。 +2. 使用 Redis 幂等 Key 抢占 `client_msg_id`。 +3. 使用 Redis Lua 原子获取 Topic 下一个 `seqid` 和全局 `active_month`。 +4. 生成服务端 UTC `created_at`，并把 Lua 返回的月份作为 `shard_month`。 +5. 把 `shard_month` 写入 Kafka；所有重试沿用原值。 +6. Kafka 生产成功后按现有策略向客户端返回 Seq。 +7. 写入 Redis 热消息并执行实时推送。 + +### 5.2 Kafka 消费端 + +消费端按 `tenant + shard_month` 分组批量写入： + +1. 校验月份格式和允许范围。 +2. 从目录确认目标表已存在。 +3. 开启事务。 +4. 插入或确认 `im_group_message_dedup`。 +5. 插入目标月表。 +6. 在同一事务中 UPSERT MySQL 路由镜像的最小/最大 Seq、时间和消息数。 +7. 更新消费幂等状态并提交事务。 +8. 提交成功后更新 Redis `route/route_meta`，并把 Redis 幂等状态改为 `done`。 + +路由 UPSERT： + +```sql +INSERT INTO im_group_message_topic_shards + (topic, shard_month, min_seq, max_seq, message_count, + min_created_at, max_created_at) +VALUES (?, ?, ?, ?, 1, ?, ?) +ON DUPLICATE KEY UPDATE + min_seq = LEAST(min_seq, VALUES(min_seq)), + max_seq = GREATEST(max_seq, VALUES(max_seq)), + message_count = message_count + 1, + min_created_at = LEAST(min_created_at, VALUES(min_created_at)), + max_created_at = GREATEST(max_created_at, VALUES(max_created_at)); +``` + +### 5.3 异常处理 + +- 目标月表不存在：消费失败、进入重试并高优告警。 +- `sealing` 月表：允许携带原 `shard_month` 的延迟事件写入。 +- `sealed` 月表：进入补偿队列，人工确认后临时开放或补写。 +- 重复 `event_id`：消费日志直接判定成功。 +- 重复 `client_msg_id`：返回原 `seqid/shard_month`，不新增消息。 +- 超过最大重试次数：进入死信队列，不能静默丢弃。 +- Redis 热缓存失败不回滚 MySQL，客户端按 Seq 回源补拉。 +- Redis 路由更新失败：进入独立修复重试；查询侧可从 MySQL 镜像重建，不影响已提交消息。 + +## 6. 消息查询 + +### 6.1 总流程 + +```text +客户端 Topic Seq 请求 + -> ACL/成员身份校验 + -> Redis 24 小时热消息 + -> Redis 是否完整覆盖请求范围 + -> Redis Topic 月段路由 + -> 月表逐张查询 + -> 按 seqid 合并、去重、排序、截断 + -> 应用 im_dellog 用户视角过滤 + -> 返回客户端 +``` + +Redis 只有完整覆盖整页时才能直接返回；有缺口必须回源 MySQL。 + +### 6.2 最新消息倒序分页 + +请求 `topic=grp_xxx&before=10088&limit=50`： + +1. 从 Redis Topic 月段路由找出 `min_seq < before` 的月份，按月倒序。 +2. 先查最新候选月： + +```sql +SELECT ... +FROM im_group_messages_202610 AS m +LEFT JOIN im_dellog AS d + ON d.topic = m.topic + AND m.seqid BETWEEN d.low AND d.hi - 1 + AND d.deletedfor = ? +WHERE m.topic = ? + AND m.seqid < ? + AND m.status != 'deleted' + AND d.deletedfor IS NULL +ORDER BY m.seqid DESC +LIMIT ?; +``` + +3. 数量不足时继续查前一个月。 +4. 服务层按 `seqid DESC` 合并并截取 `limit`。 +5. 下一页游标仍是本页最小 `seqid`，客户端不感知月份。 + +禁止对所有历史表执行无上限 `UNION ALL`。 + +### 6.3 Since 正向补拉 + +1. 从 Redis Topic 月段路由查找 `max_seq > since_seq` 的月份。 +2. 按月升序执行 `seqid > since_seq ORDER BY seqid ASC LIMIT N`。 +3. 合并后按 Seq 升序返回。 +4. 达到上限后返回最后一条 Seq，客户端继续请求。 + +### 6.4 指定消息 + +撤回、回复预览、举报取证按 `topic + seqid` 定位： + +```text +ZREVRANGEBYSCORE im:{tenant_id}:groupmsg:route:{topic} {seqid} -inf LIMIT 0 1 +HGET im:{tenant_id}:groupmsg:route_meta:{topic} {shard_month} +``` + +先找到最大 `min_seq <= seqid` 的月份，再校验该月段 `max_seq >= seqid`，随后执行 `WHERE topic=? AND seqid=? LIMIT 1`。Redis 路由 Key 缺失时读取 MySQL 镜像并回填 Redis；镜像也缺失时，仅降级扫描当前月和最近若干月并记录修复任务，不允许扫描全部历史表。 + +### 6.5 上下文和范围 + +查询目标消息前后各 20 条时： + +1. 定位目标所在月。 +2. 在本月分别查询目标前后消息。 +3. 任一方向不足时继续查询相邻月份。 +4. 合并、去重后按 Seq 正序返回。 + +### 6.6 搜索和统计 + +- 短期搜索必须带起止时间，最多覆盖 3 个月，逐月查询后归并。 +- 中长期通过 Kafka 建独立搜索索引；结果只返回 `topic + seqid`，服务端回源月表并重新校验 ACL。 +- Admin 统计使用按日/月汇总表，不实时扫描所有月表执行 `COUNT(*)`。 + +## 7. 撤回、编辑和删除 + +撤回、编辑流程： + +1. 根据 `topic + seqid` 定位目标月表。 +2. 在目标月表事务中 `SELECT ... FOR UPDATE`。 +3. 校验版本、权限和状态。 +4. 更新目标月表并写修订记录、业务事件。 +5. 提交后更新或删除 Redis 热消息。 + +修订记录建议增加月份： + +```sql +ALTER TABLE im_message_revisions + ADD COLUMN shard_month CHAR(6) NULL AFTER topic; +``` + +用户侧删除继续使用 `im_dellog` 的 Topic Seq 范围，每张月表查询都应用相同过滤。全局删除或合规处置先定位月表，再更新 `status/deleted_at/content/head`；OSS 附件仍走文件生命周期任务。 + +## 8. Store 层设计 + +所有动态表访问统一经过路由组件： + +```go +type GroupMessageShardRouter interface { + WriteTable(createdAt time.Time, shardMonth string) (string, error) + CandidateTablesBySeq(topic string, seq int64) ([]string, error) + TablesBefore(topic string, before int64) ([]string, error) + TablesSince(topic string, since int64) ([]string, error) +} +``` + +改造范围： + +- `TenantGroupMessageSave`：按事件 `shard_month` 写入。 +- `TenantGroupMessageGetOne`：路由后查询目标月。 +- `TenantGroupMessageGetAll`：逐月查询并归并。 +- `TenantGroupMessageGetByClientID`：先查 Redis/幂等表，再查询目标月。 +- `TenantGroupMessageRecall/Edit`：定位目标月后执行事务。 +- `TenantMessageSearch`：按时间选月，或转独立搜索。 +- Admin 内部消息查询、举报、收藏、回复和群文件引用同步改造。 + +动态表名必须满足 `^im_group_messages_[0-9]{6}$` 且月份已登记。SQL 参数不能绑定表名，必须严格白名单。 + +## 9. 旧表迁移 + +原 `im_group_messages` 先保留，不删除。 + +推荐步骤： + +1. 上线路由表、幂等表和 Shard Manager。 +2. 预建历史月份、当月和未来两个月表。 +3. 先发布支持“旧表 + 月表”双读的代码，写入仍走旧表。 +4. 记录切换水位 `cutover_seq/cutover_time`。 +5. 切换新消息只写月表，Kafka 开始携带 `shard_month`。 +6. 按 `created_at` 分批把旧表数据迁入对应月表。 +7. 每批使用 `INSERT IGNORE`，校验 `topic + seqid`、数量和内容哈希。 +8. 回填 MySQL Topic 月段路由镜像，并由重建任务写入 Redis 路由。 +9. 双读期间按 `topic + seqid` 去重。 +10. 全量校验通过后停止读取旧表。 +11. 旧表改名 `im_group_messages_legacy`，只读保留。 +12. 观察至少一个完整保留周期后再单独审批清理。 + +不建议长期双写旧表和月表，否则两个写目标部分成功会扩大一致性问题。 + +迁移批次要求： + +- 使用主键或 `created_at + id` 游标，禁止 OFFSET 深分页。 +- 单批 1,000 至 5,000 条，根据线上延迟调整。 +- 每批独立事务，记录开始/结束游标和成功/失败数。 +- 支持暂停、续跑和重复执行。 + +## 10. 归档与保留 + +状态流转： + +```text +prepared -> writable -> sealing -> sealed -> archived +``` + +- `writable`：当月正常写。 +- `sealing`：月末后接受延迟消息，默认 7 天。 +- `sealed`：在线 MySQL 只读。 +- `archived`：转对象存储或低成本库，在线查询走归档服务。 + +归档前确认 Kafka 无积压/死信、路由完整、消息数量和 Seq 范围一致，并确认举报、审计、收藏和群文件引用仍可定位。 + +## 11. 监控与告警 + +| 指标 | 说明 | +| --- | --- | +| `group_message_persist_total{month,status}` | 各月落库成功/失败 | +| `group_message_persist_latency` | Kafka 到 MySQL 提交耗时 | +| `group_message_shard_missing_total` | 目标月表不存在 | +| `group_message_query_tables` | 单次请求扫描表数 | +| `group_message_query_latency{source}` | Redis、单月、跨月耗时 | +| `group_message_dedup_hit_total` | 幂等命中 | +| `group_message_route_miss_total` | Redis 路由缺失 | +| `group_message_route_rebuild_total{status}` | 从 MySQL 镜像重建 Redis 路由 | +| `group_message_route_repair_backlog` | Redis 路由修复积压 | +| `group_message_dead_letter_total` | 落库死信 | + +未来月表未创建、当月不可写、`active_month` 缺失、Kafka 延迟超限、单次扫描超过 3 张表、Redis 路由缺失或修复积压增长时必须告警。 + +## 12. 验收测试 + +1. 月末和月初消息落入正确表，Seq 连续。 +2. Kafka 次月才消费时仍写事件指定月份。 +3. 同一 `client_msg_id` 跨月重试只保留一条。 +4. 最新 50 条跨两个月时数量、顺序、游标正确。 +5. `since_seq` 跨月补拉无遗漏、无重复。 +6. `topic + seq` 可定位历史消息。 +7. 回复上下文可跨月读取。 +8. 撤回、编辑、举报、收藏、群文件引用可定位旧月。 +9. 用户删除范围跨月时过滤正确。 +10. Redis 热消息不完整时正确回源。 +11. 月表缺失时进入重试/死信并告警。 +12. 迁移双读无重复，迁移任务可暂停、续跑。 +13. 万人群高并发时 Seq 唯一且落库稳定。 +14. 非法月份和非法动态表名被拒绝。 +15. 多节点跨月并发时 Lua 返回同一活动月份，Seq 不重号且月段不交叠。 +16. 删除 Redis 路由后可从 MySQL 镜像自动重建，查询结果不缺失。 + +## 13. 实施阶段 + +### 第一阶段：基础设施 + +- 建月表模板、目录、MySQL 路由镜像和跨月幂等表。 +- 实现 Shard Manager，预建未来两个月。 +- 建立长期不过期的 Redis Topic 月段路由和原子 `seqid + active_month` Lua。 +- Kafka 群消息事件增加 `shard_month`。 + +### 第二阶段：写入切换 + +- 消费者按月落库。 +- 完成幂等、重试、死信、MySQL 镜像和 Redis 路由更新。 +- Redis 热消息与实时推送保持不变。 + +### 第三阶段：查询切换 + +- 完成精确查询、倒序分页、Since 补拉和上下文跨月查询。 +- 改造撤回、编辑、删除、举报及 Admin 内部查询。 +- 增加跨月查询监控。 + +### 第四阶段：历史迁移 + +- 上线旧表/月表双读。 +- 回填历史消息与 MySQL 路由镜像，重建 Redis Topic 路由。 +- 校验后把旧表改为只读保留。 + +### 第五阶段：搜索和归档 + +- 消息搜索切换独立索引或限制时间范围。 +- 建立 `sealed/archived` 生命周期和历史恢复机制。 + +## 14. 结论 + +群消息按月分表不能只改写入表名，必须一起实现固定 `shard_month`、Redis Topic 月段路由、MySQL 路由镜像、跨月幂等、逐月游标查询、历史消息定位和旧表迁移。 + +最终链路： + +```text +发送：Redis Lua 原子分配 Seq 和 active_month -> Kafka 固定 shard_month -> MySQL 月表 +实时：Redis Pub/Sub/Stream -> WebSocket +热读：Redis 24 小时消息 +冷读：Redis Topic 月段路由 -> MySQL 月表逐月查询 +容灾：MySQL 路由镜像 -> 重建长期 Redis 路由 +补偿：客户端按 Topic Seq 拉取 +``` + +客户端协议无需增加月份字段，月表路由完全由服务端处理。 diff --git a/docs/sql/IM-用户账号编号序列表SQL.sql b/docs/sql/IM-用户账号编号序列表SQL.sql new file mode 100644 index 0000000..8c91624 --- /dev/null +++ b/docs/sql/IM-用户账号编号序列表SQL.sql @@ -0,0 +1,15 @@ +-- 在每个租户业务库执行。用于并发安全地生成 im_users.account_no。 +CREATE TABLE IF NOT EXISTS im_user_account_sequences ( + id TINYINT UNSIGNED NOT NULL, + next_value INT UNSIGNED NOT NULL, + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) + ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +-- 兼容已有用户数据，从当前最大账号编号继续递增，账号编号最小为 1000。 +INSERT INTO im_user_account_sequences (id, next_value) +SELECT 1, GREATEST(COALESCE(MAX(account_no), 999), 999) +FROM im_users +ON DUPLICATE KEY UPDATE + next_value = GREATEST(next_value, VALUES(next_value)); diff --git a/docs/sql/IM-群消息拆表SQL.sql b/docs/sql/IM-群消息拆表SQL.sql index cc100a8..cf61b51 100644 --- a/docs/sql/IM-群消息拆表SQL.sql +++ b/docs/sql/IM-群消息拆表SQL.sql @@ -1,3 +1,6 @@ +-- Legacy single-table baseline. New deployments must also run +-- IM-群消息按月分表SQL.sql; this table remains only for migration-time dual reads. + CREATE TABLE IF NOT EXISTS im_group_messages ( id BIGINT NOT NULL AUTO_INCREMENT, topic VARCHAR(64) NOT NULL COMMENT '群 Topic，如 grp_xxx', diff --git a/docs/sql/IM-群消息按月分表SQL.sql b/docs/sql/IM-群消息按月分表SQL.sql new file mode 100644 index 0000000..4624f0f --- /dev/null +++ b/docs/sql/IM-群消息按月分表SQL.sql @@ -0,0 +1,107 @@ +-- IM group-message monthly sharding, baseline month: 2026-09 (UTC). +-- Run in each tenant database. The application pre-creates subsequent months. + +CREATE TABLE IF NOT EXISTS im_group_message_shards ( + shard_month CHAR(6) NOT NULL, + table_name VARCHAR(64) NOT NULL, + shard_start DATETIME(3) NOT NULL, + shard_end DATETIME(3) NOT NULL, + status VARCHAR(16) NOT NULL DEFAULT 'prepared', + row_count BIGINT NOT NULL DEFAULT 0, + min_created_at DATETIME(3) NULL, + max_created_at DATETIME(3) NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (shard_month), + UNIQUE KEY uk_im_group_message_shards_table (table_name), + KEY idx_im_group_message_shards_status (status, shard_month) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +CREATE TABLE IF NOT EXISTS im_group_message_topic_shards ( + topic VARCHAR(64) NOT NULL, + shard_month CHAR(6) NOT NULL, + min_seq BIGINT NOT NULL, + max_seq BIGINT NOT NULL, + message_count BIGINT NOT NULL DEFAULT 0, + min_created_at DATETIME(3) NOT NULL, + max_created_at DATETIME(3) NOT NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (topic, shard_month), + KEY idx_im_group_message_topic_shards_seq (topic, max_seq), + KEY idx_im_group_message_topic_shards_month (shard_month, topic) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +CREATE TABLE IF NOT EXISTS im_group_message_dedup ( + topic VARCHAR(64) NOT NULL, + from_user BIGINT NOT NULL, + client_msg_id VARCHAR(64) NOT NULL, + seqid BIGINT NOT NULL, + shard_month CHAR(6) NOT NULL, + message_id BIGINT NULL, + status VARCHAR(16) NOT NULL DEFAULT 'pending', + expires_at DATETIME(3) NOT NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (topic, from_user, client_msg_id), + KEY idx_im_group_message_dedup_expires (expires_at), + KEY idx_im_group_message_dedup_topic_seq (topic, seqid) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +CREATE TABLE IF NOT EXISTS im_group_messages_202609 ( + message_id BIGINT NOT NULL, + topic VARCHAR(64) NOT NULL, + group_id BIGINT NOT NULL, + seqid BIGINT NOT NULL, + from_user BIGINT NOT NULL, + client_msg_id VARCHAR(64) NULL, + msg_type VARCHAR(32) NOT NULL DEFAULT 'text', + status VARCHAR(32) NOT NULL DEFAULT 'normal', + content JSON NULL, + head JSON NULL, + mentioned_user_ids JSON NULL, + mention_all TINYINT(1) NOT NULL DEFAULT 0, + reply_seqid BIGINT NULL, + root_seqid BIGINT NULL, + file_count INT NOT NULL DEFAULT 0, + content_size INT NOT NULL DEFAULT 0, + has_risk TINYINT(1) NOT NULL DEFAULT 0, + audit_status VARCHAR(32) NOT NULL DEFAULT 'pass', + version INT NOT NULL DEFAULT 1, + recalled_at DATETIME(3) NULL, + recalled_by BIGINT NULL, + deleted_at DATETIME(3) NULL, + expires_at DATETIME(3) NULL, + created_at DATETIME(3) NOT NULL, + updated_at DATETIME(3) NOT NULL, + PRIMARY KEY (message_id), + UNIQUE KEY uk_202609_group_seq (group_id, seqid), + UNIQUE KEY uk_202609_topic_seq (topic, seqid), + UNIQUE KEY uk_202609_sender_client (topic, from_user, client_msg_id), + KEY idx_202609_group_status_seq (group_id, status, seqid), + KEY idx_202609_topic_created (topic, created_at), + KEY idx_202609_topic_from_seq (topic, from_user, seqid), + KEY idx_202609_audit_created (audit_status, created_at), + KEY idx_202609_expires (expires_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +CREATE TABLE IF NOT EXISTS im_group_messages_202608 LIKE im_group_messages_202609; +CREATE TABLE IF NOT EXISTS im_group_messages_202610 LIKE im_group_messages_202609; +CREATE TABLE IF NOT EXISTS im_group_messages_202611 LIKE im_group_messages_202609; + +INSERT INTO im_group_message_shards + (shard_month,table_name,shard_start,shard_end,status) +VALUES + ('202608','im_group_messages_202608','2026-08-01 00:00:00.000','2026-09-01 00:00:00.000','sealing'), + ('202609','im_group_messages_202609','2026-09-01 00:00:00.000','2026-10-01 00:00:00.000','writable'), + ('202610','im_group_messages_202610','2026-10-01 00:00:00.000','2026-11-01 00:00:00.000','prepared'), + ('202611','im_group_messages_202611','2026-11-01 00:00:00.000','2026-12-01 00:00:00.000','prepared') +ON DUPLICATE KEY UPDATE + table_name=VALUES(table_name), + shard_start=VALUES(shard_start), + shard_end=VALUES(shard_end), + status=IF(status IN ('sealed','archived'),status,VALUES(status)); + +-- Historical migration is deliberately separate from schema deployment. +-- Backfill im_group_messages into the matching month table in bounded batches, +-- then populate im_group_message_topic_shards before removing legacy reads. diff --git a/docs/test/IM-群消息按月分表本地验收报告-20260915.md b/docs/test/IM-群消息按月分表本地验收报告-20260915.md new file mode 100644 index 0000000..0948199 --- /dev/null +++ b/docs/test/IM-群消息按月分表本地验收报告-20260915.md @@ -0,0 +1,55 @@ +# IM 群消息按月分表本地验收报告 + +## 验收环境 + +- 时间：2026-09-15 +- IM Server：`http://127.0.0.1:6060` +- WebSocket：`ws://127.0.0.1:6060/v0/channels` +- MySQL：`127.0.0.1:3306`，租户库 `tinode` +- Redis：`127.0.0.1:6379` +- Kafka：`127.0.0.1:9092` + +## 验收结果 + +| 项目 | 结果 | 说明 | +| --- | --- | --- | +| 编译、定向单测、Go Vet | 通过 | MySQL、Store、Server 本次变更相关用例全部通过 | +| WebSocket 协议审计 | 通过 | 29/29，覆盖注册、登录、P2P、群聊、文件、回执、通话与异常鉴权 | +| HTTP 接口审计 | 通过 | 98/98，未发现非预期失败或非预期 404 | +| Kafka 群消息异步落库 | 通过 | 消费成功，消费组 LAG 为 0 | +| 群消息分月写入 | 通过 | 新消息写入 `im_group_messages_202609`，未写入旧表 `im_group_messages` | +| Kafka 重放幂等 | 通过 | 同一消息使用新事件 ID 重放后仍只有一行，路由计数不重复增加 | +| Redis Seq | 通过 | 群 Topic Seq 原子增长，键 TTL 为 `-1` | +| Redis 月份路由 | 通过 | 路由和元数据完整，键 TTL 为 `-1` | +| 服务重启后历史拉取 | 通过 | 清除进程内 Topic 后，客户端可从分表恢复历史消息 | +| 跨月合并查询 | 通过 | 构造 202608、202609 两个月份数据后，一次订阅正确返回两个分表的消息 | +| 并发注册与群消息冒烟 | 通过 | 3 组并发、每组 29/29；账号编号无重复，群消息全部落入月表 | +| Manager 内部接口 | 通过 | 56/56，覆盖列表、详情、分页、筛选、用户创建及回读 | +| Manager HMAC 安全 | 通过 | 未签名、错签名、过期签名均返回 401，重复 Nonce 第二次返回 401 | + +## 测试中修复 + +1. 注册写入未生成 `im_users.account_no`，第二个新用户会命中唯一键冲突。现改为租户库单行序列表原子分配。 +2. Kafka 群消息生产前未分配持久化 `message_id`，消费者会判定消息格式不完整。现由生产端生成并随事件固定传递。 +3. HTTP 审计工具过早执行退出登录，导致后续接口误报 401。现将退出登录移到整轮审计末尾。 +4. WebSocket 审计工具并发运行时可能生成相同手机号。现将进程 ID 纳入测试数据后缀。 +5. 个人资料刷新错误地向订阅表写入不存在的 `public` 字段。现改为清理好友订阅缓存并在下次读取时重新投影。 +6. `Uid` 零值序列化为空字符串后无法反序列化，导致群配置缓存持续回源。现已补齐零值 JSON 往返支持及单测。 +7. Manager 的看板和群消息查询仍只统计历史总表。现已按分片元数据合并月表与未迁移历史数据。 +8. 群消息历史表与月表 collation 不一致，`UNION` 查询返回 500。现已在历史分支显式统一 collation。 +9. Manager 群消息详情、文件详情和审计日志仍使用旧表或旧字段口径。现已支持月表详情、19 位数字文件 ID，并完成审计字段映射。 +10. Manager 群列表混入 `p2p` 和 `slf` Topic。现已在数据源层限定仅返回 `grp` Topic。 +11. Manager 内部查询输出的用户相关 ID 已统一为数据库 `BIGINT/int64`，不再转换为 `usr...`。 + +## Manager 内部接口验收 + +- 基础查询：用户、群组、私聊消息、群消息、文件、登录日志、审计、实名、举报、反馈、配置、工作台、组织架构和看板均返回 200。 +- 群治理查询：群详情、成员、入群申请、邀请、公告、禁言、管理日志和群文件均通过。 +- 运维查询：业务 Outbox、群 Outbox、Kafka 死信列表及详情均通过。 +- 写入冒烟：`POST /internal/admin-query/tenants/default/users` 创建成功，返回数字型 `user_id`，随后详情回读成功。 +- 协议口径：`user_id`、`owner_id`、`from_user_id`、`file_id` 和 `seq_id` 均按 JSON number 返回；群列表仅包含 `grp` Topic。 +- 安全边界：HMAC 签名覆盖 GET/POST，`X-Nonce` 为一次性值，防重放状态由 Redis 保持。 + +## 结论 + +本次群消息按月分表相关的写入、读取、重启恢复、跨月查询、Kafka 幂等、Redis 路由及并发链路均已通过本地验收；Manager 内部接口 56/56 及 HMAC 安全用例也已全部通过，可以进入测试环境验证。外部 OSS、内容安全、Push 和真实音视频媒体质量不属于本轮本地验收范围。 diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index de908d8..ddc951a 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -9,7 +9,6 @@ import ( "database/sql" "encoding/json" "errors" - "fmt" "hash/fnv" "sort" "strconv" @@ -49,6 +48,7 @@ type adapter struct { tenantRegistryDB string tenantAdapters map[t.TenantID]*adapter tenantMu sync.Mutex + groupShardWindow string } const ( @@ -375,7 +375,7 @@ func (a *adapter) CheckDbVersion() error { } // Version returns adapter version. -func (adapter) Version() int { +func (*adapter) Version() int { return adpVersion } @@ -1819,26 +1819,19 @@ func (a *adapter) UserCreate(user *t.User) error { user.Phone = canonicalPhone phone = canonicalPhone } - releaseAccountNoLock, err := lockUserAccountNoSequence(ctx, tx) + accountNo, hasAccountNo, err := nextUserAccountNo(ctx, tx) if err != nil { return err } - defer releaseAccountNoLock() - accountNo, err := nextUserAccountNo(ctx, tx) - if err != nil { - return err + insert := "INSERT INTO im_users(id,username,createdat,updatedat,state,access,phone,public,trusted) VALUES(?,?,?,?,?,?,?,?,?)" + args := []any{decoded_uid, user.Username, user.CreatedAt, user.UpdatedAt, user.State, user.Access, phone, + common.ToJSON(user.Public), common.ToJSON(user.Trusted)} + if hasAccountNo { + insert = "INSERT INTO im_users(id,account_no,username,createdat,updatedat,state,access,phone,public,trusted) VALUES(?,?,?,?,?,?,?,?,?,?)" + args = []any{decoded_uid, accountNo, user.Username, user.CreatedAt, user.UpdatedAt, user.State, user.Access, phone, + common.ToJSON(user.Public), common.ToJSON(user.Trusted)} } - if _, err = tx.Exec("INSERT INTO im_users(id,account_no,username,createdat,updatedat,state,access,phone,public,trusted) VALUES(?,?,?,?,?,?,?,?,?,?)", - decoded_uid, - accountNo, - user.Username, - user.CreatedAt, - user.UpdatedAt, - user.State, - user.Access, - phone, - common.ToJSON(user.Public), - common.ToJSON(user.Trusted)); err != nil { + if _, err = tx.Exec(insert, args...); err != nil { if isDupe(err) { return t.ErrDuplicate } @@ -1859,35 +1852,36 @@ func (a *adapter) UserCreate(user *t.User) error { return tx.Commit() } -func lockUserAccountNoSequence(ctx context.Context, tx *sqlx.Tx) (func(), error) { - var locked sql.NullInt64 - if err := tx.QueryRowContext(ctx, "SELECT GET_LOCK(CONCAT(DATABASE(), ':im_users_account_no'), 10)").Scan(&locked); err != nil { - if isPXCStrictModeNamedLockError(err) { - return func() {}, nil - } - return nil, err +// nextUserAccountNo allocates a tenant-local, human-facing account number. +// The sequence row serializes concurrent registrations across server nodes. +func nextUserAccountNo(ctx context.Context, tx *sqlx.Tx) (uint32, bool, error) { + var columns int + if err := tx.GetContext(ctx, &columns, `SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='im_users' AND COLUMN_NAME='account_no'`); err != nil { + return 0, false, err } - if !locked.Valid || locked.Int64 != 1 { - return nil, fmt.Errorf("lock user account number sequence timeout") + if columns == 0 { + return 0, false, nil } - return func() { - _, _ = tx.ExecContext(ctx, "SELECT RELEASE_LOCK(CONCAT(DATABASE(), ':im_users_account_no'))") - }, nil -} -func nextUserAccountNo(ctx context.Context, tx *sqlx.Tx) (int64, error) { - var current sql.NullInt64 - err := tx.QueryRowContext(ctx, "SELECT account_no FROM im_users ORDER BY account_no DESC LIMIT 1 FOR UPDATE").Scan(¤t) - if err == sql.ErrNoRows { - return 1000, nil - } + result, err := tx.ExecContext(ctx, `UPDATE im_user_account_sequences + SET next_value=LAST_INSERT_ID(next_value+1) WHERE id=1`) if err != nil { - return 0, err + return 0, true, err } - if !current.Valid || current.Int64 < 999 { - return 1000, nil + if affected, affectedErr := result.RowsAffected(); affectedErr != nil { + return 0, true, affectedErr + } else if affected != 1 { + return 0, true, errors.New("mysql: user account sequence is not initialized") } - return current.Int64 + 1, nil + var next uint64 + if err = tx.GetContext(ctx, &next, "SELECT LAST_INSERT_ID()"); err != nil { + return 0, true, err + } + if next > uint64(^uint32(0)) { + return 0, true, errors.New("mysql: user account number exhausted") + } + return uint32(next), true, nil } // TenantAuthAddRecord adds the user's Basic authentication record. diff --git a/server/db/mysql/group_message_shards.go b/server/db/mysql/group_message_shards.go new file mode 100644 index 0000000..de92e12 --- /dev/null +++ b/server/db/mysql/group_message_shards.go @@ -0,0 +1,208 @@ +//go:build mysql +// +build mysql + +package mysql + +import ( + "context" + "fmt" + "time" + + "github.com/jmoiron/sqlx" + "github.com/tinode/chat/server/store" + t "github.com/tinode/chat/server/store/types" +) + +const legacyGroupMessageTable = "im_group_messages" + +func ensureGroupMessageShardInfrastructure(exec sqlExecutor) error { + statements := []string{ + `CREATE TABLE IF NOT EXISTS im_group_message_shards ( + shard_month CHAR(6) NOT NULL, table_name VARCHAR(64) NOT NULL, + shard_start DATETIME(3) NOT NULL, shard_end DATETIME(3) NOT NULL, + status VARCHAR(16) NOT NULL DEFAULT 'prepared', row_count BIGINT NOT NULL DEFAULT 0, + min_created_at DATETIME(3), max_created_at DATETIME(3), + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY(shard_month), UNIQUE INDEX uk_im_group_message_shards_table(table_name), + INDEX idx_im_group_message_shards_status(status, shard_month) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, + `CREATE TABLE IF NOT EXISTS im_group_message_topic_shards ( + topic VARCHAR(64) NOT NULL, shard_month CHAR(6) NOT NULL, + min_seq BIGINT NOT NULL, max_seq BIGINT NOT NULL, message_count BIGINT NOT NULL DEFAULT 0, + min_created_at DATETIME(3) NOT NULL, max_created_at DATETIME(3) NOT NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY(topic, shard_month), INDEX idx_im_group_message_topic_shards_seq(topic, max_seq), + INDEX idx_im_group_message_topic_shards_month(shard_month, topic) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, + `CREATE TABLE IF NOT EXISTS im_group_message_dedup ( + topic VARCHAR(64) NOT NULL, from_user BIGINT NOT NULL, client_msg_id VARCHAR(64) NOT NULL, + seqid BIGINT NOT NULL, shard_month CHAR(6) NOT NULL, message_id BIGINT, + status VARCHAR(16) NOT NULL DEFAULT 'pending', expires_at DATETIME(3) NOT NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY(topic, from_user, client_msg_id), INDEX idx_im_group_message_dedup_expires(expires_at), + INDEX idx_im_group_message_dedup_topic_seq(topic, seqid) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, + } + for _, statement := range statements { + if _, err := exec.Exec(statement); err != nil { + return err + } + } + + return ensureGroupMessageShardWindow(exec, time.Now().UTC()) +} + +func ensureGroupMessageShardWindow(exec sqlExecutor, now time.Time) error { + first := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.UTC) + for _, offset := range []int{-1, 0, 1, 2} { + start := first.AddDate(0, offset, 0) + status := "prepared" + if offset == -1 { + status = "sealing" + } else if offset == 0 { + status = "writable" + } + if err := ensureGroupMessageShard(exec, start, status); err != nil { + return err + } + } + _, err := exec.Exec("UPDATE im_group_message_shards SET status='sealed' WHERE status='sealing' AND shard_end '9' { + return "", false + } + } + parsed, err := time.Parse("200601", month) + if err != nil || parsed.Format("200601") != month { + return "", false + } + return "im_group_messages_" + month, true +} + +func groupMessageTablesForTopic(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID, topic string) ([]string, error) { + routes, cached, err := store.GetGroupMessageShardRoutes(tenantID, topic) + redisAvailable := err == nil + if err != nil { + cached = false + } + if !cached { + routes = nil + if err = sqlx.SelectContext(ctx, q, &routes, + `SELECT shard_month AS month,min_seq,max_seq,message_count AS count + FROM im_group_message_topic_shards WHERE topic=? ORDER BY shard_month DESC`, topic); err != nil { + return nil, err + } + if redisAvailable { + for _, route := range routes { + _ = store.PutGroupMessageShardRoute(tenantID, topic, route) + } + } + } + tables := make([]string, 0, len(routes)+1) + for _, route := range routes { + if table, ok := groupMessageShardTable(route.Month); ok { + tables = append(tables, table) + } + } + return append(tables, legacyGroupMessageTable), nil +} + +func groupMessageTablesForSeq(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID, topic string, seq int64) ([]string, error) { + routes, cached, err := store.GetGroupMessageShardRoutes(tenantID, topic) + redisAvailable := err == nil + if err != nil { + cached = false + } + if !cached { + routes = nil + if err = sqlx.SelectContext(ctx, q, &routes, + `SELECT shard_month AS month,min_seq,max_seq,message_count AS count + FROM im_group_message_topic_shards WHERE topic=? AND min_seq<=? AND max_seq>=? + ORDER BY shard_month DESC`, topic, seq, seq); err != nil { + return nil, err + } + if redisAvailable { + for _, route := range routes { + _ = store.PutGroupMessageShardRoute(tenantID, topic, route) + } + } + } + tables := make([]string, 0, 2) + for _, route := range routes { + if route.MinSeq <= seq && route.MaxSeq >= seq { + if table, ok := groupMessageShardTable(route.Month); ok { + tables = append(tables, table) + } + } + } + return append(tables, legacyGroupMessageTable), nil +} + +func groupMessageMonthFromTable(table string) string { + const prefix = "im_group_messages_" + if len(table) == len(prefix)+6 && table[:len(prefix)] == prefix { + return table[len(prefix):] + } + return "" +} diff --git a/server/db/mysql/group_schema.go b/server/db/mysql/group_schema.go index 8e94e3f..20254ee 100644 --- a/server/db/mysql/group_schema.go +++ b/server/db/mysql/group_schema.go @@ -167,5 +167,5 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error { return err } } - return nil + return ensureGroupMessageShardInfrastructure(exec) } diff --git a/server/db/mysql/group_schema_test.go b/server/db/mysql/group_schema_test.go index 3b684d1..77de2ae 100644 --- a/server/db/mysql/group_schema_test.go +++ b/server/db/mysql/group_schema_test.go @@ -22,8 +22,8 @@ func TestEnsureIMGroupBusinessTablesUsesSelectedTenantDatabase(t *testing.T) { if err := ensureIMGroupBusinessTables(exec); err != nil { t.Fatal(err) } - if len(exec.statements) != 13 { - t.Fatalf("executed %d group schema statements, want 13", len(exec.statements)) + if len(exec.statements) != 25 { + t.Fatalf("executed %d group schema statements, want 25", len(exec.statements)) } for _, statement := range exec.statements { if strings.Contains(strings.ToLower(statement), "tinode.") || strings.Contains(strings.ToLower(statement), "create database") { @@ -46,4 +46,20 @@ func TestEnsureIMGroupBusinessTablesUsesSelectedTenantDatabase(t *testing.T) { if !strings.Contains(joined, "json_set(public, '$.fn'") { t.Fatal("group schema is missing historical public.fn backfill") } + for _, expected := range []string{"im_group_message_shards", "im_group_message_topic_shards", "im_group_message_dedup", "primary key(message_id)"} { + if !strings.Contains(joined, expected) { + t.Fatalf("group shard schema is missing %q", expected) + } + } +} + +func TestGroupMessageShardTableRejectsUnsafeMonth(t *testing.T) { + if table, ok := groupMessageShardTable("202609"); !ok || table != "im_group_messages_202609" { + t.Fatalf("unexpected valid shard: table=%q ok=%v", table, ok) + } + for _, month := range []string{"", "20269", "202613", "2026_09", "202609;DROP TABLE im_users"} { + if table, ok := groupMessageShardTable(month); ok || table != "" { + t.Fatalf("unsafe shard month accepted: month=%q table=%q", month, table) + } + } } diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 42f927a..e26995c 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -270,6 +270,9 @@ func finalizeTenantMessage(id t.TenantID, msg t.Message, recalledBy sql.NullInt6 func normalizeGroupMessage(id t.TenantID, msg t.Message, recalledBy sql.NullInt64) t.Message { msg.TenantID = id + if rawID, err := strconv.ParseInt(msg.Id, 10, 64); err == nil && rawID > 0 { + msg.SetUid(store.EncodeUid(rawID)) + } msg.From = common.EncodeUidString(msg.From).String() if recalledBy.Valid { msg.RecalledBy = common.EncodeUidString(strconv.FormatInt(recalledBy.Int64, 10)).String() @@ -1390,44 +1393,70 @@ func tenantMessageDeleteList(tx *sqlx.Tx, tenantID t.TenantID, topic string, toD return nil } -func groupMessageDeleteList(tx *sqlx.Tx, topic string, toDel *t.DelMessage) error { +func groupMessageDeleteList(tx *sqlx.Tx, tenantID t.TenantID, topic string, toDel *t.DelMessage) error { var err error + ctx := context.Background() + tables, err := groupMessageTablesForTopic(tx, ctx, tenantID, topic) + if err != nil { + return err + } if toDel == nil { if _, err = tx.Exec("DELETE FROM im_dellog WHERE topic=?", topic); err != nil { return err } - _, err = tx.Exec("DELETE FROM im_group_messages WHERE topic=?", topic) + for _, table := range tables { + if _, err = tx.Exec("DELETE FROM "+table+" WHERE topic=?", topic); err != nil { + return err + } + } + if _, err = tx.Exec("DELETE FROM im_group_message_topic_shards WHERE topic=?", topic); err != nil { + return err + } + _, err = tx.Exec("DELETE FROM im_group_message_dedup WHERE topic=?", topic) return err } delRanges := toDel.SeqIdRanges if toDel.DeletedFor == "" { - where := "m.topic=?" - args := []any{topic} - if len(delRanges) > 0 { - rSQL, rArgs := common.RangesToSql(delRanges) - where += " AND m.seqid " + rSQL - args = append(args, rArgs...) - } - where += " AND m.deleted_at IS NULL" - if newerThan := toDel.GetNewerThan(); newerThan != nil { - where += " AND m.created_at>?" - args = append(args, newerThan) - } - var seqIDs []int64 - if err = tx.Select(&seqIDs, "SELECT seqid FROM im_group_messages AS m WHERE "+where, args...); err != nil { - return err + seqIDs := make([]int64, 0) + for _, table := range tables { + where := "m.topic=?" + args := []any{topic} + if len(delRanges) > 0 { + rSQL, rArgs := common.RangesToSql(delRanges) + where += " AND m.seqid " + rSQL + args = append(args, rArgs...) + } + where += " AND m.deleted_at IS NULL" + if newerThan := toDel.GetNewerThan(); newerThan != nil { + where += " AND m.created_at>?" + args = append(args, newerThan) + } + var tableSeqIDs []int64 + if err = tx.Select(&tableSeqIDs, "SELECT seqid FROM "+table+" AS m WHERE "+where, args...); err != nil { + return err + } + seqIDs = append(seqIDs, tableSeqIDs...) } if len(seqIDs) == 0 { return nil } sort.Slice(seqIDs, func(i, j int) bool { return seqIDs[i] < seqIDs[j] }) - delRanges = t.SliceToRanges(seqIDs) + uniqueSeqIDs := seqIDs[:0] + for _, seqID := range seqIDs { + if len(uniqueSeqIDs) == 0 || uniqueSeqIDs[len(uniqueSeqIDs)-1] != seqID { + uniqueSeqIDs = append(uniqueSeqIDs, seqID) + } + } + delRanges = t.SliceToRanges(uniqueSeqIDs) rSQL, rArgs := common.RangesToSql(delRanges) - where = "m.topic=? AND m.seqid " + rSQL - args = append([]any{topic}, rArgs...) - if _, err = tx.Exec("UPDATE im_group_messages AS m SET m.updated_at=?,m.deleted_at=?,m.status='deleted',m.from_user=0,m.head=NULL,m.content=NULL,m.content_size=0 WHERE "+ - where, append([]any{t.TimeNow(), t.TimeNow()}, args...)...); err != nil { - return err + where := "m.topic=? AND m.seqid " + rSQL + args := append([]any{topic}, rArgs...) + now := t.TimeNow() + for _, table := range tables { + if _, err = tx.Exec("UPDATE "+table+" AS m SET m.updated_at=?,m.deleted_at=?,m.status='deleted',m.from_user=0,m.head=NULL,m.content=NULL,m.content_size=0 WHERE "+ + where, append([]any{now, now}, args...)...); err != nil { + return err + } } } stmt, err := tx.Prepare("INSERT INTO im_dellog(topic,deletedfor,delid,low,hi) VALUES(?,?,?,?,?)") @@ -1481,7 +1510,7 @@ func (a *adapter) TenantTopicDelete(id t.TenantID, topic string, isChan, hard bo return err } if t.GetTopicCat(topic) == t.TopicCatGrp { - err = groupMessageDeleteList(tx, topic, nil) + err = groupMessageDeleteList(tx, id, topic, nil) } else { err = tenantMessageDeleteList(tx, id, topic, nil) } @@ -2053,6 +2082,9 @@ func (a *adapter) TenantGroupMessageSave(id t.TenantID, msg *t.Message) error { if tenantAdp != a { return tenantAdp.TenantGroupMessageSave(id, msg) } + if err = a.ensureCurrentGroupMessageShardWindow(); err != nil { + return err + } ctx, cancel := a.getContext() if cancel != nil { defer cancel() @@ -2061,24 +2093,77 @@ func (a *adapter) TenantGroupMessageSave(id t.TenantID, msg *t.Message) error { if err != nil { return err } + if msg.ShardMonth == "" { + msg.ShardMonth = msg.CreatedAt.UTC().Format("200601") + } + table, ok := groupMessageShardTable(msg.ShardMonth) + if !ok || msg.Uid().IsZero() { + return t.ErrMalformed + } + if msg.ShardMonth == time.Now().UTC().Format("200601") { + if _, err = a.db.ExecContext(ctx, "UPDATE im_group_message_shards SET status='writable' WHERE shard_month=? AND status='prepared'", msg.ShardMonth); err != nil { + return err + } + } + var shardStatus string + if err = a.db.QueryRowContext(ctx, "SELECT status FROM im_group_message_shards WHERE shard_month=? AND table_name=?", msg.ShardMonth, table).Scan(&shardStatus); err != nil { + if err == sql.ErrNoRows { + return fmt.Errorf("mysql: group message shard %s is not registered", msg.ShardMonth) + } + return err + } + if shardStatus != "writable" && shardStatus != "sealing" { + return fmt.Errorf("mysql: group message shard %s is %s", msg.ShardMonth, shardStatus) + } clientMsgID := sql.NullString{String: msg.ClientMsgID, Valid: msg.ClientMsgID != ""} contentJSON := common.ToJSON(msg.Content) - res, err := a.db.ExecContext(ctx, - `INSERT INTO im_group_messages - (topic,group_id,seqid,from_user,client_msg_id,msg_type,status,content,head,mentioned_user_ids,mention_all, + tx, err := a.db.BeginTxx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + messageID := store.DecodeUid(msg.Uid()) + if clientMsgID.Valid { + if _, err = tx.ExecContext(ctx, `INSERT INTO im_group_message_dedup + (topic,from_user,client_msg_id,seqid,shard_month,message_id,status,expires_at) + VALUES(?,?,?,?,?,?,'done',?)`, msg.Topic, decodeUID(t.ParseUid(msg.From)), clientMsgID, + msg.SeqId, msg.ShardMonth, messageID, time.Now().UTC().Add(30*24*time.Hour)); err != nil { + if isDupe(err) { + return t.ErrDuplicate + } + return err + } + } + query := fmt.Sprintf(`INSERT INTO %s + (message_id,topic,group_id,seqid,from_user,client_msg_id,msg_type,status,content,head,mentioned_user_ids,mention_all, file_count,content_size,version,created_at,updated_at) - VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, - msg.Topic, groupID, msg.SeqId, decodeUID(t.ParseUid(msg.From)), clientMsgID, messageHeadType(msg.Head), "normal", + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, table) + _, err = tx.ExecContext(ctx, query, + messageID, msg.Topic, groupID, msg.SeqId, decodeUID(t.ParseUid(msg.From)), clientMsgID, messageHeadType(msg.Head), "normal", contentJSON, msg.Head, messageHeadJSON(msg.Head, "mentioned_user_ids"), messageHeadBool(msg.Head, "mention_all"), 0, len(contentJSON), messageVersion(msg.Version), msg.CreatedAt, msg.UpdatedAt) - if err == nil { - lastID, _ := res.LastInsertId() - msg.SetUid(t.Uid(lastID)) - } if isDupe(err) { return t.ErrDuplicate } - return err + if err != nil { + return err + } + if _, err = tx.ExecContext(ctx, `INSERT INTO im_group_message_topic_shards + (topic,shard_month,min_seq,max_seq,message_count,min_created_at,max_created_at) + VALUES(?,?,?,?,1,?,?) ON DUPLICATE KEY UPDATE + min_seq=LEAST(min_seq,VALUES(min_seq)),max_seq=GREATEST(max_seq,VALUES(max_seq)), + message_count=message_count+1,min_created_at=LEAST(min_created_at,VALUES(min_created_at)), + max_created_at=GREATEST(max_created_at,VALUES(max_created_at))`, + msg.Topic, msg.ShardMonth, msg.SeqId, msg.SeqId, msg.CreatedAt, msg.CreatedAt); err != nil { + return err + } + if _, err = tx.ExecContext(ctx, `UPDATE im_group_message_shards SET row_count=row_count+1, + min_created_at=IF(min_created_at IS NULL,?,LEAST(min_created_at,?)), + max_created_at=IF(max_created_at IS NULL,?,GREATEST(max_created_at,?)) WHERE shard_month=?`, + msg.CreatedAt, msg.CreatedAt, msg.CreatedAt, msg.CreatedAt, msg.ShardMonth); err != nil { + return err + } + return tx.Commit() } func (a *adapter) TenantMessageGetByClientID(id t.TenantID, from t.Uid, clientMsgID string) (*t.Message, error) { @@ -2131,22 +2216,44 @@ func (a *adapter) TenantGroupMessageGetByClientID(id t.TenantID, topic string, f if cancel != nil { defer cancel() } - row := struct { - t.Message - RecalledBy sql.NullInt64 `db:"recalled_by"` - }{} - err = a.db.QueryRowxContext(ctx, - "SELECT id,created_at AS createdat,updated_at AS updatedat,deleted_at AS deletedat,0 AS delid,seqid,topic,from_user AS `from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content "+ - "FROM im_group_messages WHERE topic=? AND from_user=? AND client_msg_id=?", - topic, decodeUID(from), clientMsgID).StructScan(&row) - if err == sql.ErrNoRows { - return nil, nil + var seq int64 + if err = a.db.QueryRowContext(ctx, `SELECT seqid FROM im_group_message_dedup + WHERE topic=? AND from_user=? AND client_msg_id=?`, topic, decodeUID(from), clientMsgID).Scan(&seq); err == nil { + msg, getErr := tenantGroupMessageGetOne(a.db, ctx, id, topic, seq, "") + if getErr == nil { + return msg, nil + } + if getErr != sql.ErrNoRows { + return nil, getErr + } + } else if err != sql.ErrNoRows { + return nil, err } + tables, err := groupMessageTablesForTopic(a.db, ctx, id, topic) if err != nil { return nil, err } - msg := normalizeGroupMessage(id, row.Message, row.RecalledBy) - return &msg, nil + for _, table := range tables { + month := groupMessageMonthFromTable(table) + idColumn := "id" + if month != "" { + idColumn = "message_id" + } + row := struct { + t.Message + RecalledBy sql.NullInt64 `db:"recalled_by"` + }{} + query := fmt.Sprintf("SELECT %s AS id,created_at AS createdat,updated_at AS updatedat,deleted_at AS deletedat,0 AS delid,seqid,topic,from_user AS `from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content,'%s' AS shard_month FROM %s WHERE topic=? AND from_user=? AND client_msg_id=?", idColumn, month, table) + err = a.db.QueryRowxContext(ctx, query, topic, decodeUID(from), clientMsgID).StructScan(&row) + if err == nil { + msg := normalizeGroupMessage(id, row.Message, row.RecalledBy) + return &msg, nil + } + if err != sql.ErrNoRows { + return nil, err + } + } + return nil, nil } func (a *adapter) TenantMessageGetOne(id t.TenantID, topic string, seq int64) (*t.Message, error) { @@ -2270,7 +2377,7 @@ func (a *adapter) TenantGroupMessageRecall(id t.TenantID, topic string, seq int6 tx.Rollback() } }() - msg, err := tenantGroupMessageGetOne(tx, ctx, id, topic, seq, " FOR UPDATE") + msg, messageTable, err := tenantGroupMessageGetOneWithTable(tx, ctx, id, topic, seq, " FOR UPDATE") if err != nil { if err == sql.ErrNoRows { return nil, t.ErrNotFound @@ -2280,8 +2387,8 @@ func (a *adapter) TenantGroupMessageRecall(id t.TenantID, topic string, seq int6 if msg.Version != expectedVersion || msg.RecalledAt != nil || msg.DeletedAt != nil { return nil, t.ErrDuplicate } - _, err = tx.ExecContext(ctx, - "UPDATE im_group_messages SET updated_at=?,version=version+1,status='recalled',recalled_at=?,recalled_by=?,head=NULL,content=NULL WHERE topic=? AND seqid=? AND version=?", + _, err = tx.ExecContext(ctx, fmt.Sprintf( + "UPDATE %s SET updated_at=?,version=version+1,status='recalled',recalled_at=?,recalled_by=?,head=NULL,content=NULL WHERE topic=? AND seqid=? AND version=?", messageTable), recalledAt, recalledAt, decodeUID(actor), topic, seq, expectedVersion) if err != nil { return nil, err @@ -2375,7 +2482,7 @@ func (a *adapter) TenantGroupMessageEdit(id t.TenantID, topic string, seq int64, tx.Rollback() } }() - msg, err := tenantGroupMessageGetOne(tx, ctx, id, topic, seq, " FOR UPDATE") + msg, messageTable, err := tenantGroupMessageGetOneWithTable(tx, ctx, id, topic, seq, " FOR UPDATE") if err != nil { if err == sql.ErrNoRows { return nil, t.ErrNotFound @@ -2391,8 +2498,8 @@ func (a *adapter) TenantGroupMessageEdit(id t.TenantID, topic string, seq int64, return nil, err } contentJSON := common.ToJSON(content) - _, err = tx.ExecContext(ctx, - "UPDATE im_group_messages SET updated_at=?,version=version+1,msg_type=?,head=?,content=?,content_size=? WHERE topic=? AND seqid=? AND version=?", + _, err = tx.ExecContext(ctx, fmt.Sprintf( + "UPDATE %s SET updated_at=?,version=version+1,msg_type=?,head=?,content=?,content_size=? WHERE topic=? AND seqid=? AND version=?", messageTable), editedAt, messageHeadType(head), head, contentJSON, len(contentJSON), topic, seq, expectedVersion) if err != nil { return nil, err @@ -2483,58 +2590,84 @@ func (a *adapter) TenantGroupMessageGetAll(id t.TenantID, topic string, uid t.Ui return tenantAdp.TenantGroupMessageGetAll(id, topic, uid, opts) } limit := a.maxMessageResults - args := []any{store.DecodeUid(uid), topic} - seq := "" + seqClause := "" + seqArgs := make([]any, 0) if opts != nil { - seq = "AND m.seqid " + seqClause = "AND m.seqid " if len(opts.IdRanges) > 0 { constr, addArgs := common.RangesToSql(opts.IdRanges) - seq += constr - args = append(args, addArgs...) + seqClause += constr + seqArgs = append(seqArgs, addArgs...) } else { - seq += "BETWEEN ? AND ?" + seqClause += "BETWEEN ? AND ?" if opts.Since > 0 { - args = append(args, opts.Since) + seqArgs = append(seqArgs, opts.Since) } else { - args = append(args, 0) + seqArgs = append(seqArgs, 0) } if opts.Before > 1 { - args = append(args, opts.Before-1) + seqArgs = append(seqArgs, opts.Before-1) } else { - args = append(args, 1<<31-1) + seqArgs = append(seqArgs, int64(^uint64(0)>>1)) } } if opts.Limit > 0 && opts.Limit < limit { limit = opts.Limit } } - args = append(args, limit) ctx, cancel := a.getContext() if cancel != nil { defer cancel() } - rows, err := a.db.QueryxContext(ctx, - "SELECT m.id,m.created_at AS createdat,m.updated_at AS updatedat,m.deleted_at AS deletedat,0 AS delid,m.seqid,m.topic,m.from_user AS `from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content "+ - "FROM im_group_messages AS m LEFT JOIN im_dellog AS d "+ - "ON d.topic COLLATE utf8mb4_unicode_ci=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? "+ - "WHERE m.status!='deleted' AND m.topic=? "+seq+" AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", - args...) + tables, err := groupMessageTablesForTopic(a.db, ctx, id, topic) if err != nil { return nil, err } - defer rows.Close() msgs := make([]t.Message, 0, limit) - for rows.Next() { - row := struct { - t.Message - RecalledBy sql.NullInt64 `db:"recalled_by"` - }{} - if err = rows.StructScan(&row); err != nil { + seen := make(map[int64]struct{}, limit) + for _, table := range tables { + if len(msgs) >= limit { + break + } + month := groupMessageMonthFromTable(table) + idColumn := "id" + if month != "" { + idColumn = "message_id" + } + args := []any{store.DecodeUid(uid), topic} + args = append(args, seqArgs...) + args = append(args, limit-len(msgs)) + query := fmt.Sprintf("SELECT m.%s AS id,m.created_at AS createdat,m.updated_at AS updatedat,m.deleted_at AS deletedat,0 AS delid,m.seqid,m.topic,m.from_user AS `from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content,'%s' AS shard_month FROM %s AS m LEFT JOIN im_dellog AS d ON d.topic COLLATE utf8mb4_unicode_ci=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? WHERE m.status!='deleted' AND m.topic=? %s AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", idColumn, month, table, seqClause) + rows, queryErr := a.db.QueryxContext(ctx, query, args...) + if queryErr != nil { + return nil, queryErr + } + for rows.Next() { + row := struct { + t.Message + RecalledBy sql.NullInt64 `db:"recalled_by"` + }{} + if err = rows.StructScan(&row); err != nil { + rows.Close() + return nil, err + } + if _, exists := seen[row.SeqId]; exists { + continue + } + seen[row.SeqId] = struct{}{} + msgs = append(msgs, normalizeGroupMessage(id, row.Message, row.RecalledBy)) + } + if err = rows.Err(); err != nil { + rows.Close() return nil, err } - msgs = append(msgs, normalizeGroupMessage(id, row.Message, row.RecalledBy)) + rows.Close() } - return msgs, rows.Err() + sort.Slice(msgs, func(i, j int) bool { return msgs[i].SeqId > msgs[j].SeqId }) + if len(msgs) > limit { + msgs = msgs[:limit] + } + return msgs, nil } func (a *adapter) TenantMessageSearch(id t.TenantID, topic string, uid t.Uid, queryText, messageType, cursor string, limit int) ([]t.Message, string, error) { @@ -2619,7 +2752,7 @@ func (a *adapter) tenantGroupMessageSearch(id t.TenantID, topic string, uid t.Ui where := []string{"m.status!='deleted'", "m.topic=?"} args := []any{topic} if cursor != "" { - parsed, err := strconv.Atoi(cursor) + parsed, err := strconv.ParseInt(cursor, 10, 64) if err != nil || parsed <= 0 { return nil, "", t.ErrMalformed } @@ -2640,39 +2773,55 @@ func (a *adapter) tenantGroupMessageSearch(id t.TenantID, topic string, uid t.Ui if cancel != nil { defer cancel() } - queryArgs := append([]any{store.DecodeUid(uid)}, args...) - queryArgs = append(queryArgs, limit+1) - rows, err := a.db.QueryxContext(ctx, - "SELECT m.id,m.created_at AS createdat,m.updated_at AS updatedat,m.deleted_at AS deletedat,0 AS delid,m.seqid,m.topic,m.from_user AS `from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content "+ - "FROM im_group_messages AS m LEFT JOIN im_dellog AS d "+ - "ON d.topic COLLATE utf8mb4_unicode_ci=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? "+ - "WHERE "+strings.Join(where, " AND ")+" AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", - queryArgs...) + tables, err := groupMessageTablesForTopic(a.db, ctx, id, topic) if err != nil { return nil, "", err } - defer rows.Close() - msgs := make([]t.Message, 0, limit) - nextCursor := "" - lastSeq := int64(0) - for rows.Next() { - row := struct { - t.Message - RecalledBy sql.NullInt64 `db:"recalled_by"` - }{} - if err = rows.StructScan(&row); err != nil { + msgs := make([]t.Message, 0, limit+1) + seen := make(map[int64]struct{}, limit+1) + for _, table := range tables { + if len(msgs) > limit { + break + } + month := groupMessageMonthFromTable(table) + idColumn := "id" + if month != "" { + idColumn = "message_id" + } + queryArgs := append([]any{store.DecodeUid(uid)}, args...) + queryArgs = append(queryArgs, limit+1-len(msgs)) + query := fmt.Sprintf("SELECT m.%s AS id,m.created_at AS createdat,m.updated_at AS updatedat,m.deleted_at AS deletedat,0 AS delid,m.seqid,m.topic,m.from_user AS `from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content,'%s' AS shard_month FROM %s AS m LEFT JOIN im_dellog AS d ON d.topic COLLATE utf8mb4_unicode_ci=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? WHERE %s AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", idColumn, month, table, strings.Join(where, " AND ")) + rows, queryErr := a.db.QueryxContext(ctx, query, queryArgs...) + if queryErr != nil { + return nil, "", queryErr + } + for rows.Next() { + row := struct { + t.Message + RecalledBy sql.NullInt64 `db:"recalled_by"` + }{} + if err = rows.StructScan(&row); err != nil { + rows.Close() + return nil, "", err + } + if _, exists := seen[row.SeqId]; exists { + continue + } + seen[row.SeqId] = struct{}{} + msgs = append(msgs, normalizeGroupMessage(id, row.Message, row.RecalledBy)) + } + if err = rows.Err(); err != nil { + rows.Close() return nil, "", err } - msg := normalizeGroupMessage(id, row.Message, row.RecalledBy) - if len(msgs) == limit { - nextCursor = strconv.FormatInt(lastSeq, 10) - continue - } - msgs = append(msgs, msg) - lastSeq = msg.SeqId + rows.Close() } - if err = rows.Err(); err != nil { - return nil, "", err + sort.Slice(msgs, func(i, j int) bool { return msgs[i].SeqId > msgs[j].SeqId }) + nextCursor := "" + if len(msgs) > limit { + msgs = msgs[:limit] + lastSeq := msgs[len(msgs)-1].SeqId + nextCursor = strconv.FormatInt(lastSeq, 10) } return msgs, nextCursor, nil } @@ -7101,19 +7250,36 @@ func (a *adapter) groupIDByTopic(ctx context.Context, topic string) (int64, erro } func tenantGroupMessageGetOne(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID, topic string, seq int64, suffix string) (*t.Message, error) { - row := struct { - t.Message - RecalledBy sql.NullInt64 `db:"recalled_by"` - }{} - err := sqlx.GetContext(ctx, q, &row, - "SELECT id,created_at AS createdat,updated_at AS updatedat,deleted_at AS deletedat,0 AS delid,seqid,topic,from_user AS `from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content "+ - "FROM im_group_messages WHERE topic=? AND seqid=?"+suffix, - topic, seq) + msg, _, err := tenantGroupMessageGetOneWithTable(q, ctx, tenantID, topic, seq, suffix) + return msg, err +} + +func tenantGroupMessageGetOneWithTable(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID, topic string, seq int64, suffix string) (*t.Message, string, error) { + tables, err := groupMessageTablesForSeq(q, ctx, tenantID, topic, seq) if err != nil { - return nil, err + return nil, "", err } - msg := normalizeGroupMessage(tenantID, row.Message, row.RecalledBy) - return &msg, nil + for _, table := range tables { + month := groupMessageMonthFromTable(table) + idColumn := "id" + if month != "" { + idColumn = "message_id" + } + row := struct { + t.Message + RecalledBy sql.NullInt64 `db:"recalled_by"` + }{} + query := fmt.Sprintf("SELECT %s AS id,created_at AS createdat,updated_at AS updatedat,deleted_at AS deletedat,0 AS delid,seqid,topic,from_user AS `from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content,'%s' AS shard_month FROM %s WHERE topic=? AND seqid=?%s", idColumn, month, table, suffix) + err = sqlx.GetContext(ctx, q, &row, query, topic, seq) + if err == nil { + msg := normalizeGroupMessage(tenantID, row.Message, row.RecalledBy) + return &msg, table, nil + } + if err != sql.ErrNoRows { + return nil, "", err + } + } + return nil, "", sql.ErrNoRows } func messageVersion(version int) int { @@ -7594,7 +7760,7 @@ func (a *adapter) TenantGroupMessageDeleteList(id t.TenantID, topic string, del tx.Rollback() } }() - if err = groupMessageDeleteList(tx, topic, del); err != nil { + if err = groupMessageDeleteList(tx, id, topic, del); err != nil { return err } return tx.Commit() diff --git a/server/db/mysql/tenant_security_audit.go b/server/db/mysql/tenant_security_audit.go index 31ab894..9d1078a 100644 --- a/server/db/mysql/tenant_security_audit.go +++ b/server/db/mysql/tenant_security_audit.go @@ -98,7 +98,7 @@ func (a *adapter) TenantMessageAdminApply(id t.TenantID, action *t.MessageAdminA return tx.Commit() } - sender, beforeStatus, err := lockAdminMessage(ctx, tx, action.ConversationType, action.Topic, action.SeqID) + sender, beforeStatus, messageTable, err := lockAdminMessage(ctx, tx, id, action.ConversationType, action.Topic, action.SeqID) if err != nil { return err } @@ -108,7 +108,7 @@ func (a *adapter) TenantMessageAdminApply(id t.TenantID, action *t.MessageAdminA afterStatus = "deleted" } if action.ConversationType == "group" { - _, err = tx.ExecContext(ctx, "UPDATE im_group_messages SET status=?,deleted_at=?,updated_at=? WHERE topic=? AND seqid=?", afterStatus, action.ProcessedAt, action.ProcessedAt, action.Topic, action.SeqID) + _, err = tx.ExecContext(ctx, "UPDATE "+messageTable+" SET status=?,deleted_at=?,updated_at=? WHERE topic=? AND seqid=?", afterStatus, action.ProcessedAt, action.ProcessedAt, action.Topic, action.SeqID) } else { _, err = tx.ExecContext(ctx, "UPDATE im_messages SET deletedat=?,updatedat=? WHERE topic=? AND seqid=?", action.ProcessedAt, action.ProcessedAt, action.Topic, action.SeqID) } @@ -127,17 +127,26 @@ func (a *adapter) TenantMessageAdminApply(id t.TenantID, action *t.MessageAdminA return tx.Commit() } -func lockAdminMessage(ctx context.Context, tx *sqlx.Tx, conversationType, topic string, seqID int64) (int64, string, error) { +func lockAdminMessage(ctx context.Context, tx *sqlx.Tx, tenantID t.TenantID, conversationType, topic string, seqID int64) (int64, string, string, error) { if conversationType == "group" { + tables, err := groupMessageTablesForSeq(tx, ctx, tenantID, topic, seqID) + if err != nil { + return 0, "", "", err + } var row struct { Sender int64 `db:"from_user"` Status string `db:"status"` } - err := tx.GetContext(ctx, &row, "SELECT from_user,status FROM im_group_messages WHERE topic=? AND seqid=? FOR UPDATE", topic, seqID) - if errors.Is(err, sql.ErrNoRows) { - return 0, "", t.ErrNotFound + for _, table := range tables { + err = tx.GetContext(ctx, &row, "SELECT from_user,status FROM "+table+" WHERE topic=? AND seqid=? FOR UPDATE", topic, seqID) + if err == nil { + return row.Sender, row.Status, table, nil + } + if !errors.Is(err, sql.ErrNoRows) { + return 0, "", "", err + } } - return row.Sender, row.Status, err + return 0, "", "", t.ErrNotFound } var row struct { Sender int64 `db:"from"` @@ -146,7 +155,7 @@ func lockAdminMessage(ctx context.Context, tx *sqlx.Tx, conversationType, topic } err := tx.GetContext(ctx, &row, "SELECT `from`,deletedat,recalled_at FROM im_messages WHERE topic=? AND seqid=? FOR UPDATE", topic, seqID) if errors.Is(err, sql.ErrNoRows) { - return 0, "", t.ErrNotFound + return 0, "", "", t.ErrNotFound } status := "normal" if row.RecalledAt.Valid { @@ -155,7 +164,7 @@ func lockAdminMessage(ctx context.Context, tx *sqlx.Tx, conversationType, topic if row.DeletedAt.Valid { status = "deleted" } - return row.Sender, status, err + return row.Sender, status, "im_messages", err } func nullableJSON(raw []byte) any { diff --git a/server/db/mysql/user_identity_schema.go b/server/db/mysql/user_identity_schema.go index 77538a7..1a2ce9d 100644 --- a/server/db/mysql/user_identity_schema.go +++ b/server/db/mysql/user_identity_schema.go @@ -150,6 +150,9 @@ func ensureUserIdentitySchema(db *sqlx.DB) error { if _, err = conn.ExecContext(ctx, "ALTER TABLE im_users MODIFY COLUMN username VARCHAR(32) NOT NULL"); err != nil { return err } + if err = ensureUserAccountSequence(ctx, conn); err != nil { + return err + } if err = migrateBasicAuthSchema(ctx, conn, usersScoped); err != nil { return err } @@ -171,6 +174,23 @@ func ensureUserIdentitySchema(db *sqlx.DB) error { return nil } +func ensureUserAccountSequence(ctx context.Context, conn *sqlx.Conn) error { + exists, err := identityColumnExists(ctx, conn, "im_users", "account_no") + if err != nil || !exists { + return err + } + if _, err = conn.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS im_user_account_sequences( + id TINYINT UNSIGNED NOT NULL,next_value INT UNSIGNED NOT NULL, + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY(id)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`); err != nil { + return err + } + _, err = conn.ExecContext(ctx, `INSERT INTO im_user_account_sequences(id,next_value) + SELECT 1,GREATEST(COALESCE(MAX(account_no),999),999) FROM im_users + ON DUPLICATE KEY UPDATE next_value=GREATEST(next_value,VALUES(next_value))`) + return err +} + func backfillUsernames(ctx context.Context, conn *sqlx.Conn, scoped bool) error { prefix := "0 AS tenant_id,u.id AS user_id," join := "a.userid=u.id" diff --git a/server/internal_admin_query_resources.go b/server/internal_admin_query_resources.go index f860bf6..18ad9bd 100644 --- a/server/internal_admin_query_resources.go +++ b/server/internal_admin_query_resources.go @@ -63,19 +63,20 @@ var internalTenantResourceSpecs = map[string]*internalQuerySpec{ "private-messages": privateMessagesInternalQuerySpec(), "group-messages": groupMessagesInternalQuerySpec(), "user-login-logs": { - Select: `SELECT id,CAST(user_id AS CHAR) AS user_id,username,account_type,account_mask,login_type,event_type,session_id,device_id,platform,app_version,ip,region,user_agent,risk_level,request_id,login_at,logout_at,metadata`, + Select: `SELECT id,user_id,username,account_type,account_mask,login_type,event_type,session_id,device_id,platform,app_version,ip,region,user_agent,risk_level,request_id,login_at,logout_at,metadata`, From: "FROM im_user_login_logs", DefaultSort: "ORDER BY id DESC", DetailKey: "id", Filters: []internalQueryFilter{{"user_id", "user_id", "=", "uid"}, {"account", "account_mask", "LIKE", "string"}, {"login_type", "login_type", "=", "string"}, {"event_type", "event_type", "=", "string"}, {"platform", "platform", "=", "string"}, {"ip", "ip", "LIKE", "string"}, {"login_at_start", "login_at", ">=", "time"}, {"login_at_end", "login_at", "<=", "time"}}, }, "login-failure-logs": { - Select: `SELECT id,CAST(user_id AS CHAR) AS user_id,account_type,account_mask,failure_type,reason,detail,ip,region,device_id,platform,app_version,user_agent,risk_level,risk_action,request_id,attempt_at,metadata`, + Select: `SELECT id,user_id,account_type,account_mask,failure_type,reason,detail,ip,region,device_id,platform,app_version,user_agent,risk_level,risk_action,request_id,attempt_at,metadata`, From: "FROM im_login_failure_logs", DefaultSort: "ORDER BY id DESC", DetailKey: "id", Filters: []internalQueryFilter{{"user_id", "user_id", "=", "uid"}, {"account", "account_mask", "LIKE", "string"}, {"failure_type", "failure_type", "=", "string"}, {"platform", "platform", "=", "string"}, {"ip", "ip", "LIKE", "string"}, {"attempt_at_start", "attempt_at", ">=", "time"}, {"attempt_at_end", "attempt_at", "<=", "time"}}, }, "audit-logs": { - Select: "SELECT id,request_id,actor_user_id,action,target_type,target_id,ip,user_agent,detail,created_at", - From: "FROM im_audit_logs", DefaultSort: "ORDER BY id DESC", DetailKey: "id", - Filters: []internalQueryFilter{{"actor_user_id", "actor_user_id", "=", "uid"}, {"action", "action", "=", "string"}, {"target_type", "target_type", "=", "string"}, {"target_id", "target_id", "=", "string"}, {"created_at_start", "created_at", ">=", "time"}, {"created_at_end", "created_at", "<=", "time"}}, + Select: `SELECT id,request_id,actor_uid AS actor_user_id,action,target_type,target_id,ip,user_agent, + JSON_OBJECT('actor_role',actor_role,'before_snapshot',before_snapshot,'after_snapshot',after_snapshot,'reason',reason,'metadata',metadata) AS detail,created_at`, + From: "FROM im_audit_logs", DefaultSort: "ORDER BY id DESC", DetailKey: "id", + Filters: []internalQueryFilter{{"actor_user_id", "actor_uid", "=", "uid"}, {"action", "action", "=", "string"}, {"target_type", "target_type", "=", "string"}, {"target_id", "target_id", "=", "string"}, {"created_at_start", "created_at", ">=", "time"}, {"created_at_end", "created_at", "<=", "time"}}, }, "realname-verifications": { Select: "SELECT id,user_id,submission_id,status,masked_name AS real_name,masked_id_number AS id_card_no_masked,review_reason AS reject_reason,verified_at AS reviewed_at,created_at,updated_at", @@ -136,6 +137,14 @@ func serveInternalAdminResourceList(wrt http.ResponseWriter, req *http.Request, writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil) return } + if resource == "group-messages" { + var err error + spec, err = groupMessagesInternalQuerySpecForTenant(tenantID) + if err != nil { + writeInternalQueryError(wrt, req, err) + return + } + } query, countQuery, args, page, perPage, err := buildInternalAdminListQuery(req, spec, scope) if err != nil { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) @@ -256,7 +265,7 @@ func internalAdminQuerySpec(resource string) (*internalQuerySpec, bool) { func usersInternalQuerySpec() *internalQuerySpec { return &internalQuerySpec{ - Select: `SELECT CAST(u.id AS CHAR) AS user_id,u.account_no AS display_no,u.username,u.phone,u.state,u.stateat AS state_at,u.access,u.lastseen AS last_seen,u.useragent AS user_agent,u.public,u.trusted, + Select: `SELECT u.id AS user_id,u.account_no AS display_no,u.username,u.phone,u.state,u.stateat AS state_at,u.access,u.lastseen AS last_seen,u.useragent AS user_agent,u.public,u.trusted, p.nickname,p.public_account,p.gender,p.region,p.avatar_file_id, COALESCE( NULLIF(avatar.location,''), @@ -283,7 +292,7 @@ func groupsInternalQuerySpec() *internalQuerySpec { return &internalQuerySpec{ Select: `SELECT t.name AS topic,t.state,t.owner AS owner_id,t.seqid AS seq_id,t.delid AS del_id,t.subcnt AS sub_count,t.touchedat AS touched_at,t.createdat AS created_at,t.updatedat AS updated_at,t.public,t.trusted,t.access,t.tags,t.aux, gs.name AS group_name,gs.avatar_file_id,gs.description,gs.group_status,gs.join_approval,gs.mute_all,gs.member_upload_files,gs.member_private_chat,gs.member_invite,gs.max_member_count,gs.max_admin_count,gs.member_count AS business_member_count,gs.version AS settings_version,gs.updated_at AS settings_updated_at`, - From: "FROM im_topics AS t LEFT JOIN im_group_settings AS gs ON gs.topic=t.name", + From: "FROM (SELECT * FROM im_topics WHERE name LIKE 'grp%') AS t LEFT JOIN im_group_settings AS gs ON gs.topic=t.name", DefaultSort: "ORDER BY t.id DESC", DetailKey: "t.name", Filters: []internalQueryFilter{{"topic", "t.name", "LIKE", "string"}, {"owner_id", "t.owner", "=", "uid"}, {"state", "t.state", "=", "int"}, {"group_status", "gs.group_status", "=", "string"}, {"created_at_start", "t.createdat", ">=", "time"}, {"created_at_end", "t.createdat", "<=", "time"}, {"touched_at_start", "t.touchedat", ">=", "time"}, {"touched_at_end", "t.touchedat", "<=", "time"}}, @@ -296,9 +305,9 @@ func privateMessagesInternalQuerySpec() *internalQuerySpec { contentExcerpt := `LEFT(COALESCE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(m.content,'$.txt')),''),CASE JSON_TYPE(m.content) WHEN 'STRING' THEN JSON_UNQUOTE(m.content) ELSE CONCAT('[',` + contentType + `,']') END,''),500)` status := `CASE WHEN aa.action='delete' THEN 'deleted' WHEN aa.action='hide' THEN 'hidden' WHEN m.recalled_at IS NOT NULL THEN 'recalled' WHEN m.deletedat IS NOT NULL THEN 'deleted' ELSE 'normal' END` return &internalQuerySpec{ - Select: `SELECT m.topic,m.seqid AS seq_id,CAST(m.` + "`from`" + ` AS CHAR) AS from_user_id, + Select: `SELECT m.topic,m.seqid AS seq_id,m.` + "`from`" + ` AS from_user_id, COALESCE(su.username,'') AS sender_username,COALESCE(sp.nickname,'') AS sender_nickname, - CAST(rs.userid AS CHAR) AS to_user_id,COALESCE(ru.username,'') AS receiver_username,COALESCE(rp.nickname,'') AS receiver_nickname, + rs.userid AS to_user_id,COALESCE(ru.username,'') AS receiver_username,COALESCE(rp.nickname,'') AS receiver_nickname, ` + contentType + ` AS content_type,` + contentExcerpt + ` AS content_excerpt, ` + status + ` AS status,m.createdat AS created_at,m.updatedat AS updated_at`, From: `FROM (SELECT * FROM im_messages WHERE topic LIKE 'p2p%') AS m @@ -318,7 +327,7 @@ func groupMessagesInternalQuerySpec() *internalQuerySpec { contentType := `CASE m.msg_type WHEN 'call' THEN 'system' WHEN 'system_notice' THEN 'system' WHEN 'contact_card' THEN 'card' WHEN 'forward_bundle' THEN 'card' WHEN 'link' THEN 'text' ELSE m.msg_type END` contentExcerpt := `LEFT(COALESCE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(m.content,'$.txt')),''),CASE JSON_TYPE(m.content) WHEN 'STRING' THEN JSON_UNQUOTE(m.content) ELSE CONCAT('[',` + contentType + `,']') END,''),500)` return &internalQuerySpec{ - Select: `SELECT m.topic,m.seqid AS seq_id,CAST(m.from_user AS CHAR) AS from_user_id, + Select: `SELECT m.topic,m.seqid AS seq_id,m.from_user AS from_user_id, COALESCE(u.username,'') AS sender_username,COALESCE(p.nickname,'') AS sender_nickname, COALESCE(gs.name,'') AS group_name,m.topic AS group_no,` + contentType + ` AS content_type, ` + contentExcerpt + ` AS content_excerpt, @@ -346,14 +355,68 @@ func groupMessagesInternalQuerySpec() *internalQuerySpec { } } +func groupMessagesInternalQuerySpecForTenant(tenantID types.TenantID) (*internalQuerySpec, error) { + source, err := internalGroupMessageSource(tenantID) + if err != nil { + return nil, err + } + spec := groupMessagesInternalQuerySpec() + spec.From = strings.Replace(spec.From, "FROM im_group_messages AS m", "FROM "+source+" AS m", 1) + spec.DefaultSort = "ORDER BY m.created_at DESC,m.id DESC" + return spec, nil +} + +func internalGroupMessageSource(tenantID types.TenantID) (string, error) { + result, err := store.InternalAdminQueries.Query(tenantID, + "SELECT table_name FROM im_group_message_shards WHERE status<>'archived' ORDER BY shard_month DESC", + "SELECT COUNT(*) FROM im_group_message_shards WHERE status<>'archived'", nil, 1, 200) + if err != nil { + return "", err + } + selects := make([]string, 0) + for _, item := range result.Items { + table := strings.TrimSpace(fmt.Sprint(item["table_name"])) + if !validInternalGroupMessageTable(table) { + continue + } + selects = append(selects, fmt.Sprintf(`SELECT message_id AS id,topic,seqid,from_user,msg_type,status,content,head,reply_seqid,mentioned_user_ids,created_at,updated_at FROM %s`, table)) + } + selects = append(selects, `SELECT id,topic COLLATE utf8mb4_0900_ai_ci AS topic,seqid,from_user, + msg_type COLLATE utf8mb4_0900_ai_ci AS msg_type,status COLLATE utf8mb4_0900_ai_ci AS status, + content,head,reply_seqid,mentioned_user_ids,created_at,updated_at + FROM im_group_messages AS legacy WHERE NOT EXISTS ( + SELECT 1 FROM im_group_message_topic_shards AS route + WHERE route.topic=legacy.topic COLLATE utf8mb4_0900_ai_ci AND legacy.seqid BETWEEN route.min_seq AND route.max_seq)`) + return "(" + strings.Join(selects, " UNION ALL ") + ")", nil +} + +func validInternalGroupMessageTable(table string) bool { + const prefix = "im_group_messages_" + if len(table) != len(prefix)+6 || !strings.HasPrefix(table, prefix) { + return false + } + for _, ch := range table[len(prefix):] { + if ch < '0' || ch > '9' { + return false + } + } + return true +} + func serveInternalDashboardOverview(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) { - row, err := store.InternalAdminQueries.QueryOne(tenantID, `SELECT + groupMessages, err := internalGroupMessageSource(tenantID) + if err != nil { + writeInternalQueryError(wrt, req, err) + return + } + query := fmt.Sprintf(`SELECT (SELECT COUNT(*) FROM im_users WHERE state<>20) AS total_users, (SELECT COUNT(*) FROM im_users WHERE state=0 AND lastseen>=UTC_DATE()) AS active_users, (SELECT COUNT(*) FROM im_group_settings WHERE group_status='active') AS total_groups, - (SELECT COUNT(*) FROM im_topics WHERE state<>20 AND (name LIKE 'p2p%' OR name LIKE 'grp%')) AS total_conversations, - ((SELECT COUNT(*) FROM im_messages WHERE topic LIKE 'p2p%')+(SELECT COUNT(*) FROM im_group_messages)) AS total_messages, - ((SELECT COUNT(*) FROM im_messages WHERE topic LIKE 'p2p%' AND createdat>=UTC_DATE())+(SELECT COUNT(*) FROM im_group_messages WHERE created_at>=UTC_DATE())) AS today_messages`, nil) + (SELECT COUNT(*) FROM im_topics WHERE state<>20 AND (name LIKE 'p2p%%' OR name LIKE 'grp%%')) AS total_conversations, + ((SELECT COUNT(*) FROM im_messages WHERE topic LIKE 'p2p%%')+(SELECT COUNT(*) FROM %s AS gm)) AS total_messages, + ((SELECT COUNT(*) FROM im_messages WHERE topic LIKE 'p2p%%' AND createdat>=UTC_DATE())+(SELECT COUNT(*) FROM %s AS gm WHERE gm.created_at>=UTC_DATE())) AS today_messages`, groupMessages, groupMessages) + row, err := store.InternalAdminQueries.QueryOne(tenantID, query, nil) if err != nil { writeInternalQueryError(wrt, req, err) return @@ -366,14 +429,19 @@ func serveInternalDashboardTrends(wrt http.ResponseWriter, req *http.Request, te if days > 31 { days = 31 } + groupMessages, err := internalGroupMessageSource(tenantID) + if err != nil { + writeInternalQueryError(wrt, req, err) + return + } query := fmt.Sprintf(`WITH RECURSIVE dates AS ( SELECT UTC_DATE()-INTERVAL %d DAY AS day UNION ALL SELECT day+INTERVAL 1 DAY FROM dates WHERE day=d.day AND u.createdat=d.day AND pm.createdat=d.day AND gm.created_at=d.day AND gm.created_at 0, nil +} + +func (r *redisGroupMessageRuntime) DeleteShardRoutes(tenantID types.TenantID, topic string) error { + return r.client.Del(context.Background(), r.shardRouteKey(tenantID, topic), r.shardRouteMetaKey(tenantID, topic)).Err() +} + +func validShardMonth(month string) bool { + if len(month) != 6 { + return false + } + for i := range month { + if month[i] < '0' || month[i] > '9' { + return false + } + } + parsed, err := time.Parse("200601", month) + return err == nil && parsed.Format("200601") == month } func (r *redisGroupMessageRuntime) PeekSeq(tenantID types.TenantID, topic string) (int64, error) { @@ -1182,7 +1319,19 @@ func expandRanges(ranges []types.Range, limit int) []int64 { } func (r *redisGroupMessageRuntime) seqKey(tenantID types.TenantID, topic string) string { - return r.key(tenantID, "topic", topic, "last_seq") + return r.shardKey(tenantID, "topic", topic, "last_seq") +} + +func (r *redisGroupMessageRuntime) activeMonthKey(tenantID types.TenantID) string { + return r.shardKey(tenantID, "groupmsg", "active_month") +} + +func (r *redisGroupMessageRuntime) shardRouteKey(tenantID types.TenantID, topic string) string { + return r.shardKey(tenantID, "groupmsg", "route", topic) +} + +func (r *redisGroupMessageRuntime) shardRouteMetaKey(tenantID types.TenantID, topic string) string { + return r.shardKey(tenantID, "groupmsg", "route_meta", topic) } func (r *redisGroupMessageRuntime) uniqueKey(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) string { @@ -1276,3 +1425,11 @@ func (r *redisGroupMessageRuntime) key(tenantID types.TenantID, parts ...string) all = append(all, parts...) return strings.Join(all, ":") } + +func (r *redisGroupMessageRuntime) shardKey(tenantID types.TenantID, parts ...string) string { + prefix := strings.TrimSuffix(r.prefix, ":") + slot := "{" + strconv.FormatInt(int64(tenantID), 10) + "}" + all := []string{prefix, slot} + all = append(all, parts...) + return strings.Join(all, ":") +} diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index 086a6e8..658db78 100644 --- a/server/store/group_message_runtime_test.go +++ b/server/store/group_message_runtime_test.go @@ -35,12 +35,24 @@ func (r *testGroupMessageRuntime) Claim(tenantID types.TenantID, topic string, f return rec, true, nil } -func (r *testGroupMessageRuntime) NextSeq(_ types.TenantID, topic string, baseline int64) (int64, error) { +func (r *testGroupMessageRuntime) NextSeq(_ types.TenantID, topic string, baseline int64, candidateMonth string) (int64, string, error) { if r.seqs[topic] < baseline { r.seqs[topic] = baseline } r.seqs[topic]++ - return r.seqs[topic], nil + return r.seqs[topic], candidateMonth, nil +} + +func (r *testGroupMessageRuntime) PutShardRoute(types.TenantID, string, GroupMessageShardRoute) error { + return nil +} + +func (r *testGroupMessageRuntime) GetShardRoutes(types.TenantID, string) ([]GroupMessageShardRoute, bool, error) { + return nil, false, nil +} + +func (r *testGroupMessageRuntime) DeleteShardRoutes(types.TenantID, string) error { + return nil } func (r *testGroupMessageRuntime) PeekSeq(_ types.TenantID, topic string) (int64, error) { @@ -234,9 +246,9 @@ func TestMessageRuntimeSeqAndClientMidIdempotency(t *testing.T) { if err != nil || !ready || !claimed || claim == nil || claim.Status != "pending" { t.Fatalf("unexpected first claim: claim=%#v claimed=%v ready=%v err=%v", claim, claimed, ready, err) } - seq, ready, err := ReserveMessageSeq(tenantID, topic, 184) - if err != nil || !ready || seq != 185 { - t.Fatalf("unexpected first seq: seq=%d ready=%v err=%v", seq, ready, err) + seq, month, ready, err := ReserveMessageSeq(tenantID, topic, 184, "202609") + if err != nil || !ready || seq != 185 || month != "202609" { + t.Fatalf("unexpected first seq: seq=%d month=%s ready=%v err=%v", seq, month, ready, err) } if err = MarkMessageUniqueDone(tenantID, topic, from, "mid-1", seq, ""); err != nil { t.Fatalf("mark done: %v", err) @@ -251,9 +263,9 @@ func TestMessageRuntimeSeqAndClientMidIdempotency(t *testing.T) { if err != nil || !ready || !claimed || claim == nil { t.Fatalf("unexpected second claim: claim=%#v claimed=%v ready=%v err=%v", claim, claimed, ready, err) } - seq, ready, err = ReserveMessageSeq(tenantID, topic, 184) - if err != nil || !ready || seq != 186 { - t.Fatalf("unexpected second seq: seq=%d ready=%v err=%v", seq, ready, err) + seq, month, ready, err = ReserveMessageSeq(tenantID, topic, 184, "202609") + if err != nil || !ready || seq != 186 || month != "202609" { + t.Fatalf("unexpected second seq: seq=%d month=%s ready=%v err=%v", seq, month, ready, err) } } @@ -266,7 +278,7 @@ func TestBumpMessageSeqFloorNeverMovesBackward(t *testing.T) { tenantID := types.TenantID(7) topic := types.Uid(1).P2PName(types.Uid(2)) - seq, ready, err := ReserveMessageSeq(tenantID, topic, 0) + seq, _, ready, err := ReserveMessageSeq(tenantID, topic, 0, "") if err != nil || !ready || seq != 1 { t.Fatalf("unexpected first seq: seq=%d ready=%v err=%v", seq, ready, err) } @@ -283,7 +295,7 @@ func TestBumpMessageSeqFloorNeverMovesBackward(t *testing.T) { t.Fatalf("floor must never move backward: current=%d err=%v", current, err) } // The next allocation must continue above the bumped floor. - if seq, _, err := ReserveMessageSeq(tenantID, topic, 0); err != nil || seq != 6 { + if seq, _, _, err := ReserveMessageSeq(tenantID, topic, 0, ""); err != nil || seq != 6 { t.Fatalf("unexpected seq after bump: seq=%d err=%v", seq, err) } } diff --git a/server/store/store.go b/server/store/store.go index ec29e2a..1c6d95d 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -1271,6 +1271,9 @@ func (topicsMapper) Delete(tenantID types.TenantID, topic string, isChan, hard b return err } _ = DeleteCachedObject(tenantID, cacheNamespaceTopic, topic) + if hard && types.GetTopicCat(topic) == types.TopicCatGrp { + _ = DeleteGroupMessageShardRoutes(tenantID, topic) + } return nil } @@ -3034,9 +3037,8 @@ func refreshUserProfileProjections(tenantID types.TenantID, uid types.Uid, profi if item.Friendship.Topic == "" || item.Friendship.PeerID.IsZero() { continue } - if err := Subs.Update(tenantID, item.Friendship.Topic, item.Friendship.PeerID, map[string]any{"Public": public}); err != nil { - logs.Warn.Printf("profile %s: refresh subscription %s/%s public failed: %v", uid, item.Friendship.Topic, item.Friendship.PeerID, err) - } + _ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(item.Friendship.Topic, item.Friendship.PeerID, false)) + _ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(item.Friendship.Topic, item.Friendship.PeerID, true)) } if nextCursor == "" { break @@ -3615,6 +3617,7 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message return nil, err } msg.ClientMsgID = clientMsgID + msg.InitTimes() fromUid := types.ParseUid(msg.From) isGroupMessage := types.GetTopicCat(msg.Topic) == types.TopicCatGrp if isGroupMessage && groupMsgRuntime != nil { @@ -3631,7 +3634,7 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message return nil, types.ErrDuplicate } } - seq, err := nextMessageSeqWithFloor(tenantID, msg.Topic, msg.SeqId-1) + seq, shardMonth, err := nextMessageSeqWithFloor(tenantID, msg.Topic, msg.SeqId-1, msg.CreatedAt.UTC().Format("200601")) if err != nil { if clientMsgID != "" { _ = groupMsgRuntime.MarkFailed(tenantID, msg.Topic, fromUid, clientMsgID) @@ -3639,6 +3642,10 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message return nil, err } msg.SeqId = seq + msg.ShardMonth = shardMonth + } + if isGroupMessage && msg.ShardMonth == "" { + msg.ShardMonth = msg.CreatedAt.UTC().Format("200601") } if clientMsgID != "" { var duplicate *types.Message @@ -3657,6 +3664,7 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message msg.DeletedFor = duplicate.DeletedFor msg.SeqId = duplicate.SeqId msg.Topic = duplicate.Topic + msg.ShardMonth = duplicate.ShardMonth msg.From = duplicate.From msg.ClientMsgID = duplicate.ClientMsgID msg.Head = duplicate.Head @@ -3664,7 +3672,6 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message return &types.MessageSaveResult{SeqId: duplicate.SeqId, Duplicate: true}, nil } } - msg.InitTimes() msg.SetUid(Store.GetUid()) // Reserve topic's or user's SeqId before optional async persistence. err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg) @@ -3703,6 +3710,7 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag } if duplicate != nil { msg.SeqId = duplicate.SeqId + msg.ShardMonth = duplicate.ShardMonth if groupMsgRuntime != nil { _ = groupMsgRuntime.MarkDone(tenantID, msg.Topic, fromUid, clientMsgID, duplicate.SeqId, duplicate.Uid().String()) } @@ -3742,6 +3750,11 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag } } if isGroupMessage && groupMsgRuntime != nil { + if err := groupMsgRuntime.PutShardRoute(tenantID, msg.Topic, GroupMessageShardRoute{ + Month: msg.ShardMonth, MinSeq: msg.SeqId, MaxSeq: msg.SeqId, Count: 1, + }); err != nil { + logs.Warn.Printf("topic[%s]: failed to update group message shard route: %v", msg.Topic, err) + } if err := groupMsgRuntime.CacheHotMessage(tenantID, msg); err != nil { logs.Warn.Printf("topic[%s]: failed to cache hot group message: %v", msg.Topic, err) } @@ -3804,6 +3817,7 @@ func copyPersistedMessage(dst, src *types.Message) { dst.DeletedFor = src.DeletedFor dst.SeqId = src.SeqId dst.Topic = src.Topic + dst.ShardMonth = src.ShardMonth dst.From = src.From dst.ClientMsgID = src.ClientMsgID dst.Head = src.Head @@ -4052,6 +4066,11 @@ func (messagesMapper) DeleteList(tenantID types.TenantID, topic string, delID in if cacheErr := groupMsgRuntime.InvalidateHotMessages(tenantID, topic, ranges); cacheErr != nil { logs.Warn.Printf("topic[%s]: failed to invalidate deleted group message cache: %v", topic, cacheErr) } + if delID == 0 { + if routeErr := groupMsgRuntime.DeleteShardRoutes(tenantID, topic); routeErr != nil { + logs.Warn.Printf("topic[%s]: failed to delete group message shard route: %v", topic, routeErr) + } + } } // TODO: move to adapter. diff --git a/server/store/types/types.go b/server/store/types/types.go index c194faa..c6759e0 100644 --- a/server/store/types/types.go +++ b/server/store/types/types.go @@ -138,6 +138,10 @@ func (uid *Uid) MarshalJSON() ([]byte, error) { // UnmarshalJSON reads Uid from a double quoted string. func (uid *Uid) UnmarshalJSON(b []byte) error { size := len(b) + if size == 2 && b[0] == '"' && b[1] == '"' { + *uid = ZeroUid + return nil + } if size != (uidBase64Unpadded + 2) { return errors.New("Uid.UnmarshalJSON: invalid length") } else if b[0] != '"' || b[size-1] != '"' { @@ -1244,6 +1248,7 @@ type Message struct { DeletedFor []SoftDelete `json:"DeletedFor,omitempty" bson:",omitempty"` SeqId int64 Topic string + ShardMonth string `db:"shard_month" json:"-" bson:"shard_month,omitempty"` // Sender's user ID as string (without 'usr' prefix), could be empty. From string ClientMsgID string `db:"client_msg_id" json:"ClientMsgId,omitempty" bson:"client_msg_id,omitempty"` diff --git a/server/store/types/uid_json_test.go b/server/store/types/uid_json_test.go new file mode 100644 index 0000000..b0e9c7f --- /dev/null +++ b/server/store/types/uid_json_test.go @@ -0,0 +1,23 @@ +package types + +import ( + "encoding/json" + "testing" +) + +func TestZeroUidJSONRoundTrip(t *testing.T) { + type wrapper struct { + UID Uid + } + raw, err := json.Marshal(&wrapper{UID: ZeroUid}) + if err != nil { + t.Fatal(err) + } + var decoded wrapper + if err = json.Unmarshal(raw, &decoded); err != nil { + t.Fatal(err) + } + if decoded.UID != ZeroUid { + t.Fatalf("decoded uid=%v want zero", decoded.UID) + } +} diff --git a/server/topic_test.go b/server/topic_test.go index af7e2d0..d0fab14 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -1115,6 +1115,14 @@ func TestSaveMessageMaybeKafkaPublishesWithoutDBPersist(t *testing.T) { if publishedEnvelope.IdempotencyKey != "client-1" { t.Fatalf("unexpected idempotency key: %q", publishedEnvelope.IdempotencyKey) } + var payload kafkaMessagePersistPayload + if err := json.Unmarshal(publishedEnvelope.Payload, &payload); err != nil { + t.Fatalf("decode kafka message payload: %v", err) + } + wantMonth := msg.CreatedAt.UTC().Format("200601") + if payload.ShardMonth != wantMonth { + t.Fatalf("group message shard month was not frozen: payload=%q want=%q", payload.ShardMonth, wantMonth) + } } func TestShouldTryHotMessageFallback(t *testing.T) {