Merge remote-tracking branch 'origin/dev' into dev

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))

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) {

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{

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()

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) {

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) {