diff --git a/server/im_http.go b/server/im_http.go index 9bf985e..184682f 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}) } @@ -4580,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, "有成员通过邀请加入了群聊。", @@ -4941,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) { 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) {