From f4cedc7d75cfb1b62c823b2471f047b03912ec32 Mon Sep 17 00:00:00 2001 From: Austin Date: Fri, 18 Sep 2026 17:42:09 +0800 Subject: [PATCH 1/3] fix: push realtime del presence to topic sessions on message recall The REST recall response reached only the caller, so peers kept showing the recalled message until a full resync. broadcastIMMessageRecalled now pushes the native {pres what:del delseq} tombstone through the hub for both single and batch recall (empty pres topic for p2p, required by handlePresence's xoriginal check); clients flush the message and render the recalled placeholder. Zentao #16, #2. --- server/im_http.go | 48 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/server/im_http.go b/server/im_http.go index 9bf985e..f3ff8c0 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -3161,6 +3161,50 @@ func serveIMMessageReceipts(wrt http.ResponseWriter, req *http.Request) { }) } +// broadcastIMMessageRecalled pushes the native {pres what:"del" delseq} frame to +// the sessions currently attached to the topic so peers learn about a recall in +// real time (the REST response alone reaches only the caller). The wire record +// carries "low" only (no "hi"): client SDKs then flush the message from their +// local cache and leave a content-less tombstone behind, which the webapp +// renders as the usual "recalled" placeholder. Routed through the hub so the +// p2p topic rewrites the frame's topic name to each recipient's own view. +func broadcastIMMessageRecalled(tenantID types.TenantID, topic string, seqs []int64, actor types.Uid) { + if globals.hub == nil || topic == "" || actor.IsZero() { + return + } + delseq := make([]MsgRange, 0, len(seqs)) + for _, seq := range seqs { + if seq > 0 { + delseq = append(delseq, MsgRange{LowId: seq}) + } + } + if len(delseq) == 0 { + return + } + // P2P topics are loaded under their canonical name with xoriginal cleared, + // so the presence frame must carry an empty topic name to pass the + // handlePresence xoriginal check; any other category uses the topic name. + presTopic := topic + if types.GetTopicCat(topic) == types.TopicCatP2P { + presTopic = "" + } + select { + case globals.hub.routeSrv <- &ServerComMessage{ + TenantID: tenantID, + RcptTo: topic, + Pres: &MsgServerPres{ + Topic: presTopic, + What: "del", + Src: actor.UserId(), + DelSeq: delseq, + FilterIn: int(types.ModeRead), + }, + }: + default: + logs.Warn.Printf("im message recall: hub presence queue full topic=%s count=%d", topic, len(delseq)) + } +} + func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) { if req.Method != http.MethodPost { writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) @@ -3215,6 +3259,7 @@ func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) { return } } + broadcastIMMessageRecalled(principal.TenantID, updated.Topic, []int64{seq}, principal.Uid) writeIMSuccess(wrt, req, messageMutationPayload(updated, "recalled")) } @@ -3242,6 +3287,7 @@ func serveIMMessageBatchRecall(wrt http.ResponseWriter, req *http.Request) { return } items := make([]map[string]any, 0, len(body.Items)) + recalledSeqs := make([]int64, 0, len(body.Items)) recalledAt := time.Now().UTC().Round(time.Millisecond) for _, item := range body.Items { if item.Seq <= 0 { @@ -3281,7 +3327,9 @@ func serveIMMessageBatchRecall(wrt http.ResponseWriter, req *http.Request) { } } items = append(items, result) + recalledSeqs = append(recalledSeqs, item.Seq) } + broadcastIMMessageRecalled(principal.TenantID, topic, recalledSeqs, principal.Uid) writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": items}) } From 753a04ed12ae945682a4fba5298f8feb52811333 Mon Sep 17 00:00:00 2001 From: Austin Date: Fri, 18 Sep 2026 17:42:18 +0800 Subject: [PATCH 2/3] fix: broadcast offline presence as soon as the last session drops Logout left contacts seeing the user online until the me topic idle timeout fired presUsersOfInterest(off), and even that frame was silently dropped for p2p routes by handlePresence's xoriginal check. MarkUserOffline now returns the remaining session count and cleanUp notifies users of interest immediately once it hits zero; p2p presence is delivered straight to the contact's me topic, mirroring presSingleUserOffline. Zentao #22. --- server/pres.go | 32 ++++++++++++++++++++++ server/session.go | 12 +++++++- server/store/group_message_runtime.go | 16 +++++------ server/store/group_message_runtime_test.go | 4 +-- 4 files changed, 53 insertions(+), 11 deletions(-) diff --git a/server/pres.go b/server/pres.go index 13101b9..3f2a9e9 100644 --- a/server/pres.go +++ b/server/pres.go @@ -278,6 +278,7 @@ func (t *Topic) presUsersOfInterest(what, ua string) { // Publish user's update to his/her users of interest on their 'me' topic while user's 'me' topic is offline // Case A: user is being deleted, "gone". +// Case B: user's last session just disconnected, "off". func presUsersOfInterestOffline(tenantID types.TenantID, uid types.Uid, subs []types.Subscription, what string) { // Push update to subscriptions for i := range subs { @@ -286,6 +287,37 @@ func presUsersOfInterestOffline(tenantID types.TenantID, uid types.Uid, subs []t continue } + if types.GetTopicCat(subs[i].Topic) == types.TopicCatP2P { + // A presence request routed to the p2p topic would be silently + // dropped: the topic is loaded under its canonical p2p name while + // the request carries Topic="me" (notifyOnOrSkip), so the + // xoriginal check in handlePresence rejects it. Deliver to the + // contact's 'me' topic directly instead, the same way + // presSingleUserOffline does. + uid1, uid2, err := types.ParseP2P(subs[i].Topic) + if err != nil { + continue + } + contact := uid1 + if contact == uid { + contact = uid2 + } + if contact == uid || contact.IsZero() { + continue + } + routeUserServerMessage(contact, &ServerComMessage{ + TenantID: tenantID, + Pres: &MsgServerPres{ + Topic: notifyOn, + What: what, + Src: subs[i].Topic, + WantReply: false, + }, + RcptTo: contact.UserId(), + }) + continue + } + globals.hub.routeSrv <- &ServerComMessage{ TenantID: tenantID, Pres: &MsgServerPres{ diff --git a/server/session.go b/server/session.go index 25e021b..cc49c02 100644 --- a/server/session.go +++ b/server/session.go @@ -465,9 +465,19 @@ func (s *Session) cleanUp(expired bool) { if err := store.SecurityAudit.CloseLogin(s.tenantID, s.sid, time.Now().UTC().Round(time.Millisecond)); err != nil { logs.Warn.Println("s.cleanUp: failed to close login audit", s.uid, err, s.sid) } - if err := store.MarkUserOffline(s.tenantID, s.uid, s.sid, currentNodeID()); err != nil { + remaining, err := store.MarkUserOffline(s.tenantID, s.uid, s.sid, currentNodeID()) + if err != nil { logs.Warn.Println("s.cleanUp: failed to mark user offline", s.uid, err, s.sid) } + // 最后一个会话已断开（如主动登出、断网）：立即通知关注者该用户离线， + // 不再等待 me topic 闲置超时后的 presUsersOfInterest("off")。 + if err == nil && remaining == 0 && globals.hub != nil { + if uoi, uerr := store.Users.GetSubs(s.tenantID, s.uid); uerr == nil { + presUsersOfInterestOffline(s.tenantID, s.uid, uoi, "off") + } else { + logs.Warn.Println("s.cleanUp: failed to notify users of interest", s.uid, uerr, s.sid) + } + } } s.purgeChannels() s.inflightReqs.Wait() diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index 2803a3d..8450922 100644 --- a/server/store/group_message_runtime.go +++ b/server/store/group_message_runtime.go @@ -109,7 +109,7 @@ type groupMessageRuntime interface { InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error MarkUserOnline(tenantID types.TenantID, uid types.Uid, sid, nodeID, device, userAgent string) error RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error - MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error + MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error) GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error) MarkTopicMemberOnline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error MarkTopicMemberOffline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error @@ -300,9 +300,9 @@ func RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID str return groupMsgRuntime.RefreshUserPresence(tenantID, uid, sid, nodeID) } -func MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error { +func MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error) { if groupMsgRuntime == nil || !tenantID.IsValid() || uid.IsZero() || sid == "" { - return nil + return 0, nil } return groupMsgRuntime.MarkUserOffline(tenantID, uid, sid, nodeID) } @@ -939,7 +939,7 @@ func (r *redisGroupMessageRuntime) RefreshUserPresence(tenantID types.TenantID, return err } -func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error { +func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error) { if nodeID == "" { nodeID = "local" } @@ -956,11 +956,11 @@ func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid pipe.SRem(ctx, sessionsKey, member) pipe.SRem(ctx, nodeKey, uid.UserId()) if _, err := pipe.Exec(ctx); err != nil { - return err + return 0, err } count, err := r.client.SCard(ctx, sessionsKey).Result() if err != nil { - return err + return 0, err } if count > 0 { pipe = r.client.Pipeline() @@ -968,14 +968,14 @@ func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid pipe.Expire(ctx, userKey, ttl) pipe.Expire(ctx, sessionsKey, ttl) _, err = pipe.Exec(ctx) - return err + return int(count), err } pipe = r.client.Pipeline() pipe.HSet(ctx, userKey, "online", "0", "session_count", 0, "node_id", nodeID, "last_seen_at", now.Format(time.RFC3339Nano)) pipe.Expire(ctx, userKey, ttl) pipe.Del(ctx, sessionsKey) _, err = pipe.Exec(ctx) - return err + return 0, err } func (r *redisGroupMessageRuntime) GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error) { diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index 8a78b33..4616b5b 100644 --- a/server/store/group_message_runtime_test.go +++ b/server/store/group_message_runtime_test.go @@ -114,8 +114,8 @@ func (r *testGroupMessageRuntime) RefreshUserPresence(types.TenantID, types.Uid, return nil } -func (r *testGroupMessageRuntime) MarkUserOffline(types.TenantID, types.Uid, string, string) error { - return nil +func (r *testGroupMessageRuntime) MarkUserOffline(types.TenantID, types.Uid, string, string) (int, error) { + return 0, nil } func (r *testGroupMessageRuntime) GetUserPresence(types.TenantID, types.Uid) (*UserPresence, error) { From 0fcbbb9318bfd034fe728c088bb182722263cd11 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Fri, 18 Sep 2026 20:57:50 +0800 Subject: [PATCH 3/3] fix: publish member joined after group invitation --- server/im_http.go | 6 ++++++ server/im_http_test.go | 18 ++++++++++++++++++ 2 files changed, 24 insertions(+) diff --git a/server/im_http.go b/server/im_http.go index f3ff8c0..184682f 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -4628,6 +4628,8 @@ func serveIMGroupInvitations(wrt http.ResponseWriter, req *http.Request, topic s } if joined { item["status"] = types.GroupInvitationStatusJoined + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.member.joined", + map[string]any{"user_id": uid.UserId(), "joined": true, "source": "direct_invitation"}) if imReliableHTTPEventsEnabled(req) { publishIMGroupMemberJoinedSystemNotice(principal.TenantID, topic, principal.Uid, uid, "有成员通过邀请加入了群聊。", @@ -4989,6 +4991,10 @@ func serveIMGroupJoinRequestAction(wrt http.ResponseWriter, req *http.Request, t []imSystemMessageAction{{Type: "group_profile", Label: "查看群聊", TargetID: topic}}, map[string]any{"request_id": updated.RequestID, "source": "approval"}) } + if updated.Status == types.GroupJoinRequestStatusAccepted { + publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.member.joined", + map[string]any{"request_id": updated.RequestID, "user_id": updated.ApplicantUserID.UserId(), "joined": true, "source": "approval"}) + } publishIMGroupRealtimeEvent(principal.TenantID, topic, principal.Uid, "group.join_request."+updated.Status, map[string]any{"request_id": updated.RequestID, "user_id": updated.ApplicantUserID.UserId(), "status": updated.Status}) writeIMSuccess(wrt, req, groupJoinRequestPayload(updated)) diff --git a/server/im_http_test.go b/server/im_http_test.go index 0283f66..9d63484 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -3715,16 +3715,19 @@ func TestServeIMGroupInvitationAddsNonMember(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() expectIMBearer(t, ctrl, "token-1", 7, 42) + oldHub := globals.hub oldSubs := store.Subs oldGroupBusiness := store.GroupBusiness oldSecurity := store.Security subs := mock_store.NewMockSubsPersistenceInterface(ctrl) groupStore := &testGroupBusinessStore{} securityStore := &testUserSecurityStore{} + globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 2)} store.Subs = subs store.GroupBusiness = groupStore store.Security = securityStore defer func() { + globals.hub = oldHub store.Subs = oldSubs store.GroupBusiness = oldGroupBusiness store.Security = oldSecurity @@ -3758,6 +3761,21 @@ func TestServeIMGroupInvitationAddsNonMember(t *testing.T) { if len(securityStore.notifications) != 0 { t.Fatalf("direct invite should not create approval notification: %#v", securityStore.notifications) } + select { + case msg := <-globals.hub.routeSrv: + if msg.Info == nil || msg.Info.Event != "group.member.joined" || msg.Info.Topic != "grpAlpha" { + t.Fatalf("unexpected realtime event: %#v", msg) + } + var payload map[string]any + if err := json.Unmarshal(msg.Info.Payload, &payload); err != nil { + t.Fatalf("decode payload: %v", err) + } + if payload["user_id"] != target.UserId() || payload["member_sync"] != "single" || payload["source"] != "direct_invitation" { + t.Fatalf("unexpected member joined payload: %#v", payload) + } + default: + t.Fatal("missing group.member.joined realtime event") + } } func TestServeIMGroupInvitationCreatesApprovalRequest(t *testing.T) {