diff --git a/go.mod b/go.mod index 34987958f..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.16 + 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 3ab5684a9..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.16 h1:ciSqm2rjBdpScpkQm3wPjAFv0YbIRp8MITRkDZWVv6c= -github.com/openimsdk/protocol v0.0.69-alpha.16/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/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/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) +}