diff --git a/go.mod b/go.mod index d00f0b739..2614e0f32 100644 --- a/go.mod +++ b/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 ( -// github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol -//) +//replace github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol diff --git a/go.sum b/go.sum index 4c05500bd..fe4f0c390 100644 --- a/go.sum +++ b/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.15/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= +github.com/openimsdk/protocol v0.0.69-alpha.17 h1:pEag4ZdlovE+AyLsw1VYFU/3sk6ayvGdPzgufQfKf9M= +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= diff --git a/internal/rpc/friend/sync.go b/internal/rpc/friend/sync.go index faaa987f2..684894609 100644 --- a/internal/rpc/friend/sync.go +++ b/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 { diff --git a/internal/rpc/group/group.go b/internal/rpc/group/group.go index 4d429b3d3..5b82b8ceb 100644 --- a/internal/rpc/group/group.go +++ b/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 } diff --git a/internal/rpc/group/notification.go b/internal/rpc/group/notification.go index 5d385d8c1..ee4f00ad3 100644 --- a/internal/rpc/group/notification.go +++ b/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 } diff --git a/internal/rpc/user/user.go b/internal/rpc/user/user.go index 96f784fab..211b360b7 100644 --- a/internal/rpc/user/user.go +++ b/internal/rpc/user/user.go @@ -16,10 +16,7 @@ package user import ( "context" - "math/rand" - "strings" - "time" - + "errors" "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" @@ -120,7 +123,8 @@ func (s *userServer) GetDesignateUsers(ctx context.Context, req *pbuser.GetDesig // deprecated: -// UpdateUserInfo +//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) - 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) + oldUser, err := s.db.GetUserByID(ctx, req.UserInfo.UserID) if err != nil { return nil, err } - if req.UserInfo.Nickname != "" || req.UserInfo.FaceURL != "" { - if err = s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { - return nil, err - } - } - for _, friendID := range friends { - s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID) + 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) + //if err != nil { + // return nil, err + //} + //if req.UserInfo.Nickname != "" || req.UserInfo.FaceURL != "" { + // if err = s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID,oldUser); err != nil { + // return nil, err + // } + //} + //for _, friendID := range friends { + // 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) - if err != nil { - return nil, err - } - if req.UserInfo.Nickname != nil || req.UserInfo.FaceURL != nil { - if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { - return nil, err - } - } - for _, friendID := range friends { - s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID) - } + //friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID) + //if err != nil { + // return nil, err + //} + //if req.UserInfo.Nickname != nil || req.UserInfo.FaceURL != nil { + // if err := s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { + // return nil, err + // } + //} + //for _, friendID := range friends { + // 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, -) (*pbuser.GetSubscribeUsersStatusResp, error) { + req *pbuser.GetSubscribeUsersStatusReq) (*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 { diff --git a/pkg/common/storage/cache/redis/group.go b/pkg/common/storage/cache/redis/group.go index c4598567d..589678c50 100644 --- a/pkg/common/storage/cache/redis/group.go +++ b/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) }) } diff --git a/pkg/common/storage/controller/friend.go b/pkg/common/storage/controller/friend.go index 42c230598..4d3a96f12 100644 --- a/pkg/common/storage/controller/friend.go +++ b/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) +} diff --git a/pkg/common/storage/controller/group.go b/pkg/common/storage/controller/group.go index 71bf6ead2..27b6f4366 100644 --- a/pkg/common/storage/controller/group.go +++ b/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) +} diff --git a/pkg/common/storage/database/friend.go b/pkg/common/storage/database/friend.go index 1c9cd1033..b596411fc 100644 --- a/pkg/common/storage/database/friend.go +++ b/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 } diff --git a/pkg/common/storage/database/mgo/friend.go b/pkg/common/storage/database/mgo/friend.go index 4c3c4b97b..7f456fbda 100644 --- a/pkg/common/storage/database/mgo/friend.go +++ b/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) +} diff --git a/pkg/common/storage/database/mgo/group_member.go b/pkg/common/storage/database/mgo/group_member.go index 99a5a3ed2..30f8d63b9 100644 --- a/pkg/common/storage/database/mgo/group_member.go +++ b/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) } diff --git a/pkg/common/storage/database/mgo/version_log.go b/pkg/common/storage/database/mgo/version_log.go index 1d70f96f5..8836742f0 100644 --- a/pkg/common/storage/database/mgo/version_log.go +++ b/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)