diff --git a/docs/IM-前后端对接文档.md b/docs/IM-前后端对接文档.md index 79a884c..2959a07 100644 --- a/docs/IM-前后端对接文档.md +++ b/docs/IM-前后端对接文档.md @@ -455,6 +455,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | 修改好友备注 | 操作人本人 | `{"info":{"topic":"usr_actor","from":"usr_actor","what":"im","event":"friend.remark.updated","payload":{"user_id":"usr_actor"}}}` | 刷新好友列表和会话头显示名 | | 消息撤回 | Topic 在线成员 | `{"info":{"topic":"p2p_or_grp","from":"usr_actor","what":"im","event":"message.recalled","payload":{"topic":"p2p_or_grp","seq":12,"version":2}}}` | 按 `topic+seq+version` 重新拉取或本地标记已撤回 | | 消息编辑 | Topic 在线成员 | `{"info":{"topic":"p2p_or_grp","from":"usr_actor","what":"im","event":"message.edited","payload":{"topic":"p2p_or_grp","seq":12,"version":2}}}` | 按 `topic+seq+version` 重新拉取消息详情或修订记录 | +| 单聊消息变更脏标记 | P2P 双方在线会话 | `{"info":{"topic":"usr_target","from":"usr_actor","what":"im","event":"message.mutation","payload":{"topic":"p2p...","seq":12,"version":2,"event":"message.recalled"}}}` | 当前未订阅该 P2P 也要标记会话脏，并调用 `message-mutations` 做补偿同步 | | 群业务变化 | 群 Topic 在线成员 | `{"info":{"topic":"grp...","from":"usr_actor","what":"group","event":"group.member.joined","payload":{"topic":"grp...","user_id":"usr_xxx"}}}` | 按事件刷新群详情、成员列表、入群申请、公告、禁言或角色权限；除操作人外当前会写系统通知 | | 会议变化 | 会议参会人 | `{"info":{"topic":"usr_target","from":"usr_actor","what":"im","event":"meeting.updated","payload":{"meeting_id":"mtg_1","version":2,"state":"scheduled"}}}` | 刷新会议列表/详情；除操作者外当前会写系统通知 | | 安全风险确认 | 操作人本人 | `{"info":{"topic":"usr_actor","from":"usr_actor","what":"im","event":"security.risk.continue","payload":{"risk_challenge_id":"risk_1","decision":"continue"}}}` | 刷新安全验证页面；当前会写系统通知 | @@ -1590,7 +1591,32 @@ Body： #### 4.21.3 消息变更同步水位 -撤回、编辑属于 `seq` 不变的原地变更。在线且已订阅 Topic 时，客户端会收到 `message.recalled` / `message.edited` 实时事件；离线或未订阅时，需要靠会话级水位补偿。 +撤回、编辑属于 `seq` 不变的原地变更。在线且已订阅 Topic 时，客户端会收到 `message.recalled` / `message.edited` 实时事件；单聊中如果客户端在线但当前没有订阅该 P2P，服务端会额外向双方用户通道下发 `message.mutation` 脏标记事件；离线或错过实时事件时，需要靠会话级水位补偿。 + +单聊未订阅脏标记示例： + +```json +{ + "info": { + "topic": "usr_target", + "from": "usr_actor", + "what": "im", + "event": "message.mutation", + "payload": { + "topic": "p2pJaOWxMvALjB4i1K_h_NZgQ", + "seq": 12, + "version": 2, + "event": "message.recalled" + } + } +} +``` + +收到 `message.mutation` 后不要直接改本地消息内容，先把 `payload.topic` 标记为脏，再调用： + +```http +GET /v1/im/topics/{topic}/message-mutations?since_version={localMutationVersion}&limit=50 +``` 会话列表同步时，`me.meta.sub[].private` 会带： diff --git a/server/im_http.go b/server/im_http.go index ad965f0..97ae142 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -4234,6 +4234,32 @@ func serveIMMessageMutations(wrt http.ResponseWriter, req *http.Request) { }) } +func enqueueIMP2PMessageMutationEvents(tenantID types.TenantID, topic string, actor types.Uid, seq int64, version int, mutationEvent string) error { + if tenantID.IsZero() || topic == "" || actor.IsZero() || seq <= 0 || version <= 0 || mutationEvent == "" { + return types.ErrMalformed + } + uid1, uid2, err := types.ParseP2P(topic) + if err != nil { + return nil + } + payload := map[string]any{ + "topic": topic, + "seq": seq, + "version": version, + "event": mutationEvent, + } + for _, uid := range []types.Uid{uid1, uid2} { + if uid.IsZero() { + continue + } + eventID := stableIMEventID("message-mutation", topic, seq, version, mutationEvent, uid.UserId()) + if err := enqueueIMBusinessEvent(tenantID, eventID, "user", uid.UserId(), actor, "message.mutation", payload); err != nil { + return err + } + } + return nil +} + func serveIMMessageContext(wrt http.ResponseWriter, req *http.Request) { if req.Method != http.MethodGet { writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) @@ -4411,6 +4437,10 @@ func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return } + if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, seq, updated.Version, "message.recalled"); err != nil { + writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) + return + } } writeIMSuccess(wrt, req, messageMutationPayload(updated, "recalled")) } @@ -4473,6 +4503,8 @@ func serveIMMessageBatchRecall(wrt http.ResponseWriter, req *http.Request) { if err = enqueueIMBusinessEvent(principal.TenantID, eventID, "topic", topic, principal.Uid, "message.recalled", map[string]any{"topic": topic, "seq": item.Seq, "version": updated.Version}); err != nil { result["event_status"] = "enqueue_failed" + } else if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, item.Seq, updated.Version, "message.recalled"); err != nil { + result["event_status"] = "enqueue_failed" } } items = append(items, result) @@ -4541,6 +4573,10 @@ func serveIMMessageRevision(wrt http.ResponseWriter, req *http.Request) { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return } + if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, seq, updated.Version, "message.edited"); err != nil { + writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) + return + } } writeIMSuccess(wrt, req, messageMutationPayload(updated, "edited")) } diff --git a/server/im_http_test.go b/server/im_http_test.go index 2ef0703..b261b9a 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -7966,6 +7966,72 @@ func TestServeIMMessageRecallNormalizesP2PUserTopic(t *testing.T) { } } +func TestEnqueueIMP2PMessageMutationEventsDeliversUserDirtyMarker(t *testing.T) { + oldStore := globals.sessionStore + defer func() { globals.sessionStore = oldStore }() + + tenantID := types.TenantID(7) + actor := types.Uid(42) + peer := types.Uid(43) + topic := actor.P2PName(peer) + globals.sessionStore = realtimeTestSessionStore( + realtimeTestSession("actor-session", tenantID, actor, "me"), + realtimeTestSession("peer-session", tenantID, peer, "me"), + ) + + if err := enqueueIMP2PMessageMutationEvents(tenantID, topic, actor, 9, 2, "message.recalled"); err != nil { + t.Fatal(err) + } + + for _, tc := range []struct { + sid string + uid types.Uid + }{ + {"actor-session", actor}, + {"peer-session", peer}, + } { + sess := globals.sessionStore.Get(tc.sid) + select { + case raw := <-sess.send: + msg, ok := raw.(*ServerComMessage) + if !ok || msg.Info == nil { + t.Fatalf("unexpected outbound for %s: %#v", tc.sid, raw) + } + if msg.Info.Topic != tc.uid.UserId() || msg.Info.Event != "message.mutation" || msg.Info.What != "im" { + t.Fatalf("unexpected mutation marker for %s: %#v", tc.sid, msg.Info) + } + var payload map[string]any + if err := json.Unmarshal(msg.Info.Payload, &payload); err != nil { + t.Fatal(err) + } + if payload["topic"] != topic || payload["event"] != "message.recalled" || + payload["seq"].(float64) != 9 || payload["version"].(float64) != 2 { + t.Fatalf("unexpected mutation payload for %s: %#v", tc.sid, payload) + } + default: + t.Fatalf("expected mutation marker for %s", tc.sid) + } + } +} + +func TestEnqueueIMP2PMessageMutationEventsIgnoresGroupTopic(t *testing.T) { + oldStore := globals.sessionStore + defer func() { globals.sessionStore = oldStore }() + + tenantID := types.TenantID(7) + actor := types.Uid(42) + globals.sessionStore = realtimeTestSessionStore( + realtimeTestSession("actor-session", tenantID, actor, "me"), + ) + + if err := enqueueIMP2PMessageMutationEvents(tenantID, "grpRecall", actor, 9, 2, "message.recalled"); err != nil { + t.Fatal(err) + } + if len(globals.sessionStore.Get("actor-session").send) != 0 { + t.Fatal("group topic should not emit per-user mutation markers") + } +} + func TestServeIMMessageRevision(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish()