Merge remote-tracking branch 'origin/list' into seq38

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible // indirect
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/openimsdk/protocol v0.0.69-alpha.15
+	github.com/openimsdk/protocol v0.0.69-alpha.17
 	github.com/openimsdk/tools v0.0.49-alpha.25
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
@@ -177,6 +177,4 @@ require (
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )
 
-//replace (
+//replace github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol
-//	github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol
-//)

go.sum

@@ -270,8 +270,8 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
 github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
 github.com/openimsdk/gomake v0.0.13 h1:xLDe/moqgWpRoptHzI4packAWzs4C16b+sVY+txNJp0=
 github.com/openimsdk/gomake v0.0.13/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.69-alpha.15 h1:iiVROC2nXTc9qPOSfE7NZckQAVCZiDxBm9yY36ULdyw=
+github.com/openimsdk/protocol v0.0.69-alpha.17 h1:pEag4ZdlovE+AyLsw1VYFU/3sk6ayvGdPzgufQfKf9M=
-github.com/openimsdk/protocol v0.0.69-alpha.15/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
+github.com/openimsdk/protocol v0.0.69-alpha.17/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
 github.com/openimsdk/tools v0.0.49-alpha.25 h1:OpRPwDZ2xWX7Zj5kyfZhryu/NfZTrsRVr2GFwu1HQHI=
 github.com/openimsdk/tools v0.0.49-alpha.25/go.mod h1:rwsFI1G/nBHNfiNapbven41akRDPBbH4df0Cgy6xueU=
 github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=

internal/rpc/friend/sync.go

@@ -11,6 +11,22 @@ import (
 	"github.com/openimsdk/protocol/relation"
 )
 
+func (s *friendServer) NotificationUserInfoUpdate(ctx context.Context, req *relation.NotificationUserInfoUpdateReq) (*relation.NotificationUserInfoUpdateResp, error) {
+	userIDs, err := s.db.FindFriendUserIDs(ctx, req.UserID)
+	if err != nil {
+		return nil, err
+	}
+	for _, userID := range userIDs {
+		if err := s.db.OwnerIncrVersion(ctx, userID, []string{req.UserID}, model.VersionStateUpdate); err != nil {
+			return nil, err
+		}
+	}
+	for _, userID := range userIDs {
+		s.notificationSender.FriendInfoUpdatedNotification(ctx, req.UserID, userID)
+	}
+	return &relation.NotificationUserInfoUpdateResp{}, nil
+}
+
 func (s *friendServer) GetFullFriendUserIDs(ctx context.Context, req *relation.GetFullFriendUserIDsReq) (*relation.GetFullFriendUserIDsResp, error) {
 	vl, err := s.db.FindMaxFriendVersionCache(ctx, req.UserID)
 	if err != nil {

internal/rpc/group/group.go

@@ -132,13 +132,17 @@ func (s *groupServer) NotificationUserInfoUpdate(ctx context.Context, req *pbgro
 		}
 		groupIDs = append(groupIDs, member.GroupID)
 	}
+	for _, groupID := range groupIDs {
+		if err := s.db.MemberGroupIncrVersion(ctx, groupID, []string{req.UserID}, model.VersionStateUpdate); err != nil {
+			return nil, err
+		}
+	}
 	for _, groupID := range groupIDs {
 		s.notification.GroupMemberInfoSetNotification(ctx, groupID, req.UserID)
 	}
 	if err = s.db.DeleteGroupMemberHash(ctx, groupIDs); err != nil {
 		return nil, err
 	}
-
 	return &pbgroup.NotificationUserInfoUpdateResp{}, nil
 }
 

internal/rpc/group/notification.go

@@ -477,11 +477,16 @@ func (g *GroupNotificationSender) GroupOwnerTransferredNotification(ctx context.
 	}
 	opUserID := mcontext.GetOpUserID(ctx)
 	var member map[string]*sdkws.GroupMemberFullInfo
-	member, err = g.getGroupMemberMap(ctx, req.GroupID, []string{opUserID, req.NewOwnerUserID})
+	member, err = g.getGroupMemberMap(ctx, req.GroupID, []string{opUserID, req.NewOwnerUserID, req.OldOwnerUserID})
 	if err != nil {
 		return
 	}
-	tips := &sdkws.GroupOwnerTransferredTips{Group: group, OpUser: member[opUserID], NewGroupOwner: member[req.NewOwnerUserID]}
+	tips := &sdkws.GroupOwnerTransferredTips{
+		Group:             group,
+		OpUser:            member[opUserID],
+		NewGroupOwner:     member[req.NewOwnerUserID],
+		OldGroupOwnerInfo: member[req.OldOwnerUserID],
+	}
 	if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
 		return
 	}

internal/rpc/user/user.go

@@ -16,10 +16,7 @@ package user
 
 import (
 	"context"
-	"math/rand"
+	"errors"
-	"strings"
-	"time"
-
 	"github.com/openimsdk/open-im-server/v3/internal/rpc/friend"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
@@ -27,7 +24,13 @@ import (
 	tablerelation "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache"
+	"github.com/openimsdk/protocol/group"
+	friendpb "github.com/openimsdk/protocol/relation"
 	"github.com/openimsdk/tools/db/redisutil"
+	"math/rand"
+	"strings"
+	"sync"
+	"time"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
@@ -121,6 +124,7 @@ func (s *userServer) GetDesignateUsers(ctx context.Context, req *pbuser.GetDesig
 // deprecated:
 
 //UpdateUserInfo
+
 func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInfoReq) (resp *pbuser.UpdateUserInfoResp, err error) {
 	resp = &pbuser.UpdateUserInfoResp{}
 	err = authverify.CheckAccessV3(ctx, req.UserInfo.UserID, s.config.Share.IMAdminUserID)
@@ -131,31 +135,33 @@ func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserI
 	if err := s.webhookBeforeUpdateUserInfo(ctx, &s.config.WebhooksConfig.BeforeUpdateUserInfo, req); err != nil {
 		return nil, err
 	}
-
 	data := convert.UserPb2DBMap(req.UserInfo)
+	oldUser, err := s.db.GetUserByID(ctx, req.UserInfo.UserID)
+	if err != nil {
+		return nil, err
+	}
 	if err := s.db.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil {
 		return nil, err
 	}
 	s.friendNotificationSender.UserInfoUpdatedNotification(ctx, req.UserInfo.UserID)
-	friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID)
+	//friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID)
-	if err != nil {
+	//if err != nil {
-		return nil, err
+	//	return nil, err
-	}
+	//}
-	if req.UserInfo.Nickname != "" || req.UserInfo.FaceURL != "" {
+	//if req.UserInfo.Nickname != "" || req.UserInfo.FaceURL != "" {
-		if err = s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
+	//	if err = s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID,oldUser); err != nil {
-			return nil, err
+	//		return nil, err
-		}
+	//	}
-	}
+	//}
-	for _, friendID := range friends {
+	//for _, friendID := range friends {
-		s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID)
+	//	s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID)
-	}
+	//}
 	s.webhookAfterUpdateUserInfo(ctx, &s.config.WebhooksConfig.AfterUpdateUserInfo, req)
-	if err = s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
+	if err = s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID, oldUser); err != nil {
 		return nil, err
 	}
 	return resp, nil
 }
