diff --git a/docs/IM-前后端对接文档.md b/docs/IM-前后端对接文档.md index af666bd..79a884c 100644 --- a/docs/IM-前后端对接文档.md +++ b/docs/IM-前后端对接文档.md @@ -1420,6 +1420,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `PUT /v1/im/topics/{topic}/draft` 待实现 | `{"content":"未发送草稿","version":1}` | `{"topic":"grp...","content":"未发送草稿","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `POST /v1/im/topics/{topic}/messages/{seq}/recall` | 单条撤回；`topic` 可传 `p2p...` 或 `grp...`；Body `{}`，可选 `{"version":1,"reason":"mistake"}` | `{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `POST /v1/im/topics/{topic}/messages/recall` | 多选批量撤回；`{"items":[{"seq":12,"version":1},{"seq":13}],"reason":"mistake"}`；`items` 1～50 条，`version` 可选 | `{"topic":"grp...","items":[{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"},{"seq":13,"status":"failed","http":409,"code":30006,"reason":"VERSION_CONFLICT","message":"message version conflict"}]}` | +| `GET /v1/im/topics/{topic}/message-mutations?since_version=&limit=` | 拉取撤回/编辑等原地消息变更；`since_version` 取本地已应用的会话级 `mutation_version` | `{"topic":"grp...","items":[{"seq":12,"version":2,"event":"message.recalled","operator":"usr...","mutated_at":"2026-08-20T09:00:00Z","mutation_version":3}],"next_cursor":"3","has_more":false,"mutation_version":3,"expired":false}` | | `POST /v1/im/topics/{topic}/messages/{seq}/revisions` | `{"content":{"text":"修改后的文本"},"client_mid":"mid_1","version":1}` | `{"topic":"grp...","seq":12,"status":"edited","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `GET /v1/im/topics/{topic}/messages/{seq}/receipts?cursor=` | Query only | `{"topic":"grp...","seq":12,"items":[{"user_id":"usr...","read_at":"2026-08-20T09:00:00Z","recv_at":"2026-08-20T08:59:00Z"}],"next_cursor":""}` | | `GET /v1/im/topics/{topic}/messages/search?q=&type=text&cursor=` | Query only | `{"topic":"grp...","items":[{"seq":12,"from":"usr...","ts":"2026-08-20T09:00:00Z","snippet":"命中的文本","matches":[{"start":0,"end":2}]}],"next_cursor":""}` | @@ -1429,6 +1430,225 @@ X-MeetingIM-APIKey: <预置的普通 API Key> 消息撤回和编辑成功后，Topic 在线成员会收到 0.5.4.6 定义的 `info.what="im"` 业务事件：`message.recalled` 或 `message.edited`。批量撤回按成功项逐条产生 `message.recalled` 事件；失败项不会产生事件。前端收到后按 `topic + seq + version` 刷新本地消息，不要等待用户重新进入会话才更新。 +#### 4.21.2 群消息撤回与批量撤回 + +群聊撤回使用 Topic 消息操作接口，`topic` 必须传真实群 Topic（`grp...`），不能传用户 ID、群 ID 昵称或本地会话别名。单条撤回和多选批量撤回都要求登录态，Header 至少包含 `Authorization: Bearer `、`X-MeetingIM-APIKey`、`X-IM-Schema: 1`、`X-Request-ID`。 + +单条撤回： + +```http +POST /v1/im/topics/{topic}/messages/{seq}/recall +``` + +Path 参数： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `topic` | String | 是 | 群 Topic，例如 `grp6PIoZgdpc2g`。 | +| `seq` | Int64 | 是 | 要撤回的消息序号。 | + +Body： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `version` | Int | 否 | 期望消息版本。不传时后端使用当前消息版本；传旧版本可能返回版本冲突。 | +| `reason` | String | 否 | 撤回原因，当前主要用于客户端/审计扩展，可传空。 | + +单条请求示例： + +```json +{ + "version": 1, + "reason": "mistake" +} +``` + +单条成功响应 `data`： + +```json +{ + "topic": "grp6PIoZgdpc2g", + "seq": 102, + "status": "recalled", + "version": 2, + "updated_at": "2026-09-15T04:30:00Z", + "recalled_at": "2026-09-15T04:30:00Z", + "recalled_by": "usr_actor" +} +``` + +多选批量撤回： + +```http +POST /v1/im/topics/{topic}/messages/recall +``` + +Path 参数： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `topic` | String | 是 | 群 Topic，例如 `grp6PIoZgdpc2g`。 | + +Body： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `items` | Array | 是 | 要撤回的消息列表，长度 1～50。 | +| `items[].seq` | Int64 | 是 | 要撤回的消息序号。 | +| `items[].version` | Int | 否 | 该消息期望版本。不传时后端使用当前消息版本。 | +| `items[].reason` | String | 否 | 单条撤回原因；可不传。 | +| `reason` | String | 否 | 批量默认撤回原因；单条 `items[].reason` 可覆盖。 | + +批量请求示例： + +```json +{ + "items": [ + { + "seq": 102, + "version": 1 + }, + { + "seq": 103, + "version": 1 + } + ], + "reason": "mistake" +} +``` + +批量响应始终按条目返回处理结果。一条失败不影响其他条目；成功条目会包含最新 `version`，失败条目会包含该条失败原因。 + +```json +{ + "topic": "grp6PIoZgdpc2g", + "items": [ + { + "topic": "grp6PIoZgdpc2g", + "seq": 102, + "status": "recalled", + "version": 2, + "updated_at": "2026-09-15T04:30:00Z", + "recalled_at": "2026-09-15T04:30:00Z", + "recalled_by": "usr_actor" + }, + { + "seq": 103, + "status": "failed", + "http": 409, + "code": 30006, + "reason": "VERSION_CONFLICT", + "message": "message version conflict" + } + ] +} +``` + +常见失败项： + +| `http` | `code` | `reason` | 说明 | +| --- | --- | --- | --- | +| `400` | `30000` | `MALFORMED_REQUEST` | `seq` 非法、`items` 为空或超过 50 条。 | +| `403` | `30005` | `PERMISSION_DENIED` | 当前用户无权撤回该消息，或消息已超过撤回窗口。超过窗口时 `message` 为 `message recall window expired`。 | +| `404` | `30004` | `MALFORMED_REQUEST` | 消息不存在。 | +| `409` | `30006` | `VERSION_CONFLICT` | 请求版本与服务端当前消息版本不一致，或消息已撤回/删除。 | + +权限与时限： + +- 发送人可撤回自己的消息。 +- 群主/管理员可撤回群成员消息。 +- 当前后端撤回窗口为 120 秒；超过后返回 `message recall window expired`。 +- 被撤回消息服务端会清空普通正文，历史拉取应展示撤回占位，不继续展示旧内容。 + +实时事件： + +撤回成功后，Topic 在线成员会收到 `message.recalled`。批量撤回按成功项逐条发送事件，失败项不会发送。 + +```json +{ + "info": { + "topic": "grp6PIoZgdpc2g", + "from": "usr_actor", + "what": "im", + "event": "message.recalled", + "payload": { + "topic": "grp6PIoZgdpc2g", + "seq": 102, + "version": 2 + } + } +} +``` + +客户端处理规则： + +- 多选撤回优先调用批量接口，不要并发打很多单条接口。 +- 批量接口 HTTP `200` 只表示请求被服务端处理，仍需逐条读取 `items[].status`。 +- `status="recalled"` 时，用 `topic + seq + version` 更新本地消息为撤回状态。 +- `status="failed"` 时，只对该条提示失败原因，其他成功项正常生效。 +- 收到实时事件后也按 `topic + seq + version` 刷新本地消息，避免只依赖发起方本地状态。 + +#### 4.21.3 消息变更同步水位 + +撤回、编辑属于 `seq` 不变的原地变更。在线且已订阅 Topic 时，客户端会收到 `message.recalled` / `message.edited` 实时事件；离线或未订阅时，需要靠会话级水位补偿。 + +会话列表同步时，`me.meta.sub[].private` 会带： + +```json +{ + "mutation_version": 3 +} +``` + +客户端按账号 + topic 保存本地已应用水位。若服务端 `mutation_version` 大于本地值，调用： + +```http +GET /v1/im/topics/{topic}/message-mutations?since_version={localMutationVersion}&limit=50 +``` + +响应 `data`： + +```json +{ + "topic": "grp6PIoZgdpc2g", + "items": [ + { + "seq": 102, + "version": 2, + "event": "message.recalled", + "operator": "usr_actor", + "mutated_at": "2026-09-15T04:30:00Z", + "mutation_version": 3 + } + ], + "next_cursor": "3", + "has_more": false, + "mutation_version": 3, + "expired": false +} +``` + +字段说明： + +| 字段 | 说明 | +| --- | --- | +| `seq` | 被变更消息的序号。 | +| `version` | 变更后的消息版本，用于同一 `seq` 的幂等与乱序保护。 | +| `event` | `message.recalled` / `message.edited`，与实时 `info.event` 保持一致。 | +| `operator` | 操作人用户 ID。 | +| `mutated_at` | 变更发生时间。 | +| `mutation_version` | 会话级最新水位。客户端成功应用返回项后推进本地水位。 | +| `next_cursor` | 当前实现等同最新 `mutation_version`，客户端原样保存即可。 | +| `expired` | 当前轻量实现固定为 `false`；后续若服务端裁剪变更记录，该字段会用于提示客户端降级重拉。 | + +客户端处理规则： + +- `mutation_version` 是会话级水位，不是单条消息的 `version`。 +- 本地只做 `服务端 mutation_version > 本地 mutation_version` 比较，不要假设连续递增。 +- 拉到 `message.recalled` 后，将本地 `topic + seq` 标记为撤回，并只在 `version` 更高时覆盖。 +- 拉到 `message.edited` 后，按 `seq + version` 拉取或覆盖消息内容。 +- 实时事件和补偿接口可能重复到达，客户端必须按 `topic + seq + version` 幂等处理。 + ## 5. 通讯录、好友关系与黑名单 ### 5.1 通讯录.png diff --git a/docs/sql/IM-消息变更同步水位SQL.sql b/docs/sql/IM-消息变更同步水位SQL.sql new file mode 100644 index 0000000..153cbd4 --- /dev/null +++ b/docs/sql/IM-消息变更同步水位SQL.sql @@ -0,0 +1,5 @@ +-- IM 消息变更同步水位 +-- 用于撤回/编辑等 seq 不变的原地消息变更离线补偿。 + +ALTER TABLE im_topics + ADD COLUMN mutation_version BIGINT NOT NULL DEFAULT 0 AFTER subcnt; diff --git a/docs/对接文档最新/IM-前后端对接文档.md b/docs/对接文档最新/IM-前后端对接文档.md index 80abb3c..30886d3 100644 --- a/docs/对接文档最新/IM-前后端对接文档.md +++ b/docs/对接文档最新/IM-前后端对接文档.md @@ -1379,6 +1379,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `PUT /v1/im/topics/{topic}/draft` 待实现 | `{"content":"未发送草稿","version":1}` | `{"topic":"grp...","content":"未发送草稿","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `POST /v1/im/topics/{topic}/messages/{seq}/recall` | 单条撤回；`topic` 可传 `p2p...` 或 `grp...`；Body `{}`，可选 `{"version":1,"reason":"mistake"}` | `{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `POST /v1/im/topics/{topic}/messages/recall` | 多选批量撤回；`{"items":[{"seq":12,"version":1},{"seq":13}],"reason":"mistake"}`；`items` 1～50 条，`version` 可选 | `{"topic":"grp...","items":[{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"},{"seq":13,"status":"failed","http":409,"code":30006,"reason":"VERSION_CONFLICT","message":"message version conflict"}]}` | +| `GET /v1/im/topics/{topic}/message-mutations?since_version=&limit=` | 拉取撤回/编辑等原地消息变更；`since_version` 取本地已应用的会话级 `mutation_version` | `{"topic":"grp...","items":[{"seq":12,"version":2,"event":"message.recalled","operator":"usr...","mutated_at":"2026-08-20T09:00:00Z","mutation_version":3}],"next_cursor":"3","has_more":false,"mutation_version":3,"expired":false}` | | `POST /v1/im/topics/{topic}/messages/{seq}/revisions` | `{"content":{"text":"修改后的文本"},"client_mid":"mid_1","version":1}` | `{"topic":"grp...","seq":12,"status":"edited","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `GET /v1/im/topics/{topic}/messages/{seq}/receipts?cursor=` | Query only | `{"topic":"grp...","seq":12,"items":[{"user_id":"usr...","read_at":"2026-08-20T09:00:00Z","recv_at":"2026-08-20T08:59:00Z"}],"next_cursor":""}` | | `GET /v1/im/topics/{topic}/messages/search?q=&type=text&cursor=` | Query only | `{"topic":"grp...","items":[{"seq":12,"from":"usr...","ts":"2026-08-20T09:00:00Z","snippet":"命中的文本","matches":[{"start":0,"end":2}]}],"next_cursor":""}` | @@ -1388,6 +1389,225 @@ X-MeetingIM-APIKey: <预置的普通 API Key> 消息撤回和编辑成功后，Topic 在线成员会收到 0.5.4.6 定义的 `info.what="im"` 业务事件：`message.recalled` 或 `message.edited`。批量撤回按成功项逐条产生 `message.recalled` 事件；失败项不会产生事件。前端收到后按 `topic + seq + version` 刷新本地消息，不要等待用户重新进入会话才更新。 +#### 4.21.2 群消息撤回与批量撤回 + +群聊撤回使用 Topic 消息操作接口，`topic` 必须传真实群 Topic（`grp...`），不能传用户 ID、群 ID 昵称或本地会话别名。单条撤回和多选批量撤回都要求登录态，Header 至少包含 `Authorization: Bearer `、`X-MeetingIM-APIKey`、`X-IM-Schema: 1`、`X-Request-ID`。 + +单条撤回： + +```http +POST /v1/im/topics/{topic}/messages/{seq}/recall +``` + +Path 参数： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `topic` | String | 是 | 群 Topic，例如 `grp6PIoZgdpc2g`。 | +| `seq` | Int64 | 是 | 要撤回的消息序号。 | + +Body： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `version` | Int | 否 | 期望消息版本。不传时后端使用当前消息版本；传旧版本可能返回版本冲突。 | +| `reason` | String | 否 | 撤回原因，当前主要用于客户端/审计扩展，可传空。 | + +单条请求示例： + +```json +{ + "version": 1, + "reason": "mistake" +} +``` + +单条成功响应 `data`： + +```json +{ + "topic": "grp6PIoZgdpc2g", + "seq": 102, + "status": "recalled", + "version": 2, + "updated_at": "2026-09-15T04:30:00Z", + "recalled_at": "2026-09-15T04:30:00Z", + "recalled_by": "usr_actor" +} +``` + +多选批量撤回： + +```http +POST /v1/im/topics/{topic}/messages/recall +``` + +Path 参数： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `topic` | String | 是 | 群 Topic，例如 `grp6PIoZgdpc2g`。 | + +Body： + +| 字段 | 类型 | 必填 | 说明 | +| --- | --- | --- | --- | +| `items` | Array | 是 | 要撤回的消息列表，长度 1～50。 | +| `items[].seq` | Int64 | 是 | 要撤回的消息序号。 | +| `items[].version` | Int | 否 | 该消息期望版本。不传时后端使用当前消息版本。 | +| `items[].reason` | String | 否 | 单条撤回原因；可不传。 | +| `reason` | String | 否 | 批量默认撤回原因；单条 `items[].reason` 可覆盖。 | + +批量请求示例： + +```json +{ + "items": [ + { + "seq": 102, + "version": 1 + }, + { + "seq": 103, + "version": 1 + } + ], + "reason": "mistake" +} +``` + +批量响应始终按条目返回处理结果。一条失败不影响其他条目；成功条目会包含最新 `version`，失败条目会包含该条失败原因。 + +```json +{ + "topic": "grp6PIoZgdpc2g", + "items": [ + { + "topic": "grp6PIoZgdpc2g", + "seq": 102, + "status": "recalled", + "version": 2, + "updated_at": "2026-09-15T04:30:00Z", + "recalled_at": "2026-09-15T04:30:00Z", + "recalled_by": "usr_actor" + }, + { + "seq": 103, + "status": "failed", + "http": 409, + "code": 30006, + "reason": "VERSION_CONFLICT", + "message": "message version conflict" + } + ] +} +``` + +常见失败项： + +| `http` | `code` | `reason` | 说明 | +| --- | --- | --- | --- | +| `400` | `30000` | `MALFORMED_REQUEST` | `seq` 非法、`items` 为空或超过 50 条。 | +| `403` | `30005` | `PERMISSION_DENIED` | 当前用户无权撤回该消息，或消息已超过撤回窗口。超过窗口时 `message` 为 `message recall window expired`。 | +| `404` | `30004` | `MALFORMED_REQUEST` | 消息不存在。 | +| `409` | `30006` | `VERSION_CONFLICT` | 请求版本与服务端当前消息版本不一致，或消息已撤回/删除。 | + +权限与时限： + +- 发送人可撤回自己的消息。 +- 群主/管理员可撤回群成员消息。 +- 当前后端撤回窗口为 120 秒；超过后返回 `message recall window expired`。 +- 被撤回消息服务端会清空普通正文，历史拉取应展示撤回占位，不继续展示旧内容。 + +实时事件： + +撤回成功后，Topic 在线成员会收到 `message.recalled`。批量撤回按成功项逐条发送事件，失败项不会发送。 + +```json +{ + "info": { + "topic": "grp6PIoZgdpc2g", + "from": "usr_actor", + "what": "im", + "event": "message.recalled", + "payload": { + "topic": "grp6PIoZgdpc2g", + "seq": 102, + "version": 2 + } + } +} +``` + +客户端处理规则： + +- 多选撤回优先调用批量接口，不要并发打很多单条接口。 +- 批量接口 HTTP `200` 只表示请求被服务端处理，仍需逐条读取 `items[].status`。 +- `status="recalled"` 时，用 `topic + seq + version` 更新本地消息为撤回状态。 +- `status="failed"` 时，只对该条提示失败原因，其他成功项正常生效。 +- 收到实时事件后也按 `topic + seq + version` 刷新本地消息，避免只依赖发起方本地状态。 + +#### 4.21.3 消息变更同步水位 + +撤回、编辑属于 `seq` 不变的原地变更。在线且已订阅 Topic 时，客户端会收到 `message.recalled` / `message.edited` 实时事件；离线或未订阅时，需要靠会话级水位补偿。 + +会话列表同步时，`me.meta.sub[].private` 会带： + +```json +{ + "mutation_version": 3 +} +``` + +客户端按账号 + topic 保存本地已应用水位。若服务端 `mutation_version` 大于本地值，调用： + +```http +GET /v1/im/topics/{topic}/message-mutations?since_version={localMutationVersion}&limit=50 +``` + +响应 `data`： + +```json +{ + "topic": "grp6PIoZgdpc2g", + "items": [ + { + "seq": 102, + "version": 2, + "event": "message.recalled", + "operator": "usr_actor", + "mutated_at": "2026-09-15T04:30:00Z", + "mutation_version": 3 + } + ], + "next_cursor": "3", + "has_more": false, + "mutation_version": 3, + "expired": false +} +``` + +字段说明： + +| 字段 | 说明 | +| --- | --- | +| `seq` | 被变更消息的序号。 | +| `version` | 变更后的消息版本，用于同一 `seq` 的幂等与乱序保护。 | +| `event` | `message.recalled` / `message.edited`，与实时 `info.event` 保持一致。 | +| `operator` | 操作人用户 ID。 | +| `mutated_at` | 变更发生时间。 | +| `mutation_version` | 会话级最新水位。客户端成功应用返回项后推进本地水位。 | +| `next_cursor` | 当前实现等同最新 `mutation_version`，客户端原样保存即可。 | +| `expired` | 当前轻量实现固定为 `false`；后续若服务端裁剪变更记录，该字段会用于提示客户端降级重拉。 | + +客户端处理规则： + +- `mutation_version` 是会话级水位，不是单条消息的 `version`。 +- 本地只做 `服务端 mutation_version > 本地 mutation_version` 比较，不要假设连续递增。 +- 拉到 `message.recalled` 后，将本地 `topic + seq` 标记为撤回，并只在 `version` 更高时覆盖。 +- 拉到 `message.edited` 后，按 `seq + version` 拉取或覆盖消息内容。 +- 实时事件和补偿接口可能重复到达，客户端必须按 `topic + seq + version` 幂等处理。 + ## 5. 通讯录、好友关系与黑名单 ### 5.1 通讯录.png diff --git a/scripts/IM-租户业务库初始化.sql b/scripts/IM-租户业务库初始化.sql index 6f40d2b..db90c5f 100644 --- a/scripts/IM-租户业务库初始化.sql +++ b/scripts/IM-租户业务库初始化.sql @@ -1317,6 +1317,7 @@ CREATE TABLE IF NOT EXISTS `im_topics` ( `seqid` bigint NOT NULL DEFAULT '0', `delid` int DEFAULT '0', `subcnt` int DEFAULT '0', + `mutation_version` bigint NOT NULL DEFAULT '0', `public` json DEFAULT NULL, `trusted` json DEFAULT NULL, `tags` json DEFAULT NULL, @@ -1617,4 +1618,3 @@ VALUES (2,'system', 'system merchant', 'system default merchant', 'meeting_im', INSERT INTO `hotgo`.`im_tenant_config` (`tenant_id`, `max_users`, `max_groups`, `max_group_members`, `created_by`, `updated_by`) VALUES (2, 0, 0, 0, 0, 0) ON DUPLICATE KEY UPDATE `updated_at` = CURRENT_TIMESTAMP(3); - diff --git a/server/db/adapter.go b/server/db/adapter.go index 53b96f1..c025f62 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -332,6 +332,7 @@ type TenantBusinessAdapter interface { TenantTopicUpdateOnMessage(t.TenantID, string, *t.Message) error TenantTopicUpdateSubCnt(t.TenantID, string) error TenantTopicUpdate(t.TenantID, string, map[string]any) error + TenantTopicBumpMutationVersion(t.TenantID, string, time.Time) (int64, error) TenantTopicOwnerChange(t.TenantID, string, t.Uid) error TenantSubscriptionGet(t.TenantID, string, t.Uid, bool) (*t.Subscription, error) @@ -351,6 +352,7 @@ type TenantBusinessAdapter interface { TenantMessageDeleteList(t.TenantID, string, *t.DelMessage) error TenantMessageGetDeleted(t.TenantID, string, t.Uid, *t.QueryOpt) ([]t.DelMessage, error) TenantMessageSearch(t.TenantID, string, t.Uid, string, string, string, int) ([]t.Message, string, error) + TenantMessageMutations(t.TenantID, string, t.Uid, int64, int) ([]t.MessageMutation, string, bool, int64, error) TenantGroupMessageSave(t.TenantID, *t.Message) error TenantGroupMessageGetByClientID(t.TenantID, string, t.Uid, string) (*t.Message, error) TenantGroupMessageGetOne(t.TenantID, string, int64) (*t.Message, error) @@ -387,6 +389,7 @@ type TenantBusinessAdapter interface { TenantGroupAnnouncementPut(t.TenantID, *t.GroupAnnouncement, int) (*t.GroupAnnouncement, error) TenantGroupAnnouncementRead(t.TenantID, string, string, t.Uid, time.Time) error TenantGroupMuteList(t.TenantID, string, string, int) ([]t.GroupMute, string, error) + TenantGroupMuteGet(t.TenantID, string, t.Uid) (*t.GroupMute, error) TenantGroupMuteUpsert(t.TenantID, *t.GroupMute) (*t.GroupMute, error) TenantGroupMuteDelete(t.TenantID, string, t.Uid, t.Uid, time.Time) error TenantGroupFileList(t.TenantID, string, string, string, string, int) ([]t.GroupFile, string, error) diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 40476a9..de908d8 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -2747,7 +2747,7 @@ func (a *adapter) TopicGet(topic string) (*t.Topic, error) { // Fetch topic by name var tt = new(t.Topic) if err := a.db.GetContext(ctx, tt, - "SELECT createdat,updatedat,state,stateat,touchedat,name AS id,usebt,access,owner,seqid,delid,subcnt,public,trusted,tags,aux "+ + "SELECT createdat,updatedat,state,stateat,touchedat,name AS id,usebt,access,owner,seqid,delid,subcnt,IFNULL(mutation_version,0) AS mutation_version,public,trusted,tags,aux "+ "FROM im_topics WHERE name=?", topic); err != nil { if err == sql.ErrNoRows { // Nothing found - clear the error @@ -2901,7 +2901,7 @@ func (a *adapter) TopicsForUser(uid t.Uid, keepDeleted bool, opts *t.QueryOpt) ( // Fetch grp topics and join to subscriptions. if len(topq) > 0 { - q = "SELECT updatedat,state,touchedat,name AS id,usebt,access,seqid,delid,subcnt,public,trusted " + + q = "SELECT updatedat,state,touchedat,name AS id,usebt,access,seqid,delid,subcnt,IFNULL(mutation_version,0) AS mutation_version,public,trusted " + "FROM im_topics WHERE name IN (?)" q, args, _ = sqlx.In(q, topq) @@ -2943,6 +2943,7 @@ func (a *adapter) TopicsForUser(uid t.Uid, keepDeleted bool, opts *t.QueryOpt) ( sub.SetState(top.State) sub.SetTouchedAt(top.TouchedAt) sub.SetSeqId(top.SeqId) + sub.Private = ensurePrivateMutationVersion(sub.Private, top.MutationVersion) if t.GetTopicCat(sub.Topic) == t.TopicCatGrp { sub.SetSubCnt(top.SubCnt) sub.SetPublic(common.FromJSON(top.Public)) diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index be58b87..42f927a 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -172,6 +172,15 @@ func ensureP2PPrivatePeerRemark(private any, peerID, remark string) any { return data } +func ensurePrivateMutationVersion(private any, version int64) any { + data, _ := private.(map[string]any) + if data == nil { + data = map[string]any{} + } + data["mutation_version"] = version + return data +} + func clearProfilePublicFields(public map[string]any) { for _, key := range []string{"display_name", "nickname", "fn", "name", "avatar_file_id", "photo", "avatar", "signature", "gender", "region", "public_account"} { delete(public, key) @@ -309,6 +318,17 @@ func messageHeadBool(head t.KVMap, key string) bool { } } +func (a *adapter) tenantTopicBumpMutationVersionTx(ctx context.Context, tx *sqlx.Tx, id t.TenantID, topic string, at time.Time) error { + if a.tenantScopedDB { + _, err := tx.ExecContext(ctx, "UPDATE im_topics SET mutation_version=mutation_version+1,updatedat=?,touchedat=? WHERE name=?", + at, at, topic) + return err + } + _, err := tx.ExecContext(ctx, "UPDATE im_topics SET mutation_version=mutation_version+1,updatedat=?,touchedat=? WHERE tenant_id=? AND name=?", + at, at, id, topic) + return err +} + func stampTenantFile(id t.TenantID, file *t.FileDef) *t.FileDef { if file != nil { file.TenantID = id @@ -888,7 +908,7 @@ func (a *adapter) TenantTopicGet(id t.TenantID, topic string) (*t.Topic, error) } var top t.Topic err = a.db.GetContext(ctx, &top, - "SELECT tenant_id,createdat,updatedat,state,stateat,touchedat,name AS id,usebt,access,owner,seqid,delid,subcnt,public,trusted,tags,aux "+ + "SELECT tenant_id,createdat,updatedat,state,stateat,touchedat,name AS id,usebt,access,owner,seqid,delid,subcnt,IFNULL(mutation_version,0) AS mutation_version,public,trusted,tags,aux "+ "FROM im_topics WHERE tenant_id=? AND name=?", id, topic) if err == sql.ErrNoRows { return nil, nil @@ -1013,7 +1033,7 @@ func (a *adapter) TenantTopicsForUser(id t.TenantID, uid t.Uid, keepDeleted bool return nil, err } if len(topq) > 0 { - q = "SELECT tenant_id,updatedat,state,touchedat,name AS id,usebt,access,seqid,delid,subcnt,public,trusted FROM im_topics WHERE tenant_id=? AND name IN (?)" + q = "SELECT tenant_id,updatedat,state,touchedat,name AS id,usebt,access,seqid,delid,subcnt,IFNULL(mutation_version,0) AS mutation_version,public,trusted FROM im_topics WHERE tenant_id=? AND name IN (?)" q, args, _ = sqlx.In(q, id, topq) if !keepDeleted { q += " AND state!=?" @@ -1042,6 +1062,7 @@ func (a *adapter) TenantTopicsForUser(id t.TenantID, uid t.Uid, keepDeleted bool sub.SetState(top.State) sub.SetTouchedAt(top.TouchedAt) sub.SetSeqId(top.SeqId) + sub.Private = ensurePrivateMutationVersion(sub.Private, top.MutationVersion) if t.GetTopicCat(sub.Topic) == t.TopicCatGrp { sub.SetSubCnt(top.SubCnt) sub.SetPublic(common.FromJSON(top.Public)) @@ -1575,6 +1596,49 @@ func (a *adapter) TenantTopicUpdate(id t.TenantID, topic string, update map[stri return tx.Commit() } +func (a *adapter) TenantTopicBumpMutationVersion(id t.TenantID, topic string, at time.Time) (int64, error) { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil || topic == "" { + return 0, t.ErrMalformed + } + if tenantAdp != a { + return tenantAdp.TenantTopicBumpMutationVersion(id, topic, at) + } + if at.IsZero() { + at = time.Now().UTC().Round(time.Millisecond) + } + ctx, cancel := a.getContextForTx() + if cancel != nil { + defer cancel() + } + tx, err := a.db.BeginTxx(ctx, nil) + if err != nil { + return 0, err + } + defer func() { + if err != nil { + tx.Rollback() + } + }() + if err = a.tenantTopicBumpMutationVersionTx(ctx, tx, id, topic, at); err != nil { + return 0, err + } + var version int64 + query := "SELECT mutation_version FROM im_topics WHERE name=?" + args := []any{topic} + if !a.tenantScopedDB { + query = "SELECT mutation_version FROM im_topics WHERE tenant_id=? AND name=?" + args = []any{id, topic} + } + if err = tx.GetContext(ctx, &version, query, args...); err != nil { + return 0, err + } + if err = tx.Commit(); err != nil { + return 0, err + } + return version, nil +} + func (a *adapter) TenantTopicOwnerChange(id t.TenantID, topic string, owner t.Uid) error { tenantAdp, err := a.routedTenantAdapter(id) if err != nil || topic == "" { @@ -2173,6 +2237,9 @@ func (a *adapter) TenantMessageRecall(id t.TenantID, topic string, seq int64, ac if err != nil { return nil, err } + if err = a.tenantTopicBumpMutationVersionTx(ctx, tx, id, topic, recalledAt); err != nil { + return nil, err + } if err = tx.Commit(); err != nil { return nil, err } @@ -2219,6 +2286,9 @@ func (a *adapter) TenantGroupMessageRecall(id t.TenantID, topic string, seq int6 if err != nil { return nil, err } + if err = a.tenantTopicBumpMutationVersionTx(ctx, tx, id, topic, recalledAt); err != nil { + return nil, err + } if err = tx.Commit(); err != nil { return nil, err } @@ -2272,6 +2342,9 @@ func (a *adapter) TenantMessageEdit(id t.TenantID, topic string, seq int64, acto if err != nil { return nil, err } + if err = a.tenantTopicBumpMutationVersionTx(ctx, tx, id, topic, editedAt); err != nil { + return nil, err + } if err = tx.Commit(); err != nil { return nil, err } @@ -2324,6 +2397,9 @@ func (a *adapter) TenantGroupMessageEdit(id t.TenantID, topic string, seq int64, if err != nil { return nil, err } + if err = a.tenantTopicBumpMutationVersionTx(ctx, tx, id, topic, editedAt); err != nil { + return nil, err + } if err = tx.Commit(); err != nil { return nil, err } @@ -2601,6 +2677,129 @@ func (a *adapter) tenantGroupMessageSearch(id t.TenantID, topic string, uid t.Ui return msgs, nextCursor, nil } +func (a *adapter) TenantMessageMutations(id t.TenantID, topic string, uid t.Uid, sinceVersion int64, limit int) ([]t.MessageMutation, string, bool, int64, error) { + if !a.tenantScopedDB { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil { + return nil, "", false, 0, err + } + if tenantAdp != a { + return tenantAdp.TenantMessageMutations(id, topic, uid, sinceVersion, limit) + } + } + if err := validTenant(id); err != nil || topic == "" || uid.IsZero() || sinceVersion < 0 { + return nil, "", false, 0, t.ErrMalformed + } + if limit <= 0 || limit > 100 { + limit = 50 + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + var currentVersion int64 + if a.tenantScopedDB { + if err := a.db.GetContext(ctx, ¤tVersion, "SELECT IFNULL(mutation_version,0) FROM im_topics WHERE name=?", topic); err != nil { + if err == sql.ErrNoRows { + return nil, "", false, 0, t.ErrNotFound + } + return nil, "", false, 0, err + } + } else { + if err := a.db.GetContext(ctx, ¤tVersion, "SELECT IFNULL(mutation_version,0) FROM im_topics WHERE tenant_id=? AND name=?", id, topic); err != nil { + if err == sql.ErrNoRows { + return nil, "", false, 0, t.ErrNotFound + } + return nil, "", false, 0, err + } + } + if sinceVersion >= currentVersion { + return []t.MessageMutation{}, strconv.FormatInt(currentVersion, 10), false, currentVersion, nil + } + if t.GetTopicCat(topic) == t.TopicCatGrp { + return a.tenantGroupMessageMutations(ctx, id, topic, uid, limit, currentVersion) + } + return a.tenantP2PMessageMutations(ctx, id, topic, uid, limit, currentVersion) +} + +func (a *adapter) tenantP2PMessageMutations(ctx context.Context, id t.TenantID, topic string, uid t.Uid, limit int, currentVersion int64) ([]t.MessageMutation, string, bool, int64, error) { + rows, err := a.db.QueryxContext(ctx, + "SELECT m.seqid,m.version,m.updatedat AS mutated_at,m.recalled_at,m.recalled_by,rev.actor "+ + "FROM im_messages 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=? "+ + "LEFT JOIN im_message_revisions AS rev ON rev.topic=m.topic AND rev.seqid=m.seqid AND rev.version=m.version-1 "+ + "WHERE m.topic=? AND d.deletedfor IS NULL AND (m.version>1 OR m.recalled_at IS NOT NULL OR m.deletedat IS NOT NULL) "+ + "ORDER BY m.updatedat DESC,m.seqid DESC LIMIT ?", + decodeUID(uid), topic, limit+1) + if err != nil { + return nil, "", false, 0, err + } + defer rows.Close() + return scanTenantMessageMutations(rows, limit, currentVersion) +} + +func (a *adapter) tenantGroupMessageMutations(ctx context.Context, id t.TenantID, topic string, uid t.Uid, limit int, currentVersion int64) ([]t.MessageMutation, string, bool, int64, error) { + rows, err := a.db.QueryxContext(ctx, + "SELECT m.seqid,m.version,m.updated_at AS mutated_at,m.recalled_at,m.recalled_by,rev.actor "+ + "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=? "+ + "LEFT JOIN im_message_revisions AS rev ON rev.topic=m.topic AND rev.seqid=m.seqid AND rev.version=m.version-1 "+ + "WHERE m.topic=? AND d.deletedfor IS NULL AND (m.version>1 OR m.recalled_at IS NOT NULL OR m.deleted_at IS NOT NULL OR m.status='recalled') "+ + "ORDER BY m.updated_at DESC,m.seqid DESC LIMIT ?", + decodeUID(uid), topic, limit+1) + if err != nil { + return nil, "", false, 0, err + } + defer rows.Close() + return scanTenantMessageMutations(rows, limit, currentVersion) +} + +func scanTenantMessageMutations(rows *sqlx.Rows, limit int, currentVersion int64) ([]t.MessageMutation, string, bool, int64, error) { + items := make([]t.MessageMutation, 0, limit) + hasMore := false + for rows.Next() { + row := struct { + SeqId int64 `db:"seqid"` + Version int `db:"version"` + MutatedAt time.Time `db:"mutated_at"` + RecalledAt sql.NullTime `db:"recalled_at"` + RecalledBy sql.NullInt64 `db:"recalled_by"` + Actor sql.NullInt64 `db:"actor"` + }{} + if err := rows.StructScan(&row); err != nil { + return nil, "", false, 0, err + } + if len(items) == limit { + hasMore = true + continue + } + event := "message.edited" + operator := "" + if row.Actor.Valid { + operator = store.EncodeUid(row.Actor.Int64).String() + } + if row.RecalledAt.Valid { + event = "message.recalled" + if row.RecalledBy.Valid { + operator = store.EncodeUid(row.RecalledBy.Int64).String() + } + row.MutatedAt = row.RecalledAt.Time + } + items = append(items, t.MessageMutation{ + SeqId: row.SeqId, + Version: row.Version, + Event: event, + Operator: operator, + MutatedAt: row.MutatedAt, + MutationVersion: currentVersion, + }) + } + if err := rows.Err(); err != nil { + return nil, "", false, 0, err + } + return items, strconv.FormatInt(currentVersion, 10), hasMore, currentVersion, nil +} + func (a *adapter) TenantCallCreate(id t.TenantID, record *t.CallRecord) error { if !a.tenantScopedDB { tenantAdp, err := a.routedTenantAdapter(id) diff --git a/server/im_http.go b/server/im_http.go index 3b7e6b4..21598de 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -4074,6 +4074,8 @@ func serveIMTopicMessageAction(wrt http.ResponseWriter, req *http.Request) { serveIMTopicDraft(wrt, req) case strings.HasSuffix(req.URL.Path, "/calls"): serveIMTopicCalls(wrt, req) + case strings.HasSuffix(req.URL.Path, "/message-mutations"): + serveIMMessageMutations(wrt, req) case strings.HasSuffix(req.URL.Path, "/messages/search"): serveIMMessageSearch(wrt, req) case strings.Contains(req.URL.Path, "/messages/") && strings.HasSuffix(req.URL.Path, "/context"): @@ -4198,6 +4200,40 @@ func serveIMMessageSearch(wrt http.ResponseWriter, req *http.Request) { writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": items, "next_cursor": nextCursor}) } +func serveIMMessageMutations(wrt http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodGet { + writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) + return + } + principal, topic, ok := authenticateTopicReader(wrt, req, topicFromMessageMutationsPath(req.URL.Path)) + if !ok { + return + } + sinceVersion, err := parseIMMutationSince(req.URL.Query().Get("since_version"), req.URL.Query().Get("since")) + if err != nil { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "since_version"}) + return + } + limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 100) + mutations, nextCursor, hasMore, mutationVersion, err := store.Messages.Mutations(principal.TenantID, topic, principal.Uid, sinceVersion, limit) + if err != nil { + writeIMRelationError(wrt, req, err) + return + } + items := make([]map[string]any, 0, len(mutations)) + for _, mutation := range mutations { + items = append(items, messageMutationEventPayload(mutation)) + } + writeIMSuccess(wrt, req, map[string]any{ + "topic": topic, + "items": items, + "next_cursor": nextCursor, + "has_more": hasMore, + "mutation_version": mutationVersion, + "expired": false, + }) +} + func serveIMMessageContext(wrt http.ResponseWriter, req *http.Request) { if req.Method != http.MethodGet { writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) @@ -6324,19 +6360,34 @@ func serveIMGroupMutes(wrt http.ResponseWriter, req *http.Request, topic, tail s if !ok { return } - if !(selfSub.ModeGiven & selfSub.ModeWant).IsAdmin() { - writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil) - return - } + isAdmin := (selfSub.ModeGiven & selfSub.ModeWant).IsAdmin() switch { case tail == "mutes" && req.Method == http.MethodGet: - items, nextCursor, err := store.GroupBusiness.ListMutes(principal.TenantID, topic, strings.TrimSpace(req.URL.Query().Get("cursor")), parseIMLimit(req.URL.Query().Get("limit"), 50, 100)) - if err != nil && !errors.Is(err, types.ErrNotFound) { - writeIMRelationError(wrt, req, err) - return + var items []types.GroupMute + nextCursor := "" + if isAdmin { + var err error + items, nextCursor, err = store.GroupBusiness.ListMutes(principal.TenantID, topic, strings.TrimSpace(req.URL.Query().Get("cursor")), parseIMLimit(req.URL.Query().Get("limit"), 50, 100)) + if err != nil && !errors.Is(err, types.ErrNotFound) { + writeIMRelationError(wrt, req, err) + return + } + } else { + item, err := store.GroupBusiness.GetMute(principal.TenantID, topic, principal.Uid) + if err != nil && !errors.Is(err, types.ErrNotFound) { + writeIMRelationError(wrt, req, err) + return + } + if item != nil { + items = append(items, *item) + } } writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": groupMutePayloads(items), "next_cursor": nextCursor}) case (tail == "mutes" || tail == "mutes/batch") && req.Method == http.MethodPost: + if !isAdmin { + writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil) + return + } var body imGroupMuteBatchRequest if !decodeIMJSONBody(wrt, req, &body) { return @@ -6377,6 +6428,10 @@ func serveIMGroupMutes(wrt http.ResponseWriter, req *http.Request, topic, tail s map[string]any{"topic": topic, "items": groupMutePayloads(mutes)}) writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": groupMutePayloads(mutes)}) case strings.HasPrefix(tail, "mutes/") && req.Method == http.MethodDelete: + if !isAdmin { + writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil) + return + } uid := types.ParseUserId(strings.TrimSpace(strings.TrimPrefix(tail, "mutes/"))) if uid.IsZero() { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "user_id"}) @@ -8689,6 +8744,17 @@ func messageSearchPayload(msg types.Message, query string) map[string]any { } } +func messageMutationEventPayload(mutation types.MessageMutation) map[string]any { + return map[string]any{ + "seq": mutation.SeqId, + "version": mutation.Version, + "event": mutation.Event, + "operator": mutation.Operator, + "mutated_at": mutation.MutatedAt.UTC().Format(time.RFC3339Nano), + "mutation_version": mutation.MutationVersion, + } +} + func messageContextPayload(msg types.Message) map[string]any { return map[string]any{ "seq": msg.SeqId, @@ -11701,6 +11767,12 @@ func topicFromMessagesSearchPath(path string) string { return topicFromPathWithSuffix(path, prefix, suffix) } +func topicFromMessageMutationsPath(path string) string { + const prefix = "/v1/im/topics/" + const suffix = "/message-mutations" + return topicFromPathWithSuffix(path, prefix, suffix) +} + func topicFromMediaPath(path string) string { const prefix = "/v1/im/topics/" const suffix = "/media" @@ -11728,6 +11800,21 @@ func topicFromPathWithSuffix(path, prefix, suffix string) string { return strings.Trim(rest, "/") } +func parseIMMutationSince(primary, fallback string) (int64, error) { + value := strings.TrimSpace(primary) + if value == "" { + value = strings.TrimSpace(fallback) + } + if value == "" { + return 0, nil + } + parsed, err := strconv.ParseInt(value, 10, 64) + if err != nil || parsed < 0 { + return 0, strconv.ErrSyntax + } + return parsed, nil +} + func messageContextPath(path string) (string, int64, bool) { const prefix = "/v1/im/topics/" idx := strings.Index(path, prefix) diff --git a/server/im_http_test.go b/server/im_http_test.go index fdaed49..fd51cdc 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -548,6 +548,15 @@ func (s *testGroupBusinessStore) ListMutes(_ types.TenantID, _ string, _ string, return s.mutes, "", nil } +func (s *testGroupBusinessStore) GetMute(_ types.TenantID, _ string, uid types.Uid) (*types.GroupMute, error) { + for i := range s.mutes { + if s.mutes[i].UserID == uid && s.mutes[i].Status == types.GroupMuteStatusActive { + return &s.mutes[i], nil + } + } + return nil, types.ErrNotFound +} + func (s *testGroupBusinessStore) UpsertMute(_ types.TenantID, mute *types.GroupMute) (*types.GroupMute, error) { mute.Version = 1 s.mutes = append(s.mutes, *mute) @@ -5625,6 +5634,56 @@ func TestServeIMGroupAdminsMutesAndOwnerTransfer(t *testing.T) { } } +func TestServeIMGroupMutesMemberGetsOwnMuteOnly(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + oldStore := store.Store + oldSubs := store.Subs + oldGroupBusiness := store.GroupBusiness + mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) + tokenAuth := mock_auth.NewMockAuthHandler(ctrl) + mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth).Times(2) + tokenAuth.EXPECT().IsInitialized().Return(true).Times(2) + tokenAuth.EXPECT().Authenticate(auth.AuthContext{}, []byte("token-1")). + Return(&auth.Rec{TenantID: 7, Uid: 43, AuthLevel: auth.LevelAuth}, nil, nil).Times(2) + store.Store = mockStore + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + store.Subs = subs + memberSub := &types.Subscription{TenantID: 7, User: types.Uid(43).String(), Topic: "grpAlpha", ModeWant: types.ModeCPublic, ModeGiven: types.ModeCPublic} + subs.EXPECT().Get(types.TenantID(7), "grpAlpha", types.Uid(43), false).Return(memberSub, nil).Times(2) + groupStore := &testGroupBusinessStore{mutes: []types.GroupMute{ + {Topic: "grpAlpha", UserID: types.Uid(42), MutedBy: types.Uid(99), MutedAt: time.Now().UTC(), Status: types.GroupMuteStatusActive, Version: 1}, + {Topic: "grpAlpha", UserID: types.Uid(43), MutedBy: types.Uid(99), MutedAt: time.Now().UTC(), Reason: "test", Status: types.GroupMuteStatusActive, Version: 2}, + }} + store.GroupBusiness = groupStore + defer func() { + store.Store = oldStore + store.Subs = oldSubs + store.GroupBusiness = oldGroupBusiness + }() + + getReq := httptest.NewRequest(http.MethodGet, "/v1/im/groups/grpAlpha/mutes", nil) + getReq.Header.Set("Authorization", "Bearer token-1") + getRec := httptest.NewRecorder() + serveIMGroupResource(getRec, getReq) + if getRec.Code != http.StatusOK { + t.Fatalf("member mute get status: got %d body=%s", getRec.Code, getRec.Body.String()) + } + data := decodeIMHTTPResponse(t, getRec).Data.(map[string]any) + items := data["items"].([]any) + if len(items) != 1 || items[0].(map[string]any)["user_id"] != types.Uid(43).UserId() { + t.Fatalf("member should only see own mute item: %#v", items) + } + + postReq := httptest.NewRequest(http.MethodPost, "/v1/im/groups/grpAlpha/mutes/batch", bytes.NewBufferString(`{"user_ids":["`+types.Uid(42).UserId()+`"]}`)) + postReq.Header.Set("Authorization", "Bearer token-1") + postRec := httptest.NewRecorder() + serveIMGroupResource(postRec, postReq) + if postRec.Code != http.StatusForbidden { + t.Fatalf("member mute post status: got %d body=%s", postRec.Code, postRec.Body.String()) + } +} + func TestServeIMGroupDissolveConsumesActionToken(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -7647,6 +7706,62 @@ func TestServeIMMessageRecall(t *testing.T) { } } +func TestServeIMMessageMutations(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + actor := types.Uid(42) + expectIMBearer(t, ctrl, "token-1", 7, actor) + oldSubs := store.Subs + oldMessages := store.Messages + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) + store.Subs = subs + store.Messages = messages + defer func() { + store.Subs = oldSubs + store.Messages = oldMessages + }() + + subs.EXPECT().Get(types.TenantID(7), "grpRecall", actor, false).Return(&types.Subscription{ + TenantID: 7, + User: actor.String(), + Topic: "grpRecall", + ModeWant: types.ModeCPublic, + ModeGiven: types.ModeCPublic, + }, nil) + mutatedAt := time.Date(2026, 9, 15, 10, 0, 0, 0, time.UTC) + messages.EXPECT().Mutations(types.TenantID(7), "grpRecall", actor, int64(1), 20).Return([]types.MessageMutation{{ + SeqId: 9, + Version: 2, + Event: "message.recalled", + Operator: actor.UserId(), + MutatedAt: mutatedAt, + MutationVersion: 3, + }}, "3", false, int64(3), nil) + + req := httptest.NewRequest(http.MethodGet, "/v1/im/topics/grpRecall/message-mutations?since_version=1&limit=20", nil) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + + serveIMTopicMessageAction(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) + if data["mutation_version"].(float64) != 3 || data["next_cursor"] != "3" { + t.Fatalf("unexpected mutation response: %#v", data) + } + items := data["items"].([]any) + if len(items) != 1 { + t.Fatalf("unexpected mutation items: %#v", items) + } + item := items[0].(map[string]any) + if item["seq"].(float64) != 9 || item["version"].(float64) != 2 || item["event"] != "message.recalled" { + t.Fatalf("unexpected mutation item: %#v", item) + } +} + func TestServeIMMessageBatchRecallPartialSuccess(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() diff --git a/server/store/mock_store/mock_store.go b/server/store/mock_store/mock_store.go index a33e656..cf30df8 100644 --- a/server/store/mock_store/mock_store.go +++ b/server/store/mock_store/mock_store.go @@ -1008,6 +1008,21 @@ func (mr *MockTopicsPersistenceInterfaceMockRecorder) OwnerChange(tenantID, topi return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OwnerChange", reflect.TypeOf((*MockTopicsPersistenceInterface)(nil).OwnerChange), tenantID, topic, newOwner) } +// BumpMutationVersion mocks base method. +func (m *MockTopicsPersistenceInterface) BumpMutationVersion(tenantID types.TenantID, topic string, at time.Time) (int64, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "BumpMutationVersion", tenantID, topic, at) + ret0, _ := ret[0].(int64) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// BumpMutationVersion indicates an expected call of BumpMutationVersion. +func (mr *MockTopicsPersistenceInterfaceMockRecorder) BumpMutationVersion(tenantID, topic, at interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BumpMutationVersion", reflect.TypeOf((*MockTopicsPersistenceInterface)(nil).BumpMutationVersion), tenantID, topic, at) +} + // Update mocks base method. func (m *MockTopicsPersistenceInterface) Update(tenantID types.TenantID, topic string, update map[string]any) error { m.ctrl.T.Helper() @@ -1220,6 +1235,24 @@ func (mr *MockMessagesPersistenceInterfaceMockRecorder) Search(tenantID, topic, return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Search", reflect.TypeOf((*MockMessagesPersistenceInterface)(nil).Search), tenantID, topic, forUser, query, messageType, cursor, limit) } +// Mutations mocks base method. +func (m *MockMessagesPersistenceInterface) Mutations(tenantID types.TenantID, topic string, forUser types.Uid, sinceVersion int64, limit int) ([]types.MessageMutation, string, bool, int64, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Mutations", tenantID, topic, forUser, sinceVersion, limit) + ret0, _ := ret[0].([]types.MessageMutation) + ret1, _ := ret[1].(string) + ret2, _ := ret[2].(bool) + ret3, _ := ret[3].(int64) + ret4, _ := ret[4].(error) + return ret0, ret1, ret2, ret3, ret4 +} + +// Mutations indicates an expected call of Mutations. +func (mr *MockMessagesPersistenceInterfaceMockRecorder) Mutations(tenantID, topic, forUser, sinceVersion, limit interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Mutations", reflect.TypeOf((*MockMessagesPersistenceInterface)(nil).Mutations), tenantID, topic, forUser, sinceVersion, limit) +} + // GetOne mocks base method. func (m *MockMessagesPersistenceInterface) GetOne(tenantID types.TenantID, topic string, seq int64) (*types.Message, error) { m.ctrl.T.Helper() diff --git a/server/store/store.go b/server/store/store.go index 92b5390..ec29e2a 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -1079,6 +1079,7 @@ type TopicsPersistenceInterface interface { GetSubs(tenantID types.TenantID, topic string, opts *types.QueryOpt) ([]types.Subscription, error) GetSubsAny(tenantID types.TenantID, topic string, opts *types.QueryOpt) ([]types.Subscription, error) Update(tenantID types.TenantID, topic string, update map[string]any) error + BumpMutationVersion(tenantID types.TenantID, topic string, at time.Time) (int64, error) UpdateSubCnt(tenantID types.TenantID, topic string) error OwnerChange(tenantID types.TenantID, topic string, newOwner types.Uid) error Delete(tenantID types.TenantID, topic string, isChan, hard bool) error @@ -1235,6 +1236,18 @@ func (topicsMapper) Update(tenantID types.TenantID, topic string, update map[str return nil } +func (topicsMapper) BumpMutationVersion(tenantID types.TenantID, topic string, at time.Time) (int64, error) { + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return 0, err + } + version, err := tenantAdp.TenantTopicBumpMutationVersion(tenantID, topic, at) + if err == nil { + _ = DeleteCachedObject(tenantID, cacheNamespaceTopic, topic) + } + return version, err +} + // OwnerChange replaces the old topic owner with the new owner. func (topicsMapper) OwnerChange(tenantID types.TenantID, topic string, newOwner types.Uid) error { tenantAdp, err := tenantBusinessAdapter(tenantID) @@ -1386,6 +1399,7 @@ type MessagesPersistenceInterface interface { GetAll(tenantID types.TenantID, topic string, forUser types.Uid, opt *types.QueryOpt) ([]types.Message, error) GetDeleted(tenantID types.TenantID, topic string, forUser types.Uid, opt *types.QueryOpt) ([]types.Range, int, error) Search(tenantID types.TenantID, topic string, forUser types.Uid, query, messageType, cursor string, limit int) ([]types.Message, string, error) + Mutations(tenantID types.TenantID, topic string, forUser types.Uid, sinceVersion int64, limit int) ([]types.MessageMutation, string, bool, int64, error) } // messagesMapper is a concrete type implementing MessagesPersistenceInterface. @@ -1635,6 +1649,7 @@ type GroupBusinessPersistenceInterface interface { PutAnnouncement(types.TenantID, *types.GroupAnnouncement, int) (*types.GroupAnnouncement, error) ReadAnnouncement(types.TenantID, string, string, types.Uid, time.Time) error ListMutes(types.TenantID, string, string, int) ([]types.GroupMute, string, error) + GetMute(types.TenantID, string, types.Uid) (*types.GroupMute, error) UpsertMute(types.TenantID, *types.GroupMute) (*types.GroupMute, error) DeleteMute(types.TenantID, string, types.Uid, types.Uid, time.Time) error ListFiles(types.TenantID, string, string, string, string, int) ([]types.GroupFile, string, error) @@ -2007,6 +2022,17 @@ func (groupBusinessMapper) ListMutes(tenantID types.TenantID, topic, cursor stri return tenantAdp.TenantGroupMuteList(tenantID, strings.TrimSpace(topic), strings.TrimSpace(cursor), limit) } +func (groupBusinessMapper) GetMute(tenantID types.TenantID, topic string, uid types.Uid) (*types.GroupMute, error) { + if strings.TrimSpace(topic) == "" || uid.IsZero() { + return nil, types.ErrMalformed + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + return tenantAdp.TenantGroupMuteGet(tenantID, strings.TrimSpace(topic), uid) +} + func (groupBusinessMapper) UpsertMute(tenantID types.TenantID, mute *types.GroupMute) (*types.GroupMute, error) { if mute == nil || mute.Topic == "" || mute.UserID.IsZero() || mute.MutedBy.IsZero() { return nil, types.ErrMalformed @@ -4080,6 +4106,20 @@ func (messagesMapper) Search(tenantID types.TenantID, topic string, forUser type return tenantAdp.TenantMessageSearch(tenantID, topic, forUser, query, messageType, cursor, limit) } +func (messagesMapper) Mutations(tenantID types.TenantID, topic string, forUser types.Uid, sinceVersion int64, limit int) ([]types.MessageMutation, string, bool, int64, error) { + if topic == "" || forUser.IsZero() || sinceVersion < 0 { + return nil, "", false, 0, types.ErrMalformed + } + if limit <= 0 || limit > 100 { + limit = 50 + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, "", false, 0, err + } + return tenantAdp.TenantMessageMutations(tenantID, topic, forUser, sinceVersion, limit) +} + // GetDeleted returns the ranges of deleted messages and the largest DelId reported in the list. func (messagesMapper) GetDeleted(tenantID types.TenantID, topic string, forUser types.Uid, opt *types.QueryOpt) ([]types.Range, int, error) { tenantAdp, err := tenantBusinessAdapter(tenantID) diff --git a/server/store/types/types.go b/server/store/types/types.go index b5c3409..c194faa 100644 --- a/server/store/types/types.go +++ b/server/store/types/types.go @@ -1151,6 +1151,9 @@ type Topic struct { // Count of topic subscribers. SubCnt int + // Conversation-level version for in-place message mutations such as recall and edit. + MutationVersion int64 `db:"mutation_version" json:"MutationVersion,omitempty" bson:"mutation_version,omitempty"` + Public any Trusted any @@ -1264,6 +1267,16 @@ type MessageRevision struct { Reason string } +// MessageMutation describes an in-place message change used for offline compensation. +type MessageMutation struct { + SeqId int64 + Version int + Event string + Operator string + MutatedAt time.Time + MutationVersion int64 +} + // MessageSaveResult describes the outcome of a message persistence attempt. type MessageSaveResult struct { SeqId int64