-
 func (s *userServer) UpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUserInfoExReq) (resp *pbuser.UpdateUserInfoExResp, err error) {
 	resp = &pbuser.UpdateUserInfoExResp{}
 	err = authverify.CheckAccessV3(ctx, req.UserInfo.UserID, s.config.Share.IMAdminUserID)
@@ -165,30 +171,33 @@ func (s *userServer) UpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUse
 	if err = s.webhookBeforeUpdateUserInfoEx(ctx, &s.config.WebhooksConfig.BeforeUpdateUserInfoEx, req); err != nil {
 		return nil, err
 	}
+	oldUser, err := s.db.GetUserByID(ctx, req.UserInfo.UserID)
+	if err != nil {
+		return nil, err
+	}
 	data := convert.UserPb2DBMapEx(req.UserInfo)
 	if err = s.db.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil {
 		return nil, err
 	}
 	s.friendNotificationSender.UserInfoUpdatedNotification(ctx, req.UserInfo.UserID)
-	friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID)
+	//friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID)
-	if err != nil {
+	//if err != nil {
-		return nil, err
+	//	return nil, err
-	}
+	//}
-	if req.UserInfo.Nickname != nil || req.UserInfo.FaceURL != nil {
+	//if req.UserInfo.Nickname != nil || req.UserInfo.FaceURL != nil {
-		if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
+	//	if err := s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
-			return nil, err
+	//		return nil, err
-		}
+	//	}
-	}
+	//}
-	for _, friendID := range friends {
+	//for _, friendID := range friends {
-		s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID)
+	//	s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID)
-	}
+	//}
 	s.webhookAfterUpdateUserInfoEx(ctx, &s.config.WebhooksConfig.AfterUpdateUserInfoEx, req)
-	if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
+	if err := s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID, oldUser); err != nil {
 		return nil, err
 	}
 	return resp, nil
 }
-
 func (s *userServer) SetGlobalRecvMessageOpt(ctx context.Context, req *pbuser.SetGlobalRecvMessageOptReq) (resp *pbuser.SetGlobalRecvMessageOptResp, err error) {
 	resp = &pbuser.SetGlobalRecvMessageOptResp{}
 	if _, err := s.db.FindWithError(ctx, []string{req.UserID}); err != nil {
@@ -247,6 +256,7 @@ func (s *userServer) GetPaginationUsers(ctx context.Context, req *pbuser.GetPagi
 		return &pbuser.GetPaginationUsersResp{Total: int32(total), Users: convert.UsersDB2Pb(users)}, err
 
 	}
+
 }
 
 func (s *userServer) UserRegister(ctx context.Context, req *pbuser.UserRegisterReq) (resp *pbuser.UserRegisterResp, err error) {
@@ -343,8 +353,7 @@ func (s *userServer) SubscribeOrCancelUsersStatus(ctx context.Context, req *pbus
 
 // GetUserStatus Get the online status of the user.
 func (s *userServer) GetUserStatus(ctx context.Context, req *pbuser.GetUserStatusReq) (resp *pbuser.GetUserStatusResp,
-	err error,
+	err error) {
-) {
 	onlineStatusList, err := s.db.GetUserStatus(ctx, req.UserIDs)
 	if err != nil {
 		return nil, err
@@ -354,8 +363,7 @@ func (s *userServer) GetUserStatus(ctx context.Context, req *pbuser.GetUserStatu
 
 // SetUserStatus Synchronize user's online status.
 func (s *userServer) SetUserStatus(ctx context.Context, req *pbuser.SetUserStatusReq) (resp *pbuser.SetUserStatusResp,
-	err error,
+	err error) {
-) {
 	err = s.db.SetUserStatus(ctx, req.UserID, req.Status, req.PlatformID)
 	if err != nil {
 		return nil, err
@@ -379,8 +387,7 @@ func (s *userServer) SetUserStatus(ctx context.Context, req *pbuser.SetUserStatu
 
 // GetSubscribeUsersStatus Get the online status of subscribers.
 func (s *userServer) GetSubscribeUsersStatus(ctx context.Context,
-	req *pbuser.GetSubscribeUsersStatusReq,
+	req *pbuser.GetSubscribeUsersStatusReq) (*pbuser.GetSubscribeUsersStatusResp, error) {
-) (*pbuser.GetSubscribeUsersStatusResp, error) {
 	userList, err := s.db.GetAllSubscribeList(ctx, req.UserID)
 	if err != nil {
 		return nil, err
@@ -469,6 +476,7 @@ func (s *userServer) ProcessUserCommandUpdate(ctx context.Context, req *pbuser.P
 }
 
 func (s *userServer) ProcessUserCommandGet(ctx context.Context, req *pbuser.ProcessUserCommandGetReq) (*pbuser.ProcessUserCommandGetResp, error) {
+
 	err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID)
 	if err != nil {
 		return nil, err
@@ -687,6 +695,40 @@ func (s *userServer) userModelToResp(users []*tablerelation.User, pagination pag
 	return &pbuser.SearchNotificationAccountResp{Total: total, NotificationAccounts: notificationAccounts}
 }
 
+func (s *userServer) NotificationUserInfoUpdate(ctx context.Context, userID string, oldUser *tablerelation.User) error {
+	user, err := s.db.GetUserByID(ctx, userID)
+	if err != nil {
+		return err
+	}
+	if user.Nickname == oldUser.Nickname && user.FaceURL == oldUser.FaceURL {
+		return nil
+	}
+	oldUserInfo := convert.UserDB2Pb(oldUser)
+	newUserInfo := convert.UserDB2Pb(user)
+	var wg sync.WaitGroup
+	var es [2]error
+	wg.Add(len(es))
+	go func() {
+		defer wg.Done()
+		_, es[0] = s.groupRpcClient.Client.NotificationUserInfoUpdate(ctx, &group.NotificationUserInfoUpdateReq{
+			UserID:      userID,
+			OldUserInfo: oldUserInfo,
+			NewUserInfo: newUserInfo,
+		})
+	}()
+
+	go func() {
+		defer wg.Done()
+		_, es[1] = s.friendRpcClient.Client.NotificationUserInfoUpdate(ctx, &friendpb.NotificationUserInfoUpdateReq{
+			UserID:      userID,
+			OldUserInfo: oldUserInfo,
+			NewUserInfo: newUserInfo,
+		})
+	}()
+	wg.Wait()
+	return errors.Join(es[:]...)
+}
+
 func (s *userServer) SortQuery(ctx context.Context, req *pbuser.SortQueryReq) (*pbuser.SortQueryResp, error) {
 	users, err := s.db.SortQuery(ctx, req.UserIDName, req.Asc)
 	if err != nil {

pkg/common/storage/cache/redis/group.go

@@ -446,7 +446,7 @@ func (g *GroupCacheRedis) DelMaxJoinGroupVersion(userIDs ...string) cache.GroupC
 
 func (g *GroupCacheRedis) FindMaxGroupMemberVersion(ctx context.Context, groupID string) (*model.VersionLog, error) {
 	return getCache(ctx, g.rcClient, g.getGroupMemberMaxVersionKey(groupID), g.expireTime, func(ctx context.Context) (*model.VersionLog, error) {
-		return g.groupMemberDB.FindJoinIncrVersion(ctx, groupID, 0, 0)
+		return g.groupMemberDB.FindMemberIncrVersion(ctx, groupID, 0, 0)
 	})
 }
 

pkg/common/storage/controller/friend.go

@@ -86,7 +86,7 @@ type FriendDatabase interface {
 
 	FindFriendUserID(ctx context.Context, friendUserID string) ([]string, error)
 
-	//SearchFriend(ctx context.Context, ownerUserID, keyword string, pagination pagination.Pagination) (int64, []*model.Friend, error)
+	OwnerIncrVersion(ctx context.Context, ownerUserID string, friendUserIDs []string, state int32) error
 }
 
 type friendDatabase struct {
@@ -379,3 +379,10 @@ func (f *friendDatabase) FindFriendUserID(ctx context.Context, friendUserID stri
 //func (f *friendDatabase) SearchFriend(ctx context.Context, ownerUserID, keyword string, pagination pagination.Pagination) (int64, []*model.Friend, error) {
 //	return f.friend.SearchFriend(ctx, ownerUserID, keyword, pagination)
 //}
+
+func (f *friendDatabase) OwnerIncrVersion(ctx context.Context, ownerUserID string, friendUserIDs []string, state int32) error {
+	if err := f.friend.IncrVersion(ctx, ownerUserID, friendUserIDs, state); err != nil {
+		return err
+	}
+	return f.cache.DelMaxFriendVersion(ownerUserID).ChainExecDel(ctx)
+}

pkg/common/storage/controller/group.go

@@ -109,6 +109,7 @@ type GroupDatabase interface {
 
 	FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error)
 	FindJoinIncrVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error)
+	MemberGroupIncrVersion(ctx context.Context, groupID string, userIDs []string, state int32) error
 
 	//FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error)
 	//FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error)
@@ -518,3 +519,10 @@ func (g *groupDatabase) SearchJoinGroup(ctx context.Context, userID string, keyw
 	}
 	return g.groupDB.SearchJoin(ctx, groupIDs, keyword, pagination)
 }
+
+func (g *groupDatabase) MemberGroupIncrVersion(ctx context.Context, groupID string, userIDs []string, state int32) error {
+	if err := g.groupMemberDB.MemberGroupIncrVersion(ctx, groupID, userIDs, state); err != nil {
+		return err
+	}
+	return g.cache.DelMaxGroupMemberVersion(groupID).ChainExecDel(ctx)
+}

pkg/common/storage/database/friend.go

@@ -55,4 +55,6 @@ type Friend interface {
 	//SearchFriend(ctx context.Context, ownerUserID, keyword string, pagination pagination.Pagination) (int64, []*model.Friend, error)
 
 	FindOwnerFriendUserIds(ctx context.Context, ownerUserID string, limit int) ([]string, error)
+
+	IncrVersion(ctx context.Context, ownerUserID string, friendUserIDs []string, state int32) error
 }

pkg/common/storage/database/mgo/friend.go

@@ -244,3 +244,7 @@ func (f *FriendMgo) FindFriendUserID(ctx context.Context, friendUserID string) (
 	}
 	return mongoutil.Find[string](ctx, f.coll, filter, options.Find().SetProjection(bson.M{"_id": 0, "owner_user_id": 1}).SetSort(f.friendSort()))
 }
+
+func (f *FriendMgo) IncrVersion(ctx context.Context, ownerUserID string, friendUserIDs []string, state int32) error {
+	return f.owner.IncrVersion(ctx, ownerUserID, friendUserIDs, state)
+}

pkg/common/storage/database/mgo/group_member.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"github.com/openimsdk/tools/log"
 
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/tools/db/mongoutil"
@@ -187,9 +188,11 @@ func (g *GroupMemberMgo) MemberGroupIncrVersion(ctx context.Context, groupID str
 }
 
 func (g *GroupMemberMgo) FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) {
+	log.ZDebug(ctx, "find member incr version", "groupID", groupID, "version", version)
 	return g.member.FindChangeLog(ctx, groupID, version, limit)
 }
 
 func (g *GroupMemberMgo) FindJoinIncrVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error) {
+	log.ZDebug(ctx, "find join incr version", "userID", userID, "version", version)
 	return g.join.FindChangeLog(ctx, userID, version, limit)
 }

pkg/common/storage/database/mgo/version_log.go

@@ -8,6 +8,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/versionctx"
 	"github.com/openimsdk/tools/db/mongoutil"
 	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
@@ -169,7 +170,9 @@ func (l *VersionLogMgo) FindChangeLog(ctx context.Context, dId string, version u
 	} else if !errors.Is(err, mongo.ErrNoDocuments) {
 		return nil, err
 	}
+	log.ZDebug(ctx, "init doc", "dId", dId)
 	if res, err := l.initDoc(ctx, dId, nil, 0, time.Now()); err == nil {
+		log.ZDebug(ctx, "init doc success", "dId", dId)
 		return res, nil
 	} else if mongo.IsDuplicateKeyError(err) {
 		return l.findChangeLog(ctx, dId, version, limit